forked from genewildish/Mainline
251 lines
7.8 KiB
Python
251 lines
7.8 KiB
Python
#!/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()
|