mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-10-14 19:27:09 -04:00
768 lines
37 KiB
Python
Executable File
768 lines
37 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Control a DroidDeck build over ADB (docs/agent-control.md)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from adb_device import DeviceResolutionError, resolve_serial # noqa: E402
|
|
|
|
|
|
AUTHORITY = "content://com.droiddeck.launcher.agent"
|
|
PACKAGE = "com.droiddeck.launcher"
|
|
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 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:
|
|
"""A provider or guest answer, raised as a ControlError when it carries ok: false."""
|
|
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}/.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:
|
|
"""Poll state until done(state) is true. A failed session ends the wait at once."""
|
|
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:
|
|
"""The game has focus the way a player sees it: gamescope focuses it and it is in the list the
|
|
Steam client orders the base layer from. app_id None means any game (not the client)."""
|
|
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
|
|
|
|
|
|
# --- displays ------------------------------------------------------------------------------------
|
|
|
|
def displays(adb: str, serial: str) -> list[dict]:
|
|
"""Logical displays with the SurfaceFlinger (physical) id screencap and screenrecord want."""
|
|
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}
|
|
|
|
|
|
# --- commands ------------------------------------------------------------------------------------
|
|
|
|
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 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 > Session > 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 execute(args: argparse.Namespace, adb: str, serial: str) -> dict:
|
|
command = args.command
|
|
|
|
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:
|
|
# A scenario's per-run "before": keep a session of this mode that is already up, so only
|
|
# a run after one that ended (a crash, the low-memory killer) starts a new one.
|
|
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 == "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 == "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 == "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()
|
|
result = run_adb(adb, serial, "pull", remote_dir, str(destination), 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")
|
|
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 == "wake":
|
|
# A sleeping device stops the session and hides the app's windows from ui dump.
|
|
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, or for a game to have focus")
|
|
wait.add_argument("phase", choices=["ready", "idle", "failed", "focused", "game"])
|
|
wait.add_argument("app_id", nargs="?", type=int, help="with focused: the Steam app id")
|
|
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)
|
|
# Collecting the client's logs and redacting them took about two minutes after a Skyrim session.
|
|
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)
|
|
|
|
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="put a binary first on the next session's PATH (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, brightness, volume, sunblock, 3d, 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")
|
|
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")
|
|
|
|
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)
|
|
|
|
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",
|
|
}
|
|
EXIT_TIMEOUT = {"TIMEOUT", "ARTIFACT_TIMEOUT"}
|
|
EXIT_FILE = {"NO_SESSION_LOGS", "LOG_PULL_FAILED", "SCREENSHOT_FAILED", "FILE_IO", "RECORD_FAILED", "PUSH_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:
|
|
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: # the contract is JSON on stdout, even for a bug in this tool
|
|
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())
|