From 691d29034244d005adb245fe56058a797a1eb832 Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Fri, 23 Jan 2026 03:01:45 -0600 Subject: [PATCH 01/12] Adds protocol cancellation event. Adds SSE to let frontend know of cancellation --- src/pqnstack/app/api/routes/coordination.py | 133 +++++++++++++++++++- src/pqnstack/app/api/routes/qkd.py | 43 ++++++- src/pqnstack/app/core/config.py | 1 + 3 files changed, 171 insertions(+), 6 deletions(-) diff --git a/src/pqnstack/app/api/routes/coordination.py b/src/pqnstack/app/api/routes/coordination.py index 4c36880..a985553 100644 --- a/src/pqnstack/app/api/routes/coordination.py +++ b/src/pqnstack/app/api/routes/coordination.py @@ -1,4 +1,5 @@ import asyncio +import json import logging from fastapi import APIRouter @@ -7,12 +8,14 @@ from fastapi import WebSocket from fastapi import WebSocketDisconnect from fastapi import status +from fastapi.responses import StreamingResponse from pydantic import BaseModel from pqnstack.app.api.deps import ClientDep from pqnstack.app.api.deps import StateDep from pqnstack.app.core.config import NodeRole from pqnstack.app.core.config import ask_user_for_follow_event +from pqnstack.app.core.config import protocol_cancelled_event from pqnstack.app.core.config import settings from pqnstack.app.core.config import user_replied_event @@ -31,14 +34,50 @@ class ResetCoordinationStateResponse(BaseModel): message: str = "Coordination state reset successfully" +class ProtocolCancellationNotification(BaseModel): + reason: str = "Protocol cancelled by peer" + cancelled_by_role: str + + router = APIRouter(prefix="/coordination", tags=["coordination"]) # TODO: Send a disconnection message if I was following/leading someone. # FIXME: This is technically resetting more than just coordination state. including qkd. @router.post("/reset_coordination_state") -async def reset_coordination_state(state: StateDep) -> ResetCoordinationStateResponse: +async def reset_coordination_state(state: StateDep, http_client: ClientDep) -> ResetCoordinationStateResponse: """Reset the coordination state of the node.""" + # Notify peer node BEFORE resetting state + peer_address = None + current_role = state.role + + if state.role == NodeRole.LEADER and state.followers_address: + peer_address = state.followers_address + elif state.role == NodeRole.FOLLOWER and state.leaders_address: + peer_address = state.leaders_address + + # Try to notify peer (best-effort, don't fail if peer is unreachable) + if peer_address: + try: + logger.info("Notifying peer at %s about protocol cancellation", peer_address) + await http_client.post( + f"http://{peer_address}/coordination/protocol_cancelled", + json={ + "reason": "Protocol cancelled by user", + "cancelled_by_role": current_role.value + }, + timeout=5.0 # Short timeout to avoid hanging + ) + except Exception as e: + logger.warning( + "Failed to notify peer about cancellation: %s. Proceeding with reset.", + str(e) + ) + + # Set local cancellation event to unblock any waiting operations + protocol_cancelled_event.set() + + # Reset state state.role = NodeRole.INDEPENDENT state.followers_address = "" state.following_requested = False @@ -55,9 +94,33 @@ async def reset_coordination_state(state: StateDep) -> ResetCoordinationStateRes state.qkd_request_basis_list = [] state.qkd_request_bit_list = [] state.qkd_n_matching_bits = -1 + + # Clear the cancellation event for next use + protocol_cancelled_event.clear() + return ResetCoordinationStateResponse() +@router.post("/protocol_cancelled") +async def protocol_cancelled( + notification: ProtocolCancellationNotification, +) -> dict[str, str]: + """Receive notification that peer node cancelled the protocol.""" + logger.info( + "Received protocol cancellation from %s: %s", + notification.cancelled_by_role, + notification.reason + ) + protocol_cancelled_event.set() + + # Give waiting operations a chance to wake up and handle the cancellation + # Then clear for the next operation + await asyncio.sleep(0.5) + protocol_cancelled_event.clear() + + return {"status": "acknowledged"} + + @router.post("/collect_follower") async def collect_follower( request: Request, address: str, state: StateDep, http_client: ClientDep @@ -127,7 +190,31 @@ async def follow_requested( ask_user_for_follow_event.set() logger.debug("Asking user to accept follow request from %s (%s)", leaders_name, leaders_address) - await user_replied_event.wait() # Wait for a state change event to see if user accepted + + # Wait for EITHER user reply OR cancellation + done, pending = await asyncio.wait( + [ + asyncio.create_task(user_replied_event.wait()), + asyncio.create_task(protocol_cancelled_event.wait()) + ], + return_when=asyncio.FIRST_COMPLETED + ) + + # Cancel pending tasks + for task in pending: + task.cancel() + + # Check if protocol was cancelled + if protocol_cancelled_event.is_set(): + logger.warning("Follow request cancelled") + # Clean up state + state.leaders_address = "" + state.leaders_name = "" + state.following_requested = False + state.following_requested_user_response = None + # protocol_cancelled_event.clear() + return FollowRequestResponse(accepted=False) + user_replied_event.clear() # Reset the event for the next change if state.following_requested_user_response: logger.debug("Follow request from %s accepted.", leaders_address) @@ -193,3 +280,45 @@ async def client_message_handler() -> None: finally: state_change_task.cancel() client_message_task.cancel() + + +@router.get("/state_events") +async def state_events(state: StateDep) -> StreamingResponse: + """SSE endpoint for streaming state change events to frontend.""" + async def event_generator(): + try: + last_role = state.role + + # Send initial connection event + yield f"data: {json.dumps({'event': 'connected', 'role': state.role.value})}\n\n" + + while True: + event_sent = False + + # Check for cancellation event + try: + await asyncio.wait_for(protocol_cancelled_event.wait(), timeout=1.0) + yield f"data: {json.dumps({'event': 'protocol_cancelled', 'reason': 'Protocol cancelled by peer or user'})}\n\n" + protocol_cancelled_event.clear() + event_sent = True + except asyncio.TimeoutError: + pass + + # Send heartbeat if no event was sent to keep connection alive + if not event_sent: + yield ":\n" + + await asyncio.sleep(1.0) + + except asyncio.CancelledError: + logger.info("SSE connection closed by client") + raise + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + } + ) diff --git a/src/pqnstack/app/api/routes/qkd.py b/src/pqnstack/app/api/routes/qkd.py index 456ac4d..5adcd8b 100644 --- a/src/pqnstack/app/api/routes/qkd.py +++ b/src/pqnstack/app/api/routes/qkd.py @@ -15,6 +15,7 @@ from pqnstack.app.api.deps import StateDep from pqnstack.app.core.config import NodeRole from pqnstack.app.core.config import NodeState +from pqnstack.app.core.config import protocol_cancelled_event from pqnstack.app.core.config import qkd_result_received_event from pqnstack.app.core.config import settings from pqnstack.constants import BasisBool @@ -296,8 +297,24 @@ async def submit_result(result: QKDResult, state: StateDep) -> None: async def _wait_for_follower_ready(state: NodeState, http_client: httpx.AsyncClient) -> None: """Poll the follower until it's ready, checking every 0.5 seconds.""" ready = False + first_503 = True while not ready: + # Check if protocol was cancelled + if protocol_cancelled_event.is_set(): + logger.warning("Protocol cancelled while waiting for follower ready") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Protocol cancelled by peer or user" + ) + r = await http_client.get(f"http://{state.followers_address}/qkd/is_follower_ready") + + # If the follower disconnects while the leader is waiting, the 503 error of `Node is not a follower` error might come before we can handle the cancellation event. + if r.status_code == status.HTTP_503_SERVICE_UNAVAILABLE and first_503: + logger.warning("Received QKD result from follower: %s", r) + first_503 = False + continue + if r.status_code != status.HTTP_200_OK: logger.error("Failed to check if follower is ready: %s", r.text) raise HTTPException( @@ -308,9 +325,9 @@ async def _wait_for_follower_ready(state: NodeState, http_client: httpx.AsyncCli ready = r.json() if not ready: logger.info("Follower is not ready yet, waiting.") - await asyncio.sleep(0.5) + await asyncio.sleep(0.4) # Make sure this is smaller than the protocol_cancelled event clear timer. - logger.info("Follower ready is ready") + logger.info("Follower is ready") async def _submit_result_to_follower(state: NodeState, http_client: httpx.AsyncClient, qkd_result: QKDResult) -> None: @@ -352,8 +369,26 @@ async def _submit_basis_list_follower(state: NodeState, basis_list: list[QKDEnco # don't wait for the event if the result is already set. This avoids deadlocks in case the result was set before this function is called. if state.qkd_n_matching_bits == -1: - # Wait until the leader submits the QKD result - await qkd_result_received_event.wait() + # Wait for EITHER result OR cancellation + done, pending = await asyncio.wait( + [ + asyncio.create_task(qkd_result_received_event.wait()), + asyncio.create_task(protocol_cancelled_event.wait()) + ], + return_when=asyncio.FIRST_COMPLETED + ) + + # Cancel pending tasks + for task in pending: + task.cancel() + + # Check if protocol was cancelled + if protocol_cancelled_event.is_set(): + logger.warning("Protocol cancelled while waiting for QKD result") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Protocol cancelled by peer or user" + ) # Reassemble the QKDResult object from the state qkd_result = QKDResult( diff --git a/src/pqnstack/app/core/config.py b/src/pqnstack/app/core/config.py index 61f9f1c..380efc8 100644 --- a/src/pqnstack/app/core/config.py +++ b/src/pqnstack/app/core/config.py @@ -133,6 +133,7 @@ class NodeState(BaseModel): ask_user_for_follow_event = asyncio.Event() user_replied_event = asyncio.Event() qkd_result_received_event = asyncio.Event() +protocol_cancelled_event = asyncio.Event() def get_state() -> NodeState: From dfcbd71a38b9751954650db2aaf07ce396c89c7a Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Mon, 9 Feb 2026 14:48:43 -0600 Subject: [PATCH 02/12] Add an absolute val to chsh --- src/pqnstack/app/api/routes/chsh.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pqnstack/app/api/routes/chsh.py b/src/pqnstack/app/api/routes/chsh.py index 9a12b7a..fe69ae9 100644 --- a/src/pqnstack/app/api/routes/chsh.py +++ b/src/pqnstack/app/api/routes/chsh.py @@ -102,7 +102,7 @@ async def _chsh( # Complexity is high due to the nature of the CHSH experiment. logger.info("What are you settings? %s", settings.chsh_settings.expectation_signs) logger.info("After passing signed calculation: %s", expectation_values) - chsh_value = sum(x for x in expectation_values) + chsh_value = abs(sum(x for x in expectation_values)) chsh_error = sum(x**2 for x in expectation_errors) ** 0.5 return chsh_value, chsh_error From 5730bfc51f2bdbcd2e931ea56933075f8f6811a8 Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Mon, 9 Feb 2026 22:11:24 -0600 Subject: [PATCH 03/12] Adds SSE for measurements to allow loading bar on the frontend --- src/pqnstack/app/api/routes/chsh.py | 62 ++++++++++++++++++++++++++++- src/pqnstack/app/api/routes/rng.py | 61 ++++++++++++++++++++++++++++ src/pqnstack/app/core/config.py | 13 ++++++ 3 files changed, 135 insertions(+), 1 deletion(-) diff --git a/src/pqnstack/app/api/routes/chsh.py b/src/pqnstack/app/api/routes/chsh.py index fe69ae9..d68a9e0 100644 --- a/src/pqnstack/app/api/routes/chsh.py +++ b/src/pqnstack/app/api/routes/chsh.py @@ -1,3 +1,5 @@ +import asyncio +import json import logging from typing import TYPE_CHECKING from typing import cast @@ -5,9 +7,11 @@ from fastapi import APIRouter from fastapi import HTTPException from fastapi import status +from fastapi.responses import StreamingResponse from pqnstack.app.api.deps import ClientDep from pqnstack.app.api.deps import StateDep +from pqnstack.app.core.config import chsh_progress_event from pqnstack.app.core.config import settings from pqnstack.app.core.models import calculate_chsh_expectation_error from pqnstack.network.client import Client @@ -20,14 +24,61 @@ router = APIRouter(prefix="/chsh", tags=["chsh"]) +@router.get("/progress") +async def chsh_progress(state: StateDep) -> StreamingResponse: + """SSE endpoint for streaming CHSH measurement progress to frontend.""" + async def event_generator(): + try: + # Send initial connection event + yield f"data: {json.dumps({'event': 'connected'})}\n\n" + + while True: + event_sent = False + + # Check for progress event + try: + await asyncio.wait_for(chsh_progress_event.wait(), timeout=1.0) + yield f"data: {json.dumps({'event': 'chsh_progress', 'current': state.chsh_progress_current, 'total': state.chsh_progress_total, 'running': state.chsh_running})}\n\n" + chsh_progress_event.clear() + event_sent = True + except asyncio.TimeoutError: + pass + + # Send heartbeat if no event was sent to keep connection alive + if not event_sent: + yield ":\n" + + await asyncio.sleep(1.0) + + except asyncio.CancelledError: + logger.info("CHSH SSE connection closed by client") + raise + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + } + ) + + async def _chsh( # Complexity is high due to the nature of the CHSH experiment. basis: tuple[float, float], follower_node_address: str, http_client: ClientDep, timetagger_address: str, + state: StateDep, ) -> tuple[float, float]: logger.debug("Starting CHSH") + # Initialize progress tracking + state.chsh_running = True + state.chsh_progress_current = 0 + state.chsh_progress_total = 16 # 2 basis × 2 follower × 2 angles × 2 perp + chsh_progress_event.set() + logger.debug("Instantiating client") client = Client(host=settings.router_address, port=settings.router_port, timeout=600_000) @@ -74,6 +125,10 @@ async def _chsh( # Complexity is high due to the nature of the CHSH experiment. count = cast("int", count_ret.json()) counts.append(count) + # Update progress + state.chsh_progress_current += 1 + chsh_progress_event.set() + # Calculating expectation value numerator = counts[0] - counts[1] - counts[2] + counts[3] denominator = sum(counts) - 4 * settings.chsh_settings.measurement_config.dark_count @@ -105,6 +160,10 @@ async def _chsh( # Complexity is high due to the nature of the CHSH experiment. chsh_value = abs(sum(x for x in expectation_values)) chsh_error = sum(x**2 for x in expectation_errors) ** 0.5 + # Mark CHSH as complete + state.chsh_running = False + chsh_progress_event.set() + return chsh_value, chsh_error @@ -114,10 +173,11 @@ async def chsh( follower_node_address: str, http_client: ClientDep, timetagger_address: str, + state: StateDep, ) -> dict[str, float]: logger.info("Starting CHSH experiment with basis: %s", basis) - chsh_value, chsh_error = await _chsh(basis, follower_node_address, http_client, timetagger_address) + chsh_value, chsh_error = await _chsh(basis, follower_node_address, http_client, timetagger_address, state) return { "chsh_value": chsh_value, diff --git a/src/pqnstack/app/api/routes/rng.py b/src/pqnstack/app/api/routes/rng.py index db9d27e..a0c6ee2 100644 --- a/src/pqnstack/app/api/routes/rng.py +++ b/src/pqnstack/app/api/routes/rng.py @@ -1,3 +1,5 @@ +import asyncio +import json import logging from typing import Annotated from typing import Any @@ -6,14 +8,57 @@ from fastapi import HTTPException from fastapi import Query from fastapi import status +from fastapi.responses import StreamingResponse from pqnstack.app.api.deps import ClientDep +from pqnstack.app.api.deps import StateDep +from pqnstack.app.core.config import rng_progress_event logger = logging.getLogger(__name__) router = APIRouter(prefix="/rng", tags=["rng"]) +@router.get("/progress") +async def rng_progress(state: StateDep) -> StreamingResponse: + """SSE endpoint for streaming RNG fortune measurement progress to frontend.""" + async def event_generator(): + try: + # Send initial connection event + yield f"data: {json.dumps({'event': 'connected'})}\n\n" + + while True: + event_sent = False + + # Check for progress event + try: + await asyncio.wait_for(rng_progress_event.wait(), timeout=1.0) + yield f"data: {json.dumps({'event': 'rng_progress', 'current': state.rng_progress_current, 'total': state.rng_progress_total, 'running': state.rng_running})}\n\n" + rng_progress_event.clear() + event_sent = True + except asyncio.TimeoutError: + pass + + # Send heartbeat if no event was sent to keep connection alive + if not event_sent: + yield ":\n" + + await asyncio.sleep(1.0) + + except asyncio.CancelledError: + logger.info("RNG SSE connection closed by client") + raise + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + } + ) + + @router.get("/singles_parity") async def singles_parity( timetagger_address: str, @@ -57,12 +102,19 @@ async def fortune( integration_time_s: float, fortune_size: int, http_client: ClientDep, + state: StateDep, channels: Annotated[list[int], Query()], ) -> list[int]: """Run singles parity `fortune_size` times and, per channel, interpret the result in bitstring as a decimal number.""" if fortune_size <= 0: raise HTTPException(status_code=400, detail="fortune_size must be a positive integer") + # Initialize progress tracking + state.rng_running = True + state.rng_progress_current = 0 + state.rng_progress_total = fortune_size + rng_progress_event.set() + trials: list[list[int]] = [] for _ in range(fortune_size): params: list[tuple[str, str | int | float | bool | None]] = [ @@ -75,6 +127,10 @@ async def fortune( parities = await http_client.get(url, params=params) trials.append(parities.json()) + # Update progress + state.rng_progress_current += 1 + rng_progress_event.set() + results: list[int] = [] for bits_for_channel in zip(*trials, strict=True): value = 0 @@ -88,4 +144,9 @@ async def fortune( fortune_size, results, ) + + # Mark RNG as complete + state.rng_running = False + rng_progress_event.set() + return results diff --git a/src/pqnstack/app/core/config.py b/src/pqnstack/app/core/config.py index 380efc8..b942c2a 100644 --- a/src/pqnstack/app/core/config.py +++ b/src/pqnstack/app/core/config.py @@ -102,11 +102,17 @@ class NodeState(BaseModel): # CHSH state chsh_request_basis: list[float] = [22.5, 67.5] + chsh_progress_current: int = 0 # Current iteration in CHSH measurement + chsh_progress_total: int = 16 # Total iterations (2 basis × 2 follower × 2 angles × 2 perp) + chsh_running: bool = False # Whether CHSH measurement is currently running # QKD state # FIXME: At the moment the reset_coordination_state resets this, probably want to refactor that function out. qkd_question_order: list[int] = [] # Order of questions for QKD qkd_emoji_pick: str = "" # Emoji chosen for QKD + qkd_progress_current: int = 0 # Current iteration in QKD measurement + qkd_progress_total: int = 11 # Total iterations (bitstring length) + qkd_running: bool = False # Whether QKD measurement is currently running qkd_leader_basis_list: list[QKDEncodingBasis] = [ QKDEncodingBasis.DA, QKDEncodingBasis.DA, @@ -128,12 +134,19 @@ class NodeState(BaseModel): qkd_request_bit_list: list[int] = [] qkd_n_matching_bits: int = -1 # Leaders populate this value after qkd is done. Same with the emoji + # RNG state + rng_progress_current: int = 0 # Current iteration in RNG fortune measurement + rng_progress_total: int = 0 # Total iterations (fortune_size) + rng_running: bool = False # Whether RNG fortune measurement is currently running + state = NodeState() ask_user_for_follow_event = asyncio.Event() user_replied_event = asyncio.Event() qkd_result_received_event = asyncio.Event() protocol_cancelled_event = asyncio.Event() +chsh_progress_event = asyncio.Event() +rng_progress_event = asyncio.Event() def get_state() -> NodeState: From a3882698d1247ab22383ecc69c619dfb96c1d196 Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Tue, 10 Feb 2026 11:02:46 -0600 Subject: [PATCH 04/12] Adds config clarification --- configs/config_app_example.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/configs/config_app_example.toml b/configs/config_app_example.toml index 84c8a81..3fd6e99 100644 --- a/configs/config_app_example.toml +++ b/configs/config_app_example.toml @@ -20,6 +20,7 @@ bell_state = 0 [chsh_settings] hwp = ["provider", "instrument_hwp"] # Replace with actual HWP names request_hwp = ["provider", "instrument_hwp"] +expectation_signs = [-1, 1, 1, 1] # This is for HV + VH. for HH + VV, the signs are [1, -1, 1, 1] # CHSH measurement configuration [chsh_settings.measurement_config] From 2ffb37321c4a13d6a84729b7caa5956cd67e09b8 Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Thu, 12 Feb 2026 21:09:21 -0600 Subject: [PATCH 05/12] Adds daily chsh report to slack capabilities --- configs/config_app_example.toml | 10 +- scripts/chsh_daily_report.py | 229 ++++++++++++++++++++++++++++ src/pqnstack/app/api/routes/chsh.py | 16 +- 3 files changed, 248 insertions(+), 7 deletions(-) create mode 100644 scripts/chsh_daily_report.py diff --git a/configs/config_app_example.toml b/configs/config_app_example.toml index 3fd6e99..a7c7f42 100644 --- a/configs/config_app_example.toml +++ b/configs/config_app_example.toml @@ -43,4 +43,12 @@ integration_time_s = 5 binwidth = 500 channel1 = 1 channel2 = 2 -dark_count = 0 \ No newline at end of file +dark_count = 0 + +# Daily CHSH report settings (for automated Slack reporting) +[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 diff --git a/scripts/chsh_daily_report.py b/scripts/chsh_daily_report.py new file mode 100644 index 0000000..5bab8a0 --- /dev/null +++ b/scripts/chsh_daily_report.py @@ -0,0 +1,229 @@ +#!/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 json +import sys +import tomllib +from datetime import datetime +from pathlib import Path + +import httpx + + +def load_config() -> dict: + """Load configuration from config.toml.""" + config_path = Path(__file__).parent.parent / "config.toml" + + if not config_path.exists(): + print(f"❌ Error: config.toml not found at {config_path}") + print("Please create config.toml from configs/config_app_example.toml") + sys.exit(1) + + with open(config_path, "rb") as f: + return tomllib.load(f) + + +def get_daily_report_config(config: dict) -> dict: + """Get and validate daily_report configuration.""" + daily_report_config = config.get("daily_report", {}) + + if not daily_report_config: + print("❌ Error: [daily_report] section not found in config.toml") + print("Please add it following the example in configs/config_app_example.toml") + sys.exit(1) + + # Check required fields + required_fields = ["slack_webhook_url", "follower_node_address"] + for field in required_fields: + if not daily_report_config.get(field): + print(f"❌ Error: {field} not set in config.toml [daily_report] section") + sys.exit(1) + + return daily_report_config + + +def run_chsh_measurement(config: dict) -> dict: + """Run CHSH measurement via API.""" + daily_report_config = get_daily_report_config(config) + + api_url = daily_report_config.get("api_url", "http://localhost:8000") + timetagger_address = daily_report_config.get("timetagger_address", "127.0.0.1:8000") + follower_node_address = daily_report_config["follower_node_address"] + basis = daily_report_config.get("basis", [0, 22.5]) + + print(f"🔬 Starting CHSH measurement at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + print(f" Basis: {basis}") + print(f" Follower: {follower_node_address}") + print(f" TimeTagger: {timetagger_address}") + + try: + with httpx.Client(timeout=600.0) as client: + response = client.post( + f"{api_url}/chsh/", + json={ + "basis": basis, + "follower_node_address": follower_node_address, + "timetagger_address": timetagger_address, + } + ) + response.raise_for_status() + return response.json() + + except httpx.HTTPError as e: + print(f"❌ Failed to contact CHSH API: {e}") + sys.exit(1) + + +def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): + """Post CHSH results to Slack.""" + chsh_value = chsh_data["chsh_value"] + chsh_error = chsh_data["chsh_error"] + expectation_values = chsh_data["expectation_values"] + expectation_errors = chsh_data["expectation_errors"] + expectation_values_sign_fixed = chsh_data["expectation_values_sign_fixed"] + + # Determine emoji based on Bell inequality violation (CHSH > 2) + emoji = ":sparkles:" if chsh_value > 2 else ":thinking_face:" + + daily_report_config = config.get("daily_report", {}) + basis = daily_report_config.get("basis", [0, 22.5]) + follower_address = daily_report_config.get("follower_node_address", "unknown") + timetagger_address = daily_report_config.get("timetagger_address", "unknown") + + # Format Slack message using Block Kit + slack_message = { + "blocks": [ + { + "type": "header", + "text": { + "type": "plain_text", + "text": f"{emoji} CHSH Daily Measurement Report", + "emoji": True + } + }, + { + "type": "section", + "fields": [ + { + "type": "mrkdwn", + "text": f"*CHSH Value:*\n`{chsh_value:.4f}` ± `{chsh_error:.4f}`" + }, + { + "type": "mrkdwn", + "text": f"*Basis:*\n`{basis}`" + } + ] + }, + { + "type": "section", + "fields": [ + { + "type": "mrkdwn", + "text": f"*Expectation Values:*\n`{expectation_values}`" + }, + { + "type": "mrkdwn", + "text": f"*Expectation Errors:*\n`{expectation_errors}`" + } + ] + }, + { + "type": "section", + "fields": [ + { + "type": "mrkdwn", + "text": f"*Expectation Values (Sign Fixed):*\n`{expectation_values_sign_fixed}`" + }, + { + "type": "mrkdwn", + "text": f"*Timestamp:*\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" + } + ] + }, + { + "type": "context", + "elements": [ + { + "type": "mrkdwn", + "text": f"Follower: `{follower_address}` | TimeTagger: `{timetagger_address}`" + } + ] + } + ] + } + + print("📤 Posting to Slack...") + + try: + with httpx.Client() as client: + response = client.post(webhook_url, json=slack_message) + + if response.text == "ok": + print("✅ Successfully posted to Slack") + else: + print(f"❌ Failed to post to Slack: {response.text}") + sys.exit(1) + + except httpx.HTTPError as e: + print(f"❌ Failed to post to Slack: {e}") + sys.exit(1) + + +def post_error_to_slack(webhook_url: str, error_message: str): + """Post error message to Slack.""" + slack_message = { + "text": f":x: CHSH Daily Report Failed\n*Error:* {error_message}\n*Time:* {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" + } + + try: + with httpx.Client() as client: + client.post(webhook_url, json=slack_message) + except Exception: + pass # Silently fail if we can't post the error + + +def main(): + """Main entry point.""" + try: + # Load configuration + config = load_config() + daily_report_config = get_daily_report_config(config) + webhook_url = daily_report_config["slack_webhook_url"] + + # Run CHSH measurement + chsh_data = run_chsh_measurement(config) + + print(f"✅ CHSH measurement completed") + print(f" Value: {chsh_data['chsh_value']:.4f} ± {chsh_data['chsh_error']:.4f}") + + # Post to Slack + post_to_slack(webhook_url, chsh_data, config) + + print("✅ CHSH daily report completed successfully") + + except Exception as e: + print(f"❌ Unexpected error: {e}") + + # Try to post error to Slack if possible + try: + config = load_config() + daily_report_config = get_daily_report_config(config) + webhook_url = daily_report_config["slack_webhook_url"] + post_error_to_slack(webhook_url, str(e)) + except Exception: + pass + + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/src/pqnstack/app/api/routes/chsh.py b/src/pqnstack/app/api/routes/chsh.py index d68a9e0..9ed55ff 100644 --- a/src/pqnstack/app/api/routes/chsh.py +++ b/src/pqnstack/app/api/routes/chsh.py @@ -63,7 +63,7 @@ async def event_generator(): } ) - +# FIXME: Make the return of this function a dataclass async def _chsh( # Complexity is high due to the nature of the CHSH experiment. basis: tuple[float, float], follower_node_address: str, @@ -151,22 +151,23 @@ async def _chsh( # Complexity is high due to the nature of the CHSH experiment. logger.info("Expectation errors: %s", expectation_errors) # FIXME: This is a temporary fix for handling impossible expectation values. We should not have to rely on the settings for this. - expectation_values = [ + 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) - chsh_value = abs(sum(x for x in expectation_values)) + logger.info("After passing signed calculation: %s", expectation_values_sign_fixed) + chsh_value = abs(sum(x for x in expectation_values_sign_fixed)) chsh_error = sum(x**2 for x in expectation_errors) ** 0.5 # Mark CHSH as complete state.chsh_running = False chsh_progress_event.set() - return chsh_value, chsh_error + return chsh_value, chsh_error, expectation_values, expectation_errors, expectation_values_sign_fixed +# FIXME: make the return of this function the same dataclass as the one returned by _chsh. @router.post("/") async def chsh( basis: tuple[float, float], @@ -177,11 +178,14 @@ async def chsh( ) -> dict[str, float]: logger.info("Starting CHSH experiment with basis: %s", basis) - chsh_value, chsh_error = await _chsh(basis, follower_node_address, http_client, timetagger_address, state) + chsh_value, chsh_error, expectation_values, expectation_errors, expectation_values_sign_fixed = await _chsh(basis, follower_node_address, http_client, timetagger_address, state) return { "chsh_value": chsh_value, "chsh_error": chsh_error, + "expectation_values": expectation_values, + "expectation_errors": expectation_errors, + "expectation_values_sign_fixed": expectation_values_sign_fixed, } From 0f6323ac24c665a8534424ae37da1d42832f27c6 Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Thu, 12 Feb 2026 21:17:02 -0600 Subject: [PATCH 06/12] Fixes parameter placement in chsh_daily_report API request --- scripts/chsh_daily_report.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/scripts/chsh_daily_report.py b/scripts/chsh_daily_report.py index 5bab8a0..3221124 100644 --- a/scripts/chsh_daily_report.py +++ b/scripts/chsh_daily_report.py @@ -69,11 +69,11 @@ def run_chsh_measurement(config: dict) -> dict: with httpx.Client(timeout=600.0) as client: response = client.post( f"{api_url}/chsh/", - json={ - "basis": basis, + params={ "follower_node_address": follower_node_address, "timetagger_address": timetagger_address, - } + }, + json={"basis": basis} ) response.raise_for_status() return response.json() From cf3a54a3fbb20b53328a6a1156baf49b35d4459f Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Thu, 12 Feb 2026 21:20:58 -0600 Subject: [PATCH 07/12] Simplify API request payload in chsh_daily_report --- scripts/chsh_daily_report.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/chsh_daily_report.py b/scripts/chsh_daily_report.py index 3221124..3fc36e6 100644 --- a/scripts/chsh_daily_report.py +++ b/scripts/chsh_daily_report.py @@ -73,7 +73,7 @@ def run_chsh_measurement(config: dict) -> dict: "follower_node_address": follower_node_address, "timetagger_address": timetagger_address, }, - json={"basis": basis} + json=basis ) response.raise_for_status() return response.json() From 5d32b1d77fc2eaa3936ddc7dd11d2440d090d7ad Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Thu, 12 Feb 2026 21:30:09 -0600 Subject: [PATCH 08/12] Dynamically generate Slack message fields in `chsh_daily_report` to handle arbitrary data and streamline formatting --- scripts/chsh_daily_report.py | 98 +++++++++++++++++++----------------- 1 file changed, 52 insertions(+), 46 deletions(-) diff --git a/scripts/chsh_daily_report.py b/scripts/chsh_daily_report.py index 3fc36e6..3923b84 100644 --- a/scripts/chsh_daily_report.py +++ b/scripts/chsh_daily_report.py @@ -85,13 +85,8 @@ def run_chsh_measurement(config: dict) -> dict: def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): """Post CHSH results to Slack.""" - chsh_value = chsh_data["chsh_value"] - chsh_error = chsh_data["chsh_error"] - expectation_values = chsh_data["expectation_values"] - expectation_errors = chsh_data["expectation_errors"] - expectation_values_sign_fixed = chsh_data["expectation_values_sign_fixed"] - - # Determine emoji based on Bell inequality violation (CHSH > 2) + # Determine emoji based on Bell inequality violation (CHSH > 2) if chsh_value exists + chsh_value = chsh_data.get("chsh_value", 0) emoji = ":sparkles:" if chsh_value > 2 else ":thinking_face:" daily_report_config = config.get("daily_report", {}) @@ -99,6 +94,55 @@ def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): follower_address = daily_report_config.get("follower_node_address", "unknown") timetagger_address = daily_report_config.get("timetagger_address", "unknown") + # Build fields dynamically from all returned data + fields = [] + + # Add all fields from the API response + for key, value in chsh_data.items(): + # Format the key nicely (replace underscores with spaces, capitalize) + field_name = key.replace("_", " ").title() + + # Format the value based on type + if isinstance(value, float): + formatted_value = f"{value:.4f}" + elif isinstance(value, list): + # Format list nicely + 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`{basis}`" + }, + { + "type": "mrkdwn", + "text": f"*Timestamp:*\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" + } + ] + }) + # Format Slack message using Block Kit slack_message = { "blocks": [ @@ -110,45 +154,7 @@ def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): "emoji": True } }, - { - "type": "section", - "fields": [ - { - "type": "mrkdwn", - "text": f"*CHSH Value:*\n`{chsh_value:.4f}` ± `{chsh_error:.4f}`" - }, - { - "type": "mrkdwn", - "text": f"*Basis:*\n`{basis}`" - } - ] - }, - { - "type": "section", - "fields": [ - { - "type": "mrkdwn", - "text": f"*Expectation Values:*\n`{expectation_values}`" - }, - { - "type": "mrkdwn", - "text": f"*Expectation Errors:*\n`{expectation_errors}`" - } - ] - }, - { - "type": "section", - "fields": [ - { - "type": "mrkdwn", - "text": f"*Expectation Values (Sign Fixed):*\n`{expectation_values_sign_fixed}`" - }, - { - "type": "mrkdwn", - "text": f"*Timestamp:*\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" - } - ] - }, + *sections, { "type": "context", "elements": [ From 865202b228f43a0e2a6614f3c035b1cd191d5c48 Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Thu, 12 Feb 2026 21:50:59 -0600 Subject: [PATCH 09/12] Allows extra fields in config.py --- src/pqnstack/app/core/config.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/pqnstack/app/core/config.py b/src/pqnstack/app/core/config.py index b942c2a..798c126 100644 --- a/src/pqnstack/app/core/config.py +++ b/src/pqnstack/app/core/config.py @@ -47,7 +47,12 @@ class Settings(BaseSettings): rotary_encoder_address: str = "/dev/ttyACM0" virtual_rotator: bool = False # If True, use terminal input instead of hardware rotary encoder - model_config = SettingsConfigDict(toml_file="./config.toml", env_file=".env", env_file_encoding="utf-8") + model_config = SettingsConfigDict( + toml_file="./config.toml", + env_file=".env", + env_file_encoding="utf-8", + extra="ignore" # Allow extra fields in config.toml (e.g., daily_report) + ) @classmethod def settings_customise_sources( From 8a4deacb4c887c61ae80362f4ce5961a5ff3685f Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Thu, 12 Feb 2026 21:58:01 -0600 Subject: [PATCH 10/12] Update return type in `chsh` route to include list of floats --- src/pqnstack/app/api/routes/chsh.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pqnstack/app/api/routes/chsh.py b/src/pqnstack/app/api/routes/chsh.py index 9ed55ff..3889444 100644 --- a/src/pqnstack/app/api/routes/chsh.py +++ b/src/pqnstack/app/api/routes/chsh.py @@ -175,7 +175,7 @@ async def chsh( http_client: ClientDep, timetagger_address: str, state: StateDep, -) -> dict[str, float]: +) -> dict[str, float | list[float]]: logger.info("Starting CHSH experiment with basis: %s", basis) chsh_value, chsh_error, expectation_values, expectation_errors, expectation_values_sign_fixed = await _chsh(basis, follower_node_address, http_client, timetagger_address, state) From 6eca850afea23a8cdad524664dbb1f6a8eb44c7e Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Wed, 15 Apr 2026 11:10:09 -0300 Subject: [PATCH 11/12] ruff and mypy errors --- scripts/chsh_daily_report.py | 121 +++++++++----------- src/pqnstack/app/api/routes/chsh.py | 12 +- src/pqnstack/app/api/routes/coordination.py | 36 ++---- src/pqnstack/app/api/routes/qkd.py | 18 +-- src/pqnstack/app/api/routes/rng.py | 7 +- src/pqnstack/app/core/config.py | 4 +- 6 files changed, 86 insertions(+), 112 deletions(-) mode change 100644 => 100755 scripts/chsh_daily_report.py diff --git a/scripts/chsh_daily_report.py b/scripts/chsh_daily_report.py old mode 100644 new mode 100755 index 3923b84..561aad5 --- a/scripts/chsh_daily_report.py +++ b/scripts/chsh_daily_report.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -CHSH Daily Report Script +CHSH Daily Report Script. Runs CHSH measurement and posts results to Slack. @@ -10,25 +10,28 @@ uv run scripts/chsh_daily_report.py """ -import json +import logging import sys import tomllib +from datetime import UTC from datetime import datetime from pathlib import Path import httpx +logger = logging.getLogger(__name__) + def load_config() -> dict: """Load configuration from config.toml.""" config_path = Path(__file__).parent.parent / "config.toml" if not config_path.exists(): - print(f"❌ Error: config.toml not found at {config_path}") - print("Please create config.toml from configs/config_app_example.toml") + 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 open(config_path, "rb") as f: + with config_path.open("rb") as f: return tomllib.load(f) @@ -37,15 +40,15 @@ def get_daily_report_config(config: dict) -> dict: daily_report_config = config.get("daily_report", {}) if not daily_report_config: - print("❌ Error: [daily_report] section not found in config.toml") - print("Please add it following the example in configs/config_app_example.toml") + 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) # Check required fields required_fields = ["slack_webhook_url", "follower_node_address"] for field in required_fields: if not daily_report_config.get(field): - print(f"❌ Error: {field} not set in config.toml [daily_report] section") + logger.error("%s not set in config.toml [daily_report] section", field) sys.exit(1) return daily_report_config @@ -60,10 +63,10 @@ def run_chsh_measurement(config: dict) -> dict: follower_node_address = daily_report_config["follower_node_address"] basis = daily_report_config.get("basis", [0, 22.5]) - print(f"🔬 Starting CHSH measurement at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") - print(f" Basis: {basis}") - print(f" Follower: {follower_node_address}") - print(f" TimeTagger: {timetagger_address}") + logger.info("Starting CHSH measurement at %s", datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S")) + logger.info("Basis: %s", basis) + logger.info("Follower: %s", follower_node_address) + logger.info("TimeTagger: %s", timetagger_address) try: with httpx.Client(timeout=600.0) as client: @@ -73,21 +76,22 @@ def run_chsh_measurement(config: dict) -> dict: "follower_node_address": follower_node_address, "timetagger_address": timetagger_address, }, - json=basis + json=basis, ) response.raise_for_status() return response.json() - except httpx.HTTPError as e: - print(f"❌ Failed to contact CHSH API: {e}") + except httpx.HTTPError: + logger.exception("Failed to contact CHSH API") sys.exit(1) def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): """Post CHSH results to Slack.""" - # Determine emoji based on Bell inequality violation (CHSH > 2) if chsh_value exists + # Determine emoji based on Bell inequality violation (CHSH > classical limit) if chsh_value exists + bell_inequality_classical_limit = 2 chsh_value = chsh_data.get("chsh_value", 0) - emoji = ":sparkles:" if chsh_value > 2 else ":thinking_face:" + emoji = ":sparkles:" if chsh_value > bell_inequality_classical_limit else ":thinking_face:" daily_report_config = config.get("daily_report", {}) basis = daily_report_config.get("basis", [0, 22.5]) @@ -114,91 +118,76 @@ def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): else: formatted_value = str(value) - fields.append({ - "type": "mrkdwn", - "text": f"*{field_name}:*\n`{formatted_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 - }) + 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`{basis}`" - }, - { - "type": "mrkdwn", - "text": f"*Timestamp:*\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" - } - ] - }) + sections.append( + { + "type": "section", + "fields": [ + {"type": "mrkdwn", "text": f"*Basis:*\n`{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 - } + "text": {"type": "plain_text", "text": f"{emoji} CHSH Daily Measurement Report", "emoji": True}, }, *sections, { "type": "context", "elements": [ - { - "type": "mrkdwn", - "text": f"Follower: `{follower_address}` | TimeTagger: `{timetagger_address}`" - } - ] - } + {"type": "mrkdwn", "text": f"Follower: `{follower_address}` | TimeTagger: `{timetagger_address}`"} + ], + }, ] } - print("📤 Posting to Slack...") + logger.info("Posting to Slack...") try: with httpx.Client() as client: response = client.post(webhook_url, json=slack_message) if response.text == "ok": - print("✅ Successfully posted to Slack") + logger.info("Successfully posted to Slack") else: - print(f"❌ Failed to post to Slack: {response.text}") + logger.error("Failed to post to Slack: %s", response.text) sys.exit(1) - except httpx.HTTPError as e: - print(f"❌ Failed to post to Slack: {e}") + except httpx.HTTPError: + logger.exception("Failed to post to Slack") sys.exit(1) def post_error_to_slack(webhook_url: str, error_message: str): """Post error message to Slack.""" slack_message = { - "text": f":x: CHSH Daily Report Failed\n*Error:* {error_message}\n*Time:* {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" + "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 Exception: - pass # Silently fail if we can't post the error + except httpx.HTTPError: + logger.debug("Failed to post error notification to Slack") def main(): - """Main entry point.""" + """Execute the CHSH daily report.""" + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + try: # Load configuration config = load_config() @@ -208,16 +197,16 @@ def main(): # Run CHSH measurement chsh_data = run_chsh_measurement(config) - print(f"✅ CHSH measurement completed") - print(f" Value: {chsh_data['chsh_value']:.4f} ± {chsh_data['chsh_error']:.4f}") + logger.info("CHSH measurement completed") + logger.info("Value: %.4f ± %.4f", chsh_data["chsh_value"], chsh_data["chsh_error"]) # Post to Slack post_to_slack(webhook_url, chsh_data, config) - print("✅ CHSH daily report completed successfully") + logger.info("CHSH daily report completed successfully") except Exception as e: - print(f"❌ Unexpected error: {e}") + logger.exception("Unexpected error") # Try to post error to Slack if possible try: @@ -225,11 +214,11 @@ def main(): daily_report_config = get_daily_report_config(config) webhook_url = daily_report_config["slack_webhook_url"] post_error_to_slack(webhook_url, str(e)) - except Exception: - pass + except Exception: # noqa: BLE001 + logger.debug("Failed to post error notification to Slack") sys.exit(1) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/src/pqnstack/app/api/routes/chsh.py b/src/pqnstack/app/api/routes/chsh.py index 3889444..b4b8262 100644 --- a/src/pqnstack/app/api/routes/chsh.py +++ b/src/pqnstack/app/api/routes/chsh.py @@ -27,6 +27,7 @@ @router.get("/progress") async def chsh_progress(state: StateDep) -> StreamingResponse: """SSE endpoint for streaming CHSH measurement progress to frontend.""" + async def event_generator(): try: # Send initial connection event @@ -41,7 +42,7 @@ async def event_generator(): yield f"data: {json.dumps({'event': 'chsh_progress', 'current': state.chsh_progress_current, 'total': state.chsh_progress_total, 'running': state.chsh_running})}\n\n" chsh_progress_event.clear() event_sent = True - except asyncio.TimeoutError: + except TimeoutError: pass # Send heartbeat if no event was sent to keep connection alive @@ -60,9 +61,10 @@ async def event_generator(): headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", - } + }, ) + # FIXME: Make the return of this function a dataclass async def _chsh( # Complexity is high due to the nature of the CHSH experiment. basis: tuple[float, float], @@ -76,7 +78,7 @@ async def _chsh( # Complexity is high due to the nature of the CHSH experiment. # Initialize progress tracking state.chsh_running = True state.chsh_progress_current = 0 - state.chsh_progress_total = 16 # 2 basis × 2 follower × 2 angles × 2 perp + state.chsh_progress_total = 16 # 2 basis x 2 follower x 2 angles x 2 perp chsh_progress_event.set() logger.debug("Instantiating client") @@ -178,7 +180,9 @@ async def chsh( ) -> dict[str, float | list[float]]: logger.info("Starting CHSH experiment with basis: %s", basis) - chsh_value, chsh_error, expectation_values, expectation_errors, expectation_values_sign_fixed = await _chsh(basis, follower_node_address, http_client, timetagger_address, state) + chsh_value, chsh_error, expectation_values, expectation_errors, expectation_values_sign_fixed = await _chsh( + basis, follower_node_address, http_client, timetagger_address, state + ) return { "chsh_value": chsh_value, diff --git a/src/pqnstack/app/api/routes/coordination.py b/src/pqnstack/app/api/routes/coordination.py index a985553..764aac6 100644 --- a/src/pqnstack/app/api/routes/coordination.py +++ b/src/pqnstack/app/api/routes/coordination.py @@ -62,17 +62,11 @@ async def reset_coordination_state(state: StateDep, http_client: ClientDep) -> R logger.info("Notifying peer at %s about protocol cancellation", peer_address) await http_client.post( f"http://{peer_address}/coordination/protocol_cancelled", - json={ - "reason": "Protocol cancelled by user", - "cancelled_by_role": current_role.value - }, - timeout=5.0 # Short timeout to avoid hanging - ) - except Exception as e: - logger.warning( - "Failed to notify peer about cancellation: %s. Proceeding with reset.", - str(e) + json={"reason": "Protocol cancelled by user", "cancelled_by_role": current_role.value}, + timeout=5.0, # Short timeout to avoid hanging ) + except Exception as e: # noqa: BLE001 + logger.warning("Failed to notify peer about cancellation: %s. Proceeding with reset.", str(e)) # Set local cancellation event to unblock any waiting operations protocol_cancelled_event.set() @@ -106,11 +100,7 @@ async def protocol_cancelled( notification: ProtocolCancellationNotification, ) -> dict[str, str]: """Receive notification that peer node cancelled the protocol.""" - logger.info( - "Received protocol cancellation from %s: %s", - notification.cancelled_by_role, - notification.reason - ) + logger.info("Received protocol cancellation from %s: %s", notification.cancelled_by_role, notification.reason) protocol_cancelled_event.set() # Give waiting operations a chance to wake up and handle the cancellation @@ -192,12 +182,9 @@ async def follow_requested( logger.debug("Asking user to accept follow request from %s (%s)", leaders_name, leaders_address) # Wait for EITHER user reply OR cancellation - done, pending = await asyncio.wait( - [ - asyncio.create_task(user_replied_event.wait()), - asyncio.create_task(protocol_cancelled_event.wait()) - ], - return_when=asyncio.FIRST_COMPLETED + _done, pending = await asyncio.wait( + [asyncio.create_task(user_replied_event.wait()), asyncio.create_task(protocol_cancelled_event.wait())], + return_when=asyncio.FIRST_COMPLETED, ) # Cancel pending tasks @@ -212,7 +199,6 @@ async def follow_requested( state.leaders_name = "" state.following_requested = False state.following_requested_user_response = None - # protocol_cancelled_event.clear() return FollowRequestResponse(accepted=False) user_replied_event.clear() # Reset the event for the next change @@ -285,9 +271,9 @@ async def client_message_handler() -> None: @router.get("/state_events") async def state_events(state: StateDep) -> StreamingResponse: """SSE endpoint for streaming state change events to frontend.""" + async def event_generator(): try: - last_role = state.role # Send initial connection event yield f"data: {json.dumps({'event': 'connected', 'role': state.role.value})}\n\n" @@ -301,7 +287,7 @@ async def event_generator(): yield f"data: {json.dumps({'event': 'protocol_cancelled', 'reason': 'Protocol cancelled by peer or user'})}\n\n" protocol_cancelled_event.clear() event_sent = True - except asyncio.TimeoutError: + except TimeoutError: pass # Send heartbeat if no event was sent to keep connection alive @@ -320,5 +306,5 @@ async def event_generator(): headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", - } + }, ) diff --git a/src/pqnstack/app/api/routes/qkd.py b/src/pqnstack/app/api/routes/qkd.py index 5adcd8b..fc090bc 100644 --- a/src/pqnstack/app/api/routes/qkd.py +++ b/src/pqnstack/app/api/routes/qkd.py @@ -302,10 +302,7 @@ async def _wait_for_follower_ready(state: NodeState, http_client: httpx.AsyncCli # Check if protocol was cancelled if protocol_cancelled_event.is_set(): logger.warning("Protocol cancelled while waiting for follower ready") - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail="Protocol cancelled by peer or user" - ) + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Protocol cancelled by peer or user") r = await http_client.get(f"http://{state.followers_address}/qkd/is_follower_ready") @@ -325,7 +322,7 @@ async def _wait_for_follower_ready(state: NodeState, http_client: httpx.AsyncCli ready = r.json() if not ready: logger.info("Follower is not ready yet, waiting.") - await asyncio.sleep(0.4) # Make sure this is smaller than the protocol_cancelled event clear timer. + await asyncio.sleep(0.4) # Make sure this is smaller than the protocol_cancelled event clear timer. logger.info("Follower is ready") @@ -370,12 +367,12 @@ async def _submit_basis_list_follower(state: NodeState, basis_list: list[QKDEnco # don't wait for the event if the result is already set. This avoids deadlocks in case the result was set before this function is called. if state.qkd_n_matching_bits == -1: # Wait for EITHER result OR cancellation - done, pending = await asyncio.wait( + _done, pending = await asyncio.wait( [ asyncio.create_task(qkd_result_received_event.wait()), - asyncio.create_task(protocol_cancelled_event.wait()) + asyncio.create_task(protocol_cancelled_event.wait()), ], - return_when=asyncio.FIRST_COMPLETED + return_when=asyncio.FIRST_COMPLETED, ) # Cancel pending tasks @@ -385,10 +382,7 @@ async def _submit_basis_list_follower(state: NodeState, basis_list: list[QKDEnco # Check if protocol was cancelled if protocol_cancelled_event.is_set(): logger.warning("Protocol cancelled while waiting for QKD result") - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail="Protocol cancelled by peer or user" - ) + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Protocol cancelled by peer or user") # Reassemble the QKDResult object from the state qkd_result = QKDResult( diff --git a/src/pqnstack/app/api/routes/rng.py b/src/pqnstack/app/api/routes/rng.py index a0c6ee2..f55fdb3 100644 --- a/src/pqnstack/app/api/routes/rng.py +++ b/src/pqnstack/app/api/routes/rng.py @@ -22,6 +22,7 @@ @router.get("/progress") async def rng_progress(state: StateDep) -> StreamingResponse: """SSE endpoint for streaming RNG fortune measurement progress to frontend.""" + async def event_generator(): try: # Send initial connection event @@ -36,7 +37,7 @@ async def event_generator(): yield f"data: {json.dumps({'event': 'rng_progress', 'current': state.rng_progress_current, 'total': state.rng_progress_total, 'running': state.rng_running})}\n\n" rng_progress_event.clear() event_sent = True - except asyncio.TimeoutError: + except TimeoutError: pass # Send heartbeat if no event was sent to keep connection alive @@ -55,7 +56,7 @@ async def event_generator(): headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", - } + }, ) @@ -97,7 +98,7 @@ async def singles_parity( @router.get("/fortune") -async def fortune( +async def fortune( # noqa: PLR0913 timetagger_address: str, integration_time_s: float, fortune_size: int, diff --git a/src/pqnstack/app/core/config.py b/src/pqnstack/app/core/config.py index 798c126..bbc8f5b 100644 --- a/src/pqnstack/app/core/config.py +++ b/src/pqnstack/app/core/config.py @@ -51,7 +51,7 @@ class Settings(BaseSettings): toml_file="./config.toml", env_file=".env", env_file_encoding="utf-8", - extra="ignore" # Allow extra fields in config.toml (e.g., daily_report) + extra="ignore", # Allow extra fields in config.toml (e.g., daily_report) ) @classmethod @@ -108,7 +108,7 @@ class NodeState(BaseModel): # CHSH state chsh_request_basis: list[float] = [22.5, 67.5] chsh_progress_current: int = 0 # Current iteration in CHSH measurement - chsh_progress_total: int = 16 # Total iterations (2 basis × 2 follower × 2 angles × 2 perp) + chsh_progress_total: int = 16 # Total iterations (2 basis x 2 follower x 2 angles x 2 perp) chsh_running: bool = False # Whether CHSH measurement is currently running # QKD state From f5a4e56e3917e1c944815162fdc1a5e4e7d20156 Mon Sep 17 00:00:00 2001 From: marcosf2 Date: Wed, 15 Apr 2026 12:03:41 -0300 Subject: [PATCH 12/12] More mypy errors --- scripts/chsh_daily_report.py | 99 ++++++++------------- src/pqnstack/app/api/routes/chsh.py | 40 +++++---- src/pqnstack/app/api/routes/coordination.py | 4 +- src/pqnstack/app/api/routes/rng.py | 3 +- src/pqnstack/app/core/config.py | 8 ++ 5 files changed, 71 insertions(+), 83 deletions(-) diff --git a/scripts/chsh_daily_report.py b/scripts/chsh_daily_report.py index 561aad5..9e44574 100755 --- a/scripts/chsh_daily_report.py +++ b/scripts/chsh_daily_report.py @@ -18,12 +18,16 @@ 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() -> dict: - """Load configuration from config.toml.""" +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(): @@ -32,85 +36,60 @@ def load_config() -> dict: sys.exit(1) with config_path.open("rb") as f: - return tomllib.load(f) - - -def get_daily_report_config(config: dict) -> dict: - """Get and validate daily_report configuration.""" - daily_report_config = config.get("daily_report", {}) + raw = tomllib.load(f) - if not daily_report_config: + 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) - # Check required fields - required_fields = ["slack_webhook_url", "follower_node_address"] - for field in required_fields: - if not daily_report_config.get(field): - logger.error("%s not set in config.toml [daily_report] section", field) - sys.exit(1) - - return daily_report_config + try: + return DailyReportConfig.model_validate(daily_report_data) + except ValidationError: + logger.exception("Invalid [daily_report] configuration") + sys.exit(1) -def run_chsh_measurement(config: dict) -> dict: +def run_chsh_measurement(config: DailyReportConfig) -> ChshResult: """Run CHSH measurement via API.""" - daily_report_config = get_daily_report_config(config) - - api_url = daily_report_config.get("api_url", "http://localhost:8000") - timetagger_address = daily_report_config.get("timetagger_address", "127.0.0.1:8000") - follower_node_address = daily_report_config["follower_node_address"] - basis = daily_report_config.get("basis", [0, 22.5]) - logger.info("Starting CHSH measurement at %s", datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S")) - logger.info("Basis: %s", basis) - logger.info("Follower: %s", follower_node_address) - logger.info("TimeTagger: %s", timetagger_address) + 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"{api_url}/chsh/", + f"{config.api_url}/chsh/", params={ - "follower_node_address": follower_node_address, - "timetagger_address": timetagger_address, + "follower_node_address": config.follower_node_address, + "timetagger_address": config.timetagger_address, }, - json=basis, + json=config.basis, ) response.raise_for_status() - return response.json() + 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: dict, config: dict): +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) if chsh_value exists + # Determine emoji based on Bell inequality violation (CHSH > classical limit) bell_inequality_classical_limit = 2 - chsh_value = chsh_data.get("chsh_value", 0) - emoji = ":sparkles:" if chsh_value > bell_inequality_classical_limit else ":thinking_face:" - - daily_report_config = config.get("daily_report", {}) - basis = daily_report_config.get("basis", [0, 22.5]) - follower_address = daily_report_config.get("follower_node_address", "unknown") - timetagger_address = daily_report_config.get("timetagger_address", "unknown") + emoji = ":sparkles:" if chsh_data.chsh_value > bell_inequality_classical_limit else ":thinking_face:" # Build fields dynamically from all returned data fields = [] - - # Add all fields from the API response - for key, value in chsh_data.items(): - # Format the key nicely (replace underscores with spaces, capitalize) + for key, value in chsh_data.model_dump().items(): field_name = key.replace("_", " ").title() - # Format the value based on type if isinstance(value, float): formatted_value = f"{value:.4f}" elif isinstance(value, list): - # Format list nicely 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: @@ -131,7 +110,7 @@ def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): { "type": "section", "fields": [ - {"type": "mrkdwn", "text": f"*Basis:*\n`{basis}`"}, + {"type": "mrkdwn", "text": f"*Basis:*\n`{config.basis}`"}, {"type": "mrkdwn", "text": f"*Timestamp:*\n{datetime.now(UTC).strftime('%Y-%m-%d %H:%M:%S')}"}, ], } @@ -148,7 +127,10 @@ def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): { "type": "context", "elements": [ - {"type": "mrkdwn", "text": f"Follower: `{follower_address}` | TimeTagger: `{timetagger_address}`"} + { + "type": "mrkdwn", + "text": f"Follower: `{config.follower_node_address}` | TimeTagger: `{config.timetagger_address}`", + } ], }, ] @@ -171,7 +153,7 @@ def post_to_slack(webhook_url: str, chsh_data: dict, config: dict): sys.exit(1) -def post_error_to_slack(webhook_url: str, error_message: str): +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')}" @@ -184,24 +166,19 @@ def post_error_to_slack(webhook_url: str, error_message: str): logger.debug("Failed to post error notification to Slack") -def main(): +def main() -> None: """Execute the CHSH daily report.""" logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") try: - # Load configuration config = load_config() - daily_report_config = get_daily_report_config(config) - webhook_url = daily_report_config["slack_webhook_url"] - # Run CHSH measurement chsh_data = run_chsh_measurement(config) logger.info("CHSH measurement completed") - logger.info("Value: %.4f ± %.4f", chsh_data["chsh_value"], chsh_data["chsh_error"]) + logger.info("Value: %.4f ± %.4f", chsh_data.chsh_value, chsh_data.chsh_error) - # Post to Slack - post_to_slack(webhook_url, chsh_data, config) + post_to_slack(config.slack_webhook_url, chsh_data, config) logger.info("CHSH daily report completed successfully") @@ -211,9 +188,7 @@ def main(): # Try to post error to Slack if possible try: config = load_config() - daily_report_config = get_daily_report_config(config) - webhook_url = daily_report_config["slack_webhook_url"] - post_error_to_slack(webhook_url, str(e)) + post_error_to_slack(config.slack_webhook_url, str(e)) except Exception: # noqa: BLE001 logger.debug("Failed to post error notification to Slack") diff --git a/src/pqnstack/app/api/routes/chsh.py b/src/pqnstack/app/api/routes/chsh.py index b4b8262..c5250aa 100644 --- a/src/pqnstack/app/api/routes/chsh.py +++ b/src/pqnstack/app/api/routes/chsh.py @@ -1,6 +1,7 @@ import asyncio import json import logging +from collections.abc import AsyncGenerator from typing import TYPE_CHECKING from typing import cast @@ -8,6 +9,7 @@ from fastapi import HTTPException from fastapi import status from fastapi.responses import StreamingResponse +from pydantic import BaseModel from pqnstack.app.api.deps import ClientDep from pqnstack.app.api.deps import StateDep @@ -21,6 +23,15 @@ logger = logging.getLogger(__name__) + +class ChshResult(BaseModel): + chsh_value: float + chsh_error: float + expectation_values: list[float] + expectation_errors: list[float] + expectation_values_sign_fixed: list[float] + + router = APIRouter(prefix="/chsh", tags=["chsh"]) @@ -28,7 +39,7 @@ async def chsh_progress(state: StateDep) -> StreamingResponse: """SSE endpoint for streaming CHSH measurement progress to frontend.""" - async def event_generator(): + async def event_generator() -> AsyncGenerator[str, None]: try: # Send initial connection event yield f"data: {json.dumps({'event': 'connected'})}\n\n" @@ -65,14 +76,13 @@ async def event_generator(): ) -# FIXME: Make the return of this function a dataclass async def _chsh( # Complexity is high due to the nature of the CHSH experiment. basis: tuple[float, float], follower_node_address: str, http_client: ClientDep, timetagger_address: str, state: StateDep, -) -> tuple[float, float]: +) -> ChshResult: logger.debug("Starting CHSH") # Initialize progress tracking @@ -166,10 +176,15 @@ async def _chsh( # Complexity is high due to the nature of the CHSH experiment. state.chsh_running = False chsh_progress_event.set() - return chsh_value, chsh_error, expectation_values, expectation_errors, expectation_values_sign_fixed + return ChshResult( + chsh_value=chsh_value, + chsh_error=chsh_error, + expectation_values=expectation_values, + expectation_errors=expectation_errors, + expectation_values_sign_fixed=expectation_values_sign_fixed, + ) -# FIXME: make the return of this function the same dataclass as the one returned by _chsh. @router.post("/") async def chsh( basis: tuple[float, float], @@ -177,20 +192,9 @@ async def chsh( http_client: ClientDep, timetagger_address: str, state: StateDep, -) -> dict[str, float | list[float]]: +) -> ChshResult: logger.info("Starting CHSH experiment with basis: %s", basis) - - chsh_value, chsh_error, expectation_values, expectation_errors, expectation_values_sign_fixed = await _chsh( - basis, follower_node_address, http_client, timetagger_address, state - ) - - return { - "chsh_value": chsh_value, - "chsh_error": chsh_error, - "expectation_values": expectation_values, - "expectation_errors": expectation_errors, - "expectation_values_sign_fixed": expectation_values_sign_fixed, - } + return await _chsh(basis, follower_node_address, http_client, timetagger_address, state) @router.post("/request-angle-by-basis") diff --git a/src/pqnstack/app/api/routes/coordination.py b/src/pqnstack/app/api/routes/coordination.py index 764aac6..ecd9fb8 100644 --- a/src/pqnstack/app/api/routes/coordination.py +++ b/src/pqnstack/app/api/routes/coordination.py @@ -1,6 +1,7 @@ import asyncio import json import logging +from collections.abc import AsyncGenerator from fastapi import APIRouter from fastapi import HTTPException @@ -272,9 +273,8 @@ async def client_message_handler() -> None: async def state_events(state: StateDep) -> StreamingResponse: """SSE endpoint for streaming state change events to frontend.""" - async def event_generator(): + async def event_generator() -> AsyncGenerator[str, None]: try: - # Send initial connection event yield f"data: {json.dumps({'event': 'connected', 'role': state.role.value})}\n\n" diff --git a/src/pqnstack/app/api/routes/rng.py b/src/pqnstack/app/api/routes/rng.py index f55fdb3..200ac0a 100644 --- a/src/pqnstack/app/api/routes/rng.py +++ b/src/pqnstack/app/api/routes/rng.py @@ -1,6 +1,7 @@ import asyncio import json import logging +from collections.abc import AsyncGenerator from typing import Annotated from typing import Any @@ -23,7 +24,7 @@ async def rng_progress(state: StateDep) -> StreamingResponse: """SSE endpoint for streaming RNG fortune measurement progress to frontend.""" - async def event_generator(): + async def event_generator() -> AsyncGenerator[str, None]: try: # Send initial connection event yield f"data: {json.dumps({'event': 'connected'})}\n\n" diff --git a/src/pqnstack/app/core/config.py b/src/pqnstack/app/core/config.py index bbc8f5b..0505b17 100644 --- a/src/pqnstack/app/core/config.py +++ b/src/pqnstack/app/core/config.py @@ -17,6 +17,14 @@ logger = logging.getLogger(__name__) +class DailyReportConfig(BaseModel): + slack_webhook_url: str + follower_node_address: str + 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]) + + class CHSHSettings(BaseModel): # Specifies which half waveplate to use for the CHSH experiment. First value is the provider's name, second is the motor name. hwp: tuple[str, str] = ("", "")