render
3ce918564d
Replace per-frame file capture with the unchanged official Hypium agent, bounded JPEG decoding, independent control RPC and scoped session cleanup. Remove the HAP build and account-signing experiment. All 85 automated checks pass; moving-screen performance and integrated phone input remain unverified because the USB device went offline. Keep published 0.3.2 artifacts unchanged.
103 lines
4.2 KiB
Python
103 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Local test peer for the public HDC and Hypium protocol. Never contacts USB."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
ROOT = Path(__file__).parent
|
|
HEAD = b'_uitestkit_rpc_message_head_'
|
|
TAIL = b'_uitestkit_rpc_message_tail_'
|
|
def path(name): return ROOT / ('hdc.' + name)
|
|
def log(value):
|
|
with path('calls').open('a') as f: f.write(value + '\n')
|
|
def exact(sock, length):
|
|
data = bytearray()
|
|
while len(data) < length:
|
|
chunk = sock.recv(length-len(data))
|
|
if not chunk: raise EOFError()
|
|
data.extend(chunk)
|
|
return bytes(data)
|
|
def client(sock):
|
|
done = threading.Event()
|
|
write_lock = threading.Lock()
|
|
def send(data):
|
|
with write_lock: sock.sendall(data)
|
|
def video():
|
|
previous = None
|
|
try:
|
|
while not done.wait(0.01) and ROOT.exists():
|
|
if path('slow').exists():
|
|
path('capturing').touch()
|
|
continue
|
|
jpeg = path('jpg').read_bytes()
|
|
if jpeg != previous:
|
|
send(jpeg+b'\n')
|
|
previous=jpeg
|
|
except (OSError, EOFError): pass
|
|
try:
|
|
while True:
|
|
assert exact(sock,len(HEAD)) == HEAD
|
|
_,length=struct.unpack('>II',exact(sock,8))
|
|
assert 0 < length <= 65536
|
|
message=json.loads(exact(sock,length))
|
|
assert exact(sock,len(TAIL)) == TAIL
|
|
params=message['params']; api=params['api']
|
|
log(json.dumps({'api':api,'args':params['args']}))
|
|
result=None
|
|
if api=='Driver.create': result='Driver#fixture'
|
|
elif api=='Driver.getDisplaySize': result=json.loads(path('dimensions').read_text())
|
|
elif api=='startCaptureScreen': result=True
|
|
elif api=='stopCaptureScreen': result=True;done.set()
|
|
elif path('fail-input').exists():
|
|
send(b'{"exception":{"code":17000005,"message":"input not permitted"}}\n')
|
|
continue
|
|
send(json.dumps({'result':result}).encode()+b'\n')
|
|
if api=='startCaptureScreen': threading.Thread(target=video,daemon=True).start()
|
|
except (OSError,EOFError): pass
|
|
finally: done.set();sock.close()
|
|
def serve(port):
|
|
with socket.socket() as listener:
|
|
listener.bind(('127.0.0.1',port));listener.listen()
|
|
path('ready').touch()
|
|
while True:
|
|
sock,_=listener.accept()
|
|
threading.Thread(target=client,args=(sock,),daemon=True).start()
|
|
if sys.argv[1:2]==['server']:
|
|
serve(int(sys.argv[2]));sys.exit()
|
|
args=sys.argv[1:]
|
|
log(' '.join(args))
|
|
if args[:2]==['-t','phone']: args=args[2:]
|
|
if args==['list','targets','-v']: print('phone USB Connected device')
|
|
elif args==['shell','uitest uiInput help']: print('click doubleClick longClick swipe drag text keyEvent')
|
|
elif args==['shell','pidof uitest']:
|
|
if path('daemon').exists(): print('4242')
|
|
elif args[:2]==['file','send']: print('FileTransfer finish')
|
|
elif args[0]=='shell' and args[1].startswith('uitest start-daemon'):
|
|
path('daemon').write_text(args[1].split()[-1])
|
|
elif args[0]=='shell' and args[1]=='cat /proc/4242/cmdline':
|
|
sys.stdout.write('uitest\0start-daemon\0'+path('daemon').read_text()+'\0')
|
|
elif args[:2]==['shell','kill 4242']: path('daemon').unlink(missing_ok=True)
|
|
elif args[0]=='shell' and args[1].startswith('rm -f /data/local/tmp/'): pass
|
|
elif args[0]=='shell' and args[1].startswith('uitest uiInput text '): print('completed')
|
|
elif args[:2]==['fport','rm']:
|
|
pid=int(path('pid').read_text());os.kill(pid,signal.SIGTERM)
|
|
path('pid').unlink();path('ready').unlink(missing_ok=True)
|
|
print('Remove forward ruler success')
|
|
elif args[0]=='fport':
|
|
process=subprocess.Popen([sys.executable,__file__,'server',args[1].split(':')[1]],
|
|
stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,start_new_session=True)
|
|
path('pid').write_text(str(process.pid))
|
|
until=time.monotonic()+3
|
|
while not path('ready').exists():
|
|
if time.monotonic()>until: raise RuntimeError('Fixture listener not ready')
|
|
time.sleep(0.01)
|
|
print('Forwardport result:OK')
|
|
else: print('[Fail] unexpected invocation')
|