clientCommands/tools/devclient.py

#!/usr/bin/env python3
"""Client for the exteraGram on-device plugin dev server.

Speaks the wire protocol at 127.0.0.1:42690 (after `adb forward tcp:42690 tcp:42690`):
consecutive UTF-8 JSON objects, no length framing. Requests are
{"@": <command>, "#": <id>, ...}; responses echo "#".
"""
from __future__ import annotations

import argparse
import json
import socket
import sys
import time
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
PLUGIN_JSON = REPO_ROOT / "plugin.json"

CONNECT_ATTEMPTS = 5
CONNECT_RETRY_DELAY = 1.0
SOCKET_TIMEOUT = 1.0
RESPONSE_TIMEOUT = 30.0

UNREACHABLE_MESSAGE = (
    "dev server unreachable — run scripts/adb-connect.sh, and enable exteraGram "
    "Preferences > Plugins > plugin engine + developer mode"
)


def default_plugin_id() -> str | None:
    try:
        data = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return data.get("id")


class DevClient:
    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        self._sock: socket.socket | None = None
        self._buf = b""
        self._decoder = json.JSONDecoder()
        self._next_id = 1

    def connect(self) -> None:
        last_exc = None
        for attempt in range(CONNECT_ATTEMPTS):
            try:
                sock = socket.create_connection((self.host, self.port), timeout=SOCKET_TIMEOUT)
                sock.settimeout(SOCKET_TIMEOUT)
                self._sock = sock
                return
            except OSError as exc:
                last_exc = exc
                if attempt < CONNECT_ATTEMPTS - 1:
                    time.sleep(CONNECT_RETRY_DELAY)
        print(UNREACHABLE_MESSAGE, file=sys.stderr)
        raise SystemExit(1) from last_exc

    def close(self) -> None:
        if self._sock is not None:
            self._sock.close()
            self._sock = None

    def _read_one(self, deadline: float) -> dict:
        while True:
            self._buf = self._buf.lstrip()
            if self._buf:
                try:
                    text = self._buf.decode("utf-8")
                except UnicodeDecodeError:
                    text = None
                if text:
                    try:
                        obj, idx = self._decoder.raw_decode(text)
                        consumed = len(text[:idx].encode("utf-8"))
                        self._buf = self._buf[consumed:]
                        return obj
                    except json.JSONDecodeError:
                        pass
            if time.monotonic() >= deadline:
                print("error: timed out waiting for dev server response", file=sys.stderr)
                raise SystemExit(1)
            try:
                chunk = self._sock.recv(4096)
            except socket.timeout:
                continue
            except OSError as exc:
                print(f"error: connection lost: {exc}", file=sys.stderr)
                raise SystemExit(1)
            if not chunk:
                print("error: dev server closed the connection", file=sys.stderr)
                raise SystemExit(1)
            self._buf += chunk

    def request(self, command: str, **fields) -> dict:
        req_id = self._next_id
        self._next_id += 1
        payload = {"@": command, "#": req_id, **fields}
        self._sock.sendall(json.dumps(payload).encode("utf-8"))
        deadline = time.monotonic() + RESPONSE_TIMEOUT
        while True:
            resp = self._read_one(deadline)
            if resp.get("#") == req_id:
                if resp.get("error") or resp.get("errors"):
                    print(json.dumps(resp, indent=2), file=sys.stderr)
                    raise SystemExit(1)
                return resp
            # response to a different (stale) request id: keep draining


def cmd_ping(client: DevClient, args) -> int:
    resp = client.request("ping")
    print(json.dumps(resp, indent=2))
    return 0


def cmd_plugins(client: DevClient, args) -> int:
    resp = client.request("get_plugins")
    print(json.dumps(resp, indent=2))
    return 0


def cmd_deploy(client: DevClient, args) -> int:
    plugin_id = args.id or default_plugin_id()
    if not plugin_id:
        print("error: could not determine plugin id (pass --id or fix plugin.json)", file=sys.stderr)
        return 1
    try:
        content = Path(args.file).read_text(encoding="utf-8")
    except OSError as exc:
        print(f"error: cannot read {args.file}: {exc}", file=sys.stderr)
        return 1
    client.request("write_plugin", plugin_id=plugin_id, content=content)
    time.sleep(0.3)
    if args.enable:
        client.request("enable_plugin", plugin_id=plugin_id)
    if not args.no_reload:
        client.request("reload_plugin", plugin_id=plugin_id)
    print(f"deployed {args.file} as {plugin_id}")
    return 0


def cmd_simple_id(command: str):
    def handler(client: DevClient, args) -> int:
        client.request(command, plugin_id=args.id)
        print(f"{command} {args.id}: ok")
        return 0

    return handler


def cmd_debugger(client: DevClient, args) -> int:
    if args.action == "start":
        client.request(
            "start_debugger",
            host=args.debug_host,
            port=args.debug_port,
            platform=args.platform,
        )
        print(f"debugger start requested ({args.platform} -> {args.debug_host}:{args.debug_port})")
    else:
        client.request("stop_debugger", platform=args.platform)
        print(f"debugger stop requested ({args.platform})")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="exteraGram plugin dev server client")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=42690)
    sub = parser.add_subparsers(dest="command", required=True)

    sub.add_parser("ping").set_defaults(func=cmd_ping)
    sub.add_parser("plugins").set_defaults(func=cmd_plugins)

    p_deploy = sub.add_parser("deploy")
    p_deploy.add_argument("file")
    p_deploy.add_argument("--id")
    p_deploy.add_argument("--enable", action="store_true")
    p_deploy.add_argument("--no-reload", action="store_true")
    p_deploy.set_defaults(func=cmd_deploy)

    for name, command in (
        ("enable", "enable_plugin"),
        ("disable", "disable_plugin"),
        ("reload", "reload_plugin"),
        ("remove", "remove_plugin"),
    ):
        p = sub.add_parser(name)
        p.add_argument("id")
        p.set_defaults(func=cmd_simple_id(command))

    p_debugger = sub.add_parser("debugger")
    p_debugger.add_argument("action", choices=["start", "stop"])
    p_debugger.add_argument("--platform", choices=["vscode", "pycharm"], default="vscode")
    p_debugger.add_argument("--debug-host", default="127.0.0.1")
    p_debugger.add_argument("--debug-port", type=int, default=5678)
    p_debugger.set_defaults(func=cmd_debugger)

    return parser


def main(argv=None) -> int:
    parser = build_parser()
    args = parser.parse_args(sys.argv[1:] if argv is None else argv)
    client = DevClient(args.host, args.port)
    client.connect()
    try:
        return args.func(client, args)
    finally:
        client.close()


if __name__ == "__main__":
    sys.exit(main())