#!/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 {"@": , "#": , ...}; 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())