ascli_tool.py

"""
ascli — Native Hermes tool wrapping the `ascli` Android Studio toolchain CLI
(SDK packages, AVDs/emulator, ADB devices, Gradle builds, APK inspection,
project parsing, Android Studio IDE). Single action-dispatch tool; backend is
~/.hermes/scripts/ascli/ascli.py, which auto-detects the Android SDK + JDK and
emits compact pointer JSON. This tool passes the agent's request through and
returns the CLI's JSON inline (capped).

Author: Peter/lesterppo
"""

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any

try:
    from hermes_constants import get_hermes_home
except ImportError:
    def get_hermes_home() -> Path:
        return Path.home() / ".hermes"


# ---------------------------------------------------------------------------
# Gate
# ---------------------------------------------------------------------------

def _ascli_path() -> Path:
    here = Path(__file__).resolve().parent
    cands = [
        here / "ascli.py",
        here / "scripts" / "ascli" / "ascli.py",
        Path(get_hermes_home()) / "scripts" / "ascli" / "ascli.py",
    ]
    for c in cands:
        if c.exists():
            return c
    return cands[-1]


def _has_ascli() -> bool:
    cli = _ascli_path()
    if not cli.exists():
        return False
    # real gate: SDK with platform-tools must be discoverable
    for sdk in (os.environ.get("ANDROID_HOME"), os.environ.get("ANDROID_SDK_ROOT"),
                str(Path.home() / "android-sdk"), str(Path.home() / "Android" / "Sdk")):
        if sdk and os.path.isdir(os.path.join(sdk, "platform-tools")):
            return True
    return False


_ASCLI_BIN = str(_ascli_path())
_INLINE_CAP = 9000

_TIMEOUTS = {
    "sdk_install": 1200,
    "build_apk": 900, "build_test": 900, "build_tasks": 600, "build_clean": 600,
    "dev_wait": 300, "avd_start": 300, "dev_rec": 240,
}


def _run_ascli(argv: list[str], timeout: int) -> dict:
    try:
        r = subprocess.run([sys.executable, _ASCLI_BIN] + argv,
                           capture_output=True, text=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": f"ascli timeout after {timeout}s"}
    except Exception as e:  # noqa: BLE001
        return {"ok": False, "error": str(e)}
    out = (r.stdout or "").strip()
    if out.startswith("{"):
        try:
            return json.loads(out)
        except json.JSONDecodeError:
            pass
    err = (r.stderr or "").strip()
    return {"ok": False, "error": (err or out or "ascli failed")[:2000]}


# ---------------------------------------------------------------------------
# Action -> CLI argv mapping
# ---------------------------------------------------------------------------

def _argv(action: str, args: dict) -> list[str] | None:
    a = action.split("_", 1)
    grp, sub = a[0], a[1] if len(a) > 1 else ""
    ser = ["-s", str(args["serial"])] if args.get("serial") else []

    if action == "env":
        return ["env"]
    if action == "sdk_list":
        return ["sdk", "list"]
    if action == "sdk_install":
        pkgs = args.get("packages")
        if not pkgs:
            return None
        return ["sdk", "install"] + [str(p) for p in pkgs]
    if action == "sdk_licenses":
        return ["sdk", "licenses"]

    if grp == "avd":
        if sub == "list":
            return ["avd", "list"]
        if sub == "create":
            if not args.get("name") or not args.get("image"):
                return None
            v = ["avd", "create", "--name", str(args["name"]), "--image", str(args["image"])]
            if args.get("device"):
                v += ["--device", str(args["device"])]
            if args.get("sdcard"):
                v += ["--sdcard", str(args["sdcard"])]
            return v
        if sub == "delete":
            return ["avd", "delete", str(args.get("name", ""))]
        if sub == "start":
            v = ["avd", "start", str(args.get("name", ""))]
            if args.get("headless"):
                v.append("--headless")
            if args.get("no_snapshot"):
                v.append("--no-snapshot")
            if args.get("wait"):
                v.append("--wait")
            if args.get("force"):
                v.append("--force")
            return v
        if sub in ("stop", "status"):
            return ser + ["avd", sub]
        return None

    if grp == "dev":
        if sub == "list":
            return ["dev", "list"]
        if sub == "wait":
            return ["dev", "wait"]
        if sub == "info":
            return ser + ["dev", "info"]
        if sub == "install":
            if not args.get("apk"):
                return None
            v = ser + ["dev", "install", str(args["apk"])]
            if args.get("replace"):
                v.append("--replace")
            return v
        if sub == "uninstall":
            if not args.get("pkg"):
                return None
            return ser + ["dev", "uninstall", str(args["pkg"])]
        if sub == "launch":
            if not args.get("pkg"):
                return None
            v = ser + ["dev", "launch", str(args["pkg"])]
            if args.get("act"):
                v += ["--act", str(args["act"])]
            return v
        if sub == "stop":
            if not args.get("pkg"):
                return None
            return ser + ["dev", "stop", str(args["pkg"])]
        if sub == "shell":
            cmd = args.get("cmd")
            if not cmd:
                return None
            return ser + ["dev", "shell"] + [str(c) for c in cmd]
        if sub == "tap":
            return ser + ["dev", "tap", str(args.get("x", 0)), str(args.get("y", 0))]
        if sub == "swipe":
            v = ser + ["dev", "swipe", str(args.get("x1", 0)), str(args.get("y1", 0)),
                       str(args.get("x2", 0)), str(args.get("y2", 0))]
            if args.get("ms"):
                v += ["--ms", str(args["ms"])]
            return v
        if sub == "text":
            return ser + ["dev", "text", str(args.get("text", ""))]
        if sub == "key":
            return ser + ["dev", "key", str(args.get("keycode", 0))]
        if sub == "scr":
            return ser + ["dev", "scr"]
        if sub == "rec":
            return ser + ["dev", "rec", "--secs", str(args.get("secs", 5))]
        if sub == "ui":
            return ser + ["dev", "ui"]
        if sub == "log":
            v = ser + ["dev", "log"]
            if args.get("tag"):
                v += ["--tag", str(args["tag"])]
            if args.get("level"):
                v += ["--level", str(args["level"])]
            if args.get("lines"):
                v += ["--lines", str(args["lines"])]
            return v
        if sub == "top":
            return ser + ["dev", "top"]
        if sub == "pull":
            if not args.get("remote") or not args.get("local"):
                return None
            return ser + ["dev", "pull", str(args["remote"]), str(args["local"])]
        if sub == "push":
            if not args.get("local") or not args.get("remote"):
                return None
            return ser + ["dev", "push", str(args["local"]), str(args["remote"])]
        if sub == "battery":
            return ser + ["dev", "battery"]
        return None

    if grp == "build":
        v = ["build", sub]
        if args.get("dir"):
            v += ["--dir", str(args["dir"])]
        if sub == "apk" and args.get("variant"):
            v += ["--variant", str(args["variant"])]
        return v

    if grp == "apk":
        if not args.get("apk"):
            return None
        return ["apk", sub, str(args["apk"])]

    if action == "proj_info":
        v = ["proj", "info"]
        if args.get("dir"):
            v += ["--dir", str(args["dir"])]
        return v

    if grp == "studio":
        if sub == "open":
            v = ["studio", "open"]
            if args.get("dir"):
                v += ["--dir", str(args["dir"])]
            return v
        return ["studio", "status"]

    return None


# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------

ACTIONS = [
    "env", "sdk_list", "sdk_install", "sdk_licenses",
    "avd_list", "avd_create", "avd_delete", "avd_start", "avd_stop", "avd_status",
    "dev_list", "dev_info", "dev_install", "dev_uninstall", "dev_launch", "dev_stop",
    "dev_shell", "dev_tap", "dev_swipe", "dev_text", "dev_key", "dev_scr", "dev_rec",
    "dev_ui", "dev_log", "dev_top", "dev_wait", "dev_pull", "dev_push", "dev_battery",
    "build_apk", "build_test", "build_tasks", "build_clean", "build_info",
    "apk_info", "apk_perms", "apk_manifest", "apk_sign",
    "proj_info", "studio_open", "studio_status",
]

ASCLI_SCHEMA = {
    "name": "ascli",
    "description": ("Android Studio toolchain: SDK pkgs, AVDs/emulator, ADB devices, "
                    "Gradle builds, APK inspect, project info, Studio IDE. env=one-shot report; "
                    "sdk_*; avd_*; dev_* (list/info/install/uninstall/launch/stop/shell/tap/swipe/"
                    "text/key/scr/rec/ui/log/top/wait/pull/push/battery); build_*; apk_*; proj_info; "
                    "studio_open/status. JSON out; large payloads saved to file (f key)."),
    "parameters": {
        "type": "object",
        "properties": {
            "action": {"type": "string", "enum": ACTIONS,
                        "description": "Operation to run (see tool description)."},
            "serial": {"type": "string", "description": "Device serial; auto-picks when exactly one device."},
            "apk": {"type": "string", "description": "APK file path (install/inspect)."},
            "pkg": {"type": "string", "description": "Android package name (uninstall/launch/stop)."},
            "act": {"type": "string", "description": "Explicit component pkg/.Activity (dev_launch)."},
            "cmd": {"type": "array", "items": {"type": "string"},
                    "description": "Shell command tokens (dev_shell)."},
            "x": {"type": "integer"}, "y": {"type": "integer"},
            "x1": {"type": "integer"}, "y1": {"type": "integer"},
            "x2": {"type": "integer"}, "y2": {"type": "integer"},
            "ms": {"type": "integer", "description": "Swipe duration ms."},
            "text": {"type": "string", "description": "Text to type (dev_text)."},
            "keycode": {"type": "integer", "description": "Android keycode: 3=home 4=back 66=enter."},
            "secs": {"type": "integer", "description": "Screenrecord seconds (dev_rec)."},
            "tag": {"type": "string", "description": "Logcat tag filter."},
            "level": {"type": "string", "enum": ["v", "d", "i", "w", "e"],
                      "description": "Logcat min level."},
            "lines": {"type": "integer", "description": "Logcat tail lines."},
            "remote": {"type": "string"}, "local": {"type": "string"},
            "name": {"type": "string", "description": "AVD name."},
            "image": {"type": "string", "description": "System image pkg, e.g. system-images;android-34;google_apis;x86_64."},
            "device": {"type": "string", "description": "AVD device profile id."},
            "sdcard": {"type": "string", "description": "AVD sdcard size e.g. 512M."},
            "headless": {"type": "boolean", "description": "avd_start: -no-window + swiftshader."},
            "no_snapshot": {"type": "boolean"},
            "wait": {"type": "boolean", "description": "avd_start: block until boot_completed."},
            "force": {"type": "boolean", "description": "avd_start: attempt boot without /dev/kvm (TCG, very slow)."},
            "packages": {"type": "array", "items": {"type": "string"},
                         "description": "SDK packages to install (sdk_install)."},
            "dir": {"type": "string", "description": "Project dir (build/proj/studio_open)."},
            "variant": {"type": "string", "enum": ["debug", "release"],
                        "description": "Build variant."},
            "replace": {"type": "boolean", "description": "dev_install: -r replace."},
        },
        "required": ["action"],
    },
}


# ---------------------------------------------------------------------------
# Handler
# ---------------------------------------------------------------------------

def ascli_handler(args: dict, **kw: Any) -> str:
    if not _has_ascli():
        return json.dumps({"ok": False, "error": "ascli CLI/SDK not installed"}, ensure_ascii=False)
    action = str(args.get("action") or "")
    if action not in ACTIONS:
        return json.dumps({"ok": False, "error": f"unknown action '{action}'"}, ensure_ascii=False)
    argv = _argv(action, args)
    if argv is None:
        return json.dumps({"ok": False, "error": f"{action} is missing required params"},
                          ensure_ascii=False)
    r = _run_ascli(argv, _TIMEOUTS.get(action, 180))
    if r.get("ok") is False and "e" in r and "error" not in r:
        r["error"] = r.pop("e")
    return json.dumps(r, ensure_ascii=False)[:_INLINE_CAP]


# ---------------------------------------------------------------------------
# Registration
# ---------------------------------------------------------------------------

from tools.registry import registry  # noqa: E402

registry.register(
    name="ascli",
    toolset="ascli",
    schema=ASCLI_SCHEMA,
    handler=lambda args, **kw: ascli_handler(args, **kw),
    check_fn=_has_ascli,
    emoji="🤖",
    max_result_size_chars=9000,
)