feat(cmdline): add command-line interface for mainline control

cmdline.py

@@ -0,0 +1,250 @@
+#!/usr/bin/env python3
+"""
+Command-line utility for interacting with mainline via ntfy.
+
+Usage:
+    python cmdline.py                     # Interactive TUI mode
+    python cmdline.py --help              # Show help
+    python cmdline.py /effects list       # Send single command via ntfy
+    python cmdline.py /effects stats      # Get performance stats via ntfy
+    python cmdline.py -w /effects stats   # Watch mode (polls for stats)
+
+The TUI mode provides:
+    - Arrow keys to navigate command history
+    - Tab completion for commands
+    - Auto-refresh for performance stats
+
+C&C works like a serial port:
+    1. Send command to ntfy_cc_topic
+    2. Mainline receives, processes, responds to same topic
+    3. Cmdline polls for response
+"""
+
+import argparse
+import json
+import sys
+import time
+import threading
+import urllib.request
+from pathlib import Path
+
+from engine import config
+from engine.terminal import CLR, CURSOR_OFF, CURSOR_ON, G_DIM, G_HI, RST, W_GHOST
+
+try:
+    CC_CMD_TOPIC = config.NTFY_CC_CMD_TOPIC
+    CC_RESP_TOPIC = config.NTFY_CC_RESP_TOPIC
+except AttributeError:
+    CC_CMD_TOPIC = "https://ntfy.sh/klubhaus_terminal_mainline_cc_cmd/json"
+    CC_RESP_TOPIC = "https://ntfy.sh/klubhaus_terminal_mainline_cc_resp/json"
+
+
+class NtfyResponsePoller:
+    """Polls ntfy for command responses."""
+
+    def __init__(self, cmd_topic: str, resp_topic: str, timeout: float = 10.0):
+        self.cmd_topic = cmd_topic
+        self.resp_topic = resp_topic
+        self.timeout = timeout
+        self._last_id = None
+        self._lock = threading.Lock()
+
+    def _build_url(self) -> str:
+        from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
+
+        parsed = urlparse(self.resp_topic)
+        params = parse_qs(parsed.query, keep_blank_values=True)
+        params["since"] = [self._last_id if self._last_id else "20s"]
+        new_query = urlencode({k: v[0] for k, v in params.items()})
+        return urlunparse(parsed._replace(query=new_query))
+
+    def send_and_wait(self, cmd: str) -> str:
+        """Send command and wait for response."""
+        url = self.cmd_topic.replace("/json", "")
+        data = cmd.encode("utf-8")
+
+        req = urllib.request.Request(
+            url,
+            data=data,
+            headers={
+                "User-Agent": "mainline-cmdline/0.1",
+                "Content-Type": "text/plain",
+            },
+            method="POST",
+        )
+
+        try:
+            urllib.request.urlopen(req, timeout=5)
+        except Exception as e:
+            return f"Error sending command: {e}"
+
+        return self._wait_for_response(cmd)
+
+    def _wait_for_response(self, expected_cmd: str = "") -> str:
+        """Poll for response message."""
+        start = time.time()
+        while time.time() - start < self.timeout:
+            try:
+                url = self._build_url()
+                req = urllib.request.Request(
+                    url, headers={"User-Agent": "mainline-cmdline/0.1"}
+                )
+                with urllib.request.urlopen(req, timeout=10) as resp:
+                    for line in resp:
+                        try:
+                            data = json.loads(line.decode("utf-8", errors="replace"))
+                        except json.JSONDecodeError:
+                            continue
+                        if data.get("event") == "message":
+                            self._last_id = data.get("id")
+                            msg = data.get("message", "")
+                            if msg:
+                                return msg
+            except Exception:
+                pass
+            time.sleep(0.5)
+        return "Timeout waiting for response"
+
+
+AVAILABLE_COMMANDS = """Available commands:
+  /effects list          - List all effects and status
+  /effects <name> on    - Enable an effect
+  /effects <name> off   - Disable an effect
+  /effects <name> intensity <0.0-1.0> - Set effect intensity
+  /effects reorder <name1>,<name2>,... - Reorder pipeline
+  /effects stats        - Show performance statistics
+  /help                 - Show this help
+  /quit                 - Exit
+"""
+
+
+def print_header():
+    w = 60
+    print(CLR, end="")
+    print(CURSOR_OFF, end="")
+    print(f"\033[1;1H", end="")
+    print(f"  \033[1;38;5;231m╔{'═' * (w - 6)}╗\033[0m")
+    print(
+        f"  \033[1;38;5;231m║\033[0m \033[1;38;5;82mMAINLINE\033[0m \033[3;38;5;245mCommand Center\033[0m \033[1;38;5;231m  ║\033[0m"
+    )
+    print(f"  \033[1;38;5;231m╚{'═' * (w - 6)}╝\033[0m")
+    print(f"  \033[2;38;5;37mCMD: {CC_CMD_TOPIC.split('/')[-2]}\033[0m")
+    print(f"  \033[2;38;5;37mRESP: {CC_RESP_TOPIC.split('/')[-2]}\033[0m")
+    print()
+
+
+def print_response(response: str, is_error: bool = False) -> None:
+    """Print response with nice formatting."""
+    print()
+    if is_error:
+        print(f"  \033[1;38;5;196m✗ Error\033[0m")
+        print(f"  \033[38;5;196m{'─' * 40}\033[0m")
+    else:
+        print(f"  \033[1;38;5;82m✓ Response\033[0m")
+        print(f"  \033[38;5;37m{'─' * 40}\033[0m")
+
+    for line in response.split("\n"):
+        print(f"  {line}")
+    print()
+
+
+def interactive_mode():
+    """Interactive TUI for sending commands."""
+    import readline
+
+    print_header()
+    poller = NtfyResponsePoller(CC_CMD_TOPIC, CC_RESP_TOPIC)
+
+    print(f"  \033[38;5;245mType /help for commands, /quit to exit\033[0m")
+    print()
+
+    while True:
+        try:
+            cmd = input(f"  \033[1;38;5;82m❯\033[0m {G_HI}").strip()
+        except (EOFError, KeyboardInterrupt):
+            print()
+            break
+
+        if not cmd:
+            continue
+
+        if cmd.startswith("/"):
+            if cmd == "/quit" or cmd == "/exit":
+                print(f"\n  \033[1;38;5;245mGoodbye!{RST}\n")
+                break
+
+            if cmd == "/help":
+                print(f"\n{AVAILABLE_COMMANDS}\n")
+                continue
+
+            print(f"  \033[38;5;245m⟳ Sending to mainline...{RST}")
+            result = poller.send_and_wait(cmd)
+            print_response(result, is_error=result.startswith("Error"))
+        else:
+            print(f"\n  \033[1;38;5;196m⚠ Commands must start with /{RST}\n")
+
+    print(CURSOR_ON, end="")
+    return 0
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Mainline command-line interface",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=AVAILABLE_COMMANDS,
+    )
+    parser.add_argument(
+        "command",
+        nargs="?",
+        default=None,
+        help="Command to send (e.g., /effects list)",
+    )
+    parser.add_argument(
+        "--watch",
+        "-w",
+        action="store_true",
+        help="Watch mode: continuously poll for stats (Ctrl+C to exit)",
+    )
+
+    args = parser.parse_args()
+
+    if args.command is None:
+        return interactive_mode()
+
+    poller = NtfyResponsePoller(CC_CMD_TOPIC, CC_RESP_TOPIC)
+
+    if args.watch and "/effects stats" in args.command:
+        import signal
+
+        def handle_sigterm(*_):
+            print(f"\n  \033[1;38;5;245mStopped watching{RST}")
+            print(CURSOR_ON, end="")
+            sys.exit(0)
+
+        signal.signal(signal.SIGTERM, handle_sigterm)
+
+        print_header()
+        print(f"  \033[38;5;245mWatching /effects stats (Ctrl+C to exit)...{RST}\n")
+        try:
+            while True:
+                result = poller.send_and_wait(args.command)
+                print(f"\033[2J\033[1;1H", end="")
+                print(
+                    f"  \033[1;38;5;82m❯\033[0m Performance Stats - \033[1;38;5;245m{time.strftime('%H:%M:%S')}{RST}"
+                )
+                print(f"  \033[38;5;37m{'─' * 44}{RST}")
+                for line in result.split("\n"):
+                    print(f"  {line}")
+                time.sleep(2)
+        except KeyboardInterrupt:
+            print(f"\n  \033[1;38;5;245mStopped watching{RST}")
+            return 0
+        return 0
+
+    result = poller.send_and_wait(args.command)
+    print(result)
+    return 0
+
+
+if __name__ == "__main__":
+    main()