363 lines
13 KiB
Python
363 lines
13 KiB
Python
"""Hermes-Relay phone status CLI — pretty-prints ``GET /bridge/status``.
|
|
|
|
The symmetric counterpart to ``plugin.pair``: that tool writes phone
|
|
configuration, this tool reads phone state. Consumes the relay's
|
|
``/bridge/status`` loopback endpoint and renders it as a human-friendly
|
|
text block for operators, or raw JSON when piped / scripted.
|
|
|
|
Exit codes:
|
|
* ``0`` — relay reachable, phone connected.
|
|
* ``1`` — relay unreachable (not running, wrong port, loopback blocked).
|
|
* ``2`` — relay reachable but no phone has ever connected (the 503 case).
|
|
|
|
Usage::
|
|
|
|
python -m plugin.status # pretty ANSI output
|
|
python -m plugin.status --json # raw JSON pass-through
|
|
python -m plugin.status --port 9000 # override RELAY_PORT
|
|
hermes-status # shell shim (installs to ~/.local/bin)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Optional
|
|
|
|
|
|
# ── Exit codes ──────────────────────────────────────────────────────────────
|
|
|
|
EXIT_OK = 0
|
|
EXIT_RELAY_UNREACHABLE = 1
|
|
EXIT_NO_PHONE = 2
|
|
|
|
|
|
# ── Low-level fetch ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def fetch_status(port: int, timeout_s: float = 5.0) -> tuple[int, Optional[dict], Optional[str]]:
|
|
"""Fetch ``/bridge/status`` from the loopback relay.
|
|
|
|
Returns a ``(exit_code, data, error_message)`` triple:
|
|
|
|
* ``(EXIT_OK, data, None)`` — connected, normal success.
|
|
* ``(EXIT_NO_PHONE, data, error)`` — relay alive, returned 503,
|
|
body carries the phone-disconnected envelope.
|
|
* ``(EXIT_RELAY_UNREACHABLE, None, msg)`` — relay can't be reached
|
|
on loopback at all.
|
|
"""
|
|
url = f"http://127.0.0.1:{port}/bridge/status"
|
|
req = urllib.request.Request(url, method="GET")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout_s) as resp:
|
|
raw = resp.read().decode("utf-8")
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 503:
|
|
try:
|
|
body = exc.read().decode("utf-8")
|
|
data = json.loads(body) if body else {}
|
|
except (OSError, ValueError, json.JSONDecodeError):
|
|
data = {}
|
|
if not isinstance(data, dict):
|
|
data = {}
|
|
data.setdefault("phone_connected", False)
|
|
msg = data.get("error") or "no phone connected"
|
|
return EXIT_NO_PHONE, data, str(msg)
|
|
return (
|
|
EXIT_RELAY_UNREACHABLE,
|
|
None,
|
|
f"relay HTTP {exc.code}: {exc.reason}",
|
|
)
|
|
except (urllib.error.URLError, OSError) as exc:
|
|
return EXIT_RELAY_UNREACHABLE, None, f"relay unreachable: {exc}"
|
|
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
return (
|
|
EXIT_RELAY_UNREACHABLE,
|
|
None,
|
|
f"relay returned non-JSON body: {exc}",
|
|
)
|
|
if not isinstance(data, dict):
|
|
return (
|
|
EXIT_RELAY_UNREACHABLE,
|
|
None,
|
|
"relay returned non-object JSON",
|
|
)
|
|
return EXIT_OK, data, None
|
|
|
|
|
|
# ── Rendering ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
class _Palette:
|
|
"""ANSI color codes. All fields are empty strings when disabled."""
|
|
|
|
def __init__(self, enabled: bool) -> None:
|
|
if enabled:
|
|
self.reset = "\033[0m"
|
|
self.bold = "\033[1m"
|
|
self.dim = "\033[2m"
|
|
self.green = "\033[32m"
|
|
self.red = "\033[31m"
|
|
self.yellow = "\033[33m"
|
|
self.cyan = "\033[36m"
|
|
else:
|
|
self.reset = self.bold = self.dim = ""
|
|
self.green = self.red = self.yellow = self.cyan = ""
|
|
|
|
def ok(self, s: str) -> str:
|
|
return f"{self.green}{s}{self.reset}"
|
|
|
|
def bad(self, s: str) -> str:
|
|
return f"{self.red}{s}{self.reset}"
|
|
|
|
def warn(self, s: str) -> str:
|
|
return f"{self.yellow}{s}{self.reset}"
|
|
|
|
def label(self, s: str) -> str:
|
|
return f"{self.bold}{s}{self.reset}"
|
|
|
|
|
|
def _yn(value: object, palette: _Palette) -> str:
|
|
"""Render a boolean as a colored yes/no."""
|
|
if value is True:
|
|
return palette.ok("yes")
|
|
if value is False:
|
|
return palette.bad("no")
|
|
return palette.warn("unknown")
|
|
|
|
|
|
def _granted(value: object, palette: _Palette) -> str:
|
|
if value is True:
|
|
return palette.ok("granted")
|
|
if value is False:
|
|
return palette.bad("not granted")
|
|
return palette.warn("unknown")
|
|
|
|
|
|
def _enabled(value: object, palette: _Palette) -> str:
|
|
if value is True:
|
|
return palette.ok("enabled")
|
|
if value is False:
|
|
return palette.bad("disabled")
|
|
return palette.warn("unknown")
|
|
|
|
|
|
def _on_off(value: object, palette: _Palette) -> str:
|
|
if value is True:
|
|
return palette.ok("on")
|
|
if value is False:
|
|
return palette.dim + "off" + palette.reset
|
|
return palette.warn("unknown")
|
|
|
|
|
|
def render_status_block(data: dict, palette: _Palette) -> str:
|
|
"""Render the full /bridge/status envelope as a pretty text block."""
|
|
lines: list[str] = []
|
|
lines.append("")
|
|
lines.append(" " + palette.label("Hermes-Relay phone status"))
|
|
lines.append(" " + "─" * 25)
|
|
lines.append("")
|
|
|
|
connected = data.get("phone_connected")
|
|
last_seen = data.get("last_seen_seconds_ago")
|
|
if connected is True:
|
|
if isinstance(last_seen, (int, float)):
|
|
lines.append(
|
|
f" Connected: {_yn(True, palette)} (last seen {int(last_seen)}s ago)"
|
|
)
|
|
else:
|
|
lines.append(f" Connected: {_yn(True, palette)}")
|
|
elif connected is False:
|
|
err = data.get("error") or "no phone connected"
|
|
lines.append(f" Connected: {_yn(False, palette)} ({err})")
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
else:
|
|
lines.append(f" Connected: {_yn(None, palette)}")
|
|
|
|
# Device
|
|
device = data.get("device") or {}
|
|
if isinstance(device, dict) and device:
|
|
lines.append("")
|
|
lines.append(" " + palette.label("Device"))
|
|
name = device.get("name")
|
|
battery = device.get("battery_percent")
|
|
screen_on = device.get("screen_on")
|
|
current_app = device.get("current_app")
|
|
if name:
|
|
lines.append(f" Name: {name}")
|
|
if battery is not None:
|
|
lines.append(f" Battery: {battery}%")
|
|
lines.append(f" Screen: {_on_off(screen_on, palette)}")
|
|
if current_app:
|
|
lines.append(f" App: {current_app}")
|
|
|
|
# Bridge permissions
|
|
bridge = data.get("bridge") or {}
|
|
if isinstance(bridge, dict) and bridge:
|
|
lines.append("")
|
|
lines.append(" " + palette.label("Bridge"))
|
|
device_control = bridge.get("device_control_supported")
|
|
if device_control is False:
|
|
lines.append(
|
|
" Device control: "
|
|
+ palette.dim
|
|
+ "unavailable (Google Play Bridge Core)"
|
|
+ palette.reset
|
|
)
|
|
else:
|
|
if device_control is True:
|
|
lines.append(
|
|
f" Device control: {_enabled(True, palette)}"
|
|
)
|
|
lines.append(
|
|
f" Master: {_enabled(bridge.get('master_enabled'), palette)}"
|
|
)
|
|
lines.append(
|
|
f" Accessibility: {_granted(bridge.get('accessibility_granted'), palette)}"
|
|
)
|
|
lines.append(
|
|
f" Screen capture: {_granted(bridge.get('screen_capture_granted'), palette)}"
|
|
)
|
|
lines.append(
|
|
f" Overlay: {_granted(bridge.get('overlay_granted'), palette)}"
|
|
)
|
|
lines.append(
|
|
f" Notifications: {_granted(bridge.get('notification_listener_granted'), palette)}"
|
|
)
|
|
|
|
capabilities = data.get("capabilities") or {}
|
|
if isinstance(capabilities, dict) and capabilities:
|
|
permanent = capabilities.get("permanent") or []
|
|
timed = capabilities.get("timed") or {}
|
|
unlimited = capabilities.get("unlimited") or []
|
|
lines.append("")
|
|
lines.append(" " + palette.label("Capability grants"))
|
|
lines.append(
|
|
" Always: "
|
|
+ (", ".join(str(item) for item in permanent) if permanent else "none")
|
|
)
|
|
lines.append(
|
|
" Timed: "
|
|
+ (", ".join(sorted(str(item) for item in timed)) if isinstance(timed, dict) and timed else "none")
|
|
)
|
|
lines.append(
|
|
" Until off: "
|
|
+ (", ".join(str(item) for item in unlimited) if unlimited else "none")
|
|
)
|
|
|
|
# Safety
|
|
safety = data.get("safety") or {}
|
|
if isinstance(safety, dict) and safety:
|
|
lines.append("")
|
|
lines.append(" " + palette.label("Safety"))
|
|
blk = safety.get("blocklist_count")
|
|
verbs = safety.get("destructive_verbs_count")
|
|
auto = safety.get("auto_disable_minutes")
|
|
at_ms = safety.get("auto_disable_at_ms")
|
|
if blk is not None:
|
|
lines.append(f" Blocklist: {blk} packages")
|
|
if verbs is not None:
|
|
lines.append(f" Destructive verbs: {verbs}")
|
|
if auto is not None:
|
|
if at_ms:
|
|
lines.append(
|
|
f" Timed screen: {auto} min idle (armed)"
|
|
)
|
|
else:
|
|
lines.append(f" Timed screen: {auto} min idle")
|
|
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ── CLI entry point ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def status_command(args: argparse.Namespace) -> int:
|
|
"""Core CLI handler. Returns the process exit code."""
|
|
port = int(args.port) if args.port else int(os.getenv("RELAY_PORT", "8767"))
|
|
code, data, err = fetch_status(port)
|
|
|
|
if args.json:
|
|
# Raw JSON pass-through, scripting-friendly. When the relay
|
|
# isn't reachable we still emit a stable envelope so callers
|
|
# can jq over it.
|
|
if code == EXIT_OK and data is not None:
|
|
sys.stdout.write(json.dumps(data, separators=(",", ":")) + "\n")
|
|
elif code == EXIT_NO_PHONE and data is not None:
|
|
sys.stdout.write(json.dumps(data, separators=(",", ":")) + "\n")
|
|
else:
|
|
sys.stdout.write(
|
|
json.dumps(
|
|
{
|
|
"phone_connected": False,
|
|
"error": err or "relay unreachable",
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
+ "\n"
|
|
)
|
|
return code
|
|
|
|
palette = _Palette(sys.stdout.isatty())
|
|
|
|
if code == EXIT_RELAY_UNREACHABLE:
|
|
msg = err or "relay unreachable"
|
|
sys.stderr.write(
|
|
f"\n {palette.bad('[error]')} Cannot reach hermes-relay on "
|
|
f"127.0.0.1:{port}\n"
|
|
)
|
|
sys.stderr.write(f" {palette.dim}{msg}{palette.reset}\n")
|
|
sys.stderr.write(
|
|
f" {palette.dim}Is the relay running? "
|
|
f"Try: systemctl --user status hermes-relay{palette.reset}\n\n"
|
|
)
|
|
return code
|
|
|
|
if data is None:
|
|
# Shouldn't happen — fetch_status always returns data for
|
|
# non-unreachable codes — but be defensive.
|
|
sys.stderr.write(
|
|
f"\n {palette.bad('[error]')} Relay returned no data\n\n"
|
|
)
|
|
return EXIT_RELAY_UNREACHABLE
|
|
|
|
sys.stdout.write(render_status_block(data, palette))
|
|
return code
|
|
|
|
|
|
def main(argv: Optional[list[str]] = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
prog="hermes-status",
|
|
description=(
|
|
"Print the current state of the paired Hermes-Relay Android "
|
|
"phone — connection, device info, bridge permissions, safety "
|
|
"rails. Reads from the relay's /bridge/status loopback endpoint."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Emit raw JSON instead of the pretty text block",
|
|
)
|
|
parser.add_argument(
|
|
"--port",
|
|
type=int,
|
|
default=None,
|
|
help="Override the relay port (default: $RELAY_PORT or 8767)",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
return status_command(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|