From 1675ff1d69b430bec7c9afa51120789b7594a01f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:30:12 +0000 Subject: [PATCH 01/11] Apply remaining changes Co-authored-by: MaxPelly <17359435+MaxPelly@users.noreply.github.com> --- isis_monitor/beam.py | 35 +++++++- isis_monitor/config.py | 38 ++++++++ isis_monitor/daemon_state.py | 166 +++++++++++++++++++++++++++++++++++ isis_monitor/ipc.py | 151 +++++++++++++++++++++++++++++++ isis_monitor/mcr.py | 34 +++++-- isis_monitor/protocols.py | 20 +++++ isis_monitor/storage.py | 111 +++++++++++++++++++++++ 7 files changed, 549 insertions(+), 6 deletions(-) create mode 100644 isis_monitor/daemon_state.py create mode 100644 isis_monitor/ipc.py create mode 100644 isis_monitor/storage.py 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..0f4d1da --- /dev/null +++ b/isis_monitor/daemon_state.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import asyncio +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) -> 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 + 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) 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..9f0b593 100644 --- a/isis_monitor/mcr.py +++ b/isis_monitor/mcr.py @@ -7,7 +7,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 +22,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 +59,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 +75,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,12 +86,16 @@ 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 - ) - await asyncio.sleep(sleep_secs) + sleep_secs = self.config.mcr_poll_interval * min(2 ** consecutive_failures, 8) + await asyncio.wait_for(self._force_reconnect.wait(), timeout=sleep_secs) + self._force_reconnect.clear() + consecutive_failures = 0 + if self.sink: + self.sink.update_health("mcr", "reconnecting") except asyncio.CancelledError: return + except asyncio.TimeoutError: + pass if stop_event and stop_event.is_set(): return @@ -96,13 +107,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() From c5f630b22e645cd146579e8464c0300ebd991db8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 4 Sep 2026 14:53:58 +0000 Subject: [PATCH 02/11] Implement daemon/TUI split with IPC and SQLite retention Co-authored-by: MaxPelly <17359435+MaxPelly@users.noreply.github.com> --- README.md | 60 ++++- config.ini.example | 19 ++ isis_monitor/__init__.py | 12 +- isis_monitor/daemon_state.py | 64 ++++- isis_monitor/mcr.py | 25 +- isis_monitor/tests/test_config.py | 54 +++++ isis_monitor/tests/test_ipc.py | 64 +++++ isis_monitor/tests/test_main.py | 26 +- isis_monitor/tests/test_storage.py | 45 ++++ isis_monitor/tui.py | 33 ++- main.py | 378 ++++++++++++++++++++++++----- 11 files changed, 681 insertions(+), 99 deletions(-) create mode 100644 isis_monitor/tests/test_ipc.py create mode 100644 isis_monitor/tests/test_storage.py 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/daemon_state.py b/isis_monitor/daemon_state.py index 0f4d1da..019f6c2 100644 --- a/isis_monitor/daemon_state.py +++ b/isis_monitor/daemon_state.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import json from collections import deque from dataclasses import dataclass from datetime import datetime, timedelta, timezone @@ -76,22 +77,30 @@ def update_beam_state(self, beam: str, current: float, power: str) -> None: 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) -> None: + 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 - self._publish( - "sample", - { - "beam": beam, - "timestamp": ts.isoformat(), - "current": current, - "power": power, - }, - ) + 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: @@ -164,3 +173,38 @@ def sample_all_currents(self, ts: Optional[datetime] = None) -> None: 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/mcr.py b/isis_monitor/mcr.py index 9f0b593..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 @@ -87,15 +88,25 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): while stop_event is None or not stop_event.is_set(): try: sleep_secs = self.config.mcr_poll_interval * min(2 ** consecutive_failures, 8) - await asyncio.wait_for(self._force_reconnect.wait(), timeout=sleep_secs) - self._force_reconnect.clear() - consecutive_failures = 0 - if self.sink: - self.sink.update_health("mcr", "reconnecting") + 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, + ) + 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 - except asyncio.TimeoutError: - pass if stop_event and stop_event.is_set(): return 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() From 7a996c30d700217d0cdb04dc6ac0866bb88ee0e5 Mon Sep 17 00:00:00 2001 From: Max <17359435+MaxPelly@users.noreply.github.com> Date: Sun, 6 Sep 2026 10:03:07 +0100 Subject: [PATCH 03/11] antigravity review --- codebase_documentation.md | 44 +++++---- config.ini.example | 3 + isis_monitor/beam.py | 9 +- isis_monitor/daemon_state.py | 64 +++++++------ isis_monitor/ipc.py | 29 +++++- isis_monitor/mcr.py | 3 +- isis_monitor/notifiers.py | 2 +- isis_monitor/tests/test_daemon_state.py | 78 +++++++++++++++ isis_monitor/tests/test_ipc.py | 39 ++++++++ isis_monitor/tests/test_main.py | 43 ++++++++- isis_monitor/tui.py | 30 +++--- main.py | 121 +++++++++++++++++------- 12 files changed, 360 insertions(+), 105 deletions(-) create mode 100644 isis_monitor/tests/test_daemon_state.py diff --git a/codebase_documentation.md b/codebase_documentation.md index 2cd1e39..b1f1e39 100644 --- a/codebase_documentation.md +++ b/codebase_documentation.md @@ -7,14 +7,11 @@ This document provides a technical overview of the ISIS Beam Monitor codebase, i The ISIS Beam Monitor is a real-time monitoring system designed to track accelerator beam status and MCR (Main Control Room) news updates at the ISIS Neutron and Muon Source. It follows a decoupled, asynchronous architecture using Python's `asyncio` for concurrent operations. ### High-Level Design -The system consists of three main parts: -1. **Monitors**: Asynchronous tasks that fetch and process data from external sources (WebSockets for beam data, HTTP polling for MCR news). -2. **Notifiers**: Flexible channels for broadcasting alerts to external services like Microsoft Teams or local logs. -3. **TUI (Terminal User Interface)**: A rich, real-time display built with the `rich` library, providing visual feedback and status summaries. - -### Data Flow -- **Beam Data**: Subscribes to PV (Process Variable) updates via a WebSocket. Updates are dispatched to the TUI and broadcast to notification channels if threshold boundaries are crossed or run states change. -- **MCR News**: Periodically polls an external URL. If new text is detected, it updates the TUI and broadcasts the news to the MCR notification channel. +The system uses a two-tier architecture (daemon and client) communicating via local UNIX domain sockets: +1. **Daemon**: A long-lived background process holding the master `DaemonState` (in `daemon_state.py`). It orchestrates monitors, persists state to a local SQLite database (`storage.py`), and serves multiple clients via JSON over IPC (`ipc.py`). +2. **Monitors**: Asynchronous tasks that fetch and process data from external sources (WebSockets for beam data, HTTP polling for MCR news). They feed data into the `DaemonState` via `MonitorSinkProtocol`. +3. **Notifiers**: Flexible channels for broadcasting alerts to external services like Microsoft Teams or local logs. +4. **TUI Client**: A terminal UI built with the `rich` library. It acts as an IPC client, fetching the initial state snapshot from the daemon and then subscribing to a real-time event stream to update its display. --- @@ -43,8 +40,18 @@ The live terminal interface. - **Sparklines**: Visualizes historical beam current data using Unicode block characters, normalized against the rolling buffer's range. - **Sampler**: An independent coroutine that snapshots state at fixed intervals to ensure consistent graph pacing. +### `isis_monitor/daemon_state.py` & `storage.py` +The core state management and persistence layer. +- **`DaemonState`**: A thread-safe, lock-protected singleton holding current beam statuses, historical data buffers, MCR news, and health checks. It manages a pub/sub queue system for IPC clients. +- **`SQLiteStateStore`**: Handles synchronizing the daemon's state to disk, enabling crash recovery and historical lookups. + +### `isis_monitor/ipc.py` +Manages local communication between the daemon and clients. +- **`IPCServer`**: A UNIX domain socket server that handles requests (like fetching a state snapshot or history) and multiplexes event streams to subscribed clients using a newline-delimited JSON protocol. +- **`IPCClient`**: A resilient async client that manages connection state and reconnection backoff. + ### `isis_monitor/protocols.py` -Defines the `TUIProtocol`, allowing the monitors to interact with any TUI implementation (or a mock during testing) without being coupled to the `rich` implementation. +Defines runtime-checkable protocols (e.g., `MonitorSinkProtocol`, `TUIProtocol`) allowing monitors to interact with the daemon or the TUI interchangeably during testing. --- @@ -52,8 +59,9 @@ Defines the `TUIProtocol`, allowing the monitors to interact with any TUI implem Configuration is managed via `config.ini` files, loaded through `isis_monitor/config.py`. Key sections include: - **`[DATA]`**: WebSocket and HTTP URLs for data sources. -- **`[WEBHOOKS]`**: URLs for Teams integration. -- **`[BEAM_BOUNDARIES]`**: Thresholds for power level classification (Off/Low/Medium/High). +- **`[WEBHOOKS]`**: URLs for Teams integration (should be kept secure). +- **`[DAEMON]`** / **`[TUI_CLIENT]`**: Paths for UNIX sockets, SQLite database, and retention settings. +- **`[BEAM_BOUNDARIES]`**: Thresholds for power level classification. - **`[TUI]`**: Display settings like history length and refresh rates. --- @@ -65,11 +73,11 @@ The TUI is built using `rich.layout.Layout`. You can adjust the proportions and ### Adjusting Section Sizes In `RichTUI._make_layout()`, sections are defined using `split_column` and `split_row`. - **Fixed Height**: Use the `size` argument (e.g., `Layout(name="header", size=3)`) to set a fixed number of rows. -- **Proportional Width/Height**: Use the `ratio` argument (e.g., `Layout(name="left", ratio=1)`) to make a section take up a proportion of the available space relative to its siblings. +- **Proportional Width/Height**: Use the `ratio` argument (e.g., `Layout(name="left", ratio=1)`) to make a section take up a proportion of the available space. ### Column Widths & Internal Padding - **Table Columns**: The beam status table in `_update_beam_panel()` uses `expand=True`. To adjust individual column behaviors, modify the `table.add_column()` calls. -- **Graph Width**: If you significantly change the width of the "left" column, you may need to update `SPARK_WIDTH` in `_update_beam_graph()` to ensure the sparklines fit correctly or fill the space. +- **Graph Width**: The TUI automatically scales sparklines using `shutil.get_terminal_size()`, but you can override `SPARK_WIDTH` in `_update_beam_graph()` if you want a fixed size. --- @@ -78,14 +86,14 @@ In `RichTUI._make_layout()`, sections are defined using `split_column` and `spli ### Technical Debt & Improvements - **Error Handling**: Enhance WebSocket reconnection logic with more granular error classification (e.g., distinguishing network errors from authentication issues). - **Testing**: Expand unit tests for `tui.py` and `main.py`. Currently, core logic is well-tested, but UI rendering and orchestration could benefit from more coverage. -- **Performance**: For very large numbers of beam targets, consider moving TUI rendering to a separate thread to avoid blocking the `asyncio` event loop, though current loads are well within limits. +- **Performance**: If the SQLite persistence overhead grows, consider migrating `storage.py` to use `aiosqlite` for native async database access instead of `asyncio.to_thread`. ### Potential Features -- **Historical Logging**: Persist beam data to a local database (e.g., SQLite) for post-run analysis. +- **Prometheus Exporter**: Add a lightweight HTTP endpoint to export beam metrics and health status for ingestion by Prometheus/Grafana. - **Interactive TUI**: Add keyboard shortcuts to the TUI to toggle specific notification channels or change view modes. - **Multiple Notifiers**: Add support for Email, Slack, or SMS notifiers by implementing the `Notifier` interface. ### Best Practices for Extension -1. **Follow the Protocols**: Always use `isis_monitor.protocols` when adding new UI elements to keep monitors decoupled. -2. **Async/Await**: Ensure all blocking I/O (like networking) is handled asynchronously to prevent freezing the TUI. -3. **State Safety**: Always use the `self._lock` when modifying `RichTUI` state to prevent race conditions during rendering. +1. **Follow the Protocols**: Always use `isis_monitor.protocols` when adding new sinks to keep monitors decoupled. +2. **Async/Await**: Ensure all blocking I/O (like networking or DB access) is handled asynchronously (or wrapped in `to_thread`) to prevent freezing the TUI or Daemon. +3. **State Safety**: Always use `self._lock` when modifying `DaemonState` or `RichTUI` state to prevent race conditions. diff --git a/config.ini.example b/config.ini.example index a894158..ff0b8f5 100644 --- a/config.ini.example +++ b/config.ini.example @@ -3,6 +3,9 @@ mcr_news_url = https://www.isis.stfc.ac.uk/gallery/beam-status/mcrnews.txt isis_websocket_url = wss://ndaextweb4.nd.rl.ac.uk/pvws/pv [WEBHOOKS] +# Webhook URLs should be kept secret. Consider restricting permissions on this file (e.g. chmod 600) +# to prevent other users from reading it. +enable_teams = false news_teams_url = beam_teams_url = experiment_teams_url = diff --git a/isis_monitor/beam.py b/isis_monitor/beam.py index 2edfd30..9578214 100644 --- a/isis_monitor/beam.py +++ b/isis_monitor/beam.py @@ -190,8 +190,9 @@ 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()) + ws = self._current_ws + if ws is not None: + asyncio.create_task(ws.close()) return True async def run(self, stop_event: Optional[asyncio.Event] = None): @@ -228,8 +229,8 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): data = json.loads(raw_msg) if data.get("type") == "update": await self._handle_update(data) - except json.JSONDecodeError: - pass + except json.JSONDecodeError as e: + logger.debug(f"Failed to decode WS message: {e}") except asyncio.CancelledError: return diff --git a/isis_monitor/daemon_state.py b/isis_monitor/daemon_state.py index 019f6c2..00d76f0 100644 --- a/isis_monitor/daemon_state.py +++ b/isis_monitor/daemon_state.py @@ -2,9 +2,12 @@ import asyncio import json +import logging from collections import deque from dataclasses import dataclass from datetime import datetime, timedelta, timezone + +logger = logging.getLogger(__name__) from threading import RLock from typing import Deque, Dict, List, Optional, Tuple @@ -58,6 +61,7 @@ def _publish(self, event: str, payload: dict) -> None: try: q.put_nowait(DaemonEvent(event=event, payload=payload)) except asyncio.QueueFull: + logger.warning("Subscriber queue full; dropping subscriber") dead.append(q) for q in dead: if q in self._subscribers: @@ -67,7 +71,7 @@ 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}) + self._publish("log", {"message": message}) def update_beam_state(self, beam: str, current: float, power: str) -> None: with self._lock: @@ -75,7 +79,7 @@ def update_beam_state(self, beam: str, current: float, power: str) -> None: 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}) + self._publish("beam", {"beam": beam, "current": current, "power": power}) def append_beam_sample( self, @@ -91,16 +95,16 @@ def append_beam_sample( 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, - }, - ) + 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: @@ -115,29 +119,40 @@ 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}) + 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}) + 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}) + 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}) + self._publish("health", {"component": component, "status": status}) def snapshot(self) -> dict: with self._lock: - history_json = { + return { + "last_update": self.last_update.isoformat(), + "beam_states": dict(self.beam_states), + "mcr_news": self.mcr_news, + "run_name": self.run_name, + "current_counts": self.current_counts, + "health": dict(self.health), + } + + def get_history_snapshot(self) -> dict: + with self._lock: + return { beam: [ { "timestamp": ts.isoformat(), @@ -148,16 +163,10 @@ def snapshot(self) -> dict: ] 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 get_logs_snapshot(self) -> list: + with self._lock: + return list(self.logs) def sample_all_currents(self, ts: Optional[datetime] = None) -> None: ts = ts or datetime.now(timezone.utc) @@ -192,6 +201,7 @@ def restore_from_snapshot_json(self, raw: Optional[str]) -> None: try: snap = json.loads(raw) except json.JSONDecodeError: + logger.warning("Corrupt daemon state snapshot, starting fresh.") return with self._lock: beam_states = snap.get("beam_states", {}) diff --git a/isis_monitor/ipc.py b/isis_monitor/ipc.py index 0913473..b7913a2 100644 --- a/isis_monitor/ipc.py +++ b/isis_monitor/ipc.py @@ -3,6 +3,7 @@ import asyncio import contextlib import json +import os from pathlib import Path from typing import Awaitable, Callable, Optional @@ -27,7 +28,8 @@ 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)) + self.server = await asyncio.start_unix_server(self._handle_client, path=str(self.socket_path), limit=65536) + os.chmod(self.socket_path, 0o600) async def stop(self) -> None: if self.server is not None: @@ -67,6 +69,24 @@ async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.Str "snapshot": self.state.snapshot(), }, ) + elif method == "get_history": + await self._send( + writer, + { + "ok": True, + "version": PROTOCOL_VERSION, + "history": self.state.get_history_snapshot(), + }, + ) + elif method == "get_logs": + await self._send( + writer, + { + "ok": True, + "version": PROTOCOL_VERSION, + "logs": self.state.get_logs_snapshot(), + }, + ) elif method == "subscribe_updates": if queue is None: queue = self.state.subscribe() @@ -112,7 +132,10 @@ async def _forward_events(self, queue: asyncio.Queue, writer: asyncio.StreamWrit "event": ev.event, "payload": ev.payload, } - await self._send(writer, payload) + try: + await self._send(writer, payload) + except (ConnectionError, BrokenPipeError, OSError): + break class IPCClient: @@ -122,7 +145,7 @@ def __init__(self, socket_path: Path): self.writer: Optional[asyncio.StreamWriter] = None async def connect(self) -> None: - self.reader, self.writer = await asyncio.open_unix_connection(str(self.socket_path)) + self.reader, self.writer = await asyncio.open_unix_connection(str(self.socket_path), limit=65536) async def close(self) -> None: if self.writer: diff --git a/isis_monitor/mcr.py b/isis_monitor/mcr.py index 0100c90..f36b588 100644 --- a/isis_monitor/mcr.py +++ b/isis_monitor/mcr.py @@ -131,9 +131,10 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): consecutive_failures += 1 if self.sink: self.sink.update_health("mcr", "error") + next_retry = self.config.mcr_poll_interval * min(2 ** consecutive_failures, 8) logger.debug( f"MCR fetch failed (attempt {consecutive_failures}); " - f"next retry in {sleep_secs * min(2, 8):.0f}s." + f"next retry in {next_retry:.0f}s." ) def request_reconnect(self) -> bool: diff --git a/isis_monitor/notifiers.py b/isis_monitor/notifiers.py index e2381ed..67ec3a7 100644 --- a/isis_monitor/notifiers.py +++ b/isis_monitor/notifiers.py @@ -99,4 +99,4 @@ async def broadcast(self, message: str, channel: Optional[str] = None): f"Channel '{self.name}' has no notifiers configured; skipping broadcast." ) return - await asyncio.gather(*(n.send(message, channel) for n in self.notifiers)) + await asyncio.gather(*(n.send(message, channel) for n in self.notifiers), return_exceptions=True) diff --git a/isis_monitor/tests/test_daemon_state.py b/isis_monitor/tests/test_daemon_state.py new file mode 100644 index 0000000..eae7e5c --- /dev/null +++ b/isis_monitor/tests/test_daemon_state.py @@ -0,0 +1,78 @@ +import asyncio +import json +from datetime import datetime, timezone +import pytest + +from isis_monitor.daemon_state import DaemonState, DaemonEvent + +def test_daemon_state_snapshot(): + state = DaemonState() + state.update_beam_state("TS1", 45.0, "medium") + state.update_mcr_news("Breaking News") + state.update_health("daemon", "running") + + snap = state.snapshot() + assert snap["mcr_news"] == "Breaking News" + assert snap["beam_states"]["TS1"]["current"] == 45.0 + assert snap["health"]["daemon"] == "running" + + # Check that history and logs are NOT in snapshot + assert "history" not in snap + assert "logs" not in snap + +def test_daemon_state_get_history_and_logs(): + state = DaemonState() + ts = datetime.now(timezone.utc) + state.append_beam_sample("TS1", 10.0, "low", ts=ts) + state.update_log("Log entry 1") + + history = state.get_history_snapshot() + assert len(history["TS1"]) == 1 + assert history["TS1"][0]["current"] == 10.0 + + logs = state.get_logs_snapshot() + assert len(logs) == 1 + assert logs[0] == "Log entry 1" + +def test_daemon_state_pubsub(): + state = DaemonState() + queue = state.subscribe() + + state.update_beam_state("TS2", 100.0, "high") + + event = queue.get_nowait() + assert event.event == "beam" + assert event.payload["beam"] == "TS2" + + state.unsubscribe(queue) + state.update_log("Log entry") + + assert queue.empty() + +def test_daemon_state_subscriber_drop(caplog): + state = DaemonState() + queue = state.subscribe() + + # Fill queue past its maxsize (usually 500) + for i in range(501): + state.update_log(f"Spam {i}") + + assert "Subscriber queue full; dropping subscriber" in caplog.text + # Should be unsubscribed automatically + assert queue not in state._subscribers + +def test_restore_from_snapshot(): + state = DaemonState() + + valid_json = json.dumps({ + "mcr_news": "Restored News", + "beam_states": {"Muons": {"current": 2.0, "power": "low"}} + }) + state.restore_from_snapshot_json(valid_json) + assert state.mcr_news == "Restored News" + assert state.beam_states["Muons"]["current"] == 2.0 + + # Corrupt JSON shouldn't crash + state.restore_from_snapshot_json("{bad_json: True") + # State should remain intact + assert state.mcr_news == "Restored News" diff --git a/isis_monitor/tests/test_ipc.py b/isis_monitor/tests/test_ipc.py index eb636c6..4773461 100644 --- a/isis_monitor/tests/test_ipc.py +++ b/isis_monitor/tests/test_ipc.py @@ -62,3 +62,42 @@ async def command_handler(_name: str): await client.close() await server.stop() + +@pytest.mark.asyncio +async def test_ipc_malformed_json_and_oversized_payload(tmp_path): + socket_path = tmp_path / "daemon.sock" + state = DaemonState() + + async def command_handler(_name: str): + return {} + + server = IPCServer(socket_path, state, command_handler) + await server.start() + + # Manual socket connection to send raw bad bytes + reader, writer = await asyncio.open_unix_connection(str(socket_path)) + + # 1. Malformed JSON + writer.write(b"{bad_json\n") + await writer.drain() + + resp_line = await reader.readline() + resp = __import__("json").loads(resp_line.decode()) + assert resp["ok"] is False + assert resp["error"] == "invalid_json" + + # 2. Oversized payload (limit is 65536) + large_payload = b"{" + b'"padding": "' + b'A' * 70000 + b'"}\n' + writer.write(large_payload) + await writer.drain() + + # The server should drop the connection due to ValueError from limit + # or just close. Wait to see it drops. + try: + await asyncio.wait_for(reader.readline(), timeout=1.0) + except (asyncio.IncompleteReadError, ConnectionResetError, asyncio.TimeoutError): + pass # Expected + + writer.close() + await writer.wait_closed() + await server.stop() diff --git a/isis_monitor/tests/test_main.py b/isis_monitor/tests/test_main.py index 1292d3c..c1a9679 100644 --- a/isis_monitor/tests/test_main.py +++ b/isis_monitor/tests/test_main.py @@ -1,7 +1,13 @@ import logging import pytest from unittest.mock import MagicMock -from main import StateLogHandler +import os +import signal +import asyncio +from pathlib import Path +from isis_monitor.daemon_state import DaemonState +from isis_monitor.tui import RichTUI +from main import StateLogHandler, SingleInstanceLock, _apply_snapshot_to_tui class TestStateLogHandler: @@ -44,3 +50,38 @@ def test_emit_with_warning_level(self): ) handler.emit(record) mock_state.update_log.assert_called_once_with("WARNING - something went wrong") + +def test_single_instance_lock_success(tmp_path): + import os + from main import SingleInstanceLock + lock_file = tmp_path / "test.lock" + with SingleInstanceLock(lock_file) as lock: + assert lock_file.exists() + assert lock_file.read_text().strip() == str(os.getpid()) + assert not lock_file.exists() + +def test_single_instance_lock_failure(tmp_path): + import os + from main import SingleInstanceLock + lock_file = tmp_path / "test.lock" + lock_file.write_text("999999") + with pytest.raises(RuntimeError, match="Lock file already held|Lock held by"): + with pytest.MonkeyPatch.context() as m: + m.setattr(os, "kill", lambda pid, sig: None) + with SingleInstanceLock(lock_file): + pass + +def test_apply_snapshot_to_tui(): + from main import _apply_snapshot_to_tui + from isis_monitor.tui import RichTUI + tui = RichTUI(60, 60, 4, 50) + snap = { + "beam_states": { + "TS1": {"current": 42.0, "power": "high"} + }, + "mcr_news": "Test news" + } + _apply_snapshot_to_tui(tui, snap) + assert tui.mcr_news == "Test news" + assert "TS1" in tui.beam_states + assert tui.beam_states["TS1"]["current"] == 42.0 diff --git a/isis_monitor/tui.py b/isis_monitor/tui.py index 9905398..acedce8 100644 --- a/isis_monitor/tui.py +++ b/isis_monitor/tui.py @@ -1,4 +1,5 @@ import asyncio +import shutil from collections import deque from datetime import datetime, timezone from threading import RLock @@ -11,17 +12,16 @@ from rich.table import Table +_STATE_COLOURS = { + "off": "red", + "unknown": "red", + "low": "orange", + "medium": "yellow", + "high": "green" +} + def _get_state_colour(state): - colour = "purple" - if state in ("off", "unknown"): - colour = "red" - elif state == "low": - colour = "orange" - elif state == "medium": - colour = "yellow" - elif state == "high": - colour = "green" - return colour + return _STATE_COLOURS.get(state, "purple") # Eight Unicode block heights, index 0 = shortest _BLOCKS = " ▁▂▃▄▅▆▇█" @@ -40,10 +40,9 @@ def _render_sparkline( text.append(" " * width) return text - # Extract just the values to calculate our dynamic range - values = [v for v, _ in history_data] tail = history_data[-width:] + values = [v for v, _ in tail] min_val = min(values) max_val = max(values) span = max_val - min_val @@ -203,7 +202,7 @@ def update_log(self, message: str): # ------------------------------------------------------------------ def _update_all(self): - """Force-refresh every panel. Called once on startup (lock not held).""" + """Force-refresh every panel. Note: this method may be called while the lock is already held, which is safe due to RLock.""" with self._lock: self.layout["header"].update( Panel( @@ -247,8 +246,9 @@ def _update_beam_panel(self): def _update_beam_graph(self): """Render the rolling sparkline graph into beam_graph.""" - # Approximate usable width: panel width minus borders/label prefix. - SPARK_WIDTH = 58 + # Approximate usable width: terminal width minus half for layout split, minus borders/padding and label. + term_width = shutil.get_terminal_size((120, 24)).columns + SPARK_WIDTH = max(10, (term_width // 2) - 30) LABEL_W = 7 # "Muons: " is 7 chars content = Text() diff --git a/main.py b/main.py index 4a4833f..ac109e7 100755 --- a/main.py +++ b/main.py @@ -44,11 +44,33 @@ def __init__(self, path: Path): self._fh = None def __enter__(self): - self._fh = self.path.open("w") + self._fh = self.path.open("a+") + self._fh.seek(0) + pid_str = self._fh.read().strip() + if pid_str.isdigit(): + try: + os.kill(int(pid_str), 0) + except ProcessLookupError: + pass + except PermissionError: + raise RuntimeError(f"Lock held by another user's process: {pid_str}") + else: + # If flock is supported, let flock do its job. But if not, we rely on this check. + pass + 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 + except OSError: + # Fallback for systems without flock: rely on the PID check + if pid_str.isdigit(): + try: + os.kill(int(pid_str), 0) + raise RuntimeError(f"Lock file already held by PID {pid_str}") + except ProcessLookupError: + pass + self._fh.seek(0) self._fh.truncate(0) self._fh.write(str(os.getpid())) @@ -85,8 +107,6 @@ def install_signal_handlers(stop_event: asyncio.Event) -> None: 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) @@ -138,16 +158,22 @@ async def state_persistence_loop(config, state: DaemonState, store: SQLiteStateS ts = datetime.now(timezone.utc) state.sample_all_currents(ts) - store.write_samples(state.get_beam_rows_for_timestamp(ts)) - + + beam_rows = 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) + snap = json.dumps(state.snapshot()) + health = state.get_health() + + def _persist(): + store.write_samples(beam_rows) + store.prune_older_than(cutoff) + store.upsert_snapshot("daemon_state", snap) + for component, status in health.items(): + store.upsert_health(component, status) + store.commit() - store.upsert_snapshot("daemon_state", json.dumps(state.snapshot())) - for component, status in state.get_health().items(): - store.upsert_health(component, status) - store.commit() + await asyncio.to_thread(_persist) async def daemon_heartbeat_loop(config, state: DaemonState, stop_event: asyncio.Event): @@ -165,10 +191,17 @@ async def run_daemon(config, args, stop_event: asyncio.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): + def _init_db(): + store = SQLiteStateStore(Path(config.daemon_db_path)) + raw_snap = store.load_snapshot("daemon_state") + cutoff = state.cutoff_for_days(config.retention_days) + recent = store.load_recent_samples(cutoff) + return store, raw_snap, recent + + store, raw_snap, recent_samples = await asyncio.to_thread(_init_db) + + state.restore_from_snapshot_json(raw_snap) + for row in recent_samples: state.append_beam_sample( beam=str(row["target"]), current=float(row["current"]), @@ -221,24 +254,28 @@ async def command_handler(name: str) -> dict: ) finally: state.update_health("daemon", "stopping") - store.upsert_snapshot("daemon_state", json.dumps(state.snapshot())) - store.commit() - store.close() + snap = json.dumps(state.snapshot()) + + def _close_db(): + store.upsert_snapshot("daemon_state", snap) + store.commit() + store.close() + + await asyncio.to_thread(_close_db) await ipc_server.stop() await close_channels(beam_channel, exp_channel, mcr_channel) +import sys + 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: @@ -267,15 +304,18 @@ def _apply_event_to_tui(tui: RichTUI, message: dict) -> None: 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')}") +def tui_command_handler(client: IPCClient, stop_event: asyncio.Event, tui: RichTUI): + cmd = sys.stdin.readline().strip().lower() + if cmd == "q": + stop_event.set() + elif cmd == "r": + async def _send_reconnect(): + try: + response = await client.request({"method": "command", "name": "force_reconnect_all"}) + tui.update_log(f"Reconnect request result: {response.get('result')}") + except Exception as e: + tui.update_log(f"Reconnect request failed: {e}") + asyncio.create_task(_send_reconnect()) async def run_tui(config, stop_event: asyncio.Event): @@ -288,12 +328,13 @@ async def run_tui(config, stop_event: asyncio.Event): logs_maxlen=config.logs_maxlen, ) tui.start() + loop = asyncio.get_running_loop() 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 + has_reader = False try: tui.update_connection_state("connecting") await client.connect() @@ -302,14 +343,26 @@ async def run_tui(config, stop_event: asyncio.Event): snapshot_resp = await client.request({"method": "get_snapshot"}) if snapshot_resp.get("ok"): _apply_snapshot_to_tui(tui, snapshot_resp.get("snapshot", {})) + + history_resp = await client.request({"method": "get_history"}) + if history_resp.get("ok"): + tui.set_history_snapshot(history_resp.get("history", {})) + + logs_resp = await client.request({"method": "get_logs"}) + if logs_resp.get("ok"): + for line in logs_resp.get("logs", [])[-20:]: + tui.update_log(str(line)) 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)) + # Reset backoff only after successful sync backoff = config.tui_reconnect_initial + loop.add_reader(sys.stdin.fileno(), tui_command_handler, client, stop_event, tui) + has_reader = True + async for message in client.iter_events(): _apply_event_to_tui(tui, message) if stop_event.is_set(): @@ -323,10 +376,8 @@ async def run_tui(config, stop_event: asyncio.Event): 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 + if has_reader: + loop.remove_reader(sys.stdin.fileno()) await client.close() finally: tui.stop() From e900f0d23c3f22d4b4af4922c24e3b186c52f9db Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 6 Sep 2026 09:05:54 +0000 Subject: [PATCH 04/11] Apply remaining changes Co-authored-by: MaxPelly <17359435+MaxPelly@users.noreply.github.com> --- isis_monitor/tests/test_main.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/isis_monitor/tests/test_main.py b/isis_monitor/tests/test_main.py index c1a9679..14d6346 100644 --- a/isis_monitor/tests/test_main.py +++ b/isis_monitor/tests/test_main.py @@ -4,6 +4,7 @@ import os import signal import asyncio +import fcntl from pathlib import Path from isis_monitor.daemon_state import DaemonState from isis_monitor.tui import RichTUI @@ -61,15 +62,18 @@ def test_single_instance_lock_success(tmp_path): assert not lock_file.exists() def test_single_instance_lock_failure(tmp_path): - import os from main import SingleInstanceLock lock_file = tmp_path / "test.lock" - lock_file.write_text("999999") - with pytest.raises(RuntimeError, match="Lock file already held|Lock held by"): - with pytest.MonkeyPatch.context() as m: - m.setattr(os, "kill", lambda pid, sig: None) + lock_file.write_text(str(os.getpid())) + blocker = lock_file.open("a+") + fcntl.flock(blocker.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + try: + with pytest.raises(RuntimeError, match="Lock file already held|Lock held by"): with SingleInstanceLock(lock_file): pass + finally: + fcntl.flock(blocker.fileno(), fcntl.LOCK_UN) + blocker.close() def test_apply_snapshot_to_tui(): from main import _apply_snapshot_to_tui From 501256d2e06d839602575e790a64bd09402a4fc0 Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 6 Sep 2026 11:52:54 +0100 Subject: [PATCH 05/11] Fix quitting --- .gitignore | 1 + isis_monitor/beam.py | 61 +++++++++++++++++++++++++++++++++++------- isis_monitor/config.py | 2 +- isis_monitor/mcr.py | 4 +++ main.py | 37 +++++++++++++++++++++++++ 5 files changed, 94 insertions(+), 11 deletions(-) diff --git a/.gitignore b/.gitignore index 5bd786c..407e556 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +*.db config.ini test.py diff --git a/isis_monitor/beam.py b/isis_monitor/beam.py index 9578214..b2d3ef4 100644 --- a/isis_monitor/beam.py +++ b/isis_monitor/beam.py @@ -216,26 +216,64 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): 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") + while True: + recv_task = asyncio.create_task(ws.recv()) + reconnect_task = asyncio.create_task(self._force_reconnect.wait()) + + tasks = {recv_task, reconnect_task} + stop_task = None + + if stop_event is not None: + stop_task = asyncio.create_task(stop_event.wait()) + tasks.add(stop_task) + + try: + done, pending = await asyncio.wait( + tasks, + return_when=asyncio.FIRST_COMPLETED, + ) + + # Prioritize shutdown/reconnection if multiple tasks finish together. + if stop_task is not None and stop_task in done: + logger.warning("Deep Beam Loop Quit") + return + + if reconnect_task in done: + self._force_reconnect.clear() + logger.info("Beam reconnect requested by operator.") + + if self.sink: + self.sink.update_health("beam", "reconnecting") + + break + + # recv_task completed. + raw_msg = recv_task.result() + + except ConnectionClosedOK: break + + finally: + # Never leave recv/event tasks running into the next iteration. + for task in tasks: + if not task.done(): + task.cancel() + + await asyncio.gather(*tasks, return_exceptions=True) + try: data = json.loads(raw_msg) if data.get("type") == "update": await self._handle_update(data) - except json.JSONDecodeError as e: - logger.debug(f"Failed to decode WS message: {e}") + except json.JSONDecodeError as exc: + logger.debug("Failed to decode WS message: %s", exc) except asyncio.CancelledError: + logger.warning(f"Beam Loop Cancelled") return except (websockets.exceptions.ConnectionClosed, OSError): if stop_event and stop_event.is_set(): + logger.warning(f"Beam Loop Quit") return if self.sink: self.sink.update_health("beam", "disconnected") @@ -243,6 +281,7 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): await asyncio.sleep(self.config.beam_reconnect_interval) except Exception as e: if stop_event and stop_event.is_set(): + logger.warning(f"Error Beam Loop Quit") return if self.sink: self.sink.update_health("beam", "error") @@ -250,3 +289,5 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): await asyncio.sleep(self.config.beam_reconnect_interval) finally: self._current_ws = None + logger.warning(f"Fallthrough Beam Loop Quit") + return diff --git a/isis_monitor/config.py b/isis_monitor/config.py index a1c10cf..dc81fba 100644 --- a/isis_monitor/config.py +++ b/isis_monitor/config.py @@ -35,7 +35,7 @@ class AppConfig: muon_boundaries: tuple = (0.0, 2.0, 5.0) # TIMEOUTS_INTERVALS - mcr_poll_interval: float = 60.0 + mcr_poll_interval: float = 30.0 beam_reconnect_interval: float = 5.0 webhook_timeout: float = 10.0 diff --git a/isis_monitor/mcr.py b/isis_monitor/mcr.py index f36b588..3d94e36 100644 --- a/isis_monitor/mcr.py +++ b/isis_monitor/mcr.py @@ -106,9 +106,11 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): if self.sink: self.sink.update_health("mcr", "reconnecting") except asyncio.CancelledError: + logger.warning(f"MCR News Collection Cancelled") return if stop_event and stop_event.is_set(): + logger.warning(f"MCR News Collection Quit") return new_news = await self.get_news(session) @@ -136,6 +138,8 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): f"MCR fetch failed (attempt {consecutive_failures}); " f"next retry in {next_retry:.0f}s." ) + logger.warning(f"MCR News Collection Fall Through") + return def request_reconnect(self) -> bool: if self._force_reconnect.is_set(): diff --git a/main.py b/main.py index ac109e7..a410c98 100755 --- a/main.py +++ b/main.py @@ -174,6 +174,8 @@ def _persist(): store.commit() await asyncio.to_thread(_persist) + logger.warning("State Persistance quit") + return async def daemon_heartbeat_loop(config, state: DaemonState, stop_event: asyncio.Event): @@ -183,6 +185,8 @@ async def daemon_heartbeat_loop(config, state: DaemonState, stop_event: asyncio. await asyncio.wait_for(stop_event.wait(), timeout=config.heartbeat_interval) except asyncio.TimeoutError: continue + logger.warning("Heartbeat quit") + return async def run_daemon(config, args, stop_event: asyncio.Event): @@ -239,6 +243,9 @@ async def command_handler(name: str) -> dict: return {"beam": beam_monitor.request_reconnect()} if name == "force_reconnect_mcr": return {"mcr": mcr_monitor.request_reconnect()} + if name == "shutdown": + stop_event.set() + return {"shutdown": "ok"} return {"error": "unknown_command", "name": name} ipc_server = IPCServer(Path(config.daemon_socket_path), state, command_handler) @@ -253,6 +260,7 @@ async def command_handler(name: str) -> dict: daemon_heartbeat_loop(config, state, stop_event), ) finally: + logger.warning("Shutting down daemon") state.update_health("daemon", "stopping") snap = json.dumps(state.snapshot()) @@ -383,6 +391,30 @@ async def run_tui(config, stop_event: asyncio.Event): tui.stop() +async def run_stop(config) -> None: + """Connect to a running daemon via IPC and request a clean shutdown.""" + client = IPCClient(Path(config.daemon_socket_path)) + try: + await client.connect() + except (FileNotFoundError, ConnectionRefusedError, OSError) as exc: + print(f"Could not connect to daemon at {config.daemon_socket_path}: {exc}") + raise SystemExit(1) + + try: + response = await client.request({"method": "command", "name": "shutdown"}) + if response.get("ok"): + result = response.get("result", {}) + if result.get("shutdown") == "ok": + print("Shutdown signal sent — daemon is stopping cleanly.") + else: + print(f"Daemon responded: {result}") + else: + print(f"Daemon returned an error: {response.get('error')}") + raise SystemExit(1) + finally: + await client.close() + + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="ISIS Beam and MCR News Monitor") subparsers = parser.add_subparsers(dest="mode", required=True) @@ -408,6 +440,9 @@ def parse_args() -> argparse.Namespace: 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") + stop_parser = subparsers.add_parser("stop", help="Gracefully shut down a running daemon") + stop_parser.add_argument("config", type=Path, help="Path to .ini configuration file") + return parser.parse_args() @@ -435,6 +470,8 @@ def main(): asyncio.run(run_daemon(config, args, stop_event)) elif args.mode == "tui": asyncio.run(run_tui(config, stop_event)) + elif args.mode == "stop": + asyncio.run(run_stop(config)) except RuntimeError as exc: print(str(exc)) raise SystemExit(1) From c94a1e5c7338baaea97596fde7487c1e6cc93016 Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 6 Sep 2026 11:57:57 +0100 Subject: [PATCH 06/11] fix unimported excpetion --- isis_monitor/beam.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/isis_monitor/beam.py b/isis_monitor/beam.py index b2d3ef4..de7ef46 100644 --- a/isis_monitor/beam.py +++ b/isis_monitor/beam.py @@ -250,7 +250,8 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): # recv_task completed. raw_msg = recv_task.result() - except ConnectionClosedOK: + except websockets.ConnectionClosedOK: + logger.warning("Websocket Closed OK") break finally: From 3b4324edfdf8baa2e2b22a72213a7d80d3047437 Mon Sep 17 00:00:00 2001 From: Max <17359435+MaxPelly@users.noreply.github.com> Date: Sun, 6 Sep 2026 16:03:35 +0100 Subject: [PATCH 07/11] Change command input handling to single character Refactor tui_command_handler to read single character input for commands. --- main.py | 34 +++++++++++++++++++++------------- 1 file changed, 21 insertions(+), 13 deletions(-) diff --git a/main.py b/main.py index a410c98..86a8d23 100755 --- a/main.py +++ b/main.py @@ -11,6 +11,9 @@ from logging.handlers import RotatingFileHandler from pathlib import Path from typing import Optional +import sys +import tty +import termios from isis_monitor.beam import BeamMonitor from isis_monitor.config import ConfigError, load_config @@ -274,8 +277,6 @@ def _close_db(): await close_channels(beam_channel, exp_channel, mcr_channel) -import sys - def _apply_snapshot_to_tui(tui: RichTUI, snapshot: dict) -> None: beam_states = snapshot.get("beam_states", {}) for beam in ("TS1", "TS2", "Muons"): @@ -313,17 +314,24 @@ def _apply_event_to_tui(tui: RichTUI, message: dict) -> None: def tui_command_handler(client: IPCClient, stop_event: asyncio.Event, tui: RichTUI): - cmd = sys.stdin.readline().strip().lower() - if cmd == "q": - stop_event.set() - elif cmd == "r": - async def _send_reconnect(): - try: - response = await client.request({"method": "command", "name": "force_reconnect_all"}) - tui.update_log(f"Reconnect request result: {response.get('result')}") - except Exception as e: - tui.update_log(f"Reconnect request failed: {e}") - asyncio.create_task(_send_reconnect()) + # Read a single character immediately without waiting for enter + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + try: + tty.setcbreak(fd) + ch = sys.stdin.read(1) + if ch.lower() == 'q': + stop_event.set() + elif ch.lower() == 'r': + async def _send_reconnect(): + try: + response = await client.request({"method": "command", "name": "force_reconnect_all"}) + tui.update_log(f"Reconnect request result: {response.get('result')}") + except Exception as e: + tui.update_log(f"Reconnect request failed: {e}") + asyncio.create_task(_send_reconnect()) + finally: + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) async def run_tui(config, stop_event: asyncio.Event): From a461858443ba0d57b60b3e2984b84edde38b6127 Mon Sep 17 00:00:00 2001 From: Max <17359435+MaxPelly@users.noreply.github.com> Date: Sun, 6 Sep 2026 16:10:56 +0100 Subject: [PATCH 08/11] Refactor event handling with stop_event check Refactor event handling to race network events against stop_event for improved responsiveness. --- main.py | 26 +++++++++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/main.py b/main.py index 86a8d23..537674f 100755 --- a/main.py +++ b/main.py @@ -379,10 +379,30 @@ async def run_tui(config, stop_event: asyncio.Event): loop.add_reader(sys.stdin.fileno(), tui_command_handler, client, stop_event, tui) has_reader = True - async for message in client.iter_events(): - _apply_event_to_tui(tui, message) - if stop_event.is_set(): + # --- FIX: Race network events against stop_event --- + event_iterator = client.iter_events().__aiter__() + while not stop_event.is_set(): + get_next_event = asyncio.create_task(event_iterator.__anext__()) + wait_stop = asyncio.create_task(stop_event.wait()) + + done, pending = await asyncio.wait( + [get_next_event, wait_stop], + return_when=asyncio.FIRST_COMPLETED + ) + + for task in pending: + task.cancel() + + if wait_stop in done: break + + try: + message = get_next_event.result() + _apply_event_to_tui(tui, message) + except StopAsyncIteration: + break + + except (FileNotFoundError, ConnectionError, OSError) as exc: tui.update_connection_state("disconnected") tui.update_log(f"Daemon connection lost: {exc}") From 7c2989563024a42d852e746d17e93c6debc38c04 Mon Sep 17 00:00:00 2001 From: Max <17359435+MaxPelly@users.noreply.github.com> Date: Sun, 6 Sep 2026 16:19:50 +0100 Subject: [PATCH 09/11] Refactor tui_command_handler for better readability Refactor tui_command_handler to simplify character reading and reconnect logic. --- main.py | 36 ++++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/main.py b/main.py index 537674f..0c78107 100755 --- a/main.py +++ b/main.py @@ -312,31 +312,30 @@ def _apply_event_to_tui(tui: RichTUI, message: dict) -> None: status = str(payload.get("status", "")) tui.update_log(f"Health: {comp} -> {status}") - def tui_command_handler(client: IPCClient, stop_event: asyncio.Event, tui: RichTUI): # Read a single character immediately without waiting for enter - fd = sys.stdin.fileno() - old_settings = termios.tcgetattr(fd) - try: - tty.setcbreak(fd) - ch = sys.stdin.read(1) - if ch.lower() == 'q': - stop_event.set() - elif ch.lower() == 'r': - async def _send_reconnect(): - try: - response = await client.request({"method": "command", "name": "force_reconnect_all"}) - tui.update_log(f"Reconnect request result: {response.get('result')}") - except Exception as e: - tui.update_log(f"Reconnect request failed: {e}") - asyncio.create_task(_send_reconnect()) - finally: - termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) + ch = sys.stdin.read(1) + if ch.lower() == 'q': + stop_event.set() + elif ch.lower() == 'r': + async def _send_reconnect(): + try: + response = await client.request({"method": "command", "name": "force_reconnect_all"}) + tui.update_log(f"Reconnect request result: {response.get('result')}") + except Exception as e: + tui.update_log(f"Reconnect request failed: {e}") + asyncio.create_task(_send_reconnect()) + async def run_tui(config, stop_event: asyncio.Event): install_signal_handlers(stop_event) + # Configure terminal to read keystrokes immediately + fd = sys.stdin.fileno() + old_settings = termios.tcgetattr(fd) + tty.setcbreak(fd) + tui = RichTUI( history_maxlen=config.history_maxlen, sample_interval=config.sample_interval, @@ -417,6 +416,7 @@ async def run_tui(config, stop_event: asyncio.Event): await client.close() finally: tui.stop() + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) async def run_stop(config) -> None: From e7f0b4c36789a80fa1982aeb8447f8b947f261ce Mon Sep 17 00:00:00 2001 From: Max Pelly Date: Tue, 8 Sep 2026 13:04:27 +0000 Subject: [PATCH 10/11] Shorten timestamp in notification --- isis_monitor/beam.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/isis_monitor/beam.py b/isis_monitor/beam.py index de7ef46..292cc6c 100644 --- a/isis_monitor/beam.py +++ b/isis_monitor/beam.py @@ -114,7 +114,7 @@ async def _handle_beam_current( if new_state != prev_state: msg = ( - f"{time_now}: {bt.display_name} Beam is now {new_state}. " + f"{time_now:%Y-%m-%d %H:%M:%S}: {bt.display_name} Beam is now {new_state}. " f"Current: {beam_val:.3f} uA" ) logger.info(f"State Change: {msg}") From 33ef5f7587e40ff79f902e0a56879ec4c57d25e8 Mon Sep 17 00:00:00 2001 From: Max Pelly Date: Tue, 8 Sep 2026 13:05:01 +0000 Subject: [PATCH 11/11] Allow larger ipc messages and sql access from multiple threads --- isis_monitor/ipc.py | 4 ++-- isis_monitor/storage.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/isis_monitor/ipc.py b/isis_monitor/ipc.py index b7913a2..c8f9249 100644 --- a/isis_monitor/ipc.py +++ b/isis_monitor/ipc.py @@ -28,7 +28,7 @@ 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), limit=65536) + self.server = await asyncio.start_unix_server(self._handle_client, path=str(self.socket_path), limit=1024*1024*10) os.chmod(self.socket_path, 0o600) async def stop(self) -> None: @@ -145,7 +145,7 @@ def __init__(self, socket_path: Path): self.writer: Optional[asyncio.StreamWriter] = None async def connect(self) -> None: - self.reader, self.writer = await asyncio.open_unix_connection(str(self.socket_path), limit=65536) + self.reader, self.writer = await asyncio.open_unix_connection(str(self.socket_path), limit=1024*1024*10) async def close(self) -> None: if self.writer: diff --git a/isis_monitor/storage.py b/isis_monitor/storage.py index 38a2b02..029ebff 100644 --- a/isis_monitor/storage.py +++ b/isis_monitor/storage.py @@ -10,7 +10,7 @@ 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 = sqlite3.connect(str(self.db_path), check_same_thread=False) self.conn.row_factory = sqlite3.Row self._init_schema()