diff --git a/configs/config_app_example.toml b/configs/config_app_example.toml index a7c7f428..a0682f69 100644 --- a/configs/config_app_example.toml +++ b/configs/config_app_example.toml @@ -45,10 +45,17 @@ channel1 = 1 channel2 = 2 dark_count = 0 -# Daily CHSH report settings (for automated Slack reporting) +# RNG / Quantum Fortune settings +[rng_settings] +channels = [1, 2] # Timetagger channels to sample for singles parity +fortune_size = 8 # Number of parity measurements per fortune run + +# Daily report settings (for automated Slack reporting of hardware + games) [daily_report] slack_webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL" # Get from https://api.slack.com/apps api_url = "http://localhost:8000" # API endpoint (usually localhost if running on same machine) timetagger_address = "127.0.0.1:8000" # TimeTagger address follower_node_address = "192.168.1.100:9000" # Replace with actual follower node address -basis = [0, 22.5] # CHSH basis angles to use for daily measurements \ No newline at end of file +basis = [0, 22.5] # CHSH basis angles to use for daily measurements +overall_timeout_s = 1800 # Hard watchdog for the whole run (SIGALRM). Posts a Slack error and exits if exceeded. +per_game_timeout_s = 600 # Per-game HTTP timeout. \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 4900f0e2..9219e95f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,17 +22,14 @@ dependencies = [ "thorlabs-apt-device>=0.3.8", "tomli-w>=1.0.0", "typer>=0.15.1", + "fastapi[standard]>=0.115.14", + "httpx>=0.28.1", + "pydantic-settings>=2.10.1", ] [project.scripts] pqn = "pqnstack.cli:app" -[project.optional-dependencies] -webapp = [ - "fastapi[standard]>=0.115.14", - "httpx>=0.28.1", - "pydantic-settings>=2.10.1", -] [dependency-groups] dev = ["hypothesis", "mypy", "coverage", "pytest-randomly", "ruff"] diff --git a/scripts/chsh_daily_report.py b/scripts/chsh_daily_report.py deleted file mode 100755 index 9e445748..00000000 --- a/scripts/chsh_daily_report.py +++ /dev/null @@ -1,199 +0,0 @@ -#!/usr/bin/env python3 -""" -CHSH Daily Report Script. - -Runs CHSH measurement and posts results to Slack. - -Reads all configuration from config.toml (including Slack webhook URL). - -Usage: - uv run scripts/chsh_daily_report.py -""" - -import logging -import sys -import tomllib -from datetime import UTC -from datetime import datetime -from pathlib import Path - -import httpx -from pydantic import ValidationError - -from pqnstack.app.api.routes.chsh import ChshResult -from pqnstack.app.core.config import DailyReportConfig - -logger = logging.getLogger(__name__) - - -def load_config() -> DailyReportConfig: - """Load and validate the [daily_report] section from config.toml.""" - config_path = Path(__file__).parent.parent / "config.toml" - - if not config_path.exists(): - logger.error("config.toml not found at %s", config_path) - logger.error("Please create config.toml from configs/config_app_example.toml") - sys.exit(1) - - with config_path.open("rb") as f: - raw = tomllib.load(f) - - daily_report_data = raw.get("daily_report") - if not daily_report_data: - logger.error("[daily_report] section not found in config.toml") - logger.error("Please add it following the example in configs/config_app_example.toml") - sys.exit(1) - - try: - return DailyReportConfig.model_validate(daily_report_data) - except ValidationError: - logger.exception("Invalid [daily_report] configuration") - sys.exit(1) - - -def run_chsh_measurement(config: DailyReportConfig) -> ChshResult: - """Run CHSH measurement via API.""" - logger.info("Starting CHSH measurement at %s", datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S")) - logger.info("Basis: %s", config.basis) - logger.info("Follower: %s", config.follower_node_address) - logger.info("TimeTagger: %s", config.timetagger_address) - - try: - with httpx.Client(timeout=600.0) as client: - response = client.post( - f"{config.api_url}/chsh/", - params={ - "follower_node_address": config.follower_node_address, - "timetagger_address": config.timetagger_address, - }, - json=config.basis, - ) - response.raise_for_status() - return ChshResult.model_validate(response.json()) - - except httpx.HTTPError: - logger.exception("Failed to contact CHSH API") - sys.exit(1) - - -def post_to_slack(webhook_url: str, chsh_data: ChshResult, config: DailyReportConfig) -> None: - """Post CHSH results to Slack.""" - # Determine emoji based on Bell inequality violation (CHSH > classical limit) - bell_inequality_classical_limit = 2 - emoji = ":sparkles:" if chsh_data.chsh_value > bell_inequality_classical_limit else ":thinking_face:" - - # Build fields dynamically from all returned data - fields = [] - for key, value in chsh_data.model_dump().items(): - field_name = key.replace("_", " ").title() - - if isinstance(value, float): - formatted_value = f"{value:.4f}" - elif isinstance(value, list): - if all(isinstance(x, (int, float)) for x in value): - formatted_value = "[" + ", ".join(f"{x:.4f}" if isinstance(x, float) else str(x) for x in value) + "]" - else: - formatted_value = str(value) - else: - formatted_value = str(value) - - fields.append({"type": "mrkdwn", "text": f"*{field_name}:*\n`{formatted_value}`"}) - - # Create sections with 2 fields each (Slack limit) - sections = [] - for i in range(0, len(fields), 2): - section_fields = fields[i : i + 2] - sections.append({"type": "section", "fields": section_fields}) - - # Add configuration info section - sections.append( - { - "type": "section", - "fields": [ - {"type": "mrkdwn", "text": f"*Basis:*\n`{config.basis}`"}, - {"type": "mrkdwn", "text": f"*Timestamp:*\n{datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S')}"}, - ], - } - ) - - # Format Slack message using Block Kit - slack_message = { - "blocks": [ - { - "type": "header", - "text": {"type": "plain_text", "text": f"{emoji} CHSH Daily Measurement Report", "emoji": True}, - }, - *sections, - { - "type": "context", - "elements": [ - { - "type": "mrkdwn", - "text": f"Follower: `{config.follower_node_address}` | TimeTagger: `{config.timetagger_address}`", - } - ], - }, - ] - } - - logger.info("Posting to Slack...") - - try: - with httpx.Client() as client: - response = client.post(webhook_url, json=slack_message) - - if response.text == "ok": - logger.info("Successfully posted to Slack") - else: - logger.error("Failed to post to Slack: %s", response.text) - sys.exit(1) - - except httpx.HTTPError: - logger.exception("Failed to post to Slack") - sys.exit(1) - - -def post_error_to_slack(webhook_url: str, error_message: str) -> None: - """Post error message to Slack.""" - slack_message = { - "text": f":x: CHSH Daily Report Failed\n*Error:* {error_message}\n*Time:* {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S')}" - } - - try: - with httpx.Client() as client: - client.post(webhook_url, json=slack_message) - except httpx.HTTPError: - logger.debug("Failed to post error notification to Slack") - - -def main() -> None: - """Execute the CHSH daily report.""" - logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") - - try: - config = load_config() - - chsh_data = run_chsh_measurement(config) - - logger.info("CHSH measurement completed") - logger.info("Value: %.4f ± %.4f", chsh_data.chsh_value, chsh_data.chsh_error) - - post_to_slack(config.slack_webhook_url, chsh_data, config) - - logger.info("CHSH daily report completed successfully") - - except Exception as e: - logger.exception("Unexpected error") - - # Try to post error to Slack if possible - try: - config = load_config() - post_error_to_slack(config.slack_webhook_url, str(e)) - except Exception: # noqa: BLE001 - logger.debug("Failed to post error notification to Slack") - - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/src/pqnstack/app/api/main.py b/src/pqnstack/app/api/main.py index 34ee765e..2879dd07 100644 --- a/src/pqnstack/app/api/main.py +++ b/src/pqnstack/app/api/main.py @@ -4,6 +4,7 @@ from pqnstack.app.api.routes import coordination from pqnstack.app.api.routes import debug from pqnstack.app.api.routes import games +from pqnstack.app.api.routes import health from pqnstack.app.api.routes import qkd from pqnstack.app.api.routes import rng from pqnstack.app.api.routes import serial @@ -18,3 +19,4 @@ api_router.include_router(coordination.router) api_router.include_router(debug.router) api_router.include_router(games.router) +api_router.include_router(health.router) diff --git a/src/pqnstack/app/api/routes/chsh.py b/src/pqnstack/app/api/routes/chsh.py index c5250aa3..1fb48cd5 100644 --- a/src/pqnstack/app/api/routes/chsh.py +++ b/src/pqnstack/app/api/routes/chsh.py @@ -166,7 +166,6 @@ async def _chsh( # Complexity is high due to the nature of the CHSH experiment. expectation_values_sign_fixed = [ x * y for x, y in zip(expectation_values, settings.chsh_settings.expectation_signs, strict=False) ] - logger.info("What are you settings? %s", settings.chsh_settings.expectation_signs) logger.info("After passing signed calculation: %s", expectation_values_sign_fixed) chsh_value = abs(sum(x for x in expectation_values_sign_fixed)) diff --git a/src/pqnstack/app/api/routes/health.py b/src/pqnstack/app/api/routes/health.py new file mode 100644 index 00000000..607ebc47 --- /dev/null +++ b/src/pqnstack/app/api/routes/health.py @@ -0,0 +1,186 @@ +import logging +import time +from typing import Annotated + +import httpx +import serial +from fastapi import APIRouter +from fastapi import Query +from pydantic import BaseModel +from pydantic import Field + +from pqnstack.app.core.config import settings +from pqnstack.network.client import Client + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/health", tags=["health"]) + +_ROUTER_TIMEOUT_MS = 5000 +_FOLLOWER_TIMEOUT_S = 5.0 + + +class ComponentStatus(BaseModel): + reachable: bool + error: str | None = None + latency_ms: float | None = None + + +class DeviceStatus(ComponentStatus): + provider: str + name: str + purpose: str # human-readable label describing what the device is used for + + +class HealthStatus(BaseModel): + router: ComponentStatus + devices: list[DeviceStatus] = Field(default_factory=list) + rotary_encoder: ComponentStatus | None = None + follower_node: ComponentStatus | None = None + + @property + def all_ok(self) -> bool: + if not self.router.reachable: + return False + if any(not d.reachable for d in self.devices): + return False + if self.rotary_encoder is not None and not self.rotary_encoder.reachable: + return False + return not (self.follower_node is not None and not self.follower_node.reachable) + + +def _elapsed_ms(start: float) -> float: + return (time.perf_counter() - start) * 1000 + + +def _format_error(exc: BaseException) -> str: + return f"{type(exc).__name__}: {exc}" + + +def _connect_router() -> tuple[ComponentStatus, Client | None]: + start = time.perf_counter() + try: + client = Client( + host=settings.router_address, + port=settings.router_port, + router_name=settings.router_name, + timeout=_ROUTER_TIMEOUT_MS, + ) + except Exception as e: # noqa: BLE001 - any failure to connect must be reported, not swallowed + return ComponentStatus(reachable=False, error=_format_error(e)), None + return ComponentStatus(reachable=True, latency_ms=_elapsed_ms(start)), client + + +def _configured_devices() -> list[tuple[str, str, str]]: + """Return deduplicated (provider, name, purpose) triples for all configured devices. + + HWP fields default to ("", "") when unconfigured; those are filtered out. + Timetagger is optional (None means unused). When two settings share the same + (provider, name) pair their purposes are merged — e.g. "CHSH HWP / QKD HWP" — + so each physical device appears exactly once in the health report. + """ + labeled: list[tuple[str, str, str]] = [ + (*settings.chsh_settings.hwp, "CHSH leader HWP"), + (*settings.chsh_settings.request_hwp, "CHSH follower HWP"), + (*settings.qkd_settings.hwp, "QKD leader HWP"), + (*settings.qkd_settings.request_hwp, "QKD follower HWP"), + *([(settings.timetagger[0], settings.timetagger[1], "Timetagger")] if settings.timetagger else []), + ] + # Preserve insertion order while merging purposes for duplicate (provider, name) pairs. + merged: dict[tuple[str, str], list[str]] = {} + for provider, name, purpose in labeled: + if not provider or not name: + continue + key = (provider, name) + merged.setdefault(key, []).append(purpose) + return [(provider, name, " / ".join(purposes)) for (provider, name), purposes in merged.items()] + + +def _probe_devices(client: Client) -> list[DeviceStatus]: + configured = _configured_devices() + # Group by provider so we make one get_available_devices call per provider. + by_provider: dict[str, list[tuple[str, str]]] = {} + for provider, name, purpose in configured: + by_provider.setdefault(provider, []).append((name, purpose)) + + results: list[DeviceStatus] = [] + for provider, name_purpose_pairs in by_provider.items(): + start = time.perf_counter() + try: + available = client.get_available_devices(provider) + except Exception as e: # noqa: BLE001 - any failure must surface as device status, not a crash + err = _format_error(e) + results.extend( + DeviceStatus(provider=provider, name=name, purpose=purpose, reachable=False, error=err) + for name, purpose in name_purpose_pairs + ) + continue + latency = _elapsed_ms(start) + for name, purpose in name_purpose_pairs: + if name in available: + results.append( + DeviceStatus(provider=provider, name=name, purpose=purpose, reachable=True, latency_ms=latency) + ) + else: + results.append( + DeviceStatus( + provider=provider, + name=name, + purpose=purpose, + reachable=False, + error=f"device '{name}' not registered on provider '{provider}'", + ) + ) + return results + + +def _probe_rotary_encoder() -> ComponentStatus | None: + if settings.virtual_rotator: + return None + start = time.perf_counter() + try: + with serial.Serial(settings.rotary_encoder_address, 115200, timeout=1): + pass + except Exception as e: # noqa: BLE001 - any failure must surface, not crash the endpoint + return ComponentStatus(reachable=False, error=_format_error(e)) + return ComponentStatus(reachable=True, latency_ms=_elapsed_ms(start)) + + +def _probe_follower(follower_node_address: str) -> ComponentStatus: + start = time.perf_counter() + try: + with httpx.Client(timeout=_FOLLOWER_TIMEOUT_S) as http: + response = http.get(f"http://{follower_node_address}/") + response.raise_for_status() + except Exception as e: # noqa: BLE001 - any failure must surface, not crash the endpoint + return ComponentStatus(reachable=False, error=_format_error(e)) + return ComponentStatus(reachable=True, latency_ms=_elapsed_ms(start)) + + +@router.get("/") +def health( + follower_node_address: Annotated[str | None, Query()] = None, +) -> HealthStatus: + """Probe router, configured devices, rotary encoder, and optional follower node.""" + router_status, client = _connect_router() + + if client is not None: + try: + devices = _probe_devices(client) + finally: + client.disconnect() + else: + devices = [ + DeviceStatus(provider=provider, name=name, purpose=purpose, reachable=False, error="router unreachable") + for provider, name, purpose in _configured_devices() + ] + + rotary_encoder = _probe_rotary_encoder() + follower_node = _probe_follower(follower_node_address) if follower_node_address else None + + return HealthStatus( + router=router_status, + devices=devices, + rotary_encoder=rotary_encoder, + follower_node=follower_node, + ) diff --git a/src/pqnstack/app/api/routes/rng.py b/src/pqnstack/app/api/routes/rng.py index 200ac0ad..7823de08 100644 --- a/src/pqnstack/app/api/routes/rng.py +++ b/src/pqnstack/app/api/routes/rng.py @@ -14,6 +14,7 @@ from pqnstack.app.api.deps import ClientDep from pqnstack.app.api.deps import StateDep from pqnstack.app.core.config import rng_progress_event +from pqnstack.app.core.config import settings logger = logging.getLogger(__name__) @@ -101,28 +102,35 @@ async def singles_parity( @router.get("/fortune") async def fortune( # noqa: PLR0913 timetagger_address: str, - integration_time_s: float, - fortune_size: int, http_client: ClientDep, state: StateDep, - channels: Annotated[list[int], Query()], + fortune_size: int | None = None, + integration_time_s: float = 1.0, + channels: Annotated[list[int] | None, Query()] = None, ) -> list[int]: - """Run singles parity `fortune_size` times and, per channel, interpret the result in bitstring as a decimal number.""" - if fortune_size <= 0: + """Run singles parity `fortune_size` times and, per channel, interpret the result in bitstring as a decimal number. + + `fortune_size` and `channels` default to the node's configured `rng_settings` when omitted. + """ + resolved_fortune_size = fortune_size if fortune_size is not None else settings.rng_settings.fortune_size + if resolved_fortune_size <= 0: raise HTTPException(status_code=400, detail="fortune_size must be a positive integer") + resolved_channels = channels if channels is not None else settings.rng_settings.channels + resolved_integration_time_s = integration_time_s + # Initialize progress tracking state.rng_running = True state.rng_progress_current = 0 - state.rng_progress_total = fortune_size + state.rng_progress_total = resolved_fortune_size rng_progress_event.set() trials: list[list[int]] = [] - for _ in range(fortune_size): + for _ in range(resolved_fortune_size): params: list[tuple[str, str | int | float | bool | None]] = [ ("timetagger_address", timetagger_address), - ("integration_time_s", integration_time_s), - *[("channels", ch) for ch in channels], + ("integration_time_s", resolved_integration_time_s), + *[("channels", ch) for ch in resolved_channels], ] url = f"http://{timetagger_address}/rng/singles_parity" @@ -142,8 +150,8 @@ async def fortune( # noqa: PLR0913 logger.info( "Fortune results (channels=%s, fortune_size=%d): %s", - channels, - fortune_size, + resolved_channels, + resolved_fortune_size, results, ) diff --git a/src/pqnstack/app/core/config.py b/src/pqnstack/app/core/config.py index 412bc191..21474459 100644 --- a/src/pqnstack/app/core/config.py +++ b/src/pqnstack/app/core/config.py @@ -23,6 +23,13 @@ class DailyReportConfig(BaseModel): api_url: str = "http://localhost:8000" timetagger_address: str = "127.0.0.1:8000" basis: list[float] = Field(default_factory=lambda: [0.0, 22.5]) + overall_timeout_s: int = 1800 + per_game_timeout_s: int = 600 + + +class RNGSettings(BaseModel): + channels: list[int] = Field(default_factory=lambda: [1, 2]) + fortune_size: int = 8 class CHSHSettings(BaseModel): @@ -56,7 +63,9 @@ class Settings(BaseSettings): router_port: int = 5555 chsh_settings: CHSHSettings = CHSHSettings() qkd_settings: QKDSettings = QKDSettings() + rng_settings: RNGSettings = RNGSettings() bell_state: BellState = BellState.Phi_plus + daily_report: DailyReportConfig | None = None timetagger: tuple[str, str] | None = None # Name of the timetagger to use for the CHSH experiment. rotary_encoder_address: str = "/dev/ttyACM0" virtual_rotator: bool = False # If True, use terminal input instead of hardware rotary encoder diff --git a/src/pqnstack/app/cron_manager.py b/src/pqnstack/app/cron_manager.py new file mode 100644 index 00000000..6322d3a2 --- /dev/null +++ b/src/pqnstack/app/cron_manager.py @@ -0,0 +1,82 @@ +"""Manage the PQN daily-report cron job via the system crontab. + +Reads and writes the user's crontab using `crontab -l` / `crontab -` so that +no extra dependencies are required. The managed entry is identified by the +inline tag `# PQN_DAILY_REPORT` appended to the cron line. +""" + +from __future__ import annotations + +import shutil +import subprocess + +CRON_TAG = "# PQN_DAILY_REPORT" + +_DOW_NAMES = ["sunday", "monday", "tuesday", "wednesday", "thursday", "friday", "saturday"] + + +def _read_crontab() -> list[str]: + result = subprocess.run(["crontab", "-l"], check=False, capture_output=True, text=True) # noqa: S607 + if result.returncode != 0: + return [] + return result.stdout.splitlines() + + +def _write_crontab(lines: list[str]) -> None: + content = "\n".join(lines) + "\n" + subprocess.run(["crontab", "-"], input=content, text=True, check=True) # noqa: S607 + + +def get_daily_report_job() -> str | None: + """Return the tagged cron line, or None if not present.""" + for line in _read_crontab(): + if line.endswith(CRON_TAG): + return line + return None + + +def describe_schedule(cron_line: str) -> str: + """Parse a tagged cron line and return a human-readable schedule string.""" + fields = cron_line.split() + if len(fields) < 5: # noqa: PLR2004 + return cron_line + + minute, hour, dom, _month, dow = fields[:5] + + if minute != "*" and hour == "*" and dom == "*" and dow == "*": + return f"Hourly at :{int(minute):02d}" + + if minute != "*" and hour != "*" and dom == "*" and dow == "*": + return f"Daily at {int(hour):02d}:{int(minute):02d}" + + if minute != "*" and hour != "*" and dom == "*" and dow != "*": + day_name = _DOW_NAMES[int(dow)].capitalize() + return f"Weekly on {day_name} at {int(hour):02d}:{int(minute):02d}" + + if minute != "*" and hour != "*" and dom != "*" and dow == "*": + return f"Monthly on day {dom} at {int(hour):02d}:{int(minute):02d}" + + return cron_line + + +def set_daily_report_schedule(minute: int, hour: int | str, dow: str, dom: str) -> None: + """Replace (or create) the tagged cron entry with the given schedule.""" + pqn = shutil.which("pqn") + if pqn is None: + msg = "Could not find 'pqn' executable on PATH" + raise RuntimeError(msg) + + new_line = f"{minute} {hour} {dom} * {dow} {pqn} daily-report run {CRON_TAG}" + lines = [line for line in _read_crontab() if not line.endswith(CRON_TAG)] + lines.append(new_line) + _write_crontab(lines) + + +def remove_daily_report_job() -> bool: + """Remove the tagged cron entry. Returns True if an entry was removed.""" + lines = _read_crontab() + filtered = [line for line in lines if not line.endswith(CRON_TAG)] + if len(filtered) == len(lines): + return False + _write_crontab(filtered) + return True diff --git a/src/pqnstack/app/daily_report.py b/src/pqnstack/app/daily_report.py new file mode 100644 index 00000000..23866dff --- /dev/null +++ b/src/pqnstack/app/daily_report.py @@ -0,0 +1,486 @@ +"""Daily health + games report posted to Slack. + +Run via `pqn daily-report` (see pqnstack.cli). Loads `DailyReportConfig`, probes +hardware via `/health`, exercises each enabled game except SSM, and posts a +single consolidated Slack digest. +""" + +from __future__ import annotations + +import logging +import os +import signal +from dataclasses import dataclass +from dataclasses import field +from datetime import UTC +from datetime import datetime +from typing import TYPE_CHECKING +from typing import Any + +import httpx + +from pqnstack.app.api.routes.chsh import ChshResult +from pqnstack.app.api.routes.health import ComponentStatus +from pqnstack.app.api.routes.health import HealthStatus + +if TYPE_CHECKING: + from types import FrameType + + from pqnstack.app.core.config import DailyReportConfig + +logger = logging.getLogger(__name__) + +_SSM_SKIP_REASON = "SSM skipped in daily report (requires coordination dance; run manually)." +_BELL_CLASSICAL_LIMIT = 2.0 + +_watchdog_state: dict[str, Any] = {} + + +@dataclass +class GameResult: + name: str + title: str + status: str # "ok" | "failed" | "skipped" + data: dict[str, Any] | list[Any] | None = None + error: str | None = None + elapsed_s: float = 0.0 + emoji: str = "" + + +@dataclass +class ReportResult: + hardware: HealthStatus | None = None + hardware_error: str | None = None + games: list[GameResult] = field(default_factory=list) + skipped_notes: list[str] = field(default_factory=list) + + @property + def overall_ok(self) -> bool: + hw_ok = self.hardware_error is None and self.hardware is not None and self.hardware.all_ok + games_ok = all(g.status == "ok" for g in self.games) + return hw_ok and games_ok + + +def _watchdog_handler(_signum: int, _frame: FrameType | None) -> None: + """SIGALRM handler: post a Slack alert and force-kill the process. + + Uses os._exit instead of sys.exit so that atexit handlers, finally blocks, + and Python's own shutdown sequence can't delay or swallow the kill — a hung + process is precisely what we're trying to escape. Reads config from the + module-level dict because signal handlers can't accept arbitrary arguments. + """ + webhook = _watchdog_state.get("webhook_url") + timeout_s = _watchdog_state.get("timeout_s") + message = ( + f":x: *Daily report watchdog fired* — exceeded overall timeout of `{timeout_s}s`. " + "Process was killed; investigate the API / hardware." + ) + if webhook: + try: + _post_plain_slack(webhook, message) + except Exception: + logger.exception("Failed to post watchdog notification") + logger.error("Daily report watchdog fired after %ss; exiting", timeout_s) + os._exit(1) + + +def _arm_watchdog(config: DailyReportConfig) -> None: + """Install the SIGALRM watchdog for the entire run. + + SIGALRM is the right tool here: it fires even if the process is blocked in + a syscall (e.g. waiting on a socket), where threading-based timeouts can't + reach. Stash config in module state so the signal handler can read it. + """ + _watchdog_state["webhook_url"] = config.slack_webhook_url + _watchdog_state["timeout_s"] = config.overall_timeout_s + signal.signal(signal.SIGALRM, _watchdog_handler) + signal.alarm(config.overall_timeout_s) + + +def _disarm_watchdog() -> None: + """Cancel the watchdog once the run completes normally.""" + signal.alarm(0) + + +def _check_api(config: DailyReportConfig) -> bool: + """Return True if the API is reachable, False otherwise. + + Runs before any other probe so that a single "API is down" message replaces + the cascade of connection-refused errors that would otherwise fill the digest. + """ + try: + with httpx.Client(timeout=5.0) as client: + client.get(f"{config.api_url}/") + except httpx.ConnectError: + return False + return True + + +def _fetch_hardware(config: DailyReportConfig) -> tuple[HealthStatus | None, str | None]: + """Probe the node's hardware health and return (result, error). + + Returns a tuple rather than raising so that a failed hardware probe doesn't + abort the whole run — we still want to attempt the games and post a digest + that tells operators what's broken. + """ + url = f"{config.api_url}/health/" + try: + with httpx.Client(timeout=30.0) as client: + response = client.get(url, params={"follower_node_address": config.follower_node_address}) + response.raise_for_status() + return HealthStatus.model_validate(response.json()), None + except httpx.HTTPError as e: + logger.exception("Hardware health probe failed") + return None, f"{type(e).__name__}: {e}" + + +def _fetch_availability(config: DailyReportConfig) -> dict[str, bool]: + """Return which games are enabled according to the running node. + + Defaults to all-enabled on failure so a transient availability fetch error + doesn't silently skip every game — better to attempt and fail visibly than + to skip silently and miss the problem entirely. + """ + url = f"{config.api_url}/games/availability" + try: + with httpx.Client(timeout=10.0) as client: + response = client.get(url) + response.raise_for_status() + payload: dict[str, bool] = response.json() + except httpx.HTTPError: + logger.exception("Failed to fetch games availability; assuming all enabled") + return {"chsh": True, "qf": True, "ssm": True} + return payload + + +def _run_chsh(config: DailyReportConfig) -> GameResult: + """Run one CHSH measurement and return a structured result. + + Catches all exceptions (not just httpx) because pydantic validation errors + on an unexpected server response shape are equally fatal to the game result + and must be captured here rather than crashing the whole report. + """ + started = datetime.now(UTC) + try: + with httpx.Client(timeout=float(config.per_game_timeout_s)) as client: + response = client.post( + f"{config.api_url}/chsh/", + params={ + "follower_node_address": config.follower_node_address, + "timetagger_address": config.timetagger_address, + }, + json=config.basis, + ) + response.raise_for_status() + parsed = ChshResult.model_validate(response.json()) + except httpx.HTTPError as e: + return GameResult( + name="chsh", + title="CHSH — Verify Quantum Link", + status="failed", + error=f"{type(e).__name__}: {e}", + elapsed_s=(datetime.now(UTC) - started).total_seconds(), + ) + except Exception as e: # noqa: BLE001 - surface any parsing / validation error, don't crash the report + return GameResult( + name="chsh", + title="CHSH — Verify Quantum Link", + status="failed", + error=f"{type(e).__name__}: {e}", + elapsed_s=(datetime.now(UTC) - started).total_seconds(), + ) + + emoji = ":sparkles:" if parsed.chsh_value > _BELL_CLASSICAL_LIMIT else ":thinking_face:" + return GameResult( + name="chsh", + title="CHSH — Verify Quantum Link", + status="ok", + data=parsed.model_dump(), + elapsed_s=(datetime.now(UTC) - started).total_seconds(), + emoji=emoji, + ) + + +def _run_qf(config: DailyReportConfig) -> GameResult: + """Run one Quantum Fortune measurement and return a structured result. + + Omits channels and integration_time_s so the node falls back to its own + rng_settings — the daily report shouldn't override per-node calibration. + """ + started = datetime.now(UTC) + # channels and integration_time_s are omitted — the node uses its configured rng_settings defaults. + params: dict[str, str] = {"timetagger_address": config.timetagger_address} + try: + with httpx.Client(timeout=float(config.per_game_timeout_s)) as client: + response = client.get(f"{config.api_url}/rng/fortune", params=params) + response.raise_for_status() + payload = response.json() + except httpx.HTTPError as e: + return GameResult( + name="qf", + title="Quantum Fortune", + status="failed", + error=f"{type(e).__name__}: {e}", + elapsed_s=(datetime.now(UTC) - started).total_seconds(), + ) + + return GameResult( + name="qf", + title="Quantum Fortune", + status="ok", + data={"fortune_per_channel": payload}, + elapsed_s=(datetime.now(UTC) - started).total_seconds(), + emoji=":game_die:", + ) + + +def _format_value(value: Any) -> str: + """Render a scalar or numeric list as a human-readable string for Slack.""" + if isinstance(value, float): + return f"{value:.4f}" + if isinstance(value, list): + if all(isinstance(x, (int, float)) for x in value): + return "[" + ", ".join(f"{x:.4f}" if isinstance(x, float) else str(x) for x in value) + "]" + return str(value) + return str(value) + + +def _fields_from_dict(data: dict[str, Any]) -> list[dict[str, str]]: + """Convert an arbitrary result dict to Slack mrkdwn field blocks. + + Operates on the raw dict rather than a typed model so CHSH and QF results + share one formatting path without coupling this layer to specific schemas. + """ + fields: list[dict[str, str]] = [] + for key, value in data.items(): + label = key.replace("_", " ").title() + fields.append({"type": "mrkdwn", "text": f"*{label}:*\n`{_format_value(value)}`"}) + return fields + + +def _sections_from_fields(fields: list[dict[str, str]]) -> list[dict[str, Any]]: + """Slack section blocks cap fields at 10 per block; chunk accordingly.""" + chunk = 10 + return [{"type": "section", "fields": fields[i : i + chunk]} for i in range(0, len(fields), chunk)] + + +def _component_line(label: str, status: ComponentStatus) -> str: + """Format one hardware component as a single Slack line. + + Accepts the base ComponentStatus type so it works for both plain components + (router, rotary encoder) and DeviceStatus without needing an overload. + """ + emoji = ":white_check_mark:" if status.reachable else ":x:" + if status.reachable and status.latency_ms is not None: + suffix = f" _({status.latency_ms:.0f}ms)_" + elif status.error: + suffix = f" — `{status.error}`" + else: + suffix = "" + return f"{emoji} {label}{suffix}" + + +def _hardware_text(hardware: HealthStatus | None, error: str | None) -> str: + """Render the hardware section of the digest as a mrkdwn string. + + Takes the typed HealthStatus model (not a raw dict) so we can iterate + devices with their purpose labels and show the rotary encoder only when + it was actually probed (i.e. not virtual). + """ + if error is not None: + return f":x: *Hardware probe failed*\n```{error}```" + if hardware is None: + return ":x: *Hardware probe returned no data.*" + + lines = ["*Hardware Health*"] + lines.append(_component_line("Router", hardware.router)) + lines.extend(_component_line(f"`{d.provider}/{d.name}` _({d.purpose})_", d) for d in hardware.devices) + if hardware.rotary_encoder is not None: + lines.append(_component_line("Rotary encoder", hardware.rotary_encoder)) + else: + lines.append(":grey_question: Rotary encoder — _virtual (skipped)_") + if hardware.follower_node is not None: + lines.append(_component_line("Follower node", hardware.follower_node)) + + return "\n".join(lines) + + +def _game_blocks(game: GameResult) -> list[dict[str, Any]]: + """Build Block Kit blocks for one game result. + + Dispatches on status so skipped, failed, and successful games each get an + appropriate layout — skipped and failed collapse to a single block, while + a success expands to include the full data fields. + """ + header = f"{game.emoji} *{game.title}*" if game.emoji else f"*{game.title}*" + if game.status == "skipped": + return [{"type": "section", "text": {"type": "mrkdwn", "text": f":fast_forward: {header}\n_{game.error}_"}}] + + if game.status == "failed": + return [ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": f":x: {header}\n```{game.error}```\n_Elapsed: {game.elapsed_s:.1f}s_", + }, + } + ] + + blocks: list[dict[str, Any]] = [ + {"type": "section", "text": {"type": "mrkdwn", "text": f":white_check_mark: {header}"}}, + ] + if isinstance(game.data, dict): + blocks.extend(_sections_from_fields(_fields_from_dict(game.data))) + elif isinstance(game.data, list): + blocks.append( + { + "type": "section", + "text": {"type": "mrkdwn", "text": f"`{_format_value(game.data)}`"}, + } + ) + blocks.append( + { + "type": "context", + "elements": [{"type": "mrkdwn", "text": f":stopwatch: {game.elapsed_s:.1f}s"}], + } + ) + return blocks + + +def _build_digest(config: DailyReportConfig, result: ReportResult) -> dict[str, Any]: + """Assemble the full Block Kit digest payload. + + The header emoji distinguishes hardware failures from game failures so + operators can tell at a glance whether they need to check the hardware rack + or look at a software/timing issue. + """ + if result.overall_ok: + header_emoji = ":white_check_mark:" + header_text = "Daily Report — all systems nominal" + elif result.hardware_error is not None or result.hardware is None or not result.hardware.all_ok: + header_emoji = ":warning:" + header_text = "Daily Report — hardware issues detected" + else: + header_emoji = ":warning:" + header_text = "Daily Report — game failures detected" + + blocks: list[dict[str, Any]] = [ + { + "type": "header", + "text": {"type": "plain_text", "text": f"{header_emoji} {header_text}", "emoji": True}, + }, + { + "type": "section", + "text": {"type": "mrkdwn", "text": _hardware_text(result.hardware, result.hardware_error)}, + }, + ] + for game in result.games: + blocks.append({"type": "divider"}) + blocks.extend(_game_blocks(game)) + + if result.skipped_notes: + blocks.append({"type": "divider"}) + blocks.append( + { + "type": "context", + "elements": [{"type": "mrkdwn", "text": "\n".join(result.skipped_notes)}], + } + ) + + blocks.append( + { + "type": "context", + "elements": [ + { + "type": "mrkdwn", + "text": ( + f":clock1: {datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S UTC')} " + f"| Follower: `{config.follower_node_address}` " + f"| TimeTagger: `{config.timetagger_address}`" + ), + } + ], + } + ) + + return {"blocks": blocks} + + +def _post_plain_slack(webhook_url: str, text: str) -> None: + """Post a bare text message to Slack — used as a last-resort fallback.""" + with httpx.Client(timeout=10.0) as client: + client.post(webhook_url, json={"text": text}) + + +def _post_slack(webhook_url: str, message: dict[str, Any]) -> bool: + """Post a Block Kit message to Slack and return whether it was accepted. + + Falls back to a plain-text message if Slack rejects the Block Kit payload + so operators still get a notification even when the formatting is wrong, + rather than silently receiving nothing. + """ + try: + with httpx.Client(timeout=15.0) as client: + response = client.post(webhook_url, json=message) + except httpx.HTTPError: + logger.exception("Slack post failed") + return False + if response.text == "ok": + return True + logger.error("Slack rejected the digest: %s", response.text) + # Fallback: best-effort plain-text notice so the operator sees something. + try: + _post_plain_slack( + webhook_url, + f":warning: Daily report could not post Block Kit digest. Slack said: `{response.text[:200]}`", + ) + except httpx.HTTPError: + logger.exception("Fallback plain Slack post also failed") + return False + + +def run_daily_report(config: DailyReportConfig) -> int: + """Run the daily report end-to-end. Returns a process exit code.""" + _arm_watchdog(config) + try: + if not _check_api(config): + logger.error("API at %s is unreachable", config.api_url) + _post_plain_slack( + config.slack_webhook_url, + f":x: *Daily Report — API is down*\nCould not reach `{config.api_url}`. Start the server and re-run.", + ) + return 1 + + result = ReportResult() + + logger.info("Probing hardware via %s/health/", config.api_url) + hardware, hardware_error = _fetch_hardware(config) + result.hardware = hardware + result.hardware_error = hardware_error + + availability = _fetch_availability(config) + + if availability.get("chsh"): + logger.info("Running CHSH") + result.games.append(_run_chsh(config)) + else: + result.skipped_notes.append(":fast_forward: CHSH disabled in `games_availability`.") + + if availability.get("qf"): + logger.info("Running Quantum Fortune") + result.games.append(_run_qf(config)) + else: + result.skipped_notes.append(":fast_forward: Quantum Fortune disabled in `games_availability`.") + + if availability.get("ssm"): + result.skipped_notes.append(f":fast_forward: {_SSM_SKIP_REASON}") + + digest = _build_digest(config, result) + posted = _post_slack(config.slack_webhook_url, digest) + + if not posted: + return 1 + return 0 if result.overall_ok else 1 + finally: + _disarm_watchdog() diff --git a/src/pqnstack/cli.py b/src/pqnstack/cli.py index 988a162e..a1bad664 100644 --- a/src/pqnstack/cli.py +++ b/src/pqnstack/cli.py @@ -7,6 +7,12 @@ import tomli_w import typer +from pqnstack.app.core.config import get_settings +from pqnstack.app.cron_manager import describe_schedule +from pqnstack.app.cron_manager import get_daily_report_job +from pqnstack.app.cron_manager import remove_daily_report_job +from pqnstack.app.cron_manager import set_daily_report_schedule +from pqnstack.app.daily_report import run_daily_report from pqnstack.base.errors import InvalidNetworkConfigurationError from pqnstack.network.instrument_provider import InstrumentProvider from pqnstack.network.router import Router @@ -18,6 +24,9 @@ app = typer.Typer(no_args_is_help=True, help="CLI for PQN-Stack.") +daily_report_app = typer.Typer(no_args_is_help=True, help="Run and manage the daily health + Slack report.") +app.add_typer(daily_report_app, name="daily-report") + def _verify_instruments_config(instruments: list[dict[str, str]]) -> dict[str, dict[str, str]]: ins = {} @@ -217,5 +226,122 @@ def toggle_game( logger.info("Games %s %s in %s. Restart the server for changes to take effect.", games, status, path) +@daily_report_app.command("run") +def daily_report_run() -> None: + """ + Run the daily health + games report and post the result to Slack. + + Reads the [daily_report] section from config.toml, probes hardware via the + running API (`/health`), exercises each enabled game (except SSM), and posts a + consolidated Slack digest. Exits non-zero if anything failed. + """ + report_config = get_settings().daily_report + if report_config is None: + logger.error("[daily_report] section missing from config.toml") + raise typer.Exit(code=1) + + raise typer.Exit(code=run_daily_report(report_config)) + + +@daily_report_app.command("status") +def daily_report_status() -> None: + """Show whether the daily report cron job is active and its schedule.""" + job = get_daily_report_job() + if job is None: + typer.echo("Daily report is not scheduled.") + else: + typer.echo(f"Daily report is active. Schedule: {describe_schedule(job)}") + + +_DOW_MAP = { + "monday": "1", + "tuesday": "2", + "wednesday": "3", + "thursday": "4", + "friday": "5", + "saturday": "6", + "sunday": "0", +} + + +def _prompt_hhmm() -> tuple[int, int]: + raw_time = typer.prompt("Time (HH:MM, 24-hour)") + try: + h_str, m_str = raw_time.strip().split(":") + hour, minute = int(h_str), int(m_str) + except ValueError: + typer.echo("Invalid time format. Use HH:MM (e.g. 09:00).", err=True) + raise typer.Exit(code=1) # noqa: B904 + if not (0 <= hour <= 23 and 0 <= minute <= 59): # noqa: PLR2004 + typer.echo("Hour must be 0-23 and minute 0-59.", err=True) + raise typer.Exit(code=1) + return hour, minute + + +def _prompt_dow() -> str: + raw_day = typer.prompt("Day of week (monday-sunday)").strip().lower() + if raw_day not in _DOW_MAP: + typer.echo(f"Invalid day '{raw_day}'.", err=True) + raise typer.Exit(code=1) + return _DOW_MAP[raw_day] + + +def _prompt_dom() -> str: + raw_dom = typer.prompt("Day of month (1-28)") + dom_int = int(raw_dom) + if not 1 <= dom_int <= 28: # noqa: PLR2004 + typer.echo("Day of month must be between 1 and 28.", err=True) + raise typer.Exit(code=1) + return str(dom_int) + + +@daily_report_app.command("schedule") +def daily_report_schedule() -> None: + """Interactively schedule the daily report cron job.""" + frequency = typer.prompt("Frequency (hourly/daily/weekly/monthly)").strip().lower() + valid = {"hourly", "daily", "weekly", "monthly"} + if frequency not in valid: + typer.echo(f"Invalid frequency '{frequency}'. Choose from: {', '.join(sorted(valid))}", err=True) + raise typer.Exit(code=1) + + minute: int + hour: int | str = "*" + dow = "*" + dom = "*" + + if frequency == "hourly": + raw_minute = typer.prompt("Minute past the hour (0-59)") + minute = int(raw_minute) + if not 0 <= minute <= 59: # noqa: PLR2004 + typer.echo("Minute must be between 0 and 59.", err=True) + raise typer.Exit(code=1) + else: + hour, minute = _prompt_hhmm() + if frequency == "weekly": + dow = _prompt_dow() + elif frequency == "monthly": + dom = _prompt_dom() + + try: + set_daily_report_schedule(minute=minute, hour=hour, dow=dow, dom=dom) + except RuntimeError as e: + typer.echo(str(e), err=True) + raise typer.Exit(code=1) # noqa: B904 + + job = get_daily_report_job() + description = describe_schedule(job) if job else "unknown" + typer.echo(f"Daily report scheduled. Schedule: {description}") + + +@daily_report_app.command("unschedule") +def daily_report_unschedule() -> None: + """Remove the daily report cron job.""" + removed = remove_daily_report_job() + if removed: + typer.echo("Daily report unscheduled.") + else: + typer.echo("Daily report was not scheduled.") + + if __name__ == "__main__": app() diff --git a/uv.lock b/uv.lock index f3591ff8..38f67f46 100644 --- a/uv.lock +++ b/uv.lock @@ -729,7 +729,10 @@ name = "pqnstack" version = "0.2.0" source = { editable = "." } dependencies = [ + { name = "fastapi", extra = ["standard"] }, + { name = "httpx" }, { name = "numpy" }, + { name = "pydantic-settings" }, { name = "pyfirmata2" }, { name = "pyzmq" }, { name = "thorlabs-apt-device" }, @@ -737,13 +740,6 @@ dependencies = [ { name = "typer" }, ] -[package.optional-dependencies] -webapp = [ - { name = "fastapi", extra = ["standard"] }, - { name = "httpx" }, - { name = "pydantic-settings" }, -] - [package.dev-dependencies] dev = [ { name = "coverage" }, @@ -755,17 +751,16 @@ dev = [ [package.metadata] requires-dist = [ - { name = "fastapi", extras = ["standard"], marker = "extra == 'webapp'", specifier = ">=0.115.14" }, - { name = "httpx", marker = "extra == 'webapp'", specifier = ">=0.28.1" }, + { name = "fastapi", extras = ["standard"], specifier = ">=0.115.14" }, + { name = "httpx", specifier = ">=0.28.1" }, { name = "numpy", specifier = ">=2.3.5" }, - { name = "pydantic-settings", marker = "extra == 'webapp'", specifier = ">=2.10.1" }, + { name = "pydantic-settings", specifier = ">=2.10.1" }, { name = "pyfirmata2", specifier = ">=2.5.0" }, { name = "pyzmq", specifier = ">=26.2.0" }, { name = "thorlabs-apt-device", specifier = ">=0.3.8" }, { name = "tomli-w", specifier = ">=1.0.0" }, { name = "typer", specifier = ">=0.15.1" }, ] -provides-extras = ["webapp"] [package.metadata.requires-dev] dev = [