"""PC control for the RTOS phone bridge. Uses only Python's standard library.""" import argparse import hashlib import json from pathlib import Path import subprocess import sys import time import urllib.error import urllib.request PACKAGE = 'com.sydxky.rtos.bridge' class Bridge: def __init__(self, serial=None, port=18765, adb='adb'): self.adb = [adb] + (['-s', serial] if serial else []) self.port = port self.token = None self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) def shell(self, *args): result = subprocess.run(self.adb + list(args), capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=45) if result.returncode: raise RuntimeError(result.stderr.strip() or result.stdout.strip()) return result.stdout.strip() def prepare(self, launch=True): if launch: self.shell('shell', 'am', 'start', '-n', PACKAGE + '/.MainActivity') self.shell('forward', f'tcp:{self.port}', 'tcp:18765') for _ in range(20): try: token = self.shell('shell', 'run-as', PACKAGE, 'cat', 'files/bridge-token') if len(token) != 43 or any(c not in 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_' for c in token): raise RuntimeError('Bridge token not available; install and open the debug APK') self.token = token self.request('GET', '/status') return except (RuntimeError, OSError): time.sleep(.3) raise RuntimeError('Bridge unavailable. Unlock the phone and open RTOS 真机调试桥.') def request(self, method, path, data=None, headers=None): if data is not None and not isinstance(data, bytes): data = json.dumps(data, ensure_ascii=False).encode('utf-8') auth = {'Authorization': 'Bearer ' + self.token, 'Content-Type': 'application/json'} auth.update(headers or {}) req = urllib.request.Request(f'http://127.0.0.1:{self.port}' + path, data=data, headers=auth, method=method) try: with self.opener.open(req, timeout=35) as response: return json.load(response) except urllib.error.HTTPError as error: raise RuntimeError(f'Bridge HTTP {error.code}: {error.read().decode("utf-8", errors="replace")}') from None def connect(self, app_id='test', license_value='test', service='com.heytap.health'): self.request('POST', '/connect', {'appId': app_id, 'license': license_value, 'service': service}) for _ in range(75): result = self.request('GET', '/status') if result['state'] == 'bound': return result if result['state'] in ('failed', 'service_not_found', 'bind_rejected', 'null_binding', 'bind_timeout'): hint = ' Open the paired health app, then retry connect.' if result['state'] == 'bind_rejected' else '' raise RuntimeError('Wear Engine binding failed: ' + json.dumps(result, ensure_ascii=False) + hint) time.sleep(.2) raise TimeoutError('Service binding timed out. See events for details.') def command(self, args, timeout=180): job = self.request('POST', '/command', args) started = time.monotonic() while time.monotonic() - started < timeout: result = self.request('GET', '/jobs/' + job['id']) if result['state'] != 'pending': return result time.sleep(.35) return {'id': job['id'], 'state': 'timeout', 'result': 'Watch operation may still complete. Inspect job/events before retrying.'} def wait_link(self, timeout=45): """Read-only wait: a bound Binder can precede the paired watch link.""" if not 1 <= timeout <= 120: raise ValueError('Link wait must be 1-120 seconds') deadline = time.monotonic() + timeout while True: result = self.command({'command': 'link'}, min(5, max(.1, deadline - time.monotonic()))) if result.get('state') != 'success' or result.get('result', {}).get('connected') is True: return result remaining = deadline - time.monotonic() if remaining <= 0: return result time.sleep(min(1, remaining)) def upload(self, filename): path = Path(filename) if not 0 < path.stat().st_size <= 32 * 1024 * 1024: raise ValueError('File must be 1 byte to 32 MiB') data = path.read_bytes() # Use a simple unique name, regardless of Unicode or spaces in the local path. name = hashlib.sha256(data).hexdigest()[:24] + '.pkg' result = self.request('PUT', '/files/' + name, data, {'X-SHA256': hashlib.sha256(data).hexdigest(), 'Content-Type': 'application/octet-stream'}) if result['sha256'] != hashlib.sha256(data).hexdigest(): raise RuntimeError('Phone upload integrity mismatch') return name def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--serial', help='ADB phone serial; required when more than one device is attached') parser.add_argument('--port', type=int, default=18765) parser.add_argument('--adb', default='adb') parser.add_argument('--timeout', type=int, default=180) sub = parser.add_subparsers(dest='action', required=True) setup = sub.add_parser('setup', help='Install the debug APK and start the bridge') setup.add_argument('--apk', default=str(Path(__file__).parent / 'artifacts/RTOS-Phone-Bridge-0.2.0-debug.apk')) connect = sub.add_parser('connect') connect.add_argument('--app-id', default='test') connect.add_argument('--config', help='Local JSON with appId, license and optional service; keep credentials out of Git') for action in ('status', 'info', 'link', 'disconnect', 'cleanup', 'stop'): sub.add_parser(action) sub.add_parser('wait-link', help='Wait for the paired watch after Health starts').add_argument('--seconds', type=int, default=45) for action in ('check', 'launch'): sub.add_parser(action).add_argument('package') for action in ('install', 'send-file'): child = sub.add_parser(action) child.add_argument('file') if action == 'send-file': child.add_argument('--remote', required=True) sub.add_parser('message').add_argument('message') logs = sub.add_parser('logs') logs.add_argument('enabled', choices=['on', 'off']) rpc = sub.add_parser('rpc') rpc.add_argument('method') rpc.add_argument('--params', default='{}') events = sub.add_parser('events') events.add_argument('--after', type=int, default=0) events.add_argument('--follow', action='store_true') args = parser.parse_args() bridge = Bridge(args.serial, args.port, args.adb) if args.action == 'setup': if not Path(args.apk).is_file(): parser.error('Build APK with build.ps1 or pass --apk') bridge.shell('install', '-r', args.apk) if args.action == 'stop': bridge.shell('shell', 'am', 'force-stop', PACKAGE) bridge.shell('forward', '--remove', f'tcp:{args.port}') print('Bridge stopped; ADB forward removed.') return bridge.prepare() if args.action in ('setup', 'status'): result = bridge.request('GET', '/status') elif args.action == 'connect': config = json.loads(Path(args.config).read_text(encoding='utf-8')) if args.config else {'appId': args.app_id} result = bridge.connect(config.get('appId', 'test'), config.get('license', 'test'), config.get('service', 'com.heytap.health')) elif args.action in ('disconnect', 'cleanup'): result = bridge.request('POST', '/' + args.action, {}) elif args.action == 'wait-link': result = bridge.wait_link(args.seconds) if result.get('state') != 'success' or result.get('result', {}).get('connected') is not True: print(json.dumps(result, ensure_ascii=False, indent=2)) return 2 elif args.action == 'events': cursor = args.after while True: result = bridge.request('GET', '/events?after=' + str(cursor)) for row in result['events']: print(json.dumps(row, ensure_ascii=False), flush=True) cursor = row['seq'] if not args.follow: return time.sleep(1) else: command = {'command': args.action} if args.action in ('check', 'launch'): command['package'] = args.package elif args.action in ('install', 'send-file'): command['file'] = bridge.upload(args.file) if args.action == 'send-file': command['remote'] = args.remote elif args.action == 'message': command['message'] = args.message elif args.action == 'logs': command['enabled'] = args.enabled == 'on' elif args.action == 'rpc': command.update(method=args.method, params=args.params) result = bridge.command(command, args.timeout) print(json.dumps(result, ensure_ascii=False, indent=2)) if 'state' in result and result['state'] in ('failed', 'timeout'): sys.exit(2) if __name__ == '__main__': sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') try: sys.exit(main()) except KeyboardInterrupt: sys.exit(130) except (OSError, RuntimeError, ValueError, subprocess.SubprocessError) as error: print(str(error), file=sys.stderr) sys.exit(1)