feat(cmdline): add command-line interface for mainline control
Add cmdline.py utility for interacting with mainline: - Interactive TUI mode (default when no args) - Local command mode (--local) for testing without ntfy - Send commands via ntfy (default) - Watch mode (-w) for continuous stats polling Usage: python cmdline.py # Interactive mode python cmdline.py --local /effects list python cmdline.py /effects stats # Send via ntfy python cmdline.py -w /effects stats # Watch mode
This commit is contained in:
214
cmdline.py
Normal file
214
cmdline.py
Normal file
@@ -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 <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
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user