Creating implementation plan for daemon and TUI separation

.gitignore

@@ -1,3 +1,4 @@
+*.db
 config.ini
 test.py
 

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.

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,17 +40,28 @@ 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.
 
 ---
 
 ## Configuration
 
 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.

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

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",
+]

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,14 +114,16 @@ 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}")
             await self.beam_channel.broadcast(msg, bt.channel_label)
 
         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