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/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/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 cd3c72e..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 = @@ -22,6 +25,25 @@ experiment_teams_url = # Maximum number of log lines to show in the TUI (default = 50). # logs_maxlen = 50 +[DAEMON] +# SQLite database path for persisted history/state. +# db_path = beam_monitor.db +# Unix domain socket path for local IPC. +# socket_path = /tmp/isis_beam_monitor.sock +# Single-instance lock file for daemon mode. +# lock_file = /tmp/isis_beam_monitor.lock +# Beam history retention window in days. +# retention_days = 7 +# Heartbeat update interval in seconds. +# heartbeat_interval = 30 + +[TUI_CLIENT] +# Socket path to connect to daemon (same host). +# socket_path = /tmp/isis_beam_monitor.sock +# Reconnect backoff start and max in seconds. +# reconnect_initial = 1 +# reconnect_max = 15 + [LOGGING] # log_file = monitor.log # log_level = INFO diff --git a/isis_monitor/__init__.py b/isis_monitor/__init__.py index 99106e3..8432aeb 100644 --- a/isis_monitor/__init__.py +++ b/isis_monitor/__init__.py @@ -2,5 +2,15 @@ from isis_monitor.beam import BeamMonitor from isis_monitor.mcr import MCRNewsMonitor from isis_monitor.config import AppConfig, load_config, ConfigError +from isis_monitor.daemon_state import DaemonState +from isis_monitor.storage import SQLiteStateStore -__all__ = ["BeamMonitor", "MCRNewsMonitor", "AppConfig", "load_config", "ConfigError"] +__all__ = [ + "BeamMonitor", + "MCRNewsMonitor", + "AppConfig", + "load_config", + "ConfigError", + "DaemonState", + "SQLiteStateStore", +] diff --git a/isis_monitor/beam.py b/isis_monitor/beam.py index 31a7bb8..292cc6c 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] = { @@ -110,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}") @@ -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,15 @@ 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() + 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): subscribe_msg = json.dumps({ "type": "subscribe", @@ -192,27 +211,84 @@ 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 + + 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 websockets.ConnectionClosedOK: + logger.warning("Websocket Closed OK") + 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) - async for raw_msg in ws: - if stop_event and stop_event.is_set(): - return try: data = json.loads(raw_msg) if data.get("type") == "update": await self._handle_update(data) - except json.JSONDecodeError: - pass + 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") 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(): + logger.warning(f"Error Beam Loop Quit") 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 + logger.warning(f"Fallthrough Beam Loop Quit") + return diff --git a/isis_monitor/config.py b/isis_monitor/config.py index 72d82a5..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 @@ -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..00d76f0 --- /dev/null +++ b/isis_monitor/daemon_state.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +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 + +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: + logger.warning("Subscriber queue full; dropping subscriber") + dead.append(q) + for q in dead: + if q in self._subscribers: + self._subscribers.remove(q) + + def update_log(self, message: str) -> None: + with self._lock: + self.logs.append(message) + self.last_update = datetime.now(timezone.utc) + self._publish("log", {"message": message}) + + def update_beam_state(self, beam: str, current: float, power: str) -> None: + with self._lock: + if beam not in self.beam_states: + return + self.beam_states[beam] = {"current": float(current), "power": str(power)} + self.last_update = datetime.now(timezone.utc) + self._publish("beam", {"beam": beam, "current": current, "power": power}) + + def append_beam_sample( + self, + beam: str, + current: float, + power: str, + ts: Optional[datetime] = None, + publish: bool = True, + ) -> None: + ts = ts or datetime.now(timezone.utc) + with self._lock: + if beam not in self.history: + return + self.history[beam].append((ts, float(current), str(power))) + self.last_update = ts + if publish: + self._publish( + "sample", + { + "beam": beam, + "timestamp": ts.isoformat(), + "current": current, + "power": power, + }, + ) + + def trim_history_before(self, cutoff: datetime) -> None: + with self._lock: + for beam in self.history: + trimmed = deque( + (entry for entry in self.history[beam] if entry[0] >= cutoff), + maxlen=self.history_maxlen, + ) + self.history[beam] = trimmed + + def update_mcr_news(self, news: str) -> None: + with self._lock: + self.mcr_news = news + self.last_update = datetime.now(timezone.utc) + self._publish("mcr", {"news": news}) + + def update_run_name(self, run_name: str) -> None: + with self._lock: + self.run_name = run_name + self.last_update = datetime.now(timezone.utc) + self._publish("run", {"run_name": run_name}) + + def update_counts(self, counts: float) -> None: + with self._lock: + self.current_counts = float(counts) + self.last_update = datetime.now(timezone.utc) + self._publish("counts", {"counts": counts}) + + def update_health(self, component: str, status: str) -> None: + with self._lock: + self.health[component] = status + self.last_update = datetime.now(timezone.utc) + self._publish("health", {"component": component, "status": status}) + + def snapshot(self) -> dict: + with self._lock: + 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(), + "current": cur, + "power": power, + } + for ts, cur, power in data + ] + for beam, data in self.history.items() + } + + 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) + with self._lock: + items = list(self.beam_states.items()) + for beam, state in items: + self.append_beam_sample( + beam, + float(state["current"]), + str(state["power"]), + ts=ts, + ) + + def cutoff_for_days(self, retention_days: int) -> datetime: + return datetime.now(timezone.utc) - timedelta(days=retention_days) + + def get_beam_rows_for_timestamp(self, ts: Optional[datetime] = None) -> list[tuple[datetime, str, float, str]]: + ts = ts or datetime.now(timezone.utc) + with self._lock: + return [ + (ts, beam, float(state["current"]), str(state["power"])) + for beam, state in self.beam_states.items() + ] + + def get_health(self) -> Dict[str, str]: + with self._lock: + return dict(self.health) + + def restore_from_snapshot_json(self, raw: Optional[str]) -> None: + if not raw: + return + try: + snap = json.loads(raw) + except json.JSONDecodeError: + logger.warning("Corrupt daemon state snapshot, starting fresh.") + return + with self._lock: + beam_states = snap.get("beam_states", {}) + for beam in ("TS1", "TS2", "Muons"): + if beam in beam_states: + self.beam_states[beam] = { + "current": float(beam_states[beam].get("current", 0.0)), + "power": str(beam_states[beam].get("power", "unknown")), + } + self.mcr_news = str(snap.get("mcr_news", self.mcr_news)) + self.run_name = str(snap.get("run_name", self.run_name)) + self.current_counts = float(snap.get("current_counts", self.current_counts)) + health = snap.get("health", {}) + if isinstance(health, dict): + for k, v in health.items(): + self.health[str(k)] = str(v) diff --git a/isis_monitor/ipc.py b/isis_monitor/ipc.py new file mode 100644 index 0000000..c8f9249 --- /dev/null +++ b/isis_monitor/ipc.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +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), limit=1024*1024*10) + os.chmod(self.socket_path, 0o600) + + 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 == "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() + 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, + } + try: + await self._send(writer, payload) + except (ConnectionError, BrokenPipeError, OSError): + break + + +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), limit=1024*1024*10) + + 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..3d94e36 100644 --- a/isis_monitor/mcr.py +++ b/isis_monitor/mcr.py @@ -1,4 +1,5 @@ import asyncio +import contextlib import logging import re from datetime import datetime @@ -7,7 +8,7 @@ from isis_monitor.config import AppConfig from isis_monitor.notifiers import NotificationChannel -from isis_monitor.protocols import TUIProtocol +from isis_monitor.protocols import TUIProtocol, MonitorSinkProtocol logger = logging.getLogger(__name__) @@ -22,13 +23,16 @@ def __init__( channel: NotificationChannel, notify_current: bool = False, tui: Optional[TUIProtocol] = None, + sink: Optional[MonitorSinkProtocol] = None, ): self.config = config self.url = config.mcr_news_url self.channel = channel self.notify_current = notify_current self.tui = tui + self.sink = sink self.old_news: Optional[str] = None + self._force_reconnect = asyncio.Event() async def get_news(self, session: aiohttp.ClientSession) -> Optional[str]: try: @@ -56,6 +60,8 @@ async def get_news(self, session: aiohttp.ClientSession) -> Optional[str]: async def run(self, stop_event: Optional[asyncio.Event] = None): logger.info(f"MCR Monitor started. Watching {self.url}...") + if self.sink: + self.sink.update_health("mcr", "starting") # TCPConnector with DNS TTL avoids stale connections on long-running sessions connector = aiohttp.TCPConnector(ttl_dns_cache=300) @@ -70,6 +76,8 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): logger.info(f"Current MCR News: {self.old_news}") if self.tui: self.tui.update_mcr_news(self.old_news) + if self.sink: + self.sink.update_mcr_news(self.old_news) else: await asyncio.sleep(self.config.mcr_poll_interval) else: @@ -79,14 +87,30 @@ async def run(self, stop_event: Optional[asyncio.Event] = None): consecutive_failures = 0 while stop_event is None or not stop_event.is_set(): try: - sleep_secs = self.config.mcr_poll_interval * min( - 2 ** consecutive_failures, 8 + sleep_secs = self.config.mcr_poll_interval * min(2 ** consecutive_failures, 8) + sleep_task = asyncio.create_task(asyncio.sleep(sleep_secs)) + reconnect_task = asyncio.create_task(self._force_reconnect.wait()) + done, pending = await asyncio.wait( + {sleep_task, reconnect_task}, + return_when=asyncio.FIRST_COMPLETED, ) - await asyncio.sleep(sleep_secs) + for task in pending: + task.cancel() + for task in pending: + with contextlib.suppress(asyncio.CancelledError): + await task + + if reconnect_task in done and self._force_reconnect.is_set(): + self._force_reconnect.clear() + consecutive_failures = 0 + if self.sink: + self.sink.update_health("mcr", "reconnecting") except asyncio.CancelledError: + 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) @@ -96,13 +120,29 @@ 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") + 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." ) + logger.warning(f"MCR News Collection Fall Through") + return + + def request_reconnect(self) -> bool: + if self._force_reconnect.is_set(): + return False + self._force_reconnect.set() + return True 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/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..029ebff --- /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), check_same_thread=False) + self.conn.row_factory = sqlite3.Row + self._init_schema() + + def _init_schema(self) -> None: + self.conn.executescript( + """ + CREATE TABLE IF NOT EXISTS beam_samples ( + timestamp TEXT NOT NULL, + target TEXT NOT NULL, + current REAL NOT NULL, + power TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_beam_samples_time + ON beam_samples(timestamp); + + CREATE TABLE IF NOT EXISTS snapshot ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS health ( + component TEXT PRIMARY KEY, + status TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + """ + ) + self.conn.commit() + + def close(self) -> None: + self.conn.close() + + def write_sample(self, timestamp: datetime, target: str, current: float, power: str) -> None: + self.conn.execute( + "INSERT INTO beam_samples(timestamp, target, current, power) VALUES (?, ?, ?, ?)", + (timestamp.isoformat(), target, current, power), + ) + + def write_samples(self, rows: Iterable[Tuple[datetime, str, float, str]]) -> None: + self.conn.executemany( + "INSERT INTO beam_samples(timestamp, target, current, power) VALUES (?, ?, ?, ?)", + [(ts.isoformat(), target, current, power) for ts, target, current, power in rows], + ) + + def prune_older_than(self, cutoff: datetime) -> int: + cur = self.conn.execute( + "DELETE FROM beam_samples WHERE timestamp < ?", + (cutoff.isoformat(),), + ) + return cur.rowcount + + def load_recent_samples(self, since: datetime) -> list[sqlite3.Row]: + cur = self.conn.execute( + """ + SELECT timestamp, target, current, power + FROM beam_samples + WHERE timestamp >= ? + ORDER BY timestamp ASC + """, + (since.isoformat(),), + ) + return list(cur.fetchall()) + + def upsert_snapshot(self, key: str, value: str) -> None: + now = datetime.now(timezone.utc).isoformat() + self.conn.execute( + """ + INSERT INTO snapshot(key, value, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at + """, + (key, value, now), + ) + + def load_snapshot(self, key: str) -> Optional[str]: + cur = self.conn.execute("SELECT value FROM snapshot WHERE key = ?", (key,)) + row = cur.fetchone() + return row[0] if row else None + + def upsert_health(self, component: str, status: str) -> None: + now = datetime.now(timezone.utc).isoformat() + self.conn.execute( + """ + INSERT INTO health(component, status, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(component) DO UPDATE SET status=excluded.status, updated_at=excluded.updated_at + """, + (component, status, now), + ) + + def load_health(self) -> list[sqlite3.Row]: + cur = self.conn.execute("SELECT component, status FROM health") + return list(cur.fetchall()) + + def commit(self) -> None: + self.conn.commit() diff --git a/isis_monitor/tests/test_config.py b/isis_monitor/tests/test_config.py index b7f38aa..4bd1342 100644 --- a/isis_monitor/tests/test_config.py +++ b/isis_monitor/tests/test_config.py @@ -133,3 +133,57 @@ def test_load_config_empty_websocket_url_logs_warning(tmp_path, caplog): config = load_config(config_file) assert config.isis_websocket_url == "" assert "isis_websocket_url" in caplog.text + + +def test_load_config_daemon_and_tui_client_values(tmp_path): + config_file = tmp_path / "config.ini" + config_file.write_text("""\ +[DATA] +mcr_news_url = http://test.com/news +isis_websocket_url = wss://test.com/ws + +[WEBHOOKS] +news_teams_url = +beam_teams_url = +experiment_teams_url = + +[DAEMON] +db_path = /tmp/beam.db +socket_path = /tmp/beam.sock +lock_file = /tmp/beam.lock +retention_days = 7 +heartbeat_interval = 15 + +[TUI_CLIENT] +socket_path = /tmp/beam.sock +reconnect_initial = 2 +reconnect_max = 20 +""") + config = load_config(config_file) + assert config.daemon_db_path == "/tmp/beam.db" + assert config.daemon_socket_path == "/tmp/beam.sock" + assert config.daemon_lock_file == "/tmp/beam.lock" + assert config.retention_days == 7 + assert config.heartbeat_interval == 15 + assert config.tui_socket_path == "/tmp/beam.sock" + assert config.tui_reconnect_initial == 2 + assert config.tui_reconnect_max == 20 + + +def test_load_config_invalid_retention_days(tmp_path): + config_file = tmp_path / "config.ini" + config_file.write_text("""\ +[DATA] +mcr_news_url = http://test.com/news +isis_websocket_url = wss://test.com/ws + +[WEBHOOKS] +news_teams_url = +beam_teams_url = +experiment_teams_url = + +[DAEMON] +retention_days = 0 +""") + with pytest.raises(ConfigError, match="retention_days"): + load_config(config_file) diff --git a/isis_monitor/tests/test_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 new file mode 100644 index 0000000..4773461 --- /dev/null +++ b/isis_monitor/tests/test_ipc.py @@ -0,0 +1,103 @@ +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() + +@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 dd941ca..14d6346 100644 --- a/isis_monitor/tests/test_main.py +++ b/isis_monitor/tests/test_main.py @@ -1,14 +1,21 @@ import logging import pytest from unittest.mock import MagicMock -from main import TUILogHandler +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 +from main import StateLogHandler, SingleInstanceLock, _apply_snapshot_to_tui -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 +24,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 +41,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 +50,42 @@ 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") + +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): + from main import SingleInstanceLock + lock_file = tmp_path / "test.lock" + 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 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/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..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 @@ -103,6 +102,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() @@ -202,11 +202,15 @@ 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( - Text("ISIS Facility Monitor", justify="center", style="bold cyan"), + Text( + f"ISIS Facility Monitor [{self.connection_state}]", + justify="center", + style="bold cyan", + ), style="blue", ) ) @@ -242,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() @@ -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..0c78107 100755 --- a/main.py +++ b/main.py @@ -1,111 +1,481 @@ #!/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 +import sys +import tty +import termios -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("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())) + 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) + + beam_rows = state.get_beam_rows_for_timestamp(ts) + cutoff = state.cutoff_for_days(config.retention_days) + 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() + + await asyncio.to_thread(_persist) + logger.warning("State Persistance quit") + return + + +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 + logger.warning("Heartbeat quit") + return + +async def run_daemon(config, args, stop_event: asyncio.Event): + install_signal_handlers(stop_event) - # 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) + state = DaemonState(history_maxlen=max(config.history_maxlen, int((86400 * config.retention_days) / max(config.sample_interval, 1.0)))) + state.update_health("daemon", "starting") + + 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"]), + 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()} + 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) + 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: + logger.warning("Shutting down daemon") + state.update_health("daemon", "stopping") + 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) + + +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"))) + if snapshot.get("mcr_news"): + tui.update_mcr_news(str(snapshot["mcr_news"])) + + +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}") + +def tui_command_handler(client: IPCClient, stop_event: asyncio.Event, tui: RichTUI): + # Read a single character immediately without waiting for enter + 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, + refresh_per_second=config.refresh_per_second, + 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)) + has_reader = False + 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", {})) + + 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.") + + # 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 + + # --- 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}") + 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 has_reader: + loop.remove_reader(sys.stdin.fileno()) + await client.close() finally: tui.stop() + termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) -def main(): +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") - 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") + + 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() + + +def main(): + args = parse_args() try: config = load_config(args.config) @@ -113,29 +483,29 @@ 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)) + elif args.mode == "stop": + asyncio.run(run_stop(config)) + except RuntimeError as exc: + print(str(exc)) + raise SystemExit(1) except (KeyboardInterrupt, asyncio.CancelledError): print("\nStopping monitors...") + if __name__ == "__main__": main()