diff --git a/cmdline.py b/cmdline.py new file mode 100644 index 0000000..1945df7 --- /dev/null +++ b/cmdline.py @@ -0,0 +1,214 @@ +#!/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 + python cmdline.py /effects stats # Get performance stats + +The TUI mode provides: + - Arrow keys to navigate command history + - Tab completion for commands + - Auto-refresh for performance stats +""" + +import argparse +import json +import sys +import time +import urllib.request +from pathlib import Path + +from engine import config +from engine.effects.controller import handle_effects_command +from engine.terminal import CLR, CURSOR_OFF, CURSOR_ON, G_DIM, G_HI, RST, W_GHOST + +TOPIC = config.NTFY_TOPIC + + +def send_command(cmd: str) -> str: + """Send a command to the ntfy topic and return the response.""" + if not cmd.startswith("/"): + return "Commands must start with /" + + url = 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: + with urllib.request.urlopen(req, timeout=10) as resp: + return f"Command sent: {cmd}\n(Response would appear on mainline display)" + except Exception as e: + return f"Error sending command: {e}" + + +def local_command(cmd: str) -> str: + """Handle command locally without sending to ntfy.""" + if cmd.startswith("/effects"): + try: + import effects_plugins + from engine.effects.registry import get_registry + from engine.effects.chain import EffectChain + + effects_plugins.discover_plugins() + registry = get_registry() + chain = EffectChain(registry) + chain.set_order(["noise", "fade", "glitch", "firehose"]) + + from engine.layers import _effect_chain + + global _effect_chain_ref + _effect_chain_ref = chain + + from engine.effects.controller import handle_effects_command + + return handle_effects_command(cmd) + except Exception as e: + return f"Error: {type(e).__name__}: {e}\n(Effects require PIL - run with --send or run mainline first)" + if cmd == "/help": + return AVAILABLE_COMMANDS + if cmd == "/quit" or cmd == "/exit": + return "GOODBYE" + return f"Unknown command: {cmd}" + if cmd == "/help": + return AVAILABLE_COMMANDS + if cmd == "/quit" or cmd == "/exit": + return "GOODBYE" + return f"Unknown command: {cmd}" + + +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 + +Local commands (don't require running mainline): + /effects * - All /effects commands work locally +""" + + +def print_header(): + w = 60 + print(CLR, end="") + print(CURSOR_OFF, end="") + print(f"\033[1;1H", end="") + print(f" \033[1;38;5;231mMAINLINE COMMAND CENTER\033[0m") + print(f" \033[2;38;5;37m{'─' * (w - 4)}\033[0m") + print(f" \033[38;5;245mTopic: {TOPIC}\033[0m") + print() + + +def interactive_mode(): + """Interactive TUI for sending commands.""" + import readline + + print_header() + + history = [] + history_index = -1 + + print(f"{G_DIM}Type /help for available commands, /quit to exit{RST}") + print() + + while True: + try: + cmd = input(f"{G_HI}> {RST}").strip() + except (EOFError, KeyboardInterrupt): + print() + break + + if not cmd: + continue + + if cmd.startswith("/"): + history.append(cmd) + history_index = len(history) + + if cmd == "/quit" or cmd == "/exit": + print(f"{G_DIM}Goodbye!{RST}") + break + + result = local_command(cmd) + print(f"\n{result}\n") + + else: + print( + f"{G_DIM}Commands must start with / - type /help for available commands{RST}\n" + ) + + print(CURSOR_ON, end="") + + +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( + "--local", + "-l", + action="store_true", + help="Run command locally (no ntfy required)", + ) + 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: + interactive_mode() + return + + if args.local: + result = local_command(args.command) + print(result) + return + + if args.watch and "/effects stats" in args.command: + print_header() + print(f"{G_DIM}Watching /effects stats (Ctrl+C to exit)...{RST}\n") + try: + while True: + result = local_command(args.command) + print(f"\033[2J\033[1;1H", end="") + print(f"{G_HI}Performance Stats - {time.strftime('%H:%M:%S')}{RST}") + print(f"{G_DIM}{'─' * 40}{RST}") + print(result) + time.sleep(2) + except KeyboardInterrupt: + print(f"\n{G_DIM}Stopped watching{RST}") + return + + result = send_command(args.command) + print(result) + + +if __name__ == "__main__": + main()