diff --git a/README.md b/README.md index 9bb4989..d2fbb7f 100644 --- a/README.md +++ b/README.md @@ -51,13 +51,23 @@ The application requires an INI configuration file to set up the Teams webhook U ## Usage -Run the monitor using the main script: +Run the daemon (long-running monitor, notifications, persistence, IPC server): ```bash -python main.py path/to/config.ini [OPTIONS] +python main.py daemon path/to/config.ini [OPTIONS] ``` -### Options +Run the TUI client (attach/detach as needed, same host via SSH): + +```bash +python main.py tui path/to/config.ini +``` + +In TUI mode, operator commands are available from stdin: +- `r` + Enter: force reconnect (beam + MCR) on daemon +- `q` + Enter: quit TUI client + +### Daemon options - `config`: (Required) Path to the `.ini` configuration file. - `-nc`, `--notify_counts`: Counts threshold at which a "run about to finish" notification is sent (default: 130). @@ -66,8 +76,48 @@ python main.py path/to/config.ini [OPTIONS] ### Example -To run the monitor with a custom configuration file and enabling dummy notifications for testing: +To run the daemon with a custom configuration file and dummy notifications for testing: + +```bash +python main.py daemon config.ini --dummy +``` + +To run TUI from an SSH session on the same host: + +```bash +python main.py tui config.ini +``` + +### Linux service example (systemd) + +Create `/etc/systemd/system/isis-beam-monitor.service`: + +```ini +[Unit] +Description=ISIS Beam Monitor Daemon +After=network-online.target + +[Service] +Type=simple +WorkingDirectory=/path/to/ISIS_Beam_Monitor +ExecStart=/usr/bin/python /path/to/ISIS_Beam_Monitor/main.py daemon /path/to/ISIS_Beam_Monitor/config.ini +Restart=always +RestartSec=5 + +[Install] +WantedBy=multi-user.target +``` + +Then run: ```bash -python main.py config.ini --dummy +sudo systemctl daemon-reload +sudo systemctl enable --now isis-beam-monitor.service +sudo systemctl status isis-beam-monitor.service ``` + +### Troubleshooting + +- **`Lock file already held`**: another daemon instance is running (or stale lock path configured). +- **TUI cannot connect**: ensure daemon is running and `[DAEMON].socket_path` matches `[TUI_CLIENT].socket_path`. +- **No live updates**: check `monitor.log` for websocket/news source errors; use `r` in TUI to force reconnect. diff --git a/config.ini.example b/config.ini.example index cd3c72e..a894158 100644 --- a/config.ini.example +++ b/config.ini.example @@ -22,6 +22,25 @@ experiment_teams_url = # Maximum number of log lines to show in the TUI (default = 50). # logs_maxlen = 50 +[DAEMON] +# SQLite database path for persisted history/state. +# db_path = beam_monitor.db +# Unix domain socket path for local IPC. +# socket_path = /tmp/isis_beam_monitor.sock +# Single-instance lock file for daemon mode. +# lock_file = /tmp/isis_beam_monitor.lock +# Beam history retention window in days. +# retention_days = 7 +# Heartbeat update interval in seconds. +# heartbeat_interval = 30 + +[TUI_CLIENT] +# Socket path to connect to daemon (same host). +# socket_path = /tmp/isis_beam_monitor.sock +# Reconnect backoff start and max in seconds. +# reconnect_initial = 1 +# reconnect_max = 15 + [LOGGING] # log_file = monitor.log # log_level = INFO diff --git a/isis_monitor/__init__.py b/isis_monitor/__init__.py index 99106e3..8432aeb 100644 --- a/isis_monitor/__init__.py +++ b/isis_monitor/__init__.py @@ -2,5 +2,15 @@ from isis_monitor.beam import BeamMonitor from isis_monitor.mcr import MCRNewsMonitor from isis_monitor.config import AppConfig, load_config, ConfigError +from isis_monitor.daemon_state import DaemonState +from isis_monitor.storage import SQLiteStateStore -__all__ = ["BeamMonitor", "MCRNewsMonitor", "AppConfig", "load_config", "ConfigError"] +__all__ = [ + "BeamMonitor", + "MCRNewsMonitor", + "AppConfig", + "load_config", + "ConfigError", + "DaemonState", + "SQLiteStateStore", +] diff --git a/isis_monitor/beam.py b/isis_monitor/beam.py index 31a7bb8..2edfd30 100644 --- a/isis_monitor/beam.py +++ b/isis_monitor/beam.py @@ -11,7 +11,7 @@ from isis_monitor.config import AppConfig from isis_monitor.notifiers import NotificationChannel -from isis_monitor.protocols import TUIProtocol +from isis_monitor.protocols import TUIProtocol, MonitorSinkProtocol logger = logging.getLogger(__name__) @@ -55,6 +55,7 @@ def __init__( experiment_channel: NotificationChannel, counts_target: float, tui: Optional[TUIProtocol] = None, + sink: Optional[MonitorSinkProtocol] = None, ): self.config = config self.data_url = config.isis_websocket_url @@ -64,7 +65,10 @@ def __init__( self.experiment_channel = experiment_channel self.counts_target = counts_target self.tui = tui + self.sink = sink self.state = MonitorState() + self._force_reconnect = asyncio.Event() + self._current_ws = None # Build dynamic lookups from Config self.pv_to_beam: Dict[str, BeamTarget] = { @@ -118,6 +122,8 @@ async def _handle_beam_current( self.state.beams[bt.state_key].current = beam_val self.state.beams[bt.state_key].power = new_state + if self.sink: + self.sink.update_beam_state(bt.channel_label, beam_val, new_state) async def _handle_update(self, message: Dict[str, Any]): """Dispatch WebSocket update messages.""" @@ -148,6 +154,8 @@ async def _handle_update(self, message: Dict[str, Any]): self.state.current_counts = 0 self.state.run_name = name + if self.sink: + self.sink.update_run_name(name) case {"pv": pv, "text": text_val} if pv == self.counts_pv: if not text_val or ( @@ -161,6 +169,8 @@ async def _handle_update(self, message: Dict[str, Any]): return self.state.current_counts = counts + if self.sink: + self.sink.update_counts(counts) if self.state.end_notified and counts < (self.counts_target - 25): self.state.end_notified = False @@ -176,6 +186,14 @@ async def _handle_update(self, message: Dict[str, Any]): state = self.state.beams[bt.state_key] self.tui.update_beam_state(bt.channel_label, state.current, state.power) + def request_reconnect(self) -> bool: + if self._force_reconnect.is_set(): + return False + self._force_reconnect.set() + if self._current_ws is not None: + asyncio.create_task(self._current_ws.close()) + return True + async def run(self, stop_event: Optional[asyncio.Event] = None): subscribe_msg = json.dumps({ "type": "subscribe", @@ -192,11 +210,20 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): try: async with websockets.connect(self.data_url) as ws: logger.info("WebSocket connected.") + if self.sink: + self.sink.update_health("beam", "connected") await ws.send(subscribe_msg) + self._current_ws = ws async for raw_msg in ws: if stop_event and stop_event.is_set(): return + if self._force_reconnect.is_set(): + self._force_reconnect.clear() + logger.info("Beam reconnect requested by operator.") + if self.sink: + self.sink.update_health("beam", "reconnecting") + break try: data = json.loads(raw_msg) if data.get("type") == "update": @@ -209,10 +236,16 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): except (websockets.exceptions.ConnectionClosed, OSError): if stop_event and stop_event.is_set(): return + if self.sink: + self.sink.update_health("beam", "disconnected") logger.warning(f"WebSocket Connection lost. Reconnecting in {self.config.beam_reconnect_interval}s...") await asyncio.sleep(self.config.beam_reconnect_interval) except Exception as e: if stop_event and stop_event.is_set(): return + if self.sink: + self.sink.update_health("beam", "error") logger.error(f"Unexpected error in BeamMonitor: {e}. Reconnecting in {self.config.beam_reconnect_interval}s...") await asyncio.sleep(self.config.beam_reconnect_interval) + finally: + self._current_ws = None diff --git a/isis_monitor/config.py b/isis_monitor/config.py index 72d82a5..a1c10cf 100644 --- a/isis_monitor/config.py +++ b/isis_monitor/config.py @@ -45,6 +45,18 @@ class AppConfig: refresh_per_second: int = 4 logs_maxlen: int = 50 + # DAEMON + daemon_db_path: str = "beam_monitor.db" + daemon_socket_path: str = "/tmp/isis_beam_monitor.sock" + daemon_lock_file: str = "/tmp/isis_beam_monitor.lock" + retention_days: int = 7 + heartbeat_interval: float = 30.0 + + # TUI_CLIENT + tui_socket_path: str = "/tmp/isis_beam_monitor.sock" + tui_reconnect_initial: float = 1.0 + tui_reconnect_max: float = 15.0 + # LOGGING log_file: str = "monitor.log" log_level: str = "INFO" @@ -124,6 +136,24 @@ def _parse_tuple(section, key, default): except (ValueError, configparser.Error) as exc: raise ConfigError(f"[TUI] section contains invalid values: {exc}") from exc + # DAEMON (optional section) + daemon_db_path = config.get("DAEMON", "db_path", fallback="beam_monitor.db") + daemon_socket_path = config.get("DAEMON", "socket_path", fallback="/tmp/isis_beam_monitor.sock") + daemon_lock_file = config.get("DAEMON", "lock_file", fallback="/tmp/isis_beam_monitor.lock") + retention_days = config.getint("DAEMON", "retention_days", fallback=7) + heartbeat_interval = config.getfloat("DAEMON", "heartbeat_interval", fallback=30.0) + if retention_days <= 0: + raise ConfigError("[DAEMON] retention_days must be a positive integer") + + # TUI_CLIENT (optional section) + tui_socket_path = config.get("TUI_CLIENT", "socket_path", fallback=daemon_socket_path) + tui_reconnect_initial = config.getfloat("TUI_CLIENT", "reconnect_initial", fallback=1.0) + tui_reconnect_max = config.getfloat("TUI_CLIENT", "reconnect_max", fallback=15.0) + if tui_reconnect_initial <= 0 or tui_reconnect_max <= 0: + raise ConfigError("[TUI_CLIENT] reconnect values must be positive") + if tui_reconnect_initial > tui_reconnect_max: + raise ConfigError("[TUI_CLIENT] reconnect_initial cannot be greater than reconnect_max") + return AppConfig( mcr_news_url=mcr_news_url, isis_websocket_url=isis_websocket_url, @@ -145,6 +175,14 @@ def _parse_tuple(section, key, default): sample_interval=sample_interval, refresh_per_second=refresh_per_second, logs_maxlen=logs_maxlen, + daemon_db_path=daemon_db_path, + daemon_socket_path=daemon_socket_path, + daemon_lock_file=daemon_lock_file, + retention_days=retention_days, + heartbeat_interval=heartbeat_interval, + tui_socket_path=tui_socket_path, + tui_reconnect_initial=tui_reconnect_initial, + tui_reconnect_max=tui_reconnect_max, log_file=log_file, log_level=log_level, log_max_bytes=log_max_bytes, diff --git a/isis_monitor/daemon_state.py b/isis_monitor/daemon_state.py new file mode 100644 index 0000000..019f6c2 --- /dev/null +++ b/isis_monitor/daemon_state.py @@ -0,0 +1,210 @@ +from __future__ import annotations + +import asyncio +import json +from collections import deque +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from threading import RLock +from typing import Deque, Dict, List, Optional, Tuple + +from isis_monitor.protocols import MonitorSinkProtocol + + +@dataclass +class DaemonEvent: + event: str + payload: dict + + +class DaemonState(MonitorSinkProtocol): + def __init__(self, history_maxlen: int = 10_080, logs_maxlen: int = 200): + self._lock = RLock() + self.history_maxlen = history_maxlen + self.beam_states: Dict[str, Dict[str, object]] = { + "TS1": {"current": 0.0, "power": "unknown"}, + "TS2": {"current": 0.0, "power": "unknown"}, + "Muons": {"current": 0.0, "power": "unknown"}, + } + self.history: Dict[str, Deque[Tuple[datetime, float, str]]] = { + beam: deque(maxlen=history_maxlen) for beam in self.beam_states + } + self.mcr_news = "Waiting for initial MCR news..." + self.logs: Deque[str] = deque(maxlen=logs_maxlen) + self.run_name = "" + self.current_counts = -1.0 + self.last_update = datetime.now(timezone.utc) + self.health: Dict[str, str] = { + "daemon": "starting", + "beam": "unknown", + "mcr": "unknown", + } + self._subscribers: List[asyncio.Queue] = [] + + def subscribe(self) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=500) + with self._lock: + self._subscribers.append(q) + return q + + def unsubscribe(self, q: asyncio.Queue) -> None: + with self._lock: + if q in self._subscribers: + self._subscribers.remove(q) + + def _publish(self, event: str, payload: dict) -> None: + dead = [] + for q in list(self._subscribers): + try: + q.put_nowait(DaemonEvent(event=event, payload=payload)) + except asyncio.QueueFull: + dead.append(q) + for q in dead: + if q in self._subscribers: + self._subscribers.remove(q) + + def update_log(self, message: str) -> None: + with self._lock: + self.logs.append(message) + self.last_update = datetime.now(timezone.utc) + self._publish("log", {"message": message}) + + def update_beam_state(self, beam: str, current: float, power: str) -> None: + with self._lock: + if beam not in self.beam_states: + return + self.beam_states[beam] = {"current": float(current), "power": str(power)} + self.last_update = datetime.now(timezone.utc) + self._publish("beam", {"beam": beam, "current": current, "power": power}) + + def append_beam_sample( + self, + beam: str, + current: float, + power: str, + ts: Optional[datetime] = None, + publish: bool = True, + ) -> None: + ts = ts or datetime.now(timezone.utc) + with self._lock: + if beam not in self.history: + return + self.history[beam].append((ts, float(current), str(power))) + self.last_update = ts + if publish: + self._publish( + "sample", + { + "beam": beam, + "timestamp": ts.isoformat(), + "current": current, + "power": power, + }, + ) + + def trim_history_before(self, cutoff: datetime) -> None: + with self._lock: + for beam in self.history: + trimmed = deque( + (entry for entry in self.history[beam] if entry[0] >= cutoff), + maxlen=self.history_maxlen, + ) + self.history[beam] = trimmed + + def update_mcr_news(self, news: str) -> None: + with self._lock: + self.mcr_news = news + self.last_update = datetime.now(timezone.utc) + self._publish("mcr", {"news": news}) + + def update_run_name(self, run_name: str) -> None: + with self._lock: + self.run_name = run_name + self.last_update = datetime.now(timezone.utc) + self._publish("run", {"run_name": run_name}) + + def update_counts(self, counts: float) -> None: + with self._lock: + self.current_counts = float(counts) + self.last_update = datetime.now(timezone.utc) + self._publish("counts", {"counts": counts}) + + def update_health(self, component: str, status: str) -> None: + with self._lock: + self.health[component] = status + self.last_update = datetime.now(timezone.utc) + self._publish("health", {"component": component, "status": status}) + + def snapshot(self) -> dict: + with self._lock: + history_json = { + beam: [ + { + "timestamp": ts.isoformat(), + "current": cur, + "power": power, + } + for ts, cur, power in data + ] + for beam, data in self.history.items() + } + return { + "last_update": self.last_update.isoformat(), + "beam_states": dict(self.beam_states), + "history": history_json, + "mcr_news": self.mcr_news, + "logs": list(self.logs), + "run_name": self.run_name, + "current_counts": self.current_counts, + "health": dict(self.health), + } + + def sample_all_currents(self, ts: Optional[datetime] = None) -> None: + ts = ts or datetime.now(timezone.utc) + with self._lock: + items = list(self.beam_states.items()) + for beam, state in items: + self.append_beam_sample( + beam, + float(state["current"]), + str(state["power"]), + ts=ts, + ) + + def cutoff_for_days(self, retention_days: int) -> datetime: + return datetime.now(timezone.utc) - timedelta(days=retention_days) + + def get_beam_rows_for_timestamp(self, ts: Optional[datetime] = None) -> list[tuple[datetime, str, float, str]]: + ts = ts or datetime.now(timezone.utc) + with self._lock: + return [ + (ts, beam, float(state["current"]), str(state["power"])) + for beam, state in self.beam_states.items() + ] + + def get_health(self) -> Dict[str, str]: + with self._lock: + return dict(self.health) + + def restore_from_snapshot_json(self, raw: Optional[str]) -> None: + if not raw: + return + try: + snap = json.loads(raw) + except json.JSONDecodeError: + return + with self._lock: + beam_states = snap.get("beam_states", {}) + for beam in ("TS1", "TS2", "Muons"): + if beam in beam_states: + self.beam_states[beam] = { + "current": float(beam_states[beam].get("current", 0.0)), + "power": str(beam_states[beam].get("power", "unknown")), + } + self.mcr_news = str(snap.get("mcr_news", self.mcr_news)) + self.run_name = str(snap.get("run_name", self.run_name)) + self.current_counts = float(snap.get("current_counts", self.current_counts)) + health = snap.get("health", {}) + if isinstance(health, dict): + for k, v in health.items(): + self.health[str(k)] = str(v) diff --git a/isis_monitor/ipc.py b/isis_monitor/ipc.py new file mode 100644 index 0000000..0913473 --- /dev/null +++ b/isis_monitor/ipc.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import asyncio +import contextlib +import json +from pathlib import Path +from typing import Awaitable, Callable, Optional + +from isis_monitor.daemon_state import DaemonEvent, DaemonState + +PROTOCOL_VERSION = 1 + + +class IPCServer: + def __init__( + self, + socket_path: Path, + state: DaemonState, + command_handler: Callable[[str], Awaitable[dict]], + ): + self.socket_path = Path(socket_path) + self.state = state + self.command_handler = command_handler + self.server: Optional[asyncio.base_events.Server] = None + + async def start(self) -> None: + self.socket_path.parent.mkdir(parents=True, exist_ok=True) + if self.socket_path.exists(): + self.socket_path.unlink() + self.server = await asyncio.start_unix_server(self._handle_client, path=str(self.socket_path)) + + async def stop(self) -> None: + if self.server is not None: + self.server.close() + await self.server.wait_closed() + if self.socket_path.exists(): + self.socket_path.unlink() + + async def _send(self, writer: asyncio.StreamWriter, payload: dict) -> None: + writer.write((json.dumps(payload) + "\n").encode()) + await writer.drain() + + async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + queue = None + subscription_task = None + try: + while True: + line = await reader.readline() + if not line: + break + try: + req = json.loads(line.decode()) + except json.JSONDecodeError: + await self._send( + writer, + {"ok": False, "error": "invalid_json", "version": PROTOCOL_VERSION}, + ) + continue + + method = req.get("method") + if method == "get_snapshot": + await self._send( + writer, + { + "ok": True, + "version": PROTOCOL_VERSION, + "snapshot": self.state.snapshot(), + }, + ) + elif method == "subscribe_updates": + if queue is None: + queue = self.state.subscribe() + subscription_task = asyncio.create_task( + self._forward_events(queue, writer) + ) + await self._send( + writer, + {"ok": True, "version": PROTOCOL_VERSION, "subscribed": True}, + ) + elif method == "command": + command = str(req.get("name", "")) + result = await self.command_handler(command) + await self._send( + writer, + {"ok": True, "version": PROTOCOL_VERSION, "result": result}, + ) + else: + await self._send( + writer, + { + "ok": False, + "version": PROTOCOL_VERSION, + "error": "unknown_method", + }, + ) + finally: + if subscription_task: + subscription_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await subscription_task + if queue is not None: + self.state.unsubscribe(queue) + writer.close() + await writer.wait_closed() + + async def _forward_events(self, queue: asyncio.Queue, writer: asyncio.StreamWriter) -> None: + while True: + ev: DaemonEvent = await queue.get() + payload = { + "ok": True, + "version": PROTOCOL_VERSION, + "event": ev.event, + "payload": ev.payload, + } + await self._send(writer, payload) + + +class IPCClient: + def __init__(self, socket_path: Path): + self.socket_path = Path(socket_path) + self.reader: Optional[asyncio.StreamReader] = None + self.writer: Optional[asyncio.StreamWriter] = None + + async def connect(self) -> None: + self.reader, self.writer = await asyncio.open_unix_connection(str(self.socket_path)) + + async def close(self) -> None: + if self.writer: + self.writer.close() + await self.writer.wait_closed() + self.reader = None + self.writer = None + + async def request(self, payload: dict) -> dict: + if not self.writer or not self.reader: + raise RuntimeError("IPC client is not connected") + self.writer.write((json.dumps(payload) + "\n").encode()) + await self.writer.drain() + line = await self.reader.readline() + if not line: + raise ConnectionError("Daemon closed IPC connection") + return json.loads(line.decode()) + + async def iter_events(self): + if not self.reader: + raise RuntimeError("IPC client is not connected") + while True: + line = await self.reader.readline() + if not line: + raise ConnectionError("Daemon closed IPC stream") + yield json.loads(line.decode()) diff --git a/isis_monitor/mcr.py b/isis_monitor/mcr.py index c6eaa18..0100c90 100644 --- a/isis_monitor/mcr.py +++ b/isis_monitor/mcr.py @@ -1,4 +1,5 @@ import asyncio +import contextlib import logging import re from datetime import datetime @@ -7,7 +8,7 @@ from isis_monitor.config import AppConfig from isis_monitor.notifiers import NotificationChannel -from isis_monitor.protocols import TUIProtocol +from isis_monitor.protocols import TUIProtocol, MonitorSinkProtocol logger = logging.getLogger(__name__) @@ -22,13 +23,16 @@ def __init__( channel: NotificationChannel, notify_current: bool = False, tui: Optional[TUIProtocol] = None, + sink: Optional[MonitorSinkProtocol] = None, ): self.config = config self.url = config.mcr_news_url self.channel = channel self.notify_current = notify_current self.tui = tui + self.sink = sink self.old_news: Optional[str] = None + self._force_reconnect = asyncio.Event() async def get_news(self, session: aiohttp.ClientSession) -> Optional[str]: try: @@ -56,6 +60,8 @@ async def get_news(self, session: aiohttp.ClientSession) -> Optional[str]: async def run(self, stop_event: Optional[asyncio.Event] = None): logger.info(f"MCR Monitor started. Watching {self.url}...") + if self.sink: + self.sink.update_health("mcr", "starting") # TCPConnector with DNS TTL avoids stale connections on long-running sessions connector = aiohttp.TCPConnector(ttl_dns_cache=300) @@ -70,6 +76,8 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): logger.info(f"Current MCR News: {self.old_news}") if self.tui: self.tui.update_mcr_news(self.old_news) + if self.sink: + self.sink.update_mcr_news(self.old_news) else: await asyncio.sleep(self.config.mcr_poll_interval) else: @@ -79,10 +87,24 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): consecutive_failures = 0 while stop_event is None or not stop_event.is_set(): try: - sleep_secs = self.config.mcr_poll_interval * min( - 2 ** consecutive_failures, 8 + sleep_secs = self.config.mcr_poll_interval * min(2 ** consecutive_failures, 8) + sleep_task = asyncio.create_task(asyncio.sleep(sleep_secs)) + reconnect_task = asyncio.create_task(self._force_reconnect.wait()) + done, pending = await asyncio.wait( + {sleep_task, reconnect_task}, + return_when=asyncio.FIRST_COMPLETED, ) - await asyncio.sleep(sleep_secs) + for task in pending: + task.cancel() + for task in pending: + with contextlib.suppress(asyncio.CancelledError): + await task + + if reconnect_task in done and self._force_reconnect.is_set(): + self._force_reconnect.clear() + consecutive_failures = 0 + if self.sink: + self.sink.update_health("mcr", "reconnecting") except asyncio.CancelledError: return @@ -96,13 +118,26 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): logger.info(f"New MCR Update: {new_news}") if self.tui: self.tui.update_mcr_news(new_news) + if self.sink: + self.sink.update_mcr_news(new_news) + self.sink.update_health("mcr", "connected") await self.channel.broadcast(new_news) elif new_news: consecutive_failures = 0 + if self.sink: + self.sink.update_health("mcr", "connected") logger.debug("No new MCR news.") else: consecutive_failures += 1 + if self.sink: + self.sink.update_health("mcr", "error") logger.debug( f"MCR fetch failed (attempt {consecutive_failures}); " f"next retry in {sleep_secs * min(2, 8):.0f}s." ) + + def request_reconnect(self) -> bool: + if self._force_reconnect.is_set(): + return False + self._force_reconnect.set() + return True diff --git a/isis_monitor/protocols.py b/isis_monitor/protocols.py index 67d5fd4..a810074 100644 --- a/isis_monitor/protocols.py +++ b/isis_monitor/protocols.py @@ -42,3 +42,23 @@ def stop(self) -> None: async def run_sampler(self, stop_event: asyncio.Event) -> None: """Coroutine that periodically snapshots beam state into history.""" ... + + +@runtime_checkable +class MonitorSinkProtocol(Protocol): + """Interface for receiving monitor updates without coupling to RichTUI.""" + + def update_beam_state(self, beam: str, current: float, power: str) -> None: + ... + + def update_mcr_news(self, news: str) -> None: + ... + + def update_run_name(self, run_name: str) -> None: + ... + + def update_counts(self, counts: float) -> None: + ... + + def update_health(self, component: str, status: str) -> None: + ... diff --git a/isis_monitor/storage.py b/isis_monitor/storage.py new file mode 100644 index 0000000..38a2b02 --- /dev/null +++ b/isis_monitor/storage.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Optional, Tuple + + +class SQLiteStateStore: + def __init__(self, db_path: Path): + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self.conn = sqlite3.connect(str(self.db_path)) + self.conn.row_factory = sqlite3.Row + self._init_schema() + + def _init_schema(self) -> None: + self.conn.executescript( + """ + CREATE TABLE IF NOT EXISTS beam_samples ( + timestamp TEXT NOT NULL, + target TEXT NOT NULL, + current REAL NOT NULL, + power TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_beam_samples_time + ON beam_samples(timestamp); + + CREATE TABLE IF NOT EXISTS snapshot ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS health ( + component TEXT PRIMARY KEY, + status TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + """ + ) + self.conn.commit() + + def close(self) -> None: + self.conn.close() + + def write_sample(self, timestamp: datetime, target: str, current: float, power: str) -> None: + self.conn.execute( + "INSERT INTO beam_samples(timestamp, target, current, power) VALUES (?, ?, ?, ?)", + (timestamp.isoformat(), target, current, power), + ) + + def write_samples(self, rows: Iterable[Tuple[datetime, str, float, str]]) -> None: + self.conn.executemany( + "INSERT INTO beam_samples(timestamp, target, current, power) VALUES (?, ?, ?, ?)", + [(ts.isoformat(), target, current, power) for ts, target, current, power in rows], + ) + + def prune_older_than(self, cutoff: datetime) -> int: + cur = self.conn.execute( + "DELETE FROM beam_samples WHERE timestamp < ?", + (cutoff.isoformat(),), + ) + return cur.rowcount + + def load_recent_samples(self, since: datetime) -> list[sqlite3.Row]: + cur = self.conn.execute( + """ + SELECT timestamp, target, current, power + FROM beam_samples + WHERE timestamp >= ? + ORDER BY timestamp ASC + """, + (since.isoformat(),), + ) + return list(cur.fetchall()) + + def upsert_snapshot(self, key: str, value: str) -> None: + now = datetime.now(timezone.utc).isoformat() + self.conn.execute( + """ + INSERT INTO snapshot(key, value, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at + """, + (key, value, now), + ) + + def load_snapshot(self, key: str) -> Optional[str]: + cur = self.conn.execute("SELECT value FROM snapshot WHERE key = ?", (key,)) + row = cur.fetchone() + return row[0] if row else None + + def upsert_health(self, component: str, status: str) -> None: + now = datetime.now(timezone.utc).isoformat() + self.conn.execute( + """ + INSERT INTO health(component, status, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(component) DO UPDATE SET status=excluded.status, updated_at=excluded.updated_at + """, + (component, status, now), + ) + + def load_health(self) -> list[sqlite3.Row]: + cur = self.conn.execute("SELECT component, status FROM health") + return list(cur.fetchall()) + + def commit(self) -> None: + self.conn.commit() diff --git a/isis_monitor/tests/test_config.py b/isis_monitor/tests/test_config.py index b7f38aa..4bd1342 100644 --- a/isis_monitor/tests/test_config.py +++ b/isis_monitor/tests/test_config.py @@ -133,3 +133,57 @@ def test_load_config_empty_websocket_url_logs_warning(tmp_path, caplog): config = load_config(config_file) assert config.isis_websocket_url == "" assert "isis_websocket_url" in caplog.text + + +def test_load_config_daemon_and_tui_client_values(tmp_path): + config_file = tmp_path / "config.ini" + config_file.write_text("""\ +[DATA] +mcr_news_url = http://test.com/news +isis_websocket_url = wss://test.com/ws + +[WEBHOOKS] +news_teams_url = +beam_teams_url = +experiment_teams_url = + +[DAEMON] +db_path = /tmp/beam.db +socket_path = /tmp/beam.sock +lock_file = /tmp/beam.lock +retention_days = 7 +heartbeat_interval = 15 + +[TUI_CLIENT] +socket_path = /tmp/beam.sock +reconnect_initial = 2 +reconnect_max = 20 +""") + config = load_config(config_file) + assert config.daemon_db_path == "/tmp/beam.db" + assert config.daemon_socket_path == "/tmp/beam.sock" + assert config.daemon_lock_file == "/tmp/beam.lock" + assert config.retention_days == 7 + assert config.heartbeat_interval == 15 + assert config.tui_socket_path == "/tmp/beam.sock" + assert config.tui_reconnect_initial == 2 + assert config.tui_reconnect_max == 20 + + +def test_load_config_invalid_retention_days(tmp_path): + config_file = tmp_path / "config.ini" + config_file.write_text("""\ +[DATA] +mcr_news_url = http://test.com/news +isis_websocket_url = wss://test.com/ws + +[WEBHOOKS] +news_teams_url = +beam_teams_url = +experiment_teams_url = + +[DAEMON] +retention_days = 0 +""") + with pytest.raises(ConfigError, match="retention_days"): + load_config(config_file) diff --git a/isis_monitor/tests/test_ipc.py b/isis_monitor/tests/test_ipc.py new file mode 100644 index 0000000..eb636c6 --- /dev/null +++ b/isis_monitor/tests/test_ipc.py @@ -0,0 +1,64 @@ +import asyncio +from pathlib import Path + +import pytest + +from isis_monitor.daemon_state import DaemonState +from isis_monitor.ipc import IPCClient, IPCServer + + +@pytest.mark.asyncio +async def test_ipc_snapshot_and_command(tmp_path): + socket_path = tmp_path / "daemon.sock" + state = DaemonState() + state.update_mcr_news("hello") + + async def command_handler(name: str): + if name == "force_reconnect_all": + return {"beam": True, "mcr": True} + return {"error": "unknown"} + + server = IPCServer(socket_path, state, command_handler) + await server.start() + + client = IPCClient(socket_path) + await client.connect() + + snap = await client.request({"method": "get_snapshot"}) + assert snap["ok"] is True + assert snap["snapshot"]["mcr_news"] == "hello" + + cmd = await client.request({"method": "command", "name": "force_reconnect_all"}) + assert cmd["ok"] is True + assert cmd["result"] == {"beam": True, "mcr": True} + + await client.close() + await server.stop() + + +@pytest.mark.asyncio +async def test_ipc_subscribe_updates(tmp_path): + socket_path = tmp_path / "daemon.sock" + state = DaemonState() + + async def command_handler(_name: str): + return {"ok": True} + + server = IPCServer(socket_path, state, command_handler) + await server.start() + + client = IPCClient(socket_path) + await client.connect() + + sub = await client.request({"method": "subscribe_updates"}) + assert sub["ok"] is True + + state.update_beam_state("TS1", 12.3, "low") + + msg = await asyncio.wait_for(client.reader.readline(), timeout=1.0) + payload = __import__("json").loads(msg.decode()) + assert payload["event"] == "beam" + assert payload["payload"]["beam"] == "TS1" + + await client.close() + await server.stop() diff --git a/isis_monitor/tests/test_main.py b/isis_monitor/tests/test_main.py index dd941ca..1292d3c 100644 --- a/isis_monitor/tests/test_main.py +++ b/isis_monitor/tests/test_main.py @@ -1,14 +1,14 @@ import logging import pytest from unittest.mock import MagicMock -from main import TUILogHandler +from main import StateLogHandler -class TestTUILogHandler: +class TestStateLogHandler: def test_emit_calls_update_log(self): - """TUILogHandler.emit should forward the formatted message to tui.update_log.""" - mock_tui = MagicMock() - handler = TUILogHandler(mock_tui) + """StateLogHandler.emit should forward the formatted message.""" + mock_state = MagicMock() + handler = StateLogHandler(mock_state) handler.setFormatter(logging.Formatter("%(levelname)s - %(message)s")) record = logging.LogRecord( @@ -17,13 +17,13 @@ def test_emit_calls_update_log(self): ) handler.emit(record) - mock_tui.update_log.assert_called_once_with("INFO - hello world") + mock_state.update_log.assert_called_once_with("INFO - hello world") def test_emit_handles_exception_gracefully(self, caplog): - """If tui.update_log raises, handleError should be called and not propagate.""" - mock_tui = MagicMock() - mock_tui.update_log.side_effect = RuntimeError("TUI broken") - handler = TUILogHandler(mock_tui) + """If state.update_log raises, handleError should be called and not propagate.""" + mock_state = MagicMock() + mock_state.update_log.side_effect = RuntimeError("State broken") + handler = StateLogHandler(mock_state) record = logging.LogRecord( name="test", level=logging.INFO, pathname="", lineno=0, @@ -34,8 +34,8 @@ def test_emit_handles_exception_gracefully(self, caplog): def test_emit_with_warning_level(self): """Formatter applied correctly for WARNING level messages.""" - mock_tui = MagicMock() - handler = TUILogHandler(mock_tui) + mock_state = MagicMock() + handler = StateLogHandler(mock_state) handler.setFormatter(logging.Formatter("%(levelname)s - %(message)s")) record = logging.LogRecord( @@ -43,4 +43,4 @@ def test_emit_with_warning_level(self): msg="something went wrong", args=(), exc_info=None, ) handler.emit(record) - mock_tui.update_log.assert_called_once_with("WARNING - something went wrong") + mock_state.update_log.assert_called_once_with("WARNING - something went wrong") diff --git a/isis_monitor/tests/test_storage.py b/isis_monitor/tests/test_storage.py new file mode 100644 index 0000000..c14d84d --- /dev/null +++ b/isis_monitor/tests/test_storage.py @@ -0,0 +1,45 @@ +from datetime import datetime, timedelta, timezone + +from isis_monitor.storage import SQLiteStateStore + + +def test_storage_write_load_and_prune(tmp_path): + db = tmp_path / "state.db" + store = SQLiteStateStore(db) + + now = datetime.now(timezone.utc) + old = now - timedelta(days=8) + + store.write_sample(old, "TS1", 1.0, "low") + store.write_sample(now, "TS1", 2.0, "medium") + store.commit() + + rows = store.load_recent_samples(now - timedelta(days=7)) + assert len(rows) == 1 + assert rows[0]["current"] == 2.0 + + deleted = store.prune_older_than(now - timedelta(days=7)) + store.commit() + assert deleted == 1 + + rows2 = store.load_recent_samples(now - timedelta(days=30)) + assert len(rows2) == 1 + store.close() + + +def test_storage_snapshot_and_health(tmp_path): + db = tmp_path / "state.db" + store = SQLiteStateStore(db) + + store.upsert_snapshot("daemon_state", '{"ok":true}') + store.upsert_health("beam", "connected") + store.commit() + + snap = store.load_snapshot("daemon_state") + assert snap == '{"ok":true}' + + health = store.load_health() + assert len(health) == 1 + assert health[0]["component"] == "beam" + assert health[0]["status"] == "connected" + store.close() diff --git a/isis_monitor/tui.py b/isis_monitor/tui.py index 6d265fc..9905398 100644 --- a/isis_monitor/tui.py +++ b/isis_monitor/tui.py @@ -103,6 +103,7 @@ def __init__( self.mcr_news = "Waiting for initial MCR news..." self._logs: Deque[str] = deque(maxlen=self.logs_maxlen) self.last_update = datetime.now(timezone.utc) + self.connection_state = "DISCONNECTED" self._lock = RLock() self.layout = self._make_layout() @@ -206,7 +207,11 @@ def _update_all(self): with self._lock: self.layout["header"].update( Panel( - Text("ISIS Facility Monitor", justify="center", style="bold cyan"), + Text( + f"ISIS Facility Monitor [{self.connection_state}]", + justify="center", + style="bold cyan", + ), style="blue", ) ) @@ -291,6 +296,32 @@ def _update_mcr_panel(self): ) ) + def add_history_sample(self, beam: str, timestamp: datetime, current: float, power: str) -> None: + with self._lock: + if beam in self._history: + self._history[beam].append((timestamp, current, power)) + self.last_update = datetime.now(timezone.utc) + self._update_beam_graph() + + def set_history_snapshot(self, history: dict[str, list[dict]]) -> None: + with self._lock: + for beam in self._history.keys(): + self._history[beam].clear() + for beam, rows in history.items(): + if beam not in self._history: + continue + for row in rows: + ts = datetime.fromisoformat(str(row["timestamp"])) + self._history[beam].append( + (ts, float(row["current"]), str(row["power"])) + ) + self._update_beam_graph() + + def update_connection_state(self, state: str) -> None: + with self._lock: + self.connection_state = state.upper() + self._update_all() + def _update_logs_panel(self): # Only show the latest few logs that fit in the panel height (split size 8) # NOTE: caller must hold self._lock (consistent with all other _update_* helpers) diff --git a/main.py b/main.py index bc31372..4a4833f 100755 --- a/main.py +++ b/main.py @@ -1,111 +1,367 @@ #!/usr/bin/env python3 +import argparse import asyncio +import contextlib +import fcntl +import json import logging +import os import signal +from datetime import datetime, timezone from logging.handlers import RotatingFileHandler -import argparse from pathlib import Path +from typing import Optional -from isis_monitor.config import load_config, ConfigError -from isis_monitor.notifiers import NotificationChannel, TeamsNotifier, DummyNotifier from isis_monitor.beam import BeamMonitor +from isis_monitor.config import ConfigError, load_config +from isis_monitor.daemon_state import DaemonState +from isis_monitor.ipc import IPCClient, IPCServer from isis_monitor.mcr import MCRNewsMonitor +from isis_monitor.notifiers import DummyNotifier, NotificationChannel, TeamsNotifier +from isis_monitor.storage import SQLiteStateStore from isis_monitor.tui import RichTUI -# Logger is configured dynamically in main() based on config logger = logging.getLogger("MAIN") -class TUILogHandler(logging.Handler): - def __init__(self, tui): + +class StateLogHandler(logging.Handler): + def __init__(self, state: DaemonState): super().__init__() - self.tui = tui + self.state = state def emit(self, record): try: msg = self.format(record) - self.tui.update_log(msg) + self.state.update_log(msg) except Exception: self.handleError(record) -async def run_all(config, args, stop_event: asyncio.Event): - # Install signal handlers: set the stop event AND cancel all running tasks so - # the TUI (Rich Live) is torn down immediately and the terminal is restored. +class SingleInstanceLock: + def __init__(self, path: Path): + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + self._fh = None + + def __enter__(self): + self._fh = self.path.open("w") + try: + fcntl.flock(self._fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + raise RuntimeError(f"Lock file already held: {self.path}") from exc + self._fh.seek(0) + self._fh.truncate(0) + self._fh.write(str(os.getpid())) + self._fh.flush() + return self + + def __exit__(self, exc_type, exc, tb): + if self._fh: + try: + fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN) + except OSError: + pass + self._fh.close() + with contextlib.suppress(OSError): + self.path.unlink() + + +def configure_logging(log_file: str, log_level: str, max_bytes: int, backup_count: int) -> None: + log_path = Path(log_file) + if not log_path.is_absolute(): + log_path = Path(__file__).parent / log_path + numeric_level = getattr(logging, log_level.upper(), logging.WARNING) + logging.basicConfig( + level=numeric_level, + format="%(asctime)s - %(levelname)s - %(message)s", + handlers=[ + RotatingFileHandler(log_path, maxBytes=max_bytes, backupCount=backup_count) + ], + ) + + +def install_signal_handlers(stop_event: asyncio.Event) -> None: loop = asyncio.get_running_loop() + def _on_signal(): stop_event.set() for task in asyncio.all_tasks(loop): task.cancel() + for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, _on_signal) - # Initialize TUI - tui = RichTUI( - history_maxlen=config.history_maxlen, - sample_interval=config.sample_interval, - refresh_per_second=config.refresh_per_second, - logs_maxlen=config.logs_maxlen, - ) - tui.start() - # Route logs to TUI - tui_handler = TUILogHandler(tui) - tui_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) - logging.getLogger().addHandler(tui_handler) - - # Setup Notification Channels +def build_channels(config, dummy: bool): beam_channel = NotificationChannel("Beam Updates") exp_channel = NotificationChannel("Experiment Updates") mcr_channel = NotificationChannel("MCR News") - if args.dummy: - logger.info("Initializing Dummy Notifier (logs to console)") + if dummy: beam_channel.add_notifier(DummyNotifier()) exp_channel.add_notifier(DummyNotifier()) mcr_channel.add_notifier(DummyNotifier()) else: - # Configure Teams Notifiers Only If Not Dummy if config.beam_teams_url: - beam_channel.add_notifier(TeamsNotifier(config.beam_teams_url, timeout=config.webhook_timeout)) + beam_channel.add_notifier( + TeamsNotifier(config.beam_teams_url, timeout=config.webhook_timeout) + ) if config.experiment_teams_url: - exp_channel.add_notifier(TeamsNotifier(config.experiment_teams_url, timeout=config.webhook_timeout)) + exp_channel.add_notifier( + TeamsNotifier(config.experiment_teams_url, timeout=config.webhook_timeout) + ) if config.news_teams_url: - mcr_channel.add_notifier(TeamsNotifier(config.news_teams_url, timeout=config.webhook_timeout)) + mcr_channel.add_notifier( + TeamsNotifier(config.news_teams_url, timeout=config.webhook_timeout) + ) + return beam_channel, exp_channel, mcr_channel + + +async def close_channels(*channels: NotificationChannel) -> None: + to_close = [] + for channel in channels: + for notifier in channel.notifiers: + close_fn = getattr(notifier, "close", None) + if close_fn is not None: + to_close.append(close_fn()) + if to_close: + await asyncio.gather(*to_close, return_exceptions=True) + + +async def state_persistence_loop(config, state: DaemonState, store: SQLiteStateStore, stop_event: asyncio.Event): + while not stop_event.is_set(): + try: + await asyncio.wait_for(stop_event.wait(), timeout=config.sample_interval) + break + except asyncio.TimeoutError: + pass + + ts = datetime.now(timezone.utc) + state.sample_all_currents(ts) + store.write_samples(state.get_beam_rows_for_timestamp(ts)) + + cutoff = state.cutoff_for_days(config.retention_days) + store.prune_older_than(cutoff) + state.trim_history_before(cutoff) + + store.upsert_snapshot("daemon_state", json.dumps(state.snapshot())) + for component, status in state.get_health().items(): + store.upsert_health(component, status) + store.commit() + + +async def daemon_heartbeat_loop(config, state: DaemonState, stop_event: asyncio.Event): + while not stop_event.is_set(): + state.update_health("daemon", "running") + try: + await asyncio.wait_for(stop_event.wait(), timeout=config.heartbeat_interval) + except asyncio.TimeoutError: + continue + + +async def run_daemon(config, args, stop_event: asyncio.Event): + install_signal_handlers(stop_event) + + state = DaemonState(history_maxlen=max(config.history_maxlen, int((86400 * config.retention_days) / max(config.sample_interval, 1.0)))) + state.update_health("daemon", "starting") + + store = SQLiteStateStore(Path(config.daemon_db_path)) + state.restore_from_snapshot_json(store.load_snapshot("daemon_state")) + cutoff = state.cutoff_for_days(config.retention_days) + for row in store.load_recent_samples(cutoff): + state.append_beam_sample( + beam=str(row["target"]), + current=float(row["current"]), + power=str(row["power"]), + ts=datetime.fromisoformat(str(row["timestamp"])), + publish=False, + ) + + state_log_handler = StateLogHandler(state) + state_log_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) + logging.getLogger().addHandler(state_log_handler) + beam_channel, exp_channel, mcr_channel = build_channels(config, args.dummy) + beam_monitor = BeamMonitor( + config, + beam_channel, + exp_channel, + args.notify_counts, + sink=state, + ) + mcr_monitor = MCRNewsMonitor( + config, + mcr_channel, + args.notify_current, + sink=state, + ) + + async def command_handler(name: str) -> dict: + if name in {"force_reconnect", "force_reconnect_all"}: + return { + "beam": beam_monitor.request_reconnect(), + "mcr": mcr_monitor.request_reconnect(), + } + if name == "force_reconnect_beam": + return {"beam": beam_monitor.request_reconnect()} + if name == "force_reconnect_mcr": + return {"mcr": mcr_monitor.request_reconnect()} + return {"error": "unknown_command", "name": name} - # Initialize Monitors - beam_monitor = BeamMonitor(config, beam_channel, exp_channel, args.notify_counts, tui=tui) - mcr_monitor = MCRNewsMonitor(config, mcr_channel, args.notify_current, tui=tui) + ipc_server = IPCServer(Path(config.daemon_socket_path), state, command_handler) + await ipc_server.start() + state.update_health("daemon", "running") - logger.info("Starting monitors concurrently...") try: await asyncio.gather( beam_monitor.run(stop_event), mcr_monitor.run(stop_event), - tui.run_sampler(stop_event), + state_persistence_loop(config, state, store, stop_event), + daemon_heartbeat_loop(config, state, stop_event), ) + finally: + state.update_health("daemon", "stopping") + store.upsert_snapshot("daemon_state", json.dumps(state.snapshot())) + store.commit() + store.close() + await ipc_server.stop() + await close_channels(beam_channel, exp_channel, mcr_channel) + + +def _apply_snapshot_to_tui(tui: RichTUI, snapshot: dict) -> None: + beam_states = snapshot.get("beam_states", {}) + for beam in ("TS1", "TS2", "Muons"): + state = beam_states.get(beam) + if state: + tui.update_beam_state(beam, float(state.get("current", 0.0)), str(state.get("power", "unknown"))) + tui.set_history_snapshot(snapshot.get("history", {})) + if snapshot.get("mcr_news"): + tui.update_mcr_news(str(snapshot["mcr_news"])) + for line in snapshot.get("logs", [])[-20:]: + tui.update_log(str(line)) + + +def _apply_event_to_tui(tui: RichTUI, message: dict) -> None: + ev = message.get("event") + payload = message.get("payload", {}) + if ev == "beam": + tui.update_beam_state(str(payload.get("beam", "")), float(payload.get("current", 0.0)), str(payload.get("power", "unknown"))) + elif ev == "mcr": + tui.update_mcr_news(str(payload.get("news", ""))) + elif ev == "log": + tui.update_log(str(payload.get("message", ""))) + elif ev == "sample": + ts_raw = payload.get("timestamp") + if not ts_raw: + return + ts = datetime.fromisoformat(str(ts_raw)) + tui.add_history_sample( + str(payload.get("beam", "")), + ts, + float(payload.get("current", 0.0)), + str(payload.get("power", "unknown")), + ) + elif ev == "health": + comp = str(payload.get("component", "")) + status = str(payload.get("status", "")) + tui.update_log(f"Health: {comp} -> {status}") + + +async def tui_command_loop(client: IPCClient, stop_event: asyncio.Event, tui: RichTUI): + while not stop_event.is_set(): + cmd = (await asyncio.to_thread(input, "Command [r=reconnect,q=quit]: ")).strip().lower() + if cmd == "q": + stop_event.set() + return + if cmd == "r": + response = await client.request({"method": "command", "name": "force_reconnect_all"}) + tui.update_log(f"Reconnect request result: {response.get('result')}") + + +async def run_tui(config, stop_event: asyncio.Event): + install_signal_handlers(stop_event) + + tui = RichTUI( + history_maxlen=config.history_maxlen, + sample_interval=config.sample_interval, + refresh_per_second=config.refresh_per_second, + logs_maxlen=config.logs_maxlen, + ) + tui.start() + + backoff = config.tui_reconnect_initial + try: + while not stop_event.is_set(): + client = IPCClient(Path(config.tui_socket_path)) + cmd_task: Optional[asyncio.Task] = None + try: + tui.update_connection_state("connecting") + await client.connect() + tui.update_connection_state("connected") + + snapshot_resp = await client.request({"method": "get_snapshot"}) + if snapshot_resp.get("ok"): + _apply_snapshot_to_tui(tui, snapshot_resp.get("snapshot", {})) + + sub_resp = await client.request({"method": "subscribe_updates"}) + if sub_resp.get("ok"): + tui.update_log("Subscribed to daemon updates.") + + cmd_task = asyncio.create_task(tui_command_loop(client, stop_event, tui)) + backoff = config.tui_reconnect_initial + + async for message in client.iter_events(): + _apply_event_to_tui(tui, message) + if stop_event.is_set(): + break + except (FileNotFoundError, ConnectionError, OSError) as exc: + tui.update_connection_state("disconnected") + tui.update_log(f"Daemon connection lost: {exc}") + try: + await asyncio.wait_for(stop_event.wait(), timeout=backoff) + except asyncio.TimeoutError: + pass + backoff = min(config.tui_reconnect_max, max(config.tui_reconnect_initial, backoff * 2)) + finally: + if cmd_task: + cmd_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await cmd_task + await client.close() finally: tui.stop() -def main(): +def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="ISIS Beam and MCR News Monitor") - parser.add_argument("config", type=Path, help="Path to .ini configuration file") - parser.add_argument( - "-nc", "--notify_counts", type=float, default=130, - help="Counts threshold for notification", + subparsers = parser.add_subparsers(dest="mode", required=True) + + daemon_parser = subparsers.add_parser("daemon", help="Run the long-lived daemon process") + daemon_parser.add_argument("config", type=Path, help="Path to .ini configuration file") + daemon_parser.add_argument( + "-nc", "--notify_counts", type=float, default=130, help="Counts threshold for notification" ) - parser.add_argument( - "-n", "--notify_current", + daemon_parser.add_argument( + "-n", + "--notify_current", help="Send a notification for the current news immediately.", action=argparse.BooleanOptionalAction, ) - parser.add_argument( - "-d", "--dummy", + daemon_parser.add_argument( + "-d", + "--dummy", help="Use a dummy notifier that logs to console instead of sending webhooks.", action=argparse.BooleanOptionalAction, ) - args = parser.parse_args() + + tui_parser = subparsers.add_parser("tui", help="Run the TUI client attached to daemon") + tui_parser.add_argument("config", type=Path, help="Path to .ini configuration file") + + return parser.parse_args() + + +def main(): + args = parse_args() try: config = load_config(args.config) @@ -113,29 +369,27 @@ def main(): print(f"Configuration error: {e}") raise SystemExit(1) - # Configure logging based on config - log_path = Path(config.log_file) - if not log_path.is_absolute(): - log_path = Path(__file__).parent / log_path - - numeric_level = getattr(logging, config.log_level.upper(), logging.WARNING) - - logging.basicConfig( - level=numeric_level, - format="%(asctime)s - %(levelname)s - %(message)s", - handlers=[RotatingFileHandler( - log_path, - maxBytes=config.log_max_bytes, - backupCount=config.log_backup_count - )], + configure_logging( + config.log_file, + config.log_level, + config.log_max_bytes, + config.log_backup_count, ) stop_event = asyncio.Event() try: - asyncio.run(run_all(config, args, stop_event)) + if args.mode == "daemon": + with SingleInstanceLock(Path(config.daemon_lock_file)): + asyncio.run(run_daemon(config, args, stop_event)) + elif args.mode == "tui": + asyncio.run(run_tui(config, stop_event)) + except RuntimeError as exc: + print(str(exc)) + raise SystemExit(1) except (KeyboardInterrupt, asyncio.CancelledError): print("\nStopping monitors...") + if __name__ == "__main__": main()