From c08a7d3cb0917249644d010a2ac5ce02d5c55c79 Mon Sep 17 00:00:00 2001 From: David Gwilliam Date: Sun, 15 Mar 2026 18:42:54 -0700 Subject: [PATCH] feat(cmdline): add command-line interface for mainline control --- cmdline.py | 250 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 250 insertions(+) create mode 100644 cmdline.py diff --git a/cmdline.py b/cmdline.py new file mode 100644 index 0000000..9ee9ba6 --- /dev/null +++ b/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 on - Enable an effect + /effects off - Disable an effect + /effects intensity <0.0-1.0> - Set effect intensity + /effects reorder ,,... - 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()