cosmic-mobile

Latest change 510a3b926dcaa185e6de4717c0705c927f90181d - Add pointer move action so screenshots exclude the cursor by AkurAI Build

#!/usr/bin/env python3
"""Guest input via one QMP connection. Coordinates use --size, not host window pixels.
Remote: ssh midget 'cd ~/Emulators/cosmic-mobile && python3 qmp.py tap 225 495 --size 450 990'
Batch stdin: [{"action":"tap","x":225,"y":495},{"action":"text","text":"hello"}]
Text uses US keyboard layout; unsupported characters rejected before sending.
"""
import argparse
import json
import socket
import sys
import time
from pathlib import Path


def tap_events(x, y, width, height):
    if not (width > 1 and height > 1 and 0 <= x < width and 0 <= y < height):
        raise ValueError('Tap outside guest coordinate bounds')
    return [{'type': 'abs', 'data': {'axis': axis, 'value': round(value * 32767 / (extent - 1))}}
            for axis, value, extent in [('x', x, width), ('y', y, height)]]


def text_keys(text):
    plain = dict(zip(' -=[]\\;\',./`\n\t', ['spc', 'minus', 'equal', 'bracket_left', 'bracket_right', 'backslash', 'semicolon', 'apostrophe', 'comma', 'dot', 'slash', 'grave_accent', 'ret', 'tab']))
    shifted = dict(zip('!@#$%^&*()_+{}|:"<>?~', ['1','2','3','4','5','6','7','8','9','0','minus','equal','bracket_left','bracket_right','backslash','semicolon','apostrophe','comma','dot','slash','grave_accent']))
    result = []
    for c in text:
        if c in 'abcdefghijklmnopqrstuvwxyz0123456789': result.append(c)
        elif c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ': result.append('shift-' + c.lower())
        elif c in plain: result.append(plain[c])
        elif c in shifted: result.append('shift-' + shifted[c])
        else: raise ValueError(f'Unsupported US-layout character: U+{ord(c):04X}')
    return result


class QMP:
    def __init__(self, path):
        self.sock = socket.socket(socket.AF_UNIX)
        self.sock.settimeout(15)
        self.sock.connect(str(path))
        self.stream = self.sock.makefile('rwb')
        json.loads(self.stream.readline())
        self.call('qmp_capabilities')

    def call(self, name, args=None):
        self.stream.write((json.dumps({'execute': name, 'arguments': args or {}}) + '\n').encode())
        self.stream.flush()
        while True:
            line = self.stream.readline()
            if not line: raise ConnectionError('QMP disconnected')
            reply = json.loads(line)
            if 'error' in reply: raise RuntimeError(reply['error'])
            if 'return' in reply: return reply['return']

    def close(self):
        self.stream.close()
        self.sock.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--socket', default=str(Path(__file__).resolve().parent / 'runtime/qmp.sock'))
    parser.add_argument('--size', nargs=2, type=int, default=[900, 1980], metavar=('WIDTH', 'HEIGHT'))
    parser.add_argument('action', choices=['tap','move','key','text','batch','shot','status','stop'])
    parser.add_argument('args', nargs='*')
    opts = parser.parse_args()
    if opts.action == 'batch': actions = json.load(sys.stdin)
    elif opts.action in ['tap','move']:
        if len(opts.args) != 2: parser.error(f'{opts.action} needs X Y')
        actions = [{'action':opts.action, 'x':float(opts.args[0]), 'y':float(opts.args[1])}]
    elif opts.action in ['key','text','shot']:
        if len(opts.args) != 1: parser.error(f'{opts.action} needs one argument')
        actions = [{'action':opts.action, opts.action:opts.args[0]}]
    else: actions = [{'action':opts.action}]
    # Validate entire batch before any input; never partially type unsupported text.
    if not isinstance(actions, list) or len(actions) > 1000: raise ValueError('Expected at most 1000 actions')
    for action in actions:
        kind = action['action']
        if kind in ['tap','move']: tap_events(action['x'], action['y'], *opts.size)
        elif kind == 'text': action['keys'] = text_keys(action['text'])
        elif kind == 'key':
            if not action['key'] or not all(c in 'abcdefghijklmnopqrstuvwxyz0123456789_-' for c in action['key']):
                raise ValueError('Invalid QCODE chord')
        elif kind not in ['shot','status','stop']: raise ValueError('Unknown action')
    client = QMP(opts.socket)
    try:
        for action in actions:
            kind = action['action']
            if kind == 'tap':
                client.call('input-send-event', {'events':tap_events(action['x'], action['y'], *opts.size)})
                try:
                    client.call('input-send-event', {'events':[{'type':'btn','data':{'down':True,'button':'left'}}]})
                    time.sleep(.04)
                finally:
                    client.call('input-send-event', {'events':[{'type':'btn','data':{'down':False,'button':'left'}}]})
            elif kind == 'move':
                client.call('input-send-event', {'events':tap_events(action['x'], action['y'], *opts.size)})
            elif kind in ['key','text']:
                for key in action.get('keys', [action.get('key')]):
                    client.call('send-key', {'keys':[{'type':'qcode','data':k} for k in key.split('-')], 'hold-time':20})
                    time.sleep(.03)
            elif kind == 'shot': client.call('screendump', {'filename':str(Path(action['shot']).resolve()), 'format':'png'})
            elif kind == 'status': print(json.dumps(client.call('query-status')))
            elif kind == 'stop': client.call('quit')
        print(json.dumps({'ok':True, 'actions':len(actions)}))
    finally: client.close()


if __name__ == '__main__':
    try: main()
    except (ValueError, KeyError, OSError, RuntimeError) as exc:
        print(json.dumps({'error':str(exc)}), file=sys.stderr)
        sys.exit(1)