206 lines
7.8 KiB
Python
Executable File
206 lines
7.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Small ADB-command compatibility shim backed by maa-online HTTP API.
|
|
|
|
It lets an unmodified MaaCore/maa-cli use the cloud-game backend, preserving
|
|
maa-cli's normal `maa self update` and `maa update` mechanisms.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gzip
|
|
import json
|
|
import os
|
|
import shlex
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from PIL import Image
|
|
except ModuleNotFoundError:
|
|
root = Path(__file__).resolve().parents[1]
|
|
venv_python = root / "runtime/venv/bin/python"
|
|
if venv_python.is_file() and Path(sys.executable) != venv_python:
|
|
os.execv(venv_python, [str(venv_python), str(Path(__file__).resolve()), *sys.argv[1:]])
|
|
raise
|
|
|
|
BASE_URL = os.getenv("MAA_ONLINE_API", "http://127.0.0.1:22888").rstrip("/")
|
|
SERIAL = os.getenv("MAA_ONLINE_ADB_SERIAL", "cloud-game:22888")
|
|
TIMEOUT = float(os.getenv("MAA_ONLINE_ADB_TIMEOUT", "40"))
|
|
|
|
|
|
def request(path: str, payload: dict | None = None) -> tuple[int, bytes, dict[str, str]]:
|
|
data = None
|
|
headers: dict[str, str] = {}
|
|
method = "GET"
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
method = "POST"
|
|
req = urllib.request.Request(BASE_URL + path, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=TIMEOUT) as response:
|
|
return response.status, response.read(), dict(response.headers.items())
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code, exc.read(), dict(exc.headers.items())
|
|
|
|
|
|
def fetch_png() -> bytes:
|
|
status, body, headers = request("/screencap")
|
|
if status != 200:
|
|
raise RuntimeError(f"screencap failed: HTTP {status}: {body.decode(errors='replace')}")
|
|
content_type = headers.get("Content-Type", "")
|
|
if "image/png" not in content_type:
|
|
raise RuntimeError(f"unexpected screenshot content type: {content_type}")
|
|
return body
|
|
|
|
|
|
def fetch_raw_screencap() -> bytes:
|
|
image = Image.open(BytesIO(fetch_png())).convert("RGBA")
|
|
# Android screencap raw format: little-endian width, height, format=RGBA_8888.
|
|
return struct.pack("<III", image.width, image.height, 1) + image.tobytes()
|
|
|
|
|
|
def emit(data: bytes | str) -> None:
|
|
if isinstance(data, str):
|
|
data = data.encode("utf-8")
|
|
sys.stdout.buffer.write(data)
|
|
|
|
|
|
def fail(message: str, code: int = 1) -> int:
|
|
print(message, file=sys.stderr)
|
|
return code
|
|
|
|
|
|
def strip_global_options(argv: list[str]) -> list[str]:
|
|
result: list[str] = []
|
|
index = 0
|
|
options_with_value = {"-s", "-H", "-P", "-L", "--one-device"}
|
|
while index < len(argv):
|
|
if argv[index] in options_with_value and index + 1 < len(argv):
|
|
index += 2
|
|
continue
|
|
if argv[index] in {"-d", "-e", "-a"}:
|
|
index += 1
|
|
continue
|
|
result.extend(argv[index:])
|
|
break
|
|
return result
|
|
|
|
|
|
def shell(command: str) -> int:
|
|
normalized = " ".join(command.strip().split())
|
|
if "settings get secure android_id" in normalized:
|
|
emit("6d61616f6e6c696e\n")
|
|
elif "getprop ro.build.version.release" in normalized:
|
|
emit("9\n")
|
|
elif "getprop ro.product.cpu.abilist" in normalized:
|
|
emit("x86_64,arm64-v8a,armeabi-v7a\n")
|
|
elif normalized.startswith("wm size"):
|
|
status, body, _ = request("/info")
|
|
if status != 200:
|
|
return fail(f"cloud backend unavailable: HTTP {status}")
|
|
info = json.loads(body)
|
|
if info.get("status") != "ok":
|
|
return fail(f"cloud backend is {info.get('status', 'not ready')}")
|
|
emit(f"{int(info['width'])}\n{int(info['height'])}\n")
|
|
elif "SurfaceOrientation" in normalized:
|
|
emit("0\n")
|
|
elif "SurfaceFlinger --latency" in normalized:
|
|
emit("33333333\n")
|
|
elif normalized.startswith("input tap "):
|
|
fields = normalized.split()
|
|
status, body, _ = request("/click", {"x": int(fields[-2]), "y": int(fields[-1])})
|
|
if status != 200:
|
|
return fail(f"click failed: HTTP {status}: {body.decode(errors='replace')}")
|
|
elif normalized.startswith("input swipe "):
|
|
fields = normalized.split()
|
|
status, body, _ = request(
|
|
"/swipe",
|
|
{
|
|
"x1": int(fields[-5]), "y1": int(fields[-4]),
|
|
"x2": int(fields[-3]), "y2": int(fields[-2]),
|
|
"duration": int(fields[-1]),
|
|
},
|
|
)
|
|
if status != 200:
|
|
return fail(f"swipe failed: HTTP {status}: {body.decode(errors='replace')}")
|
|
elif normalized.startswith("input text "):
|
|
text = normalized.removeprefix("input text ")
|
|
status, body, _ = request("/input", {"text": text})
|
|
if status != 200:
|
|
return fail(f"text input failed: HTTP {status}: {body.decode(errors='replace')}")
|
|
elif normalized.startswith("input keyevent "):
|
|
return fail("cloud backend key events are not implemented")
|
|
elif normalized.startswith("am start "):
|
|
status, body, _ = request("/start", {})
|
|
if status not in {200, 202}:
|
|
return fail(f"cloud start failed: HTTP {status}: {body.decode(errors='replace')}")
|
|
# MaaCore's legacy startup probe treats a sufficiently long `am start`
|
|
# response as confirmation that Android accepted the launch request.
|
|
emit(
|
|
"Starting: Intent { act=android.intent.action.MAIN cat=[android.intent.category.LAUNCHER] "
|
|
"flg=0x10200000 cmp=com.hypergryph.arknights/com.u8.sdk.U8UnityContext }\n"
|
|
"Status: ok\nActivity: com.hypergryph.arknights/com.u8.sdk.U8UnityContext\n"
|
|
)
|
|
elif normalized.startswith("am force-stop "):
|
|
status, body, _ = request("/exit", {})
|
|
if status != 200:
|
|
return fail(f"cloud stop failed: HTTP {status}: {body.decode(errors='replace')}")
|
|
elif normalized.startswith(("dumpsys input", "cat /proc/net/arp")):
|
|
emit("0\n")
|
|
else:
|
|
return fail(f"unsupported adb shell command: {normalized}")
|
|
return 0
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
args = strip_global_options(argv)
|
|
if not args:
|
|
return fail("adb-cloud: missing command")
|
|
if args[0] == "devices":
|
|
emit(f"List of devices attached\n{SERIAL}\tdevice\n")
|
|
return 0
|
|
if args[0] == "connect":
|
|
emit(f"already connected to {args[1] if len(args) > 1 else SERIAL}\n")
|
|
return 0
|
|
if args[0] in {"start-server", "kill-server", "wait-for-device"}:
|
|
return 0
|
|
if args[0] == "get-state":
|
|
emit("device\n")
|
|
return 0
|
|
if args[0] == "exec-out":
|
|
command = " ".join(args[1:])
|
|
if "screencap" not in command:
|
|
return fail(f"unsupported adb exec-out command: {command}")
|
|
if "nc " in command:
|
|
fields = shlex.split(command)
|
|
nc_index = fields.index("nc")
|
|
# MaaCore passes the host address as seen from an Android guest
|
|
# (often 10.0.2.2). The shim itself runs on the host, so loop back.
|
|
host, port = "127.0.0.1", int(fields[-1])
|
|
if nc_index < 0:
|
|
return fail(f"invalid nc screencap command: {command}")
|
|
with socket.create_connection((host, port), timeout=TIMEOUT) as connection:
|
|
connection.sendall(fetch_raw_screencap())
|
|
return 0
|
|
if "gzip" in command:
|
|
emit(gzip.compress(fetch_raw_screencap(), compresslevel=1))
|
|
return 0
|
|
emit(fetch_png())
|
|
return 0
|
|
if args[0] == "shell":
|
|
return shell(" ".join(args[1:]))
|
|
return fail(f"unsupported adb command: {' '.join(args)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main(sys.argv[1:]))
|
|
except (OSError, RuntimeError, ValueError, KeyError, json.JSONDecodeError) as exc:
|
|
raise SystemExit(f"adb-cloud: {exc}")
|