"""CLI entry point: ``relay ``.""" from __future__ import annotations import argparse import json import logging import signal import sys from relay.config import load_settings from relay.daemon import Daemon from relay.logs import configure as configure_logs from relay.ntfy import topic_url from relay.state import read_json def _cmd_run(args: argparse.Namespace) -> int: settings = load_settings() configure_logs(settings.logs_dir, level=logging.DEBUG if args.verbose else logging.INFO) daemon = Daemon(settings) def _stop(signum: int, _frame: object) -> None: logging.getLogger(__name__).info("received signal %s; shutting down", signum) daemon.stop() signal.signal(signal.SIGINT, _stop) signal.signal(signal.SIGTERM, _stop) try: daemon.run() except RuntimeError as exc: # Most commonly the lock file: another daemon is running. print(f"error: {exc}", file=sys.stderr) return 1 return 0 def _cmd_status(_args: argparse.Namespace) -> int: settings = load_settings() status_path = settings.state_dir / "status.json" payload = read_json(status_path, default=None) if payload is None: print("no status file yet — has the daemon run?", file=sys.stderr) return 1 print(json.dumps(payload, indent=2, sort_keys=False)) return 0 def _cmd_topic(_args: argparse.Namespace) -> int: settings = load_settings() print(topic_url(settings.ntfy_topic)) return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="relay", description="risv3-relay daemon") sub = parser.add_subparsers(dest="cmd", required=True) run = sub.add_parser("run", help="Run the daemon in the foreground") run.add_argument("-v", "--verbose", action="store_true") run.set_defaults(func=_cmd_run) status = sub.add_parser("status", help="Print the current daemon status") status.set_defaults(func=_cmd_status) topic = sub.add_parser("topic", help="Print the ntfy subscription URL") topic.set_defaults(func=_cmd_topic) args = parser.parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main())