#!/usr/bin/env python3 """Control a DroidDeck build over ADB (docs/agent-control.md).""" from __future__ import annotations import argparse import base64 import gzip import json import os import re import shutil import subprocess import struct import sys import tempfile import time import uuid import zipfile import zlib from pathlib import Path, PurePosixPath sys.path.insert(0, str(Path(__file__).resolve().parent)) from adb_device import DeviceResolutionError, resolve_serial from proot_profile import report as proot_profile_report PACKAGE = os.environ.get("DROIDDECK_PACKAGE", "com.droiddeck.launcher") AUTHORITY = f"content://{PACKAGE}.agent" PACKAGE_NAME = re.compile(r"[A-Za-z][A-Za-z0-9_]*(\.[A-Za-z][A-Za-z0-9_]*)+") STEAM_CLIENT_APP = 769 REQUEST_CAP = 256 * 1024 class ControlError(RuntimeError): def __init__(self, message: str, code: str = "COMMAND_FAILED") -> None: super().__init__(message) self.code = code def profile_slow_us(value: str) -> int: number = int(value) if not 1 <= number <= 1_000_000_000: raise argparse.ArgumentTypeError("--slow-us must be between 1 and 1000000000") return number def adb_path(configured: str | None) -> str: if configured: return configured environment_adb = os.environ.get("ADB") if environment_adb: return environment_adb found = shutil.which("adb") if found: return found sdk = os.environ.get("ANDROID_HOME") or os.environ.get("ANDROID_SDK_ROOT") if sdk: candidate = Path(sdk) / "platform-tools" / "adb" if candidate.is_file(): return str(candidate) home_dir = Path.home() candidate = home_dir / "Library" / "Android" / "sdk" / "platform-tools" / "adb" if candidate.is_file(): return str(candidate) raise ControlError("adb not found; set ADB, ANDROID_HOME, or ANDROID_SDK_ROOT", "ADB_NOT_FOUND") def run_adb(adb: str, serial: str, *args: str, timeout: float = 30) -> subprocess.CompletedProcess[bytes]: try: return subprocess.run( [adb, "-s", serial, *args], capture_output=True, check=False, timeout=timeout ) except (OSError, subprocess.TimeoutExpired) as exc: raise ControlError(str(exc), "ADB_FAILED") from exc def _content_result(output: str) -> dict: marker = "json=" start = output.find(marker) if start < 0: detail = output.strip() or "The agent provider returned no result" raise ControlError(detail, "INVALID_PROVIDER_RESPONSE") raw = output[start + len(marker):].lstrip() try: response, _ = json.JSONDecoder().raw_decode(raw) if not isinstance(response, dict): raise ValueError("expected an object") return response except (json.JSONDecodeError, ValueError) as exc: raise ControlError("Could not decode the agent provider response", "INVALID_PROVIDER_RESPONSE") from exc def encode_request(request: dict) -> str: encoded = base64.b64encode(json.dumps(request, separators=(",", ":")).encode()).decode() if len(encoded) > REQUEST_CAP: raise ControlError("The encoded request exceeds 256 KiB", "INVALID_REQUEST") return encoded def provider_call(adb: str, serial: str, method: str, arg: str | None = None, request: dict | None = None, timeout: float = 30) -> dict: command = ["shell", "content", "call", "--uri", AUTHORITY, "--method", method] if arg is not None: command.extend(["--arg", arg]) if request is not None: encoded = encode_request(request) command.extend(["--extra", f"request:s:{encoded}"]) result = run_adb(adb, serial, *command, timeout=timeout) output = (result.stdout + result.stderr).decode(errors="replace") response = _content_result(output) if result.returncode != 0 and response.get("ok") is not False: raise ControlError(output.strip() or "ADB content call failed", "ADB_FAILED") return response def checked(response: dict, default_code: str) -> dict: if response.get("ok") is False: error = response.get("error", {}) raise ControlError(error.get("message", default_code), error.get("code", default_code)) return response def start_agent_activity(adb: str, serial: str, request: dict) -> None: encoded = encode_request(request) result = run_adb( adb, serial, "shell", "am", "start", "-W", "-n", f"{PACKAGE}/com.droiddeck.launcher.agent.AgentStartActivity", "--es", "request", encoded, timeout=90, ) output = (result.stdout + result.stderr).decode(errors="replace") if result.returncode != 0 or "status: ok" not in output.lower(): raise ControlError(output.strip() or "The agent start Activity did not launch", "ACTIVITY_START_FAILED") def confirm_start(adb: str, serial: str, request: dict, previous: dict) -> dict: deadline = time.monotonic() + 95 expected_mode = {"steam": "steam", "desktop": "lxqt", "run": "run"}[request["mode"]] while True: state = get_state(adb, serial) session = state.get("session", {}) phase = session.get("phase") if session.get("mode") == expected_mode and session.get("id") != previous.get("id"): return state remaining = deadline - time.monotonic() if remaining <= 0: raise ControlError( f"The agent Activity launched but no new session appeared; phase is {phase}", "ACTIVITY_START_FAILED", ) time.sleep(min(0.25, remaining)) def get_state(adb: str, serial: str) -> dict: return checked(provider_call(adb, serial, "state"), "STATE_FAILED") def wait_until(adb: str, serial: str, describe: str, done, timeout: float, poll: float = 1.0) -> dict: deadline = time.monotonic() + timeout while True: state = get_state(adb, serial) session = state.get("session", {}) if done(state): return state if session.get("phase") == "FAILED": failure = session.get("failure") or {} raise ControlError( f"Session failed ({failure.get('code', 'FAILED')}): {failure.get('message', 'unknown error')}", failure.get("code", "SESSION_FAILED"), ) remaining = deadline - time.monotonic() if remaining <= 0: focus = session.get("focus") or {} raise ControlError( f"Timed out waiting for {describe}; phase {session.get('phase')}, focused app {focus.get('focusedApp')}", "TIMEOUT", ) time.sleep(min(poll, remaining)) def wait_for_phase(adb: str, serial: str, wanted: str, timeout: float) -> dict: if wanted == "FAILED": deadline = time.monotonic() + timeout while True: state = get_state(adb, serial) if state.get("session", {}).get("phase") == "FAILED": return state if time.monotonic() > deadline: raise ControlError("Timed out waiting for FAILED", "TIMEOUT") time.sleep(1) return wait_until(adb, serial, wanted, lambda s: s.get("session", {}).get("phase") == wanted, timeout) def focused(state: dict, app_id: int | None) -> bool: focus = state.get("session", {}).get("focus") or {} current = focus.get("focusedApp") if app_id is None: return current not in (None, 0, STEAM_CLIENT_APP) return current == app_id and app_id in (focus.get("focusableApps") or []) def wait_for_artifacts(adb: str, serial: str, state: dict, timeout: float) -> dict: deadline = time.monotonic() + timeout while True: session = state.get("session", {}) if session.get("artifactsComplete") or not session.get("artifactsAvailable"): return state remaining = deadline - time.monotonic() if remaining <= 0: raise ControlError("Session stopped, but artifact collection did not complete", "ARTIFACT_TIMEOUT") time.sleep(min(1.0, remaining)) state = get_state(adb, serial) def add_serial(response: dict, serial: str) -> dict: response["deviceSerial"] = serial return response def displays(adb: str, serial: str) -> list[dict]: result = run_adb(adb, serial, "shell", "dumpsys", "display", timeout=30) text = result.stdout.decode(errors="replace") devices = dict(re.findall(r'DisplayDeviceInfo\{"([^"]+)": uniqueId="local:(\d+)"', text)) out = [] for block in re.split(r"\n Display (?=\d+:)", text)[1:]: logical = int(block.split(":", 1)[0]) device = re.search(r"mPrimaryDisplayDevice=(.+)", block) size = re.search(r"real (\d+) x (\d+)", block) name = device.group(1).strip() if device else None out.append({ "display": logical, "name": name, "physicalId": devices.get(name) if name else None, "size": [int(size.group(1)), int(size.group(2))] if size else None, }) return out def physical_display(adb: str, serial: str, logical: int | None) -> str | None: if logical is None: return None for d in displays(adb, serial): if d["display"] == logical: if not d["physicalId"]: raise ControlError(f"Display {logical} has no physical id to capture", "NO_SUCH_DISPLAY") return d["physicalId"] raise ControlError(f"No display {logical}; see `droiddeckctl displays`", "NO_SUCH_DISPLAY") def screenshot(adb: str, serial: str, path: str, display: int | None) -> dict: destination = Path(path).expanduser().resolve() physical = physical_display(adb, serial, display) command = ["exec-out", "screencap", "-p"] + (["-d", physical] if physical else []) result = run_adb(adb, serial, *command, timeout=30) if result.returncode != 0 or not result.stdout.startswith(b"\x89PNG"): detail = result.stderr.decode(errors="replace").strip() raise ControlError(detail or "screencap failed", "SCREENSHOT_FAILED") try: destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(result.stdout) except OSError as exc: raise ControlError(str(exc), "FILE_IO") from exc return {"ok": True, "command": "screenshot", "path": str(destination), "display": display} def record(adb: str, serial: str, path: str, seconds: float, display: int | None) -> dict: destination = Path(path).expanduser().resolve() remote = "/data/local/tmp/droiddeckctl-record.mp4" physical = physical_display(adb, serial, display) command = ["shell", "screenrecord", "--time-limit", str(max(1, min(180, int(seconds))))] if physical: command += ["--display-id", physical] result = run_adb(adb, serial, *command, remote, timeout=seconds + 30) if result.returncode != 0: raise ControlError(result.stderr.decode(errors="replace").strip() or "screenrecord failed", "RECORD_FAILED") try: destination.parent.mkdir(parents=True, exist_ok=True) except OSError as exc: raise ControlError(str(exc), "FILE_IO") from exc pulled = run_adb(adb, serial, "pull", remote, str(destination), timeout=120) run_adb(adb, serial, "shell", "rm", "-f", remote) if pulled.returncode != 0: raise ControlError(pulled.stderr.decode(errors="replace").strip() or "adb pull failed", "RECORD_FAILED") return {"ok": True, "command": "record", "path": str(destination), "seconds": seconds, "display": display} def pacing_summary(text: str, package: str = PACKAGE) -> dict: text = text.replace("\r\n", "\n") records = {"game": [], "surface": []} for part in re.split(r"^displayRefreshRate = ", text, flags=re.M)[1:]: rate = re.match(r"(\d+)", part) layer = re.search(r"^layerName = (.*)$", part, re.M) frames = re.search(r"^totalFrames = (\d+)", part, re.M) hist = re.search(r"^present2present histogram is as below:\n(.*)", part, re.M) if not (rate and layer and frames and hist): continue name = layer.group(1) kind = ("game" if "droiddeck_wayland_game" in name else "surface" if f"SurfaceView[{package}/" in name else None) if kind and int(frames.group(1)): bins = {int(ms): int(n) for ms, n in re.findall(r"(\d+)ms=(\d+)", hist.group(1)) if int(n)} records[kind].append((re.sub(r"#\d+$", "", name.strip()), int(frames.group(1)), bins, int(rate.group(1)))) chosen = records["game"] or records["surface"] histogram: dict[int, int] = {} off = 0 for _, _, bins, _ in chosen: for ms, count in bins.items(): histogram[ms] = histogram.get(ms, 0) + count if bins: mode = max(bins, key=bins.get) off += sum(count for ms, count in bins.items() if abs(ms - mode) > 1) intervals = sum(histogram.values()) cadence = max(histogram, key=histogram.get) if histogram else None return { "layers": sorted({name for name, _, _, _ in chosen}), "frames": sum(frames for _, frames, _, _ in chosen), "refreshRates": sorted({rate for _, _, _, rate in chosen}), "presentToPresentMs": {str(ms): histogram[ms] for ms in sorted(histogram)}, "cadenceMs": cadence, "offCadence": off, "offCadencePercent": round(100.0 * off / intervals, 1) if intervals else None, } def pacing(adb: str, serial: str, seconds: float) -> dict: seconds = max(1.0, min(600.0, seconds)) timestats = ("shell", "dumpsys", "SurfaceFlinger", "--timestats") try: enabled = run_adb(adb, serial, *timestats, "-clear", "-enable") except ControlError as exc: raise ControlError(str(exc), "PACING_FAILED") from exc if enabled.returncode != 0: raise ControlError("could not enable SurfaceFlinger timestats", "PACING_FAILED") started = time.monotonic() try: time.sleep(seconds) dump = run_adb(adb, serial, *timestats, "-dump", timeout=60) except ControlError as exc: raise ControlError(str(exc), "PACING_FAILED") from exc finally: elapsed = time.monotonic() - started try: disabled = run_adb(adb, serial, *timestats, "-disable").returncode == 0 except ControlError: disabled = False if dump.returncode != 0: raise ControlError(dump.stderr.decode(errors="replace").strip() or "timestats dump failed", "PACING_FAILED") if not disabled: raise ControlError("could not turn SurfaceFlinger timestats off", "PACING_FAILED") summary = pacing_summary(dump.stdout.decode(errors="replace")) if not summary["layers"] or not summary["frames"]: raise ControlError("No DroidDeck layer drew during the capture: is a session on screen?", "NO_FRAMES") summary["fpsOnScreen"] = round(summary["frames"] / elapsed, 1) return {"ok": True, "command": "pacing", "seconds": round(elapsed, 1), **summary} THERMAL_STATUS = ["NONE", "LIGHT", "MODERATE", "SEVERE", "CRITICAL", "EMERGENCY", "SHUTDOWN"] CPUFREQ = "/sys/devices/system/cpu/cpufreq" def thermal_summary(battery: str, service: str, cpufreq: str) -> dict: tenths = re.search(r"^\s*temperature:\s*(-?\d+)", battery, re.M) status = re.search(r"^\s*Thermal Status:\s*(\d+)", service, re.M) level = int(status.group(1)) if status else None cpu: dict[str, dict] = {} for policy, name, khz in re.findall(r"/(policy\d+)/(scaling_max_freq|cpuinfo_max_freq):(\d+)", cpufreq): cpu.setdefault(policy, {})["maxKHz" if name == "scaling_max_freq" else "hardwareMaxKHz"] = int(khz) for limits in cpu.values(): if "maxKHz" in limits and "hardwareMaxKHz" in limits: limits["capped"] = limits["maxKHz"] < limits["hardwareMaxKHz"] return { "batteryC": int(tenths.group(1)) / 10 if tenths else None, "status": level, "statusName": THERMAL_STATUS[level] if level is not None and level < len(THERMAL_STATUS) else None, "cpu": dict(sorted(cpu.items(), key=lambda item: int(item[0][6:]))), } def thermal(adb: str, serial: str) -> dict: def shell(*command: str) -> str: return run_adb(adb, serial, "shell", *command).stdout.decode(errors="replace") summary = thermal_summary(shell("dumpsys", "battery"), shell("dumpsys", "thermalservice"), shell("grep", ".", f"{CPUFREQ}/policy*/scaling_max_freq", f"{CPUFREQ}/policy*/cpuinfo_max_freq")) return {"ok": True, "command": "thermal", **summary} def wait_sleep(seconds: float) -> None: time.sleep(seconds) def wait_cool(adb: str, serial: str, below: float, timeout: float) -> dict: started = time.monotonic() while True: reading = thermal(adb, serial) if reading["batteryC"] is None: raise ControlError("dumpsys battery reports no temperature", "THERMAL_FAILED") if reading["batteryC"] < below: return {**reading, "command": "wait", "waitedSeconds": round(time.monotonic() - started, 1)} remaining = started + timeout - time.monotonic() if remaining <= 0: raise ControlError(f"Battery still at {reading['batteryC']:g} °C after {timeout:g} s, wanted below {below:g}", "TIMEOUT") wait_sleep(min(10.0, remaining)) def start_session(adb: str, serial: str, request: dict) -> dict: if request.get("mode") == "run": require_commands(adb, serial) current = get_state(adb, serial) if current["session"].get("phase") not in {"IDLE", "FAILED"} or current["session"].get("running"): raise ControlError("A session is already starting or running", "SESSION_ACTIVE") start_agent_activity(adb, serial, request) return confirm_start(adb, serial, request, current["session"]) def unpack_logs(archive: Path, destination: Path, session_id: str) -> None: """Unpack a Share logs zip so the session's own files sit at the top of destination.""" destination = destination.resolve() try: destination.mkdir(parents=True, exist_ok=True) except OSError as exc: raise ControlError(f"Could not unpack the session logs: {exc}", "LOG_PULL_FAILED") from exc manifest = destination / ".droiddeckctl-managed-logs.json" try: previous = json.loads(manifest.read_text(encoding="utf-8")) if not isinstance(previous, list) or any(not isinstance(name, str) for name in previous): previous = [] except (OSError, ValueError): previous = [] previous = set(previous) previous.discard(manifest.name) extracted = set() def managed_target(name: str) -> Path | None: relative = Path(name) if relative.is_absolute() or not relative.parts or any(part in {".", ".."} for part in relative.parts): return None target = destination for part in relative.parts: target = target / part if target.is_symlink(): return None return target if target.resolve().is_relative_to(destination) else None try: with tempfile.TemporaryDirectory(prefix=".droiddeckctl-logs-", dir=destination) as temp, zipfile.ZipFile(archive) as z: staged = Path(temp) staged_files = staged / "new" backups = staged / "old" staged_files.mkdir() for entry in z.infolist(): name = entry.filename if name.startswith(session_id + "/"): name = name[len(session_id) + 1:] target = managed_target(name) if entry.is_dir() or target is None or target == manifest: continue relative = target.relative_to(destination).as_posix() staged_file = staged_files / relative staged_file.parent.mkdir(parents=True, exist_ok=True) with z.open(entry) as source, staged_file.open("wb") as copy: shutil.copyfileobj(source, copy) extracted.add(relative) for name in extracted - previous: target = managed_target(name) if target is None or target.exists(): raise ControlError(f"Refusing to overwrite an unrelated file: {name}", "LOG_PULL_FAILED") if manifest.is_symlink() or (manifest.exists() and not manifest.is_file()): raise ControlError("The managed log list is not a regular file", "LOG_PULL_FAILED") backed_up = {} installed = set() manifest_backup = backups / manifest.name manifest_fd, manifest_path = tempfile.mkstemp( prefix=".droiddeckctl-managed-logs-", suffix=".tmp", dir=destination ) temp_manifest = Path(manifest_path) try: with os.fdopen(manifest_fd, "w", encoding="utf-8") as manifest_file: manifest_file.write(json.dumps(sorted(extracted), indent=2) + "\n") except Exception: temp_manifest.unlink(missing_ok=True) raise try: for name in sorted(previous): target = managed_target(name) if target is not None and target.is_file(): backup = backups / name backup.parent.mkdir(parents=True, exist_ok=True) os.replace(target, backup) backed_up[name] = backup if manifest.is_file(): manifest_backup.parent.mkdir(parents=True, exist_ok=True) os.replace(manifest, manifest_backup) for name in sorted(extracted): target = managed_target(name) if target is None: raise ControlError(f"A log destination became unsafe: {name}", "LOG_PULL_FAILED") target.parent.mkdir(parents=True, exist_ok=True) os.replace(staged_files / name, target) installed.add(name) os.replace(temp_manifest, manifest) except Exception: temp_manifest.unlink(missing_ok=True) for name in installed: target = managed_target(name) if target is not None and target.is_file(): target.unlink() if manifest_backup.is_file() and manifest.exists() and not manifest.is_symlink(): manifest.unlink() for name, backup in backed_up.items(): target = managed_target(name) if target is not None: target.parent.mkdir(parents=True, exist_ok=True) os.replace(backup, target) if manifest_backup.is_file(): os.replace(manifest_backup, manifest) raise except (OSError, zipfile.BadZipFile) as exc: raise ControlError(f"Could not unpack the session logs: {exc}", "LOG_PULL_FAILED") from exc def require_commands(adb: str, serial: str) -> dict: access = checked(provider_call(adb, serial, "access"), "ACCESS_FAILED") if access.get("commands") is not True: raise ControlError("Debugging tools are off: turn on Setup > Diagnostics > Debugging tools on the device", "AGENT_COMMANDS_DISABLED") return access def input_request(args: argparse.Namespace) -> dict: kind = args.input_type if kind == "key": return {"type": "key", "key": args.key, "action": args.action, "modifiers": args.mod or [], "holdMs": args.hold} if kind == "text": return {"type": "text", "text": args.text, "commit": args.commit} if kind == "tap": return {"type": "tap", "x": args.x, "y": args.y, "space": args.space, "holdMs": args.hold} if kind == "swipe": return {"type": "swipe", "x": args.x, "y": args.y, "x2": args.x2, "y2": args.y2, "space": args.space, "durationMs": args.duration} if kind == "pad": return {"type": "pad", "button": args.button, "action": args.action, "holdMs": args.hold} if kind == "stick": return {"type": "stick", "stick": args.stick, "x": args.x, "y": args.y, "holdMs": args.hold} if kind == "trigger": return {"type": "trigger", "trigger": args.trigger, "value": args.value, "holdMs": args.hold} raise ControlError(f"Unknown input type {kind}", "INVALID_INPUT") def xr_ppm_to_png(data: bytes, linear: bool = False) -> bytes: header = re.match(rb"P6\n([0-9]+) ([0-9]+)\n255\n", data) if not header: raise ControlError("Invalid XR image header", "XR_CAPTURE_FAILED") width, height = map(int, header.groups()) pixels = data[header.end():] if not (0 < width <= 16384 and 0 < height <= 16384) or len(pixels) != width * height * 3: raise ControlError("Invalid XR image dimensions or incomplete pixels", "XR_CAPTURE_FAILED") def chunk(kind: bytes, value: bytes) -> bytes: return struct.pack('>I', len(value)) + kind + value + struct.pack('>I', zlib.crc32(kind + value)) stride = width * 3 rows = b''.join(b'\0' + pixels[start:start + stride] for start in range(0, len(pixels), stride)) transfer = chunk(b'gAMA', struct.pack('>I', 100000)) if linear else chunk(b'sRGB', b'\0') return (b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', width, height, 8, 2, 0, 0, 0)) + transfer + chunk(b'IDAT', zlib.compress(rows)) + chunk(b'IEND', b'')) def xr_capture(adb: str, serial: str, destination: Path, timeout: float) -> dict: if not 0 < timeout <= 120: raise ControlError("XR capture timeout must be between 0 and 120 seconds", "INVALID_TIMEOUT") remote = f'/sdcard/Android/data/{PACKAGE}/files/xr-capture' token = uuid.uuid4().hex with tempfile.TemporaryDirectory() as temporary: request = Path(temporary) / 'request' request.write_text(token, encoding='ascii') pushed = run_adb(adb, serial, 'push', str(request), remote + '.request.tmp') if pushed.returncode: raise ControlError(pushed.stderr.decode(errors='replace'), 'XR_CAPTURE_FAILED') moved = run_adb(adb, serial, 'shell', 'mv', remote + '.request.tmp', remote + '.request') if moved.returncode: raise ControlError(moved.stderr.decode(errors='replace'), 'XR_CAPTURE_FAILED') deadline = time.monotonic() + timeout metadata = None while time.monotonic() < deadline: remaining = deadline - time.monotonic() if remaining <= 0: break try: result = run_adb(adb, serial, 'shell', 'cat', remote + '.json', timeout=remaining) except ControlError as error: if not isinstance(error.__cause__, subprocess.TimeoutExpired): raise break try: answer = json.loads(result.stdout) if answer.get('id') == token: metadata = answer break except (ValueError, AttributeError): pass time.sleep(0.2) if metadata is None: try: run_adb(adb, serial, 'shell', 'rm', '-f', remote + '.request', timeout=1) except ControlError: pass raise ControlError('No fresh XR capture. Enter SteamVR using a build with XR capture support.', 'TIMEOUT') if not metadata.get('ok'): detail = 'No valid XR projection layer is being submitted' if not metadata.get('validLayer') else 'GPU readback or image write failed' raise ControlError(detail, 'XR_CAPTURE_FAILED') images = {} for eye in (('left', 'right', 'source') if metadata.get('sourceImage') else ('left', 'right')): pulled = run_adb(adb, serial, 'exec-out', 'gzip', '-c', remote + '-' + eye + '.ppm', timeout=120) if pulled.returncode: raise ControlError(pulled.stderr.decode(errors='replace'), 'XR_CAPTURE_FAILED') try: pixels = gzip.decompress(pulled.stdout) except (OSError, EOFError, zlib.error) as error: raise ControlError('Incomplete compressed XR image', 'XR_CAPTURE_FAILED') from error images[eye] = xr_ppm_to_png(pixels, linear=eye == 'source' and metadata.get('sourceColorSpace') == 'linear') check = run_adb(adb, serial, 'shell', 'cat', remote + '.json') try: unchanged = json.loads(check.stdout).get('id') == token except (ValueError, AttributeError): unchanged = False if not unchanged: raise ControlError('XR capture was replaced by another request; retry', 'XR_CAPTURE_FAILED') try: destination.mkdir(parents=True, exist_ok=True) for eye, image in images.items(): (destination / (eye + '.png')).write_bytes(image) (destination / 'capture.json').write_text(json.dumps(metadata, indent=2), encoding='utf-8') except OSError as error: raise ControlError(str(error), 'FILE_IO') from error return {'ok': True, 'command': 'xr-frame', 'deviceSerial': serial, 'capture': metadata, 'left': str((destination / 'left.png').resolve()), 'right': str((destination / 'right.png').resolve()), **({'source': str((destination / 'source.png').resolve())} if 'source' in images else {}), 'scope': 'App eye textures before headset compositor reprojection; excludes Meta system overlays'} def execute(args: argparse.Namespace, adb: str, serial: str) -> dict: command = args.command if command == 'xr-frame': return xr_capture(adb, serial, Path(args.path), args.timeout) if command == "state": return add_serial(get_state(adb, serial), serial) if command == "start": request = {"mode": args.mode} if args.mode == "steam": if args.ui == "desktop": request["steamUi"] = "desktop" if args.url: if not args.url.startswith("steam://"): raise ControlError("Steam URL must start with steam://", "INVALID_STEAM_URL") request["steamUrl"] = args.url if args.steamvr: request["steamVr"] = True if args.reuse: current = get_state(adb, serial)["session"] wanted = {"steam": "steam", "desktop": "lxqt"}[args.mode] if current.get("running") and current.get("mode") == wanted and current.get("phase") not in {"STOPPING", "FAILED"}: state = wait_for_phase(adb, serial, "READY", args.timeout) if args.wait else get_state(adb, serial) return add_serial(dict(state, reused=True), serial) started = start_session(adb, serial, request) if args.wait: return add_serial(wait_for_phase(adb, serial, "READY", args.timeout), serial) session = started["session"] return add_serial({"ok": True, "command": "start", "mode": args.mode, "sessionId": session.get("id"), "phase": session.get("phase")}, serial) if command == "run": program_args = args.arguments[1:] if args.arguments[:1] == ["--"] else args.arguments started = start_session(adb, serial, {"mode": "run", "program": args.program, "programArgs": program_args}) session = started["session"] return add_serial({"ok": True, "command": "start", "mode": "run", "sessionId": session.get("id"), "phase": session.get("phase")}, serial) if command == "wait": if args.phase == "cool": if args.below is None: raise ControlError("wait cool needs --below <°C>", "INVALID_ARGUMENTS") return add_serial(wait_cool(adb, serial, args.below, args.timeout), serial) if args.phase == "focused": if args.app_id is None: raise ControlError("wait focused needs an app id", "INVALID_ARGUMENTS") return add_serial(wait_until(adb, serial, f"app {args.app_id} to have focus", lambda s: focused(s, args.app_id), args.timeout, 0.5), serial) if args.phase == "game": return add_serial(wait_until(adb, serial, "a game to have focus", lambda s: focused(s, None), args.timeout, 0.5), serial) phases = {"ready": "READY", "idle": "IDLE", "failed": "FAILED"} return add_serial(wait_for_phase(adb, serial, phases[args.phase], args.timeout), serial) if command == "stop": current = get_state(adb, serial) if current["session"].get("phase") in {"IDLE", "FAILED"}: return add_serial(wait_for_artifacts(adb, serial, current, args.artifact_timeout), serial) checked(provider_call(adb, serial, "stop"), "STOP_FAILED") state = wait_for_phase(adb, serial, "IDLE", args.timeout) return add_serial(wait_for_artifacts(adb, serial, state, args.artifact_timeout), serial) if command == "resume": return add_serial(checked(provider_call(adb, serial, "resume"), "RESUME_FAILED"), serial) if command == "launch": response = checked(provider_call(adb, serial, "launch", request={"appId": str(args.app_id)}), "LAUNCH_FAILED") if args.wait: state = wait_until(adb, serial, f"app {args.app_id} to have focus", lambda s: focused(s, args.app_id), args.timeout, 0.5) response["focus"] = state["session"].get("focus") return add_serial(response, serial) if command == "quit": response = provider_call(adb, serial, "quit", request={"appId": args.app_id, "timeout": args.timeout}, timeout=args.timeout + 40) return add_serial(checked(response, "QUIT_FAILED"), serial) if command == "focus": return add_serial(checked(provider_call(adb, serial, "focus", timeout=40), "FOCUS_FAILED"), serial) if command == "guest": argv = args.argv[1:] if args.argv[:1] == ["--"] else args.argv if not argv: raise ControlError("guest needs a command after --", "INVALID_ARGUMENTS") env = {} for item in args.env or []: key, sep, value = item.partition("=") if not sep: raise ControlError(f"--env takes KEY=VALUE, not {item}", "INVALID_ARGUMENTS") env[key] = value request = {"argv": argv, "timeout": args.timeout, "env": env} if args.cwd: request["cwd"] = args.cwd if args.stdin is not None: request["stdin"] = args.stdin response = checked(provider_call(adb, serial, "guest", request=request, timeout=args.timeout + 40), "GUEST_FAILED") if args.check and response.get("exitCode") != 0: raise ControlError(f"{argv[0]} exited with {response.get('exitCode')}: " f"{(response.get('stderr') or {}).get('text', '')[-400:]}", "GUEST_EXIT") return add_serial(response, serial) if command == "cdp": if args.cdp_op == "targets": request = {"op": "targets"} else: request = {"op": "eval", "expression": args.expression, "awaitPromise": not args.no_await, "timeout": args.timeout} if args.target: request["target"] = args.target timeout = getattr(args, "timeout", 15) return add_serial(checked(provider_call(adb, serial, "cdp", request=request, timeout=timeout + 40), "CDP_FAILED"), serial) if command == "input": return add_serial(checked(provider_call(adb, serial, "input", request=input_request(args), timeout=60), "INPUT_FAILED"), serial) if command == "ui": if args.ui_op == "dump": request = {"op": "dump", "all": args.all} else: request = {"op": args.ui_op, "index": args.index} if args.match: request["match"] = args.match else: request["tag"] = args.tag if args.ui_op == "set-text": request["text"] = args.text return add_serial(checked(provider_call(adb, serial, "ui", request=request), "UI_FAILED"), serial) if command == "proot-profile": lines = ["PROOT_PROFILE=1", f"PROOT_PROFILE_SLOW_US={args.slow_us}"] if args.profile_op == "enable" else ["PROOT_PROFILE=0"] return add_serial(checked(provider_call(adb, serial, "env", request={"op": "set", "lines": lines, "persistent": False}), "ENV_FAILED"), serial) if command == "env": if args.env_op == "set": request = {"op": "set", "lines": args.lines, "persistent": args.persistent} elif args.env_op == "clear": request = {"op": "clear", "scope": args.scope} else: request = {"op": "list"} return add_serial(checked(provider_call(adb, serial, "env", request=request), "ENV_FAILED"), serial) if command == "override": access = require_commands(adb, serial) if not re.fullmatch(r"[A-Za-z0-9._-]{1,64}", args.name) or args.name in {".", ".."}: raise ControlError("Override names must be simple filenames", "INVALID_NAME") source = Path(args.path).expanduser() if not source.is_file(): raise ControlError(f"No such file: {source}", "FILE_IO") inbox = access["inbox"] run_adb(adb, serial, "shell", "mkdir", "-p", inbox) pushed = run_adb(adb, serial, "push", str(source), f"{inbox}/{args.name}", timeout=300) if pushed.returncode != 0: raise ControlError(pushed.stderr.decode(errors="replace").strip() or "adb push failed", "PUSH_FAILED") return add_serial(checked(provider_call(adb, serial, "override", request={"name": args.name}), "OVERRIDE_FAILED"), serial) if command == "prefs": if args.prefs_op == "set": request = {"op": "set", "file": args.file, "key": args.key, "type": args.type, "value": args.value} else: request = {"op": "get", "file": args.file} return add_serial(checked(provider_call(adb, serial, "prefs", request=request), "PREFS_FAILED"), serial) if command == "access": return add_serial(checked(provider_call(adb, serial, "access"), "ACCESS_FAILED"), serial) if command == "glasses": request = {"op": args.glasses_op or "status"} if args.glasses_op == "set": request.update(key=args.key, value=args.value) elif args.glasses_op == "simulate": request["seconds"] = args.seconds if args.yaw is not None: request["yaw"] = args.yaw if args.pitch is not None: request["pitch"] = args.pitch elif args.glasses_op == "mode": request["mode"] = args.mode elif args.glasses_op == "imu": request["ms"] = args.ms elif args.glasses_op == "spatial-diagnostics": request["enabled"] = not args.off elif args.glasses_op == "spatial-recenter": request["screen"] = args.screen elif args.glasses_op == "raw": request.update(msgId=args.msg_id, payload=args.payload or "") return add_serial(checked(provider_call(adb, serial, "glasses", request=request), "GLASSES_FAILED"), serial) if command == "logs": state = get_state(adb, serial) session = state.get("session", {}) if not session.get("running") and session.get("artifactsAvailable") and not session.get("artifactsComplete"): state = wait_for_artifacts(adb, serial, state, args.timeout) session = state.get("session", {}) remote_dir = session.get("logDir") if not remote_dir or not session.get("id") or not session.get("artifactsAvailable"): raise ControlError("No session artifact directory is available", "NO_SESSION_LOGS") destination = Path(args.destination or session["id"]).expanduser().resolve() exported = provider_call(adb, serial, "logs", timeout=args.timeout + 40) cleanup_error = None if (exported.get("error") or {}).get("code") == "UNKNOWN_COMMAND": # Older builds have no logs method; their private folder only pulls on a rooted device. result = run_adb(adb, serial, "pull", remote_dir, str(destination), timeout=120) else: zip_path = checked(exported, "LOG_PULL_FAILED")["zip"] remote_zip = PurePosixPath(zip_path) if (not remote_zip.is_absolute() or ".." in remote_zip.parts or not re.fullmatch(r"/[A-Za-z0-9_./-]+", zip_path)): raise ControlError("The exported logs path is invalid", "LOG_PULL_FAILED") requested_dir = exported.get("cleanupDir") request_dir = PurePosixPath(requested_dir) if isinstance(requested_dir, str) else None cleanup = ["shell", "rm", "-f", zip_path] if (request_dir and request_dir.is_absolute() and ".." not in request_dir.parts and request_dir == remote_zip.parent and request_dir.parent.name == "agent-logs" and re.fullmatch(r"[0-9a-fA-F-]{36}", request_dir.name)): cleanup = ["shell", "rm", "-rf", str(request_dir)] def cleanup_export(): try: cleaned = run_adb(adb, serial, *cleanup, timeout=30) if cleaned.returncode == 0: return None detail = (cleaned.stderr or cleaned.stdout).decode(errors="replace").strip() except Exception as error: detail = str(error) return ControlError( "Could not remove the exported log archive; it may remain in shared storage" + (f": {detail}" if detail else ""), "LOG_CLEANUP_FAILED", ) try: if exported.get("sessionId", session["id"]) != session["id"]: raise ControlError("A new session started while the logs were exported; run logs again", "LOG_PULL_FAILED") with tempfile.TemporaryDirectory() as temp: local_zip = Path(temp) / "logs.zip" result = run_adb(adb, serial, "pull", zip_path, str(local_zip), timeout=120) if result.returncode != 0: detail = (result.stderr or result.stdout).decode(errors="replace").strip() raise ControlError(detail or "adb pull failed", "LOG_PULL_FAILED") unpack_logs(local_zip, destination, session["id"]) except Exception: cleanup_error = cleanup_export() if cleanup_error: print(cleanup_error, file=sys.stderr) raise else: # The copy sits in shared storage, so it does not outlive the pull. cleanup_error = cleanup_export() if result.returncode != 0: detail = (result.stderr or result.stdout).decode(errors="replace").strip() raise ControlError(detail or "adb pull failed", "LOG_PULL_FAILED") if cleanup_error: raise cleanup_error return add_serial({"ok": True, "command": "logs", "sessionId": session["id"], "path": str(destination)}, serial) if command == "screenshot": return add_serial(screenshot(adb, serial, args.path, args.display), serial) if command == "record": return add_serial(record(adb, serial, args.path, args.seconds, args.display), serial) if command == "pacing": return add_serial(pacing(adb, serial, args.seconds), serial) if command == "thermal": return add_serial(thermal(adb, serial), serial) if command == "wake": run_adb(adb, serial, "shell", "input", "keyevent", "KEYCODE_WAKEUP") run_adb(adb, serial, "shell", "wm", "dismiss-keyguard") power = run_adb(adb, serial, "shell", "dumpsys", "power").stdout.decode(errors="replace") awake = re.search(r"mWakefulness=(\w+)", power) return add_serial({"ok": True, "command": "wake", "wakefulness": awake.group(1) if awake else None}, serial) if command == "displays": return add_serial({"ok": True, "command": "displays", "displays": displays(adb, serial)}, serial) raise ControlError(f"Unknown command: {command}", "UNKNOWN_COMMAND") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--adb", help="adb executable (defaults to ADB, PATH, or the Android SDK)") parser.add_argument("--serial", help="ADB serial; defaults to ADB_SERIAL or ANDROID_SERIAL") commands = parser.add_subparsers(dest="command", required=True) state = commands.add_parser("state", help="print machine-readable session state") state.add_argument("--json", action="store_true", help="accepted for explicit machine-readable usage") start = commands.add_parser("start", help="start Steam or the desktop") start.add_argument("mode", choices=["steam", "desktop"]) start.add_argument("--ui", choices=["bigpicture", "desktop"], default="bigpicture") start.add_argument("--steamvr", action="store_true", help="start SteamVR with the session (VITURE glasses)") start.add_argument("--url", help="steam:// URL to pass to the client") start.add_argument("--wait", action="store_true", help="wait for READY after starting") start.add_argument("--reuse", action="store_true", help="keep a running session of this mode instead of failing") start.add_argument("--timeout", type=float, default=90) run = commands.add_parser("run", help="run a guest program under Gamescope") run.add_argument("program") run.add_argument("arguments", nargs=argparse.REMAINDER) wait = commands.add_parser("wait", help="wait for a session phase, for a game to have focus, or for the battery to cool") wait.add_argument("phase", choices=["ready", "idle", "failed", "focused", "game", "cool"]) wait.add_argument("app_id", nargs="?", type=int, help="with focused: the Steam app id") wait.add_argument("--below", type=float, help="with cool: the battery temperature to get under, in °C") wait.add_argument("--timeout", type=float, default=90) stop = commands.add_parser("stop", help="stop the session and wait for IDLE") stop.add_argument("--timeout", type=float, default=90) stop.add_argument("--artifact-timeout", type=float, default=300) commands.add_parser("resume", help="resume a suspended session") launch = commands.add_parser("launch", help="launch a Steam game in the running Steam session") launch.add_argument("app_id", type=int) launch.add_argument("--wait", action="store_true", help="wait until the game has focus") launch.add_argument("--timeout", type=float, default=180) quit_ = commands.add_parser("quit", help="quit a running game (through the client, then signals)") quit_.add_argument("app_id", type=int) quit_.add_argument("--timeout", type=float, default=15) commands.add_parser("focus", help="gamescope's focus and every Xwayland server's windows, fresh") guest = commands.add_parser("guest", help="run a command inside the session (debugging tools)") guest.add_argument("--timeout", type=float, default=30) guest.add_argument("--env", action="append", metavar="KEY=VALUE") guest.add_argument("--cwd") guest.add_argument("--stdin") guest.add_argument("--check", action="store_true", help="fail unless the command exits 0") guest.add_argument("argv", nargs=argparse.REMAINDER) cdp = commands.add_parser("cdp", help="the Steam client's DevTools (debugging tools)") cdp_ops = cdp.add_subparsers(dest="cdp_op", required=True) cdp_ops.add_parser("targets", help="list DevTools targets") evaluate = cdp_ops.add_parser("eval", help="evaluate JavaScript (default target SharedJSContext)") evaluate.add_argument("expression") evaluate.add_argument("--target", help="title or URL substring of the target") evaluate.add_argument("--no-await", action="store_true", help="do not await a returned promise") evaluate.add_argument("--timeout", type=float, default=15) inp = commands.add_parser("input", help="keys, text, touch and pad input into the session") kinds = inp.add_subparsers(dest="input_type", required=True) key = kinds.add_parser("key", help="a key by name (ENTER, ESC, A, F1...) or evdev code") key.add_argument("key") key.add_argument("--action", choices=["tap", "down", "up"], default="tap") key.add_argument("--mod", action="append", help="modifier held around a tap (CTRL, SHIFT, ALT, META)") key.add_argument("--hold", type=int, default=40, help="ms between down and up") text = kinds.add_parser("text", help="type text (US layout keys; --commit for text-input)") text.add_argument("text") text.add_argument("--commit", action="store_true") tap = kinds.add_parser("tap", help="touch at a point of the session's output") tap.add_argument("x", type=float) tap.add_argument("y", type=float) tap.add_argument("--space", choices=["output", "fraction"], default="output") tap.add_argument("--hold", type=int, default=60) swipe = kinds.add_parser("swipe", help="drag a touch from one point to another") for name in ("x", "y", "x2", "y2"): swipe.add_argument(name, type=float) swipe.add_argument("--space", choices=["output", "fraction"], default="output") swipe.add_argument("--duration", type=int, default=300) pad = kinds.add_parser("pad", help="a virtual pad button (a, b, x, y, lb, rb, start, select, guide, qam, up...)") pad.add_argument("button") pad.add_argument("--action", choices=["tap", "down", "up"], default="tap") pad.add_argument("--hold", type=int, default=120) stick = kinds.add_parser("stick", help="hold a stick at x, y (-1..1)") stick.add_argument("stick", choices=["left", "right"]) stick.add_argument("x", type=float) stick.add_argument("y", type=float) stick.add_argument("--hold", type=int, default=300) trigger = kinds.add_parser("trigger", help="hold a trigger (0..1)") trigger.add_argument("trigger", choices=["left", "right"]) trigger.add_argument("value", type=float, nargs="?", default=1.0) trigger.add_argument("--hold", type=int, default=200) ui = commands.add_parser("ui", help="the app's own UI: dump nodes, click or type by test tag") ui_ops = ui.add_subparsers(dest="ui_op", required=True) dump = ui_ops.add_parser("dump", help="tagged, labelled and clickable nodes of every app window") dump.add_argument("--all", action="store_true", help="every node") for name in ("click", "set-text"): op = ui_ops.add_parser(name) op.add_argument("tag", nargs="?", help="test tag, e.g. rail-steam") op.add_argument("--match", help="instead of a tag: text or description containing this") op.add_argument("--index", type=int, default=0, help="which match, in tree order") if name == "set-text": op.add_argument("--text", required=True) profile = commands.add_parser("proot-profile", help="profile PRoot syscall handlers in the next session, or rank a saved log") profile_ops = profile.add_subparsers(dest="profile_op", required=True) profile_enable = profile_ops.add_parser("enable", help="enable counters and bounded slow samples for the next session") profile_enable.add_argument("--slow-us", type=profile_slow_us, default=1000, metavar="US") profile_ops.add_parser("disable", help="disable profiling for the next session") profile_report = profile_ops.add_parser("report", help="read a local session.log without a device") profile_report.add_argument("path", type=Path) profile_report.add_argument("--sort", choices=["wall", "cpu", "calls"], default="wall") profile_report.add_argument("--limit", type=int, default=20) env = commands.add_parser("env", help="session environment lines (debugging tools)") env_ops = env.add_subparsers(dest="env_op", required=True) env_ops.add_parser("list") env_set = env_ops.add_parser("set", help="KEY=VALUE lines for the next session") env_set.add_argument("lines", nargs="+") env_set.add_argument("--persistent", action="store_true", help="every session until cleared") env_clear = env_ops.add_parser("clear") env_clear.add_argument("scope", nargs="?", choices=["next", "persistent", "all"], default="next") override = commands.add_parser("override", help="override a program or libblfastpath.so for the next session (debugging tools)") override.add_argument("name", help="the program name, e.g. gamescope") override.add_argument("path", help="the local file to push") prefs = commands.add_parser("prefs", help="the app's preferences (debugging tools)") prefs_ops = prefs.add_subparsers(dest="prefs_op", required=True) prefs_get = prefs_ops.add_parser("get") prefs_get.add_argument("file", help="preferences file, e.g. session") prefs_set = prefs_ops.add_parser("set") prefs_set.add_argument("file") prefs_set.add_argument("key") prefs_set.add_argument("value", nargs="?") prefs_set.add_argument("--type", choices=["bool", "int", "long", "float", "string", "remove"], default="string") commands.add_parser("access", help="whether debugging tools are enabled on the device") glasses = commands.add_parser("glasses", help="VITURE glasses: status, settings, head tracking test") glasses_ops = glasses.add_subparsers(dest="glasses_op") glasses_ops.add_parser("status") glasses_ops.add_parser("allow", help="ask for USB control of the glasses again") glasses_set = glasses_ops.add_parser("set", help="refreshHz, sbsHz, resolution2d, resolution3d, resolutionVr (auto or WIDTHxHEIGHT per eye), brightness, volume, sunblock, 3d, 3dof, gameOnGlasses, headTracking, auto3d, picture3d, screenWidthDm, screenDistanceDm, vrFovDeg") glasses_set.add_argument("key") glasses_set.add_argument("value") glasses_sim = glasses_ops.add_parser("simulate", help="feed a moving head pose (yaw, pitch and roll sweeps), or hold one") glasses_sim.add_argument("--seconds", type=int, default=20) glasses_sim.add_argument("--yaw", type=float, help="hold this yaw (degrees, left positive) instead of sweeping") glasses_sim.add_argument("--pitch", type=float, help="hold this pitch (degrees, down positive) instead of sweeping") glasses_ops.add_parser("stop-simulate") glasses_ops.add_parser("query", help="ask the glasses for their display mode, brightness, volume and IMU state") spatial_diagnostics = glasses_ops.add_parser("spatial-diagnostics", help="start stereo diagnostics without saving images") spatial_diagnostics.add_argument("--off", action="store_true") spatial_recenter = glasses_ops.add_parser("spatial-recenter", help="recenter the current spatial origin") spatial_recenter.add_argument("--screen", action="store_true") glasses_ops.add_parser("spatial-floor", help="set floor using configured eye height") glasses_ops.add_parser("spatial-restart", help="restart the provider for explicit tracking recovery") glasses_mode = glasses_ops.add_parser("mode", help="send a display mode byte, e.g. 0x31 (1080p60), 0x34 (1080p120), 0x32 (3D)") glasses_mode.add_argument("mode") glasses_imu = glasses_ops.add_parser("imu", help="stream the glasses' IMU for a moment and show what it reports") glasses_imu.add_argument("--ms", type=int, default=2000) glasses_raw = glasses_ops.add_parser("raw", help="send one MCU command (debugging tools)") glasses_raw.add_argument("msg_id", help="hex message id, e.g. 0x0007") glasses_raw.add_argument("payload", nargs="?", help="hex bytes") logs = commands.add_parser("logs", help="pull the latest session artifact directory") logs.add_argument("which", choices=["latest"]) logs.add_argument("destination", nargs="?") logs.add_argument("--timeout", type=float, default=90) shot = commands.add_parser("screenshot", help="save a device screenshot") shot.add_argument("path") shot.add_argument("--display", type=int, help="logical display id (see displays); default the default display") xr = commands.add_parser('xr-frame', help='capture both submitted XR eyes and pose metadata') xr.add_argument('path', help='output directory for left.png, right.png and capture.json') xr.add_argument('--timeout', type=float, default=15) rec = commands.add_parser("record", help="record the screen to an mp4") rec.add_argument("path") rec.add_argument("--seconds", type=float, default=10) rec.add_argument("--display", type=int) pace = commands.add_parser("pacing", help="frames that reached the screen and their intervals (SurfaceFlinger)") pace.add_argument("--seconds", type=float, default=30) commands.add_parser("thermal", help="battery temperature, thermal status and CPU frequency caps") commands.add_parser("displays", help="the device's displays and their capture ids") commands.add_parser("wake", help="wake the device and dismiss the keyguard") return parser EXIT_ADB = {"ADB_NOT_FOUND", "ADB_FAILED", "INVALID_PROVIDER_RESPONSE"} EXIT_SESSION = { "SESSION_ACTIVE", "NO_ACTIVE_SESSION", "SESSION_NOT_SUSPENDED", "SESSION_FAILED", "GUEST_EXIT", "GUEST_START_FAILED", "RUNTIME_INSTALL_FAILED", "DESKTOP_INSTALL_FAILED", "ACTIVITY_START_FAILED", "NO_STEAM_SESSION", "SESSION_NOT_READY", "AGENT_UNAVAILABLE", "LAUNCH_REJECTED", "NO_PAD", "NO_FRAMES", } EXIT_TIMEOUT = {"TIMEOUT", "ARTIFACT_TIMEOUT"} EXIT_FILE = {"NO_SESSION_LOGS", "LOG_PULL_FAILED", "LOG_CLEANUP_FAILED", "SCREENSHOT_FAILED", "FILE_IO", "RECORD_FAILED", "PUSH_FAILED", "PACING_FAILED", "XR_CAPTURE_FAILED"} EXIT_DENIED = {"AGENT_COMMANDS_DISABLED", "PROTECTED_PREFS", "PROTECTED_CONTROL"} def exit_code(exc: Exception) -> int: if isinstance(exc, DeviceResolutionError): return 3 code = getattr(exc, "code", "") if code in EXIT_ADB: return 3 if code in EXIT_SESSION: return 4 if code in EXIT_TIMEOUT: return 5 if code in EXIT_FILE: return 6 if code in EXIT_DENIED: return 7 return 2 def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) serial = None try: if not PACKAGE_NAME.fullmatch(PACKAGE): raise ControlError(f"DROIDDECK_PACKAGE is not an Android package name: {PACKAGE!r}", "INVALID_REQUEST") if args.command == "proot-profile" and args.profile_op == "report": if args.limit < 1: raise ControlError("--limit must be positive", "INVALID_ARGUMENTS") try: result = proot_profile_report(args.path.expanduser(), args.sort, args.limit) except OSError as exc: raise ControlError(str(exc), "FILE_IO") from exc print(json.dumps(result, indent=2, sort_keys=True)) return 0 adb = adb_path(args.adb) serial = resolve_serial(adb, args.serial, log=True) result = execute(args, adb, serial) print(json.dumps(result, indent=2, sort_keys=True)) return 0 except (ControlError, DeviceResolutionError) as exc: error = {"code": getattr(exc, "code", "DEVICE_RESOLUTION_FAILED"), "message": str(exc)} code = exit_code(exc) except Exception as exc: error = {"code": "INTERNAL_ERROR", "message": f"{type(exc).__name__}: {exc}"} code = 2 response = {"ok": False, "error": error} if serial: response["deviceSerial"] = serial print(json.dumps(response, indent=2, sort_keys=True)) return code if __name__ == "__main__": raise SystemExit(main())