diff --git a/software/acquisition_watchdog/README.md b/software/acquisition_watchdog/README.md new file mode 100644 index 000000000..e8d16457b --- /dev/null +++ b/software/acquisition_watchdog/README.md @@ -0,0 +1,134 @@ +# Acquisition Watchdog + +An independent process that watches Squid acquisitions and posts a **single Slack alert** +when one ends prematurely — a process **crash / hang / kill**, a **fatal error**, or a +**user abort**. Clean completions stay silent ("no news is good news"). It covers +acquisitions started from the **GUI** and from the **MCP control server**, on **Ubuntu and +Windows**. + +Because it runs as a *separate* process, it can report failures the in-app Slack notifier +can't — a segfault in a camera SDK, an OOM-kill, a frozen UI, or the whole process dying. + +## How it works + +The Squid GUI writes a small `run.json` breadcrumb into a shared state dir: `status=running` +at acquisition start, a throttled heartbeat (+ progress) during the run, and `status=ended` +with a reason at the end. This watchdog polls that file and alerts when a run's process has +died / gone silent, or ended with a non-clean reason. One alert per run (de-duplicated). + +## Prerequisites + +1. **Slack configured in the GUI.** Open *Settings → Slack Notifications*, enter your **Bot + Token** (`xoxb-…`) and **Channel ID** (`C…`), and click *Save*. That writes + `cache/slack_settings.yaml`, which the watchdog reads — there is no separate config. + (Need to create the token? See [`../docs/slack_notifications.md`](../docs/slack_notifications.md).) +2. **"Enable watchdog alerts" checked** (the default) in that same dialog — or the + `watchdog_enabled` key in `cache/slack_settings.yaml`. + +## Run it + +From the `software/` directory: + +```bash +cd software +python3 -m acquisition_watchdog +``` + +Leave it running. Start an acquisition; if it crashes / hangs / aborts / errors, you get a +Slack alert. A clean finish produces nothing. + +### Options + +| Flag | Default | Purpose | +|---|---|---| +| `--heartbeat-timeout` | `120` | Seconds of heartbeat silence (with a live process) before declaring a hang. Raise it if you have very long single exposures / fluidics steps. | +| `--poll-interval` | `5` | Seconds between checks. | +| `--once` | — | Run a single check and exit (handy for testing or a cron probe). | +| `--slack-settings ` | `cache/slack_settings.yaml` | Only needed if you don't run from `software/`. | +| `--state-dir ` | platformdirs user-state dir | Must match the GUI's; override here or via `$SQUID_WATCHDOG_STATE_DIR`. | + +## Run it always-on (recommended for a lab microscope) + +A manual run stops when you close the terminal or reboot. To keep it up independently of the +GUI: + +- **Linux (systemd user service)** — run these from the `software/` directory: + ```bash + mkdir -p ~/.config/systemd/user + cp acquisition_watchdog/systemd/squid-acquisition-watchdog.service ~/.config/systemd/user/ + # The shipped unit's WorkingDirectory is a placeholder (%h/Squid/software); point it here: + sed -i "s#^WorkingDirectory=.*#WorkingDirectory=$PWD#" ~/.config/systemd/user/squid-acquisition-watchdog.service + systemctl --user daemon-reload + systemctl --user enable --now squid-acquisition-watchdog # auto-start at login + start now + systemctl --user status squid-acquisition-watchdog # verify it's active + ``` + Enable it once and it comes up at every login and restarts on failure (`Restart=always`) — + no need to launch it by hand. Logs: `journalctl --user -u squid-acquisition-watchdog -f`. + To keep it running before/without a graphical login, also run `loginctl enable-linger $USER` + once. (The unit runs `/usr/bin/python3`; if Squid runs on a different interpreter/venv, edit + the `ExecStart=` line to that python.) +- **Windows (Task Scheduler):** run `windows/install.ps1` in PowerShell from `software\`. It + registers a logon-triggered task (via `pythonw.exe`; make sure it's on `PATH`, or edit the + path in the script). + +## Verify it works + +```bash +python3 -m acquisition_watchdog --once # one check, then exits — no error means it's healthy +``` + +End-to-end: start a `--simulation` acquisition, `kill -9` the GUI process, and you should get +a crash alert — within one poll (~5 s) when `psutil` is installed (it is, by default), +otherwise within `--heartbeat-timeout` seconds. + +## What triggers an alert + +| Situation | Alert | +|---|---| +| Process died — `running` breadcrumb + PID gone (crash / OOM-kill / power loss) | 🔴 crash | +| Process alive but no heartbeat past the timeout | 🟠 hang | +| Fatal error / auto-abort (timeout, failed save job, camera/frame failure) | 🔴 error | +| Finished, but some save/job errors occurred | 🟠 completed-with-errors | +| Aborted by the user, the MCP server, or by closing the GUI mid-run | 🟡 aborted | +| Finished cleanly | *(silent)* | + +Each alert includes the machine name, experiment, reason, and progress (e.g. "stopped at +timepoint 3/10"). + +## Good to know + +- **Independent of the GUI.** Restarting the software neither starts nor stops the watchdog — + it just picks up the next run. That decoupling is the whole point: a watchdog spawned by the + GUI couldn't survive the GUI crashing. +- **Start order doesn't matter.** Start it before, during, or after the GUI. If it starts + mid-run it monitors from there; if it starts *after* a crash already happened, it reads the + stale `running` breadcrumb, sees the PID is dead, and alerts once. +- **One alert per run.** Alerted run IDs persist in `/alerted.json`, so it never + double-alerts and never re-alerts after a restart. +- **Turn alerts off** on a machine (without disabling the GUI notifier): uncheck *"Enable + watchdog alerts"*, or set `watchdog_enabled: false` in `cache/slack_settings.yaml`. Takes + effect on the next check — no watchdog restart needed. +- **Runs on the same machine as the GUI** (it reads local breadcrumb files). For coverage of a + full machine death / power loss, run it on another host pointed at a shared/synced state dir + (see *Remote / power-loss coverage*). + +## Troubleshooting + +| Symptom | Check | +|---|---| +| No alerts at all | Is the watchdog process actually running? Are `bot_token`/`channel_id` set (GUI → *Test Connection*)? Is *"Enable watchdog alerts"* checked? | +| Log says Slack not configured | Run from `software/` so `cache/slack_settings.yaml` resolves, or pass `--slack-settings `. | +| False "hang" alerts | Raise `--heartbeat-timeout` — a single very long exposure/fluidics step can exceed the default. | +| Crash reported slowly (~2 min) | `psutil` missing → falls back to the heartbeat timeout. Install `psutil` for instant PID-based detection. | + +## State dir + +Defaults to `platformdirs.user_state_path("squid", "cephla")/watchdog`. The GUI (writer) and +the watchdog (reader) must agree on it — run both as the same user, or set +`SQUID_WATCHDOG_STATE_DIR` on both (or `--state-dir` on the watchdog). + +## Remote / power-loss coverage (future) + +Point `--state-dir` at a shared/synced mount on another host and run this process there. +Per-machine `run-.json` naming and a clock-skew tolerance are needed first (see the +design spec). diff --git a/software/acquisition_watchdog/__init__.py b/software/acquisition_watchdog/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/software/acquisition_watchdog/__main__.py b/software/acquisition_watchdog/__main__.py new file mode 100644 index 000000000..dcbcb9aaa --- /dev/null +++ b/software/acquisition_watchdog/__main__.py @@ -0,0 +1,50 @@ +# acquisition_watchdog/__main__.py +"""CLI entry point: python -m acquisition_watchdog""" +import argparse +import time +from pathlib import Path +from typing import Optional, Sequence + +import squid.logging +from acquisition_watchdog.monitor import Monitor + + +def main(argv: Optional[Sequence[str]] = None) -> None: + parser = argparse.ArgumentParser( + prog="acquisition_watchdog", + description="Alert on prematurely-ended Squid acquisitions (crash/hang/abort/error).", + ) + parser.add_argument( + "--slack-settings", + help="Path to the Slack settings YAML (defaults to ./cache/slack_settings.yaml, " + "the same file the GUI writes).", + ) + parser.add_argument("--state-dir", help="Override the watchdog state directory.") + parser.add_argument("--poll-interval", type=float, default=5.0, help="Seconds between checks (default 5).") + parser.add_argument( + "--heartbeat-timeout", + type=float, + default=120.0, + help="Seconds of heartbeat silence (with a live PID) before declaring a hang (default 120).", + ) + parser.add_argument("--once", action="store_true", help="Run a single check and exit.") + args = parser.parse_args(argv) + + log = squid.logging.get_logger("acquisition_watchdog") + monitor = Monitor( + state_dir=Path(args.state_dir) if args.state_dir else None, + slack_settings=args.slack_settings, + poll_interval=args.poll_interval, + heartbeat_timeout=args.heartbeat_timeout, + ) + if args.once: + monitor.check_once(time.time()) + else: + try: + monitor.run_forever() + except KeyboardInterrupt: + log.info("Acquisition watchdog stopped.") + + +if __name__ == "__main__": + main() diff --git a/software/acquisition_watchdog/alerts.py b/software/acquisition_watchdog/alerts.py new file mode 100644 index 000000000..a4cbee647 --- /dev/null +++ b/software/acquisition_watchdog/alerts.py @@ -0,0 +1,49 @@ +# acquisition_watchdog/alerts.py +"""Format watchdog Slack alerts (text + Block Kit blocks).""" +from datetime import datetime, timezone +from typing import Optional, Tuple + +_KIND_TITLE = { + "crash": ":red_circle: Acquisition process died", + "hang": ":large_orange_circle: Acquisition hung (no heartbeat)", + "error": ":red_circle: Acquisition ended with a fatal error", + "completed_with_errors": ":large_orange_circle: Acquisition finished with errors", + "user_abort": ":large_yellow_circle: Acquisition aborted", +} + + +def _fmt_ts(epoch: Optional[float]) -> str: + if not epoch: + return "unknown" + return datetime.fromtimestamp(epoch, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + + +def _progress_line(run: dict) -> str: + prog = run.get("progress") or {} + expected = run.get("expected") or {} + tp = prog.get("timepoint", "?") + exp_tp = prog.get("expected_timepoints", expected.get("timepoints", "?")) + images = prog.get("images", "?") + return f"timepoint {tp}/{exp_tp}, {images} images" + + +def format_alert(kind: str, run: dict) -> Tuple[str, list]: + title = _KIND_TITLE.get(kind, f"Acquisition alert: {kind}") + experiment = run.get("experiment_id", "unknown") + machine = run.get("machine", "unknown") + text = f"{title}: {experiment} on {machine}" + + last_seen = run.get("ended_at") or run.get("heartbeat_at") + detail = ( + f"*Experiment:* {experiment}\n" + f"*Machine:* {machine}\n" + f"*Progress:* {_progress_line(run)}\n" + f"*Started:* {_fmt_ts(run.get('started_at'))}\n" + f"*Last seen:* {_fmt_ts(last_seen)}\n" + f"*Output:* {run.get('output_path', 'unknown')}" + ) + blocks = [ + {"type": "section", "text": {"type": "mrkdwn", "text": f"*{title}*"}}, + {"type": "section", "text": {"type": "mrkdwn", "text": detail}}, + ] + return text, blocks diff --git a/software/acquisition_watchdog/config.py b/software/acquisition_watchdog/config.py new file mode 100644 index 000000000..090096a99 --- /dev/null +++ b/software/acquisition_watchdog/config.py @@ -0,0 +1,51 @@ +# acquisition_watchdog/config.py +"""Load Slack credentials from the same source the Squid GUI uses. + +The GUI stores Slack settings (bot token, channel, enabled) in +`cache/slack_settings.yaml` (written by the Slack settings dialog and loaded at +GUI startup via control.widgets_slack.load_slack_settings_from_cache). This module +reads that same YAML so the watchdog alerts to the same workspace — without +importing the heavy control stack. +""" +import os +from pathlib import Path +from typing import NamedTuple, Optional + +import yaml + + +class SlackConfig(NamedTuple): + bot_token: Optional[str] + channel_id: Optional[str] + watchdog_enabled: bool + + +DEFAULT_SLACK_SETTINGS = "cache/slack_settings.yaml" + + +def resolve_slack_settings_path(cli_path: Optional[str]) -> Path: + """Priority: --slack-settings > $SQUID_SLACK_SETTINGS > cache/slack_settings.yaml (cwd-relative).""" + if cli_path: + return Path(cli_path) + env = os.environ.get("SQUID_SLACK_SETTINGS") + if env: + return Path(env) + return Path(DEFAULT_SLACK_SETTINGS) + + +def load_slack_config(path: Optional[Path]) -> SlackConfig: + p = Path(path) if path else Path(DEFAULT_SLACK_SETTINGS) + if not p.exists(): + return SlackConfig(None, None, True) + try: + with open(p) as f: + data = yaml.safe_load(f) or {} + except Exception: + return SlackConfig(None, None, True) + if not isinstance(data, dict): + return SlackConfig(None, None, True) + return SlackConfig( + bot_token=(data.get("bot_token") or None), + channel_id=(data.get("channel_id") or None), + watchdog_enabled=bool(data.get("watchdog_enabled", True)), + ) diff --git a/software/acquisition_watchdog/monitor.py b/software/acquisition_watchdog/monitor.py new file mode 100644 index 000000000..5f49d3f79 --- /dev/null +++ b/software/acquisition_watchdog/monitor.py @@ -0,0 +1,124 @@ +# acquisition_watchdog/monitor.py +"""Poll the acquisition run-state and alert on premature ends.""" +import json +import os +import time +from pathlib import Path +from typing import Optional, Set + +import squid.acquisition_state as acquisition_state +import squid.logging +import squid.slack +from acquisition_watchdog import alerts, config + +_log = squid.logging.get_logger("acquisition_watchdog") + +ALERT_REASONS = {"completed_with_errors", "error", "user_abort"} + + +def pid_alive(pid: Optional[int]) -> bool: + if not pid: + return False + try: + import psutil + + return psutil.pid_exists(pid) + except ImportError: + pass + if os.name == "posix": + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + # Windows without psutil: cannot check reliably; rely on the heartbeat instead. + return True + + +class Monitor: + def __init__( + self, + state_dir: Optional[Path] = None, + slack_settings: Optional[str] = None, + poll_interval: float = 5.0, + heartbeat_timeout: float = 120.0, + ): + self._state_dir = Path(state_dir) if state_dir else None + self._slack_settings = slack_settings + self._poll = poll_interval + self._timeout = heartbeat_timeout + self._base = self._state_dir or acquisition_state.default_state_dir() + self._alerted_path = self._base / "alerted.json" + self._alerted = self._load_alerted() + + def _load_alerted(self) -> Set[str]: + try: + with open(self._alerted_path) as f: + return set(json.load(f)) + except (FileNotFoundError, json.JSONDecodeError): + return set() + + def _save_alerted(self) -> None: + try: + self._alerted_path.parent.mkdir(parents=True, exist_ok=True) + with open(self._alerted_path, "w") as f: + json.dump(sorted(self._alerted), f) + except OSError as e: + _log.warning(f"Could not persist alerted set: {e}") + + def classify(self, run: Optional[dict], now: float) -> Optional[str]: + """Return an alert kind ('crash'|'hang'|) or None.""" + if not run or run.get("run_id") in self._alerted: + return None + status = run.get("status") + if status == "running": + if not pid_alive(run.get("pid")): + return "crash" + if (now - (run.get("heartbeat_at") or 0)) > self._timeout: + return "hang" + return None + if status == "ended" and run.get("reason") in ALERT_REASONS: + return run["reason"] + return None + + def check_once(self, now: float) -> None: + run = acquisition_state.read_run(self._state_dir) + kind = self.classify(run, now) + if kind is None: + return + + cfg_path = config.resolve_slack_settings_path(self._slack_settings) + slack_cfg = config.load_slack_config(cfg_path) + if not (slack_cfg.bot_token and slack_cfg.channel_id and slack_cfg.watchdog_enabled): + _log.warning( + f"Premature end ({kind}) for run_id={run.get('run_id')} but Slack is not " + f"configured/enabled; not alerting." + ) + self._mark_alerted(run["run_id"]) + return + + text, blocks = alerts.format_alert(kind, run) + ok, _ = squid.slack.post_message(slack_cfg.bot_token, slack_cfg.channel_id, text, blocks) + if ok: + _log.info(f"Sent watchdog alert ({kind}) for run_id={run.get('run_id')}") + self._mark_alerted(run["run_id"]) + else: + # Leave unmarked so a transient Slack failure retries on the next poll. + _log.warning(f"Failed to send watchdog alert ({kind}) for run_id={run.get('run_id')}; will retry") + + def _mark_alerted(self, run_id: str) -> None: + self._alerted.add(run_id) + self._save_alerted() + + def run_forever(self) -> None: + _log.info(f"Acquisition watchdog started. state_dir={self._base} heartbeat_timeout={self._timeout}s") + while True: + try: + self.check_once(time.time()) + except Exception as e: + _log.exception(f"Watchdog poll error: {e}") + time.sleep(self._poll) diff --git a/software/acquisition_watchdog/systemd/squid-acquisition-watchdog.service b/software/acquisition_watchdog/systemd/squid-acquisition-watchdog.service new file mode 100644 index 000000000..3749cc194 --- /dev/null +++ b/software/acquisition_watchdog/systemd/squid-acquisition-watchdog.service @@ -0,0 +1,19 @@ +# Install (per user): +# mkdir -p ~/.config/systemd/user +# cp acquisition_watchdog/systemd/squid-acquisition-watchdog.service ~/.config/systemd/user/ +# # edit WorkingDirectory below to match this machine, then: +# systemctl --user daemon-reload +# systemctl --user enable --now squid-acquisition-watchdog +[Unit] +Description=Squid acquisition watchdog (alerts on prematurely-ended acquisitions) +After=default.target + +[Service] +Type=simple +WorkingDirectory=%h/Squid/software +ExecStart=/usr/bin/python3 -m acquisition_watchdog +Restart=always +RestartSec=5 + +[Install] +WantedBy=default.target diff --git a/software/acquisition_watchdog/windows/install.ps1 b/software/acquisition_watchdog/windows/install.ps1 new file mode 100644 index 000000000..36fe1ab49 --- /dev/null +++ b/software/acquisition_watchdog/windows/install.ps1 @@ -0,0 +1,34 @@ +# Run in PowerShell (as the user who runs the Squid GUI), from software\ : +# .\acquisition_watchdog\windows\install.ps1 +# Registers a logon-triggered scheduled task that runs the acquisition watchdog. +# Edit $workingDir below if your install path differs. +$ErrorActionPreference = "Stop" + +$taskName = "SquidAcquisitionWatchdog" +$workingDir = "C:\Squid\software" + +$action = New-ScheduledTaskAction ` + -Execute "pythonw.exe" ` + -Argument "-m acquisition_watchdog" ` + -WorkingDirectory $workingDir + +$trigger = New-ScheduledTaskTrigger -AtLogOn + +$settings = New-ScheduledTaskSettingsSet ` + -MultipleInstances IgnoreNew ` + -RestartInterval (New-TimeSpan -Minutes 1) ` + -RestartCount 999 ` + -ExecutionTimeLimit (New-TimeSpan -Seconds 0) ` + -AllowStartIfOnBatteries ` + -DontStopIfGoingOnBatteries + +Write-Host "Registering scheduled task '$taskName'..." +Register-ScheduledTask ` + -TaskName $taskName ` + -Action $action ` + -Trigger $trigger ` + -Settings $settings ` + -Description "Squid acquisition watchdog (alerts on prematurely-ended acquisitions)" ` + -Force + +Write-Host "Done. It starts at next logon. Run now with: Start-ScheduledTask -TaskName $taskName" diff --git a/software/control/_def.py b/software/control/_def.py index c803655ae..a122b8ea1 100644 --- a/software/control/_def.py +++ b/software/control/_def.py @@ -1288,6 +1288,7 @@ class SlackNotifications: NOTIFY_ON_ACQUISITION_START = False NOTIFY_ON_ACQUISITION_FINISHED = True SEND_MOSAIC_SNAPSHOTS = True + WATCHDOG_ENABLED = True # Standalone acquisition watchdog: alert on crash / hang / error / abort try: diff --git a/software/control/core/multi_point_controller.py b/software/control/core/multi_point_controller.py index 680c22b8d..9f117d989 100644 --- a/software/control/core/multi_point_controller.py +++ b/software/control/core/multi_point_controller.py @@ -28,6 +28,7 @@ from control.microcontroller import Microcontroller from control.piezo import PiezoStage from squid.abc import CameraFrame, AbstractCamera, AbstractStage +import squid.acquisition_state import squid.logging @@ -692,6 +693,7 @@ def run_acquisition(self, acquire_current_fov=False): log_memory("ACQUISITION START", include_children=True) thread_started = False + self._run_state_writer = squid.acquisition_state.NullRunStateWriter() try: self._log.info("start multipoint") self._start_position = self.stage.get_pos() @@ -872,6 +874,28 @@ def finish_fn(): self.overlap_percent, ) + # Acquisition watchdog: drop the "running" breadcrumb (covers GUI + MCP-server runs). + # self._run_state_writer is already a NullRunStateWriter (set before the outer try); it + # stays one if start() below fails, so a breadcrumb write failure never breaks acquisition. + try: + expected = { + "timepoints": self.Nt, + "regions": len(scan_position_information.scan_region_coords_mm), + "fovs": sum(len(c) for c in scan_position_information.scan_region_fov_coords_mm.values()), + "channels": len(self.selected_configurations), + "z": self.NZ, + } + config_path = (getattr(control._def, "CACHED_CONFIG_FILE_PATH", None) or "").strip() or None + self._run_state_writer = squid.acquisition_state.RunStateWriter.start( + experiment_id=self.experiment_ID, + pid=os.getpid(), + config_path=config_path, + output_path=experiment_path, + expected=expected, + ) + except Exception as e: + self._log.warning(f"Failed to write acquisition watchdog start state: {e}") + # Get pre-warmed job runner and its shared backpressure values # (starts a new one warming for next acquisition) prewarmed_runner, prewarmed_bp_values = self.get_prewarmed_job_runner() @@ -893,6 +917,7 @@ def finish_fn(): slack_notifier=self._slack_notifier, prewarmed_job_runner=prewarmed_runner, prewarmed_bp_values=prewarmed_bp_values, + run_state_writer=self._run_state_writer, ) except Exception: # Clean up pre-warmed runner if worker creation failed. @@ -912,6 +937,9 @@ def finish_fn(): self.thread.start() finally: if not thread_started: + # Acquisition never launched a worker — close out the breadcrumb so the + # watchdog doesn't later misread the lingering "running" state as a hang. + self._run_state_writer.end("error", None) self._stop_per_acquisition_log() # Stop memory monitor if acquisition setup failed if self._memory_monitor is not None: diff --git a/software/control/core/multi_point_worker.py b/software/control/core/multi_point_worker.py index 9c834ed83..4ce276411 100644 --- a/software/control/core/multi_point_worker.py +++ b/software/control/core/multi_point_worker.py @@ -30,6 +30,7 @@ from control.piezo import PiezoStage from control.models import AcquisitionChannel from squid.abc import AbstractCamera, CameraFrame, CameraFrameFormat +import squid.acquisition_state import squid.logging import control.core.job_processing from control.core.job_processing import ZarrWriteResult @@ -81,6 +82,7 @@ def __init__( slack_notifier=None, prewarmed_job_runner: Optional[JobRunner] = None, prewarmed_bp_values: Optional["BackpressureValues"] = None, + run_state_writer=None, ): self._log = squid.logging.get_logger(__class__.__name__) self._timing = utils.TimingManager("MultiPointWorker Timer Manager") @@ -110,6 +112,8 @@ def __init__( self.callbacks: MultiPointControllerFunctions = callbacks self.abort_requested_fn: Callable[[], bool] = abort_requested_fn self.request_abort_fn: Callable[[], None] = request_abort_fn + self._run_state = run_state_writer or squid.acquisition_state.NullRunStateWriter() + self._abort_cause = None # set to "error" by auto-abort paths (timeout / failed jobs) self.NZ = acquisition_parameters.NZ self.deltaZ = acquisition_parameters.deltaZ @@ -424,8 +428,37 @@ def _is_well_based_acquisition(self) -> bool: ) return True + def _abort_due_to_error(self) -> None: + """Abort the run due to an internal error (vs a user abort). + + The worker only ever aborts itself on error conditions; user aborts arrive + via the external abort flag. Tagging the cause here lets _compute_end_reason + classify the end as "error" instead of "user_abort". + """ + self._abort_cause = "error" + self.request_abort_fn() + + def _run_state_beat(self) -> None: + self._run_state.beat( + { + "timepoint": self.time_point, + "fov": self._timepoint_fov_count, + "images": self.image_count, + } + ) + + def _compute_end_reason(self) -> str: + if self._run_state_fatal: + return "error" + if self.abort_requested_fn(): + return "error" if self._abort_cause == "error" else "user_abort" + if self._acquisition_error_count > 0: + return "completed_with_errors" + return "completed" + def run(self): this_image_callback_id = None + self._run_state_fatal = False try: start_time = time.perf_counter_ns() self.camera.start_streaming() @@ -457,6 +490,7 @@ def run(self): if self.abort_requested_fn(): self._log.debug("In run, abort_acquisition_requested=True") break + self._run_state_beat() # Gate on laser engine readiness for the channels this acquisition will fire. # Re-checked every timepoint so dt-induced sleep gaps are handled. @@ -502,6 +536,7 @@ def run(self): if self.abort_requested_fn(): self._log.debug("In run wait loop, abort_acquisition_requested=True") break + self._run_state_beat() self._sleep(sleep_time) elapsed_time = time.perf_counter_ns() - start_time @@ -513,9 +548,10 @@ def run(self): except TimeoutError as te: self._log.error(f"Operation timed out during acquisition, aborting acquisition!") self._log.error(te) - self.request_abort_fn() + self._abort_due_to_error() except Exception as e: self._log.exception(e) + self._run_state_fatal = True raise finally: # We do this above, but there are some paths that skip the proper end of the acquisition so make @@ -527,16 +563,29 @@ def run(self): self._finish_jobs() + # Determine why the acquisition ended (drives the watchdog + the in-process finish msg). + reason = self._compute_end_reason() + total_duration = time.time() - self.timestamp_acquisition_started + self._run_state.end( + reason, + { + "total_images": self.image_count, + "total_timepoints": self.time_point, + "total_duration_seconds": total_duration, + "errors_encountered": self._acquisition_error_count, + }, + ) + # Send Slack acquisition finished notification via callback (ensures ordering with timepoint notifications) if self._slack_notifier is not None: try: - total_duration = time.time() - self.timestamp_acquisition_started stats = AcquisitionStats( total_images=self.image_count, total_timepoints=self.time_point, total_duration_seconds=total_duration, errors_encountered=self._acquisition_error_count, experiment_id=self.experiment_ID or "unknown", + reason=reason, ) self.callbacks.signal_slack_acquisition_finished(stats) except Exception as e: @@ -1029,7 +1078,7 @@ def run_coordinate_acquisition(self, current_path): result = self._summarize_runner_outputs() if not result.none_failed and self._abort_on_failed_job: self._log.error("Some jobs failed, aborting acquisition because abort_on_failed_job=True") - self.request_abort_fn() + self._abort_due_to_error() return with self._timing.get_timer("move_to_coordinate"): @@ -1321,18 +1370,19 @@ def _image_callback(self, camera_frame: CameraFrame): self._ready_for_next_trigger.set() if not info: self._log.error("In image callback, no current capture info! Something is wrong. Aborting.") - self.request_abort_fn() + self._abort_due_to_error() return image = camera_frame.frame if not camera_frame or image is None: self._log.warning("image in frame callback is None. Something is really wrong, aborting!") - self.request_abort_fn() + self._abort_due_to_error() return # Increment image counter for Slack notification stats self._timepoint_image_count += 1 self.image_count += 1 + self._run_state_beat() with self._timing.get_timer("job creation and dispatch"): # Wait for subprocess to be ready before first dispatch @@ -1356,7 +1406,7 @@ def _image_callback(self, camera_frame: CameraFrame): if job_runner is not None: if not job_runner.dispatch(job): self._log.error("Failed to dispatch multiprocessing job!") - self.request_abort_fn() + self._abort_due_to_error() return else: try: @@ -1365,7 +1415,7 @@ def _image_callback(self, camera_frame: CameraFrame): result = job.run() except Exception: self._log.exception("Failed to execute job, abandoning acquisition!") - self.request_abort_fn() + self._abort_due_to_error() return height, width = image.shape[:2] @@ -1408,7 +1458,7 @@ def acquire_camera_image( with self._timing.get_timer("_ready_for_next_trigger.wait"): if not self._ready_for_next_trigger.wait(self._frame_wait_timeout_s()): self._log.error("Frame callback never set _have_last_triggered_image callback! Aborting acquisition.") - self.request_abort_fn() + self._abort_due_to_error() return # Backpressure check AFTER previous frame dispatched, BEFORE next trigger @@ -1468,7 +1518,7 @@ def acquire_camera_image( non_hw_frame_timeout = 5 * self.camera.get_total_frame_time() / 1e3 + 2 if not self._ready_for_next_trigger.wait(non_hw_frame_timeout): self._log.error("Timed out waiting {non_hw_frame_timeout} [s] for a frame, aborting acquisition.") - self.request_abort_fn() + self._abort_due_to_error() # Let this fall through so we still turn off illumination. Let the caller actually break out # of the acquisition. diff --git a/software/control/slack_notifier.py b/software/control/slack_notifier.py index 343cca5f1..7dad6d564 100644 --- a/software/control/slack_notifier.py +++ b/software/control/slack_notifier.py @@ -23,6 +23,7 @@ import control._def import squid.logging +import squid.slack log = squid.logging.get_logger(__name__) @@ -51,6 +52,7 @@ class AcquisitionStats: total_duration_seconds: float errors_encountered: int experiment_id: str + reason: str = "completed" @dataclass @@ -158,58 +160,7 @@ def _worker_loop(self): log.warning(f"Error in Slack worker loop: {e}") def _post_message(self, text: str, blocks: Optional[list] = None) -> Tuple[bool, Optional[str]]: - """Post a message to Slack using chat.postMessage API. - - Args: - text: Fallback text for the message. - blocks: Optional Block Kit blocks for rich formatting. - - Returns: - Tuple of (success, message_ts) where message_ts is the timestamp - of the posted message (used for threading). - """ - token = self.bot_token - channel = self.channel_id - - if not token or not channel: - log.debug("No Slack bot token or channel configured") - return False, None - - try: - log.info(f"Sending Slack message: {text[:50]}") - payload = { - "channel": channel, - "text": text, - } - if blocks: - payload["blocks"] = blocks - - data = json.dumps(payload).encode("utf-8") - request = urllib.request.Request( - f"{self.SLACK_API_BASE}/chat.postMessage", - data=data, - headers={ - "Content-Type": "application/json; charset=utf-8", - "Authorization": f"Bearer {token}", - }, - method="POST", - ) - - with urllib.request.urlopen(request, timeout=15) as response: - result = json.loads(response.read().decode("utf-8")) - if result.get("ok"): - log.info("Slack message sent successfully") - return True, result.get("ts") - else: - log.warning(f"Slack API error: {result.get('error')}") - return False, None - - except urllib.error.URLError as e: - log.warning(f"Failed to send Slack message: {e}") - return False, None - except Exception as e: - log.warning(f"Unexpected error sending Slack message: {e}") - return False, None + return squid.slack.post_message(self.bot_token, self.channel_id, text, blocks) def _upload_image( self, @@ -611,6 +562,9 @@ def notify_acquisition_finished(self, stats: AcquisitionStats): """Send an acquisition completion notification to Slack.""" if not control._def.SlackNotifications.NOTIFY_ON_ACQUISITION_FINISHED: return + if stats.reason != "completed": + # Premature/degraded ends are reported once by the acquisition watchdog. + return duration_str = self._format_duration(stats.total_duration_seconds) timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") diff --git a/software/control/widgets_slack.py b/software/control/widgets_slack.py index 65e13869c..dada77cc9 100644 --- a/software/control/widgets_slack.py +++ b/software/control/widgets_slack.py @@ -41,6 +41,7 @@ "send_mosaic_snapshots": ("SEND_MOSAIC_SNAPSHOTS", lambda: True), "notify_on_acquisition_start": ("NOTIFY_ON_ACQUISITION_START", lambda: True), "notify_on_acquisition_finished": ("NOTIFY_ON_ACQUISITION_FINISHED", lambda: True), + "watchdog_enabled": ("WATCHDOG_ENABLED", lambda: True), } @@ -156,6 +157,14 @@ def _setup_ui(self): self.checkbox_notify_finished.setToolTip("Send a Slack message when an acquisition completes") notif_layout.addWidget(self.checkbox_notify_finished) + self.checkbox_watchdog = QCheckBox("Enable watchdog alerts") + self.checkbox_watchdog.setToolTip( + "Let the standalone acquisition watchdog process post a Slack alert when a run ends " + "prematurely (crash / hang / error / abort). Uses the bot token + channel above and is " + "independent of the master toggle; requires the watchdog process to be running." + ) + notif_layout.addWidget(self.checkbox_watchdog) + notif_group.setLayout(notif_layout) layout.addWidget(notif_group) @@ -216,6 +225,7 @@ def _load_settings(self): "send_mosaic_snapshots": self.checkbox_send_mosaic, "notify_on_acquisition_start": self.checkbox_notify_start, "notify_on_acquisition_finished": self.checkbox_notify_finished, + "watchdog_enabled": self.checkbox_watchdog, } lineedit_map = { "bot_token": self.lineedit_bot_token, @@ -254,6 +264,7 @@ def _save_settings(self): control._def.SlackNotifications.SEND_MOSAIC_SNAPSHOTS = self.checkbox_send_mosaic.isChecked() control._def.SlackNotifications.NOTIFY_ON_ACQUISITION_START = self.checkbox_notify_start.isChecked() control._def.SlackNotifications.NOTIFY_ON_ACQUISITION_FINISHED = self.checkbox_notify_finished.isChecked() + control._def.SlackNotifications.WATCHDOG_ENABLED = self.checkbox_watchdog.isChecked() # Update notifier if available if self._slack_notifier: @@ -270,6 +281,7 @@ def _save_settings(self): "send_mosaic_snapshots": self.checkbox_send_mosaic.isChecked(), "notify_on_acquisition_start": self.checkbox_notify_start.isChecked(), "notify_on_acquisition_finished": self.checkbox_notify_finished.isChecked(), + "watchdog_enabled": self.checkbox_watchdog.isChecked(), } try: diff --git a/software/docs/slack_notifications.md b/software/docs/slack_notifications.md index 4157ce9fe..145857377 100644 --- a/software/docs/slack_notifications.md +++ b/software/docs/slack_notifications.md @@ -68,6 +68,7 @@ In the channel, type: | Include mosaic snapshots | Attach screenshot with timepoint notifications | | Notify on acquisition start | Send message when acquisition begins | | Notify on acquisition finished | Send summary when acquisition completes | +| Enable watchdog alerts | Let the standalone acquisition watchdog alert on a crashed / hung / aborted / errored run — independent of the master switch, and requires the watchdog process to be running (see `acquisition_watchdog/README.md`) | ## Notification Examples diff --git a/software/main_hcs.py b/software/main_hcs.py index 47ee7b886..dddc81816 100644 --- a/software/main_hcs.py +++ b/software/main_hcs.py @@ -435,5 +435,23 @@ def launch_claude_code(): # All hardware cleanup (camera, stage, microcontroller) happens in closeEvent, # which completes before os._exit() is called. exit_code = app.exec_() + + # If the app is quitting mid-acquisition, request the normal abort and let the worker + # write its end breadcrumb so the watchdog reports "aborted" rather than a crash. + try: + mpc = getattr(win, "multipointController", None) + if mpc is not None and mpc.acquisition_in_progress(): + log.info("Acquisition in progress at shutdown; requesting abort before exit.") + mpc.request_abort_aquisition() + if getattr(mpc, "thread", None) is not None: + mpc.thread.join(timeout=15.0) + if mpc.thread.is_alive(): + log.warning( + "Acquisition thread still alive 15s after abort request; exiting anyway — " + "the watchdog may report this run as a crash." + ) + except Exception as e: + log.warning(f"Error during shutdown abort handling: {e}") + logging.shutdown() # Flush log handlers before os._exit() bypasses Python cleanup os._exit(exit_code) diff --git a/software/squid/acquisition_state.py b/software/squid/acquisition_state.py new file mode 100644 index 000000000..fb1f09b64 --- /dev/null +++ b/software/squid/acquisition_state.py @@ -0,0 +1,148 @@ +# squid/acquisition_state.py +"""On-disk acquisition run-state breadcrumbs, shared by the acquisition engine +(writer) and the standalone acquisition watchdog (reader). + +Stdlib-only leaf module: must NOT import anything from `control`. +""" +import json +import os +import socket +import tempfile +import time +import uuid +from pathlib import Path +from typing import Optional + +import platformdirs + +import squid.logging + +_log = squid.logging.get_logger(__name__) + +SCHEMA_VERSION = 1 +HEARTBEAT_INTERVAL_S = 5.0 +RUN_FILE_NAME = "run.json" + + +def default_state_dir() -> Path: + """Per-user watchdog state dir, shared by writer and reader. + + Overridable via SQUID_WATCHDOG_STATE_DIR (honored by both processes). + """ + override = os.environ.get("SQUID_WATCHDOG_STATE_DIR") + if override: + return Path(override) + return Path(platformdirs.user_state_path("squid", "cephla")) / "watchdog" + + +def run_file_path(state_dir: Optional[Path] = None) -> Path: + return Path(state_dir) / RUN_FILE_NAME if state_dir else default_state_dir() / RUN_FILE_NAME + + +def _atomic_write_json(path: Path, data: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".run-", suffix=".tmp") + try: + with os.fdopen(fd, "w") as f: + json.dump(data, f) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, path) # atomic on POSIX and Windows + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def read_run(state_dir: Optional[Path] = None) -> Optional[dict]: + try: + with open(run_file_path(state_dir)) as f: + return json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + return None + + +class RunStateWriter: + """Writes/updates the single run.json for the current acquisition.""" + + def __init__(self, record: dict, state_dir: Optional[Path] = None): + self._record = record + self._state_dir = state_dir + self._last_beat = 0.0 + + @classmethod + def start( + cls, + *, + experiment_id: str, + pid: int, + config_path: Optional[str], + output_path: str, + expected: dict, + machine: Optional[str] = None, + state_dir: Optional[Path] = None, + ) -> "RunStateWriter": + now = time.time() + record = { + "schema_version": SCHEMA_VERSION, + "run_id": uuid.uuid4().hex, + "experiment_id": experiment_id, + "machine": machine or socket.gethostname(), + "pid": pid, + "config_path": config_path, + "output_path": output_path, + "started_at": now, + "heartbeat_at": now, + "progress": {}, + "expected": expected, + "status": "running", + "reason": None, + "ended_at": None, + "stats": None, + } + writer = cls(record, state_dir=state_dir) + writer._flush() + writer._last_beat = now + return writer + + @property + def run_id(self) -> Optional[str]: + return self._record.get("run_id") + + def beat(self, progress: Optional[dict] = None, force: bool = False) -> None: + now = time.time() + if not force and (now - self._last_beat) < HEARTBEAT_INTERVAL_S: + return + if progress: + self._record["progress"] = progress + self._last_beat = now + self._record["heartbeat_at"] = now + self._flush() + + def end(self, reason: str, stats: Optional[dict] = None) -> None: + self._record["status"] = "ended" + self._record["reason"] = reason + self._record["ended_at"] = time.time() + if stats is not None: + self._record["stats"] = stats + self._flush() + + def _flush(self) -> None: + try: + _atomic_write_json(run_file_path(self._state_dir), dict(self._record)) + except OSError as e: + _log.warning(f"Failed to write acquisition run state: {e}") + + +class NullRunStateWriter: + """No-op writer used when breadcrumbs are not wired (tests, side paths).""" + + run_id = None + + def beat(self, progress: Optional[dict] = None, force: bool = False) -> None: + pass + + def end(self, reason: str, stats: Optional[dict] = None) -> None: + pass diff --git a/software/squid/slack.py b/software/squid/slack.py new file mode 100644 index 000000000..6bd3eafb3 --- /dev/null +++ b/software/squid/slack.py @@ -0,0 +1,57 @@ +# squid/slack.py +"""Dependency-free Slack chat.postMessage sender. + +Shared by the in-process SlackNotifier (control/slack_notifier.py) and the +standalone acquisition watchdog. Stdlib only — safe to import without the +control/Qt/hardware stack. +""" +import json +import urllib.error +import urllib.request +from typing import Optional, Tuple + +import squid.logging + +_log = squid.logging.get_logger(__name__) + +SLACK_API_BASE = "https://slack.com/api" + + +def post_message( + bot_token: Optional[str], + channel_id: Optional[str], + text: str, + blocks: Optional[list] = None, + timeout: float = 15.0, +) -> Tuple[bool, Optional[str]]: + """Post a message to Slack. Returns (ok, message_ts).""" + if not bot_token or not channel_id: + _log.debug("No Slack bot token or channel configured") + return False, None + + payload = {"channel": channel_id, "text": text} + if blocks: + payload["blocks"] = blocks + data = json.dumps(payload).encode("utf-8") + request = urllib.request.Request( + f"{SLACK_API_BASE}/chat.postMessage", + data=data, + headers={ + "Content-Type": "application/json; charset=utf-8", + "Authorization": f"Bearer {bot_token}", + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + result = json.loads(response.read().decode("utf-8")) + if result.get("ok"): + return True, result.get("ts") + _log.warning(f"Slack API error: {result.get('error')}") + return False, None + except urllib.error.URLError as e: + _log.warning(f"Failed to send Slack message: {e}") + return False, None + except Exception as e: + _log.warning(f"Unexpected error sending Slack message: {e}") + return False, None diff --git a/software/tests/acquisition_watchdog/__init__.py b/software/tests/acquisition_watchdog/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/software/tests/acquisition_watchdog/test_alerts.py b/software/tests/acquisition_watchdog/test_alerts.py new file mode 100644 index 000000000..6acc33da4 --- /dev/null +++ b/software/tests/acquisition_watchdog/test_alerts.py @@ -0,0 +1,30 @@ +# tests/acquisition_watchdog/test_alerts.py +from acquisition_watchdog import alerts + + +def _run(): + return { + "experiment_id": "plateA_2026", + "machine": "micro-1", + "output_path": "/data/plateA_2026", + "progress": {"timepoint": 3, "expected_timepoints": 10, "images": 360}, + "expected": {"timepoints": 10}, + "started_at": 1_700_000_000.0, + "heartbeat_at": 1_700_000_100.0, + } + + +def test_format_alert_includes_key_facts(): + text, blocks = alerts.format_alert("crash", _run()) + assert "plateA_2026" in text + assert "micro-1" in text + blob = str(blocks) + assert "plateA_2026" in blob + assert "3" in blob and "10" in blob # progress vs expected + assert isinstance(blocks, list) and blocks + + +def test_format_alert_each_kind_has_title(): + for kind in ("crash", "hang", "error", "completed_with_errors", "user_abort"): + text, _ = alerts.format_alert(kind, _run()) + assert text # non-empty title line for every kind diff --git a/software/tests/acquisition_watchdog/test_cli.py b/software/tests/acquisition_watchdog/test_cli.py new file mode 100644 index 000000000..3c5d7bc51 --- /dev/null +++ b/software/tests/acquisition_watchdog/test_cli.py @@ -0,0 +1,19 @@ +# tests/acquisition_watchdog/test_cli.py +from unittest.mock import patch + +from acquisition_watchdog.__main__ import main + + +def test_once_runs_single_check(tmp_path): + with patch("acquisition_watchdog.monitor.Monitor.check_once") as check, patch( + "acquisition_watchdog.monitor.Monitor.run_forever" + ) as forever: + main(["--once", "--state-dir", str(tmp_path)]) + check.assert_called_once() + forever.assert_not_called() + + +def test_default_runs_forever(tmp_path): + with patch("acquisition_watchdog.monitor.Monitor.run_forever") as forever: + main(["--state-dir", str(tmp_path)]) + forever.assert_called_once() diff --git a/software/tests/acquisition_watchdog/test_config.py b/software/tests/acquisition_watchdog/test_config.py new file mode 100644 index 000000000..060104b7f --- /dev/null +++ b/software/tests/acquisition_watchdog/test_config.py @@ -0,0 +1,49 @@ +# tests/acquisition_watchdog/test_config.py +from pathlib import Path + +import yaml + +from acquisition_watchdog import config as wdconfig + + +def _write_yaml(path, data): + path.write_text(yaml.safe_dump(data)) + return path + + +def test_resolve_prefers_cli_then_env_then_default(monkeypatch): + monkeypatch.delenv("SQUID_SLACK_SETTINGS", raising=False) + assert wdconfig.resolve_slack_settings_path("/cli.yaml") == Path("/cli.yaml") + monkeypatch.setenv("SQUID_SLACK_SETTINGS", "/env.yaml") + assert wdconfig.resolve_slack_settings_path(None) == Path("/env.yaml") + monkeypatch.delenv("SQUID_SLACK_SETTINGS", raising=False) + assert wdconfig.resolve_slack_settings_path(None) == Path(wdconfig.DEFAULT_SLACK_SETTINGS) + + +def test_load_reads_credentials_from_yaml(tmp_path): + p = _write_yaml( + tmp_path / "slack_settings.yaml", + {"enabled": True, "bot_token": "xoxb-xyz", "channel_id": "C42"}, + ) + cfg = wdconfig.load_slack_config(p) + assert cfg.bot_token == "xoxb-xyz" + assert cfg.channel_id == "C42" + assert cfg.watchdog_enabled is True # default when key absent + + +def test_load_watchdog_enabled_false(tmp_path): + p = _write_yaml(tmp_path / "s.yaml", {"bot_token": "x", "channel_id": "C", "watchdog_enabled": False}) + assert wdconfig.load_slack_config(p).watchdog_enabled is False + + +def test_load_defaults_when_missing(tmp_path): + cfg = wdconfig.load_slack_config(tmp_path / "nope.yaml") + assert cfg.bot_token is None and cfg.channel_id is None + assert cfg.watchdog_enabled is True + assert wdconfig.load_slack_config(None).bot_token is None + + +def test_load_empty_strings_become_none(tmp_path): + p = _write_yaml(tmp_path / "s.yaml", {"bot_token": "", "channel_id": ""}) + cfg = wdconfig.load_slack_config(p) + assert cfg.bot_token is None and cfg.channel_id is None diff --git a/software/tests/acquisition_watchdog/test_monitor.py b/software/tests/acquisition_watchdog/test_monitor.py new file mode 100644 index 000000000..b59b82711 --- /dev/null +++ b/software/tests/acquisition_watchdog/test_monitor.py @@ -0,0 +1,75 @@ +# tests/acquisition_watchdog/test_monitor.py +import os +import time + +import squid.acquisition_state as ast +from acquisition_watchdog.config import SlackConfig +from acquisition_watchdog.monitor import Monitor + + +def _running(tmp_path, pid, heartbeat_age=0.0, run_id="r1"): + rec = { + "schema_version": 1, + "run_id": run_id, + "experiment_id": "e", + "machine": "m", + "pid": pid, + "config_path": None, + "output_path": "o", + "started_at": time.time() - 100, + "heartbeat_at": time.time() - heartbeat_age, + "progress": {}, + "expected": {}, + "status": "running", + "reason": None, + "ended_at": None, + "stats": None, + } + ast._atomic_write_json(ast.run_file_path(tmp_path), rec) + return rec + + +def _mon(tmp_path): + return Monitor(state_dir=tmp_path, heartbeat_timeout=120.0) + + +def test_running_with_live_pid_and_fresh_heartbeat_is_silent(tmp_path): + _running(tmp_path, pid=os.getpid(), heartbeat_age=1.0) + assert _mon(tmp_path).classify(ast.read_run(tmp_path), time.time()) is None + + +def test_dead_pid_is_crash(tmp_path): + _running(tmp_path, pid=2_000_000_000, heartbeat_age=1.0) # impossible pid + assert _mon(tmp_path).classify(ast.read_run(tmp_path), time.time()) == "crash" + + +def test_stale_heartbeat_with_live_pid_is_hang(tmp_path): + _running(tmp_path, pid=os.getpid(), heartbeat_age=999.0) + assert _mon(tmp_path).classify(ast.read_run(tmp_path), time.time()) == "hang" + + +def test_ended_reasons(tmp_path): + mon = _mon(tmp_path) + for reason, expect in [ + ("completed", None), + ("completed_with_errors", "completed_with_errors"), + ("error", "error"), + ("user_abort", "user_abort"), + ]: + run = {"run_id": f"x-{reason}", "status": "ended", "reason": reason} + assert mon.classify(run, time.time()) == expect + + +def test_dedup_persists_across_restart(tmp_path, monkeypatch): + _running(tmp_path, pid=2_000_000_000, run_id="dup1") + sent = [] + monkeypatch.setattr("squid.slack.post_message", lambda *a, **k: (sent.append(a) or (True, "1"))) + monkeypatch.setattr( + "acquisition_watchdog.config.load_slack_config", + lambda p: SlackConfig("xoxb", "C1", True), + ) + Monitor(state_dir=tmp_path).check_once(time.time()) + assert len(sent) == 1 + # Fresh Monitor (simulated restart) must not re-alert the same run_id. + Monitor(state_dir=tmp_path).check_once(time.time()) + assert len(sent) == 1 diff --git a/software/tests/control/conftest.py b/software/tests/control/conftest.py index 6ca8247f3..485098188 100644 --- a/software/tests/control/conftest.py +++ b/software/tests/control/conftest.py @@ -80,3 +80,9 @@ def firmware_sim_nonstrict(): sim = FirmwareSimSerial(strict=False) yield sim sim.close() + + +@pytest.fixture(autouse=True) +def _watchdog_state_to_tmp(tmp_path, monkeypatch): + # Keep acquisition breadcrumbs out of the real user state dir during tests. + monkeypatch.setenv("SQUID_WATCHDOG_STATE_DIR", str(tmp_path / "watchdog")) diff --git a/software/tests/control/test_notifier_trim.py b/software/tests/control/test_notifier_trim.py new file mode 100644 index 000000000..d8a8f50d0 --- /dev/null +++ b/software/tests/control/test_notifier_trim.py @@ -0,0 +1,30 @@ +# tests/control/test_notifier_trim.py +from unittest.mock import patch + +import control._def +from control.slack_notifier import SlackNotifier, AcquisitionStats + + +def _stats(reason): + return AcquisitionStats( + total_images=10, + total_timepoints=2, + total_duration_seconds=5.0, + errors_encountered=0, + experiment_id="e", + reason=reason, + ) + + +def test_finish_message_sent_only_on_clean_completion(monkeypatch): + monkeypatch.setattr(control._def.SlackNotifications, "NOTIFY_ON_ACQUISITION_FINISHED", True) + n = SlackNotifier(bot_token="x", channel_id="C") + with patch.object(n, "_queue_message") as q: + n.notify_acquisition_finished(_stats("completed")) + assert q.call_count == 1 + + with patch.object(n, "_queue_message") as q: + n.notify_acquisition_finished(_stats("error")) + n.notify_acquisition_finished(_stats("user_abort")) + n.notify_acquisition_finished(_stats("completed_with_errors")) + assert q.call_count == 0 # watchdog owns these alerts diff --git a/software/tests/control/test_slack_notifier_send.py b/software/tests/control/test_slack_notifier_send.py new file mode 100644 index 000000000..262d95bb0 --- /dev/null +++ b/software/tests/control/test_slack_notifier_send.py @@ -0,0 +1,11 @@ +# tests/control/test_slack_notifier_send.py +from unittest.mock import patch +from control.slack_notifier import SlackNotifier + + +def test_post_message_delegates_to_squid_slack(): + n = SlackNotifier(bot_token="xoxb-abc", channel_id="C999") + with patch("squid.slack.post_message", return_value=(True, "1.0")) as m: + ok, ts = n._post_message("hello", blocks=[{"type": "section"}]) + assert ok is True and ts == "1.0" + m.assert_called_once_with("xoxb-abc", "C999", "hello", [{"type": "section"}]) diff --git a/software/tests/control/test_watchdog_breadcrumbs.py b/software/tests/control/test_watchdog_breadcrumbs.py new file mode 100644 index 000000000..fc9b27daa --- /dev/null +++ b/software/tests/control/test_watchdog_breadcrumbs.py @@ -0,0 +1,19 @@ +# tests/control/test_watchdog_breadcrumbs.py +import os + +import squid.acquisition_state as ast +import control.microscope +import tests.control.gui_test_stubs as gts + + +def test_run_acquisition_writes_running_breadcrumb(qtbot): + scope = control.microscope.Microscope.build_from_global_config(True) + mpc = gts.get_test_qt_multi_point_controller(microscope=scope) + mpc.run_acquisition() + rec = ast.read_run(os.environ["SQUID_WATCHDOG_STATE_DIR"]) + assert rec is not None + assert rec["status"] == "running" + assert rec["pid"] == os.getpid() + assert rec["expected"]["timepoints"] >= 1 + mpc.request_abort_aquisition() + scope.close() diff --git a/software/tests/control/test_watchdog_integration.py b/software/tests/control/test_watchdog_integration.py new file mode 100644 index 000000000..1cf35bd78 --- /dev/null +++ b/software/tests/control/test_watchdog_integration.py @@ -0,0 +1,36 @@ +# tests/control/test_watchdog_integration.py +import os +import time + +import squid.acquisition_state as ast +import control.microscope +import tests.control.gui_test_stubs as gts + + +def _wait_for(predicate, timeout=30.0, interval=0.2): + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(interval) + return False + + +def test_simulated_acquisition_writes_ended_breadcrumb(qtbot): + state_dir = os.environ["SQUID_WATCHDOG_STATE_DIR"] + scope = control.microscope.Microscope.build_from_global_config(True) + mpc = gts.get_test_qt_multi_point_controller(microscope=scope) + + mpc.run_acquisition() + assert _wait_for(lambda: ast.read_run(state_dir) is not None) + assert ast.read_run(state_dir)["status"] == "running" + + # Abort and confirm the worker writes the end breadcrumb (deterministic path). + mpc.request_abort_aquisition() + assert _wait_for(lambda: (ast.read_run(state_dir) or {}).get("status") == "ended", timeout=30.0) + + rec = ast.read_run(state_dir) + assert rec["status"] == "ended" + assert rec["reason"] in {"completed", "completed_with_errors", "user_abort", "error"} + assert rec["ended_at"] is not None + scope.close() diff --git a/software/tests/control/test_worker_reason.py b/software/tests/control/test_worker_reason.py new file mode 100644 index 000000000..61ed5cd70 --- /dev/null +++ b/software/tests/control/test_worker_reason.py @@ -0,0 +1,41 @@ +# tests/control/test_worker_reason.py +from control.core.multi_point_worker import MultiPointWorker + + +def _make_worker(): + # Bypass __init__; we only exercise _compute_end_reason()'s pure logic. + w = MultiPointWorker.__new__(MultiPointWorker) + w._run_state_fatal = False + w._abort_cause = None + w._acquisition_error_count = 0 + w.abort_requested_fn = lambda: False + return w + + +def test_reason_completed(): + assert _make_worker()._compute_end_reason() == "completed" + + +def test_reason_user_abort(): + w = _make_worker() + w.abort_requested_fn = lambda: True + assert w._compute_end_reason() == "user_abort" + + +def test_reason_error_on_timeout_abort(): + w = _make_worker() + w.abort_requested_fn = lambda: True + w._abort_cause = "error" + assert w._compute_end_reason() == "error" + + +def test_reason_error_on_fatal_exception(): + w = _make_worker() + w._run_state_fatal = True + assert w._compute_end_reason() == "error" + + +def test_reason_completed_with_errors(): + w = _make_worker() + w._acquisition_error_count = 3 + assert w._compute_end_reason() == "completed_with_errors" diff --git a/software/tests/squid/test_acquisition_state.py b/software/tests/squid/test_acquisition_state.py new file mode 100644 index 000000000..552e4d934 --- /dev/null +++ b/software/tests/squid/test_acquisition_state.py @@ -0,0 +1,77 @@ +# tests/squid/test_acquisition_state.py +import json + +import squid.acquisition_state as ast + + +def _expected(): + return {"timepoints": 3, "regions": 1, "fovs": 4, "channels": 2, "z": 1} + + +def test_start_writes_running_record(tmp_path): + w = ast.RunStateWriter.start( + experiment_id="exp1", + pid=4321, + config_path="/cfg.ini", + output_path=str(tmp_path / "exp1"), + expected=_expected(), + machine="micro-1", + state_dir=tmp_path, + ) + rec = ast.read_run(tmp_path) + assert rec["status"] == "running" + assert rec["experiment_id"] == "exp1" + assert rec["pid"] == 4321 + assert rec["machine"] == "micro-1" + assert rec["expected"] == _expected() + assert rec["run_id"] == w.run_id + assert rec["reason"] is None + + +def test_beat_is_throttled_but_updates_progress_on_flush(tmp_path): + w = ast.RunStateWriter.start( + experiment_id="e", + pid=1, + config_path=None, + output_path="o", + expected=_expected(), + state_dir=tmp_path, + ) + first = ast.read_run(tmp_path)["heartbeat_at"] + # Immediate beat is throttled (< HEARTBEAT_INTERVAL_S since start) -> file unchanged. + w.beat({"timepoint": 1}) + assert ast.read_run(tmp_path)["heartbeat_at"] == first + # Forced beat flushes and records progress. + w.beat({"timepoint": 2}, force=True) + rec = ast.read_run(tmp_path) + assert rec["heartbeat_at"] >= first + assert rec["progress"] == {"timepoint": 2} + + +def test_end_records_reason_and_stats(tmp_path): + w = ast.RunStateWriter.start( + experiment_id="e", + pid=1, + config_path=None, + output_path="o", + expected=_expected(), + state_dir=tmp_path, + ) + w.end("user_abort", {"total_images": 7, "errors_encountered": 0}) + rec = ast.read_run(tmp_path) + assert rec["status"] == "ended" + assert rec["reason"] == "user_abort" + assert rec["ended_at"] is not None + assert rec["stats"]["total_images"] == 7 + + +def test_read_run_missing_returns_none(tmp_path): + assert ast.read_run(tmp_path) is None + + +def test_null_writer_is_noop(tmp_path): + w = ast.NullRunStateWriter() + w.beat({"timepoint": 1}) + w.end("completed", {}) + assert ast.read_run(tmp_path) is None + assert w.run_id is None diff --git a/software/tests/squid/test_slack.py b/software/tests/squid/test_slack.py new file mode 100644 index 000000000..2b5265cf2 --- /dev/null +++ b/software/tests/squid/test_slack.py @@ -0,0 +1,56 @@ +# tests/squid/test_slack.py +import json +from unittest.mock import patch, MagicMock + +import squid.slack as slack + + +def test_post_message_returns_false_without_credentials(): + assert slack.post_message(None, "C123", "hi") == (False, None) + assert slack.post_message("xoxb-1", None, "hi") == (False, None) + + +def test_post_message_builds_authorized_request_and_parses_ok(): + captured = {} + + class FakeResp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return json.dumps({"ok": True, "ts": "111.222"}).encode() + + def fake_urlopen(request, timeout=15): + captured["url"] = request.full_url + captured["headers"] = request.headers + captured["body"] = json.loads(request.data.decode()) + return FakeResp() + + with patch("urllib.request.urlopen", side_effect=fake_urlopen): + ok, ts = slack.post_message("xoxb-token", "C123", "hello", blocks=[{"type": "section"}]) + + assert ok is True and ts == "111.222" + assert captured["url"].endswith("/chat.postMessage") + assert captured["headers"]["Authorization"] == "Bearer xoxb-token" + assert captured["body"]["channel"] == "C123" + assert captured["body"]["text"] == "hello" + assert captured["body"]["blocks"] == [{"type": "section"}] + + +def test_post_message_handles_api_error(): + class FakeResp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return json.dumps({"ok": False, "error": "channel_not_found"}).encode() + + with patch("urllib.request.urlopen", return_value=FakeResp()): + ok, ts = slack.post_message("xoxb", "C1", "x") + assert ok is False and ts is None