mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-11-13 03:51:10 +00:00
maxjivi05
1e66ff17c1
Glasses get their own controller over the MCU (permission, 120 Hz default, brightness, volume, sunblock, 2D/3D) and the game moves onto them when plugged in. The IMU drives Cardboard head tracking and the SteamVR driver. 3D falls back to one eye's view on phones that can't send the 3840x1080 signal. Options open from a glasses button in game and on the Steam page.
759 lines
36 KiB
Python
Executable File
759 lines
36 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.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
|
|
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("--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")
|
|
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)")
|
|
glasses_sim.add_argument("--seconds", type=int, default=20)
|
|
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())
|