adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-10 19:03:22 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: add CLI command endpoint and standalone CLI client for pyMC Repeater

Browse Source
This commit is contained in:
Lloyd
2026-03-24 14:02:51 +00:00
parent d82c90a04d
commit 91918e7cfc
4 changed files with 293 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -61,6 +61,7 @@ dev = [
[project.scripts]
pymc-repeater = "repeater.main:main"
pymc-cli = "repeater.local_cli:main"
[tool.setuptools.packages.find]
where = ["."]
repeater/handler_helpers/repeater_cli.py
+105 -2
View File
@@ -88,8 +88,12 @@ class MeshCLI:
def _route_command(self, command: str) -> str:
"""Route command to appropriate handler method."""
# Help
if command == "help" or command.startswith("help "):
return self._cmd_help(command)
# System commands
if command == "reboot":
elif command == "reboot":
return self._cmd_reboot()
elif command == "advert":
return self._cmd_advert()
@@ -156,7 +160,106 @@ class MeshCLI:
else:
return "Unknown command"
# ==================== System Commands ====================
# ==================== Help Command ====================
def _cmd_help(self, command: str) -> str:
"""Show available commands or detailed help for a specific command."""
parts = command.split(None, 1)
if len(parts) == 2:
return self._help_detail(parts[1])
lines = [
"=== pyMC CLI Commands ===",
"",
"System:",
" reboot Restart the repeater service",
" advert Send self advertisement",
" clock Show current UTC time",
" clock sync Sync clock (no-op, uses system time)",
" ver Show version info",
" password <pw> Change admin password",
" clear stats Clear statistics",
"",
"Get:",
" get name Node name",
" get radio Radio params (freq,bw,sf,cr)",
" get freq Frequency (MHz)",
" get tx TX power",
" get af Airtime factor",
" get repeat Repeat mode (on/off)",
" get lat / get lon GPS coordinates",
" get role Identity role",
" get guest.password Guest password",
" get allow.read.only Read-only access setting",
" get advert.interval Advert interval (minutes)",
" get flood.advert.interval Flood advert interval (hours)",
" get flood.max Max flood hops",
" get rxdelay RX delay base",
" get txdelay TX delay factor",
" get direct.txdelay Direct TX delay factor",
" get multi.acks Multi-ack count",
" get int.thresh Interference threshold",
" get agc.reset.interval AGC reset interval",
"",
"Set: (use 'help set' for details)",
" set <param> <value>",
"",
"Other:",
" neighbors List neighbors",
" neighbor.remove <key> Remove neighbor by pubkey",
" tempradio <freq> <bw> <sf> <cr> <timeout_mins>",
" setperm <pubkey> <perm> Set ACL permissions",
" log start|stop|erase Logging control",
]
if self.enable_regions:
lines.append(" region ... Region commands")
lines += ["", "Type 'help <command>' for details on a specific command."]
return "\n".join(lines)
def _help_detail(self, topic: str) -> str:
"""Return detailed help for a specific command topic."""
topic = topic.strip()
details = {
"set": (
"Set commands — set <param> <value>:\n"
" set name <name> Set node name\n"
" set radio <f> <bw> <sf> <cr> Set radio (restart required)\n"
" set freq <mhz> Set frequency (restart required)\n"
" set tx <power> Set TX power\n"
" set af <factor> Airtime factor\n"
" set repeat on|off Enable/disable repeating\n"
" set lat <deg> Latitude\n"
" set lon <deg> Longitude\n"
" set guest.password <pw> Guest password\n"
" set allow.read.only on|off Read-only access\n"
" set advert.interval <min> 60-240 minutes\n"
" set flood.advert.interval <hr> 3-48 hours\n"
" set flood.max <hops> Max flood hops (max 64)\n"
" set rxdelay <val> RX delay base (>=0)\n"
" set txdelay <val> TX delay factor (>=0)\n"
" set direct.txdelay <val> Direct TX delay (>=0)\n"
" set multi.acks <n> Multi-ack count\n"
" set int.thresh <dbm> Interference threshold\n"
" set agc.reset.interval <n> AGC reset (rounded to x4)"
),
"get": "Get commands — type 'help' to see all 'get' parameters.",
"reboot": "Restart the repeater service via systemd.",
"advert": "Trigger a self-advertisement flood packet.",
"clock": "'clock' shows UTC time. 'clock sync' is a no-op (system time used).",
"ver": "Show repeater version and identity type.",
"password": "password <new_password> — Change the admin password.",
"tempradio": (
"tempradio <freq_mhz> <bw_khz> <sf> <cr> <timeout_mins>\n"
" Apply temporary radio parameters that revert after timeout.\n"
" freq: 300-2500 MHz, bw: 7-500 kHz, sf: 5-12, cr: 5-8"
),
"neighbors": "List known neighbor nodes from the routing table.",
"setperm": "setperm <pubkey_hex> <permission_int> — Set ACL permissions for a node.",
"log": "log start|stop|erase — Control logging.",
}
return details.get(topic, f"No detailed help for '{topic}'. Type 'help' for command list.")
# ==================== System Commands ==
def _cmd_reboot(self) -> str:
"""Reboot the repeater process."""
repeater/local_cli.py
+146
View File
@@ -0,0 +1,146 @@
"""
CLI client for pyMC Repeater.
Connects to an already-running repeater daemon via its HTTP API.
Reads admin password and HTTP port from the local config.yaml automatically.
"""
import sys
CONFIG_PATHS = [
"/etc/pymc_repeater/config.yaml",
"config.yaml",
]
def _load_config(config_path=None):
"""Load repeater config.yaml, trying common paths."""
import yaml
from pathlib import Path
paths = [config_path] if config_path else CONFIG_PATHS
for p in paths:
path = Path(p)
if path.is_file():
with open(path) as f:
return yaml.safe_load(f) or {}
return {}
def run_client_cli(host: str = "127.0.0.1", port: int = 8000, password: str = ""):
"""
Standalone CLI client that connects to a running repeater's HTTP API.
"""
import urllib.request
import urllib.error
import json
base_url = f"http://{host}:{port}"
# Authenticate to get JWT token
token = None
if password:
try:
auth_data = json.dumps({
"username": "admin",
"password": password,
"client_id": "pymc-cli",
}).encode()
req = urllib.request.Request(
f"{base_url}/auth/login",
data=auth_data,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=5) as resp:
result = json.loads(resp.read())
token = result.get("token") or result.get("data", {}).get("token")
except urllib.error.URLError as e:
print(f"Error: Cannot connect to repeater at {base_url} — {e.reason}")
sys.exit(1)
except Exception as e:
print(f"Authentication failed: {e}")
sys.exit(1)
if not token:
print("Error: Authentication failed. Check password or repeater status.")
sys.exit(1)
print(f"\npyMC Repeater CLI (connected to {base_url})")
print("Type 'help' for available commands, 'exit' to quit.\n")
while True:
try:
command = input(">> ").strip()
except (EOFError, KeyboardInterrupt):
print()
break
if not command:
continue
if command in ("exit", "quit"):
break
try:
payload = json.dumps({"command": command}).encode()
req = urllib.request.Request(
f"{base_url}/api/cli",
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
method="POST",
)
with urllib.request.urlopen(req, timeout=10) as resp:
result = json.loads(resp.read())
if result.get("success"):
print(result["data"]["reply"])
else:
print(f"Error: {result.get('error', 'Unknown error')}")
except urllib.error.URLError as e:
print(f"Connection error: {e.reason}")
except Exception as e:
print(f"Error: {e}")
def main():
"""Entry point for pymc-cli command."""
import argparse
parser = argparse.ArgumentParser(
description="Connect to a running pyMC Repeater and issue CLI commands"
)
parser.add_argument(
"--config", default=None,
help="Path to config.yaml (auto-detected if not set)",
)
parser.add_argument(
"--host", default=None,
help="Repeater HTTP host (default: 127.0.0.1)",
)
parser.add_argument(
"--port", type=int, default=None,
help="Repeater HTTP port (default: from config or 8000)",
)
args = parser.parse_args()
# Load config to get password and port automatically
config = _load_config(args.config)
repeater_cfg = config.get("repeater", {})
security_cfg = repeater_cfg.get("security", {})
password = security_cfg.get("admin_password", "")
if not password:
print("Error: No admin_password found in config.yaml.")
print("Searched: " + ", ".join(CONFIG_PATHS))
sys.exit(1)
host = args.host or "127.0.0.1"
port = args.port or config.get("http", {}).get("port", 8000)
run_client_cli(host=host, port=port, password=password)
if __name__ == "__main__":
main()
repeater/web/api_endpoints.py
+41
View File
@@ -4307,6 +4307,47 @@ class APIEndpoints:
logger.error(f"Error clearing room messages: {e}")
return self._error(e)
# ======================
# CLI Command Endpoint
# ======================
@cherrypy.expose
@cherrypy.tools.json_out()
@cherrypy.tools.json_in()
@require_auth
def cli(self):
"""Execute a CLI command on the running repeater.
POST /api/cli {"command": "get name"}
Returns {"success": true, "reply": "..."}
"""
self._set_cors_headers()
if cherrypy.request.method == "OPTIONS":
return ""
try:
self._require_post()
data = cherrypy.request.json
command = data.get("command", "").strip()
if not command:
return self._error("Missing 'command' field")
if not self.daemon_instance or not hasattr(self.daemon_instance, "text_helper"):
return self._error("Repeater not initialized")
text_helper = self.daemon_instance.text_helper
if not text_helper or not hasattr(text_helper, "cli") or not text_helper.cli:
return self._error("CLI handler not available")
reply = text_helper.cli.handle_command(
sender_pubkey=b"api-cli",
command=command,
is_admin=True,
)
return self._success({"reply": reply})
except cherrypy.HTTPError:
raise
except Exception as e:
logger.error(f"CLI endpoint error: {e}", exc_info=True)
return self._error(str(e))
# ======================
# OpenAPI Documentation
# ======================