From 8db14b94822d2eafc81f51767582809bc2a0d366 Mon Sep 17 00:00:00 2001 From: Ayzen Date: Tue, 23 Jun 2026 12:19:31 +0300 Subject: [PATCH] added data saving feature --- python_app/gui/app_window.py | 25 ++ .../app_window_config/profile_io_mixin.py | 2 + .../app_window_config/state_builders.py | 2 + .../controllers/app_window_pipeline_mixin.py | 7 + .../controllers/app_window_recording_mixin.py | 334 ++++++++++++++++++ .../controllers/app_window_snapshot_mixin.py | 42 ++- .../gui/controllers/app_window_web_mixin.py | 153 ++++++-- .../sections/data_actions_section.py | 18 + .../hardware_full/kamil_adc/processing.py | 75 +++- python_app/models/gui_profile_codec.py | 7 + python_app/models/gui_profile_schema.py | 3 + python_app/storage/npz/snapshot_numpy.py | 25 +- python_app/storage/npz/store.py | 73 ++++ python_app/tests/test_kamil_adc_processing.py | 49 +++ python_app/tests/test_recording_mixin.py | 265 ++++++++++++++ python_app/tests/test_storage_webui.py | 42 ++- python_app/webui/controller.py | 27 +- python_app/webui/routes.py | 48 ++- python_app/webui/static/app.js | 50 ++- python_app/webui/static/index.html | 8 +- 20 files changed, 1192 insertions(+), 63 deletions(-) create mode 100644 python_app/gui/controllers/app_window_recording_mixin.py create mode 100644 python_app/tests/test_recording_mixin.py diff --git a/python_app/gui/app_window.py b/python_app/gui/app_window.py index 3d4f54b..476a822 100644 --- a/python_app/gui/app_window.py +++ b/python_app/gui/app_window.py @@ -25,6 +25,7 @@ from python_app.gui.controllers.app_window_config_mixin import AppWindowConfigMi from python_app.gui.controllers.app_window_control_button_mixin import AppWindowControlButtonMixin from python_app.gui.controllers.app_window_pipeline_mixin import AppWindowPipelineMixin from python_app.gui.controllers.app_window_plot_mixin import AppWindowPlotMixin +from python_app.gui.controllers.app_window_recording_mixin import AppWindowRecordingMixin from python_app.gui.controllers.app_window_preprocess_mixin import AppWindowPreprocessMixin from python_app.gui.controllers.app_window_snapshot_mixin import AppWindowSnapshotMixin from python_app.gui.controllers.app_window_ui_mixin import AppWindowUiMixin @@ -96,6 +97,7 @@ class AppWindow( AppWindowPlotMixin, AppWindowPipelineMixin, AppWindowSnapshotMixin, + AppWindowRecordingMixin, AppWindowControlButtonMixin, AppWindowWebMixin, QMainWindow, @@ -114,6 +116,7 @@ class AppWindow( self._init_preprocess_state() self._init_capture_state() self._init_history_state() + self._init_recording_state() self._init_runtime_limits() self._init_polling_timer() self._init_control_button_state() @@ -676,9 +679,27 @@ class AppWindow( return max(0, int(raw_value)) return 0 + def _capture_web_action_error(self, message: str) -> bool: + """Record the first error of an in-flight web action so the browser can show it. + + Returns ``True`` if a web-triggered action is currently running (see + ``AppWindowWebMixin._run_web_action``). Callers use that to skip the blocking + desktop modal for web errors — the browser shows the message instead, and the + operator at the browser must not have to dismiss a popup on the (often headless) + host before the HTTP response returns. Desktop-only errors are unaffected. + """ + capture = getattr(self, "_web_action_error_capture", None) + if capture is None: + return False + if not capture: + capture.append(message) + return True + def _show_error(self, message: str, *, details: str | None = None) -> None: """Log and present an error in a modal dialog with optional detail text.""" self._log_error(message, details=details) + if self._capture_web_action_error(message): + return # web-triggered: surfaced to the browser; no blocking desktop modal if self._is_truthy_env("RADAR_SYSTEM_HEADLESS"): return dialog = QMessageBox(self) @@ -692,6 +713,8 @@ class AppWindow( def _show_exception(self, context: str, exc: Exception) -> None: """Log full exception details and show modal dialog with expandable traceback.""" message, details = self._log_exception(context, exc, level="ERROR") + if self._capture_web_action_error(message): + return # web-triggered: surfaced to the browser; no blocking desktop modal if self._is_truthy_env("RADAR_SYSTEM_HEADLESS"): return dialog = QMessageBox(self) @@ -717,6 +740,8 @@ class AppWindow( try: # 0) Stop the web server first so a late request cannot start work. self._shutdown_web_ui() + # 0) Stop the disk-recording writer thread so it is not orphaned. + self._shutdown_recording() # 0) Stop the GPIO button watcher so a late press cannot start work. self._stop_control_button_watcher() self._resume_pipeline_after_capture = False diff --git a/python_app/gui/controllers/app_window_config/profile_io_mixin.py b/python_app/gui/controllers/app_window_config/profile_io_mixin.py index 9571c9e..bfdd418 100644 --- a/python_app/gui/controllers/app_window_config/profile_io_mixin.py +++ b/python_app/gui/controllers/app_window_config/profile_io_mixin.py @@ -336,6 +336,7 @@ class AppWindowConfigProfileIOMixin: self._legacy_gpr_visible_z_min_m, self._legacy_gpr_visible_z_max_m, self._save_count, + self._record_count, self._save_path_input, self._save_name_input, self._adc_project_dir_input, @@ -522,6 +523,7 @@ class AppWindowConfigProfileIOMixin: self._legacy_gpr_visible_z_max_m.setValue(float(gui_state.processing.legacy_gpr.visible_z_max_m)) self._save_count.setValue(int(gui_state.data_actions.save_count)) + self._record_count.setValue(int(gui_state.data_actions.record_count)) self._save_path_input.setText(str(gui_state.data_actions.save_path)) self._save_name_input.setText(str(gui_state.data_actions.save_name)) diff --git a/python_app/gui/controllers/app_window_config/state_builders.py b/python_app/gui/controllers/app_window_config/state_builders.py index 8690ef8..8a0776d 100644 --- a/python_app/gui/controllers/app_window_config/state_builders.py +++ b/python_app/gui/controllers/app_window_config/state_builders.py @@ -273,6 +273,7 @@ class AppWindowConfigStateBuildersMixin: save_count=10, save_path=str(self._project_root / "python_app/data/snapshots"), save_name="snapshot_manual", + record_count=100, ), preprocess_dialog=GuiPreprocessDialogStateModel( set_name="set_001", @@ -404,6 +405,7 @@ class AppWindowConfigStateBuildersMixin: save_count=int(self._save_count.value()), save_path=self._save_path_input.text().strip(), save_name=self._save_name_input.text().strip(), + record_count=int(self._record_count.value()), ), preprocess_dialog=GuiPreprocessDialogStateModel( set_name=self._current_preprocess_set_name(), diff --git a/python_app/gui/controllers/app_window_pipeline_mixin.py b/python_app/gui/controllers/app_window_pipeline_mixin.py index d672713..aa82914 100644 --- a/python_app/gui/controllers/app_window_pipeline_mixin.py +++ b/python_app/gui/controllers/app_window_pipeline_mixin.py @@ -281,6 +281,9 @@ class AppWindowPipelineMixin: else: self._supervisor.stop() self._drain_rings_once_for_history() + # Flush any in-progress disk recording so its buffered (not-yet-full) chunk is + # written rather than lost; the drains above already captured the last data. + self._finalize_recording_on_stop() keep_results_reader = self._supervisor.is_processor_running() self._close_readers(keep_results=keep_results_reader) self._single_capture_active = False @@ -340,6 +343,7 @@ class AppWindowPipelineMixin: self._read_all_raw() self._read_all_preprocessed() result_latest = self._read_all_results() if self._result_reader is not None else None + self._poll_recording_writer() # finalize a disk recording once its writer drains self._update_history_indicator() self._last_reader_error_signature = None self._reader_error_repeat_count = 0 @@ -550,6 +554,7 @@ class AppWindowPipelineMixin: if collection is None: break self._raw_history.append(collection) + self._record_collection("raw", collection) latest = collection # `capture_*_ns` are populated by the C++ sweep_orchestrator with # wallclocks captured around the actual device read. Pre-orchestrator @@ -573,6 +578,7 @@ class AppWindowPipelineMixin: if collection is None: break self._pre_history.append(collection) + self._record_collection("preprocessed", collection) def _read_all_results(self) -> ResultCollection | None: """Read available result collections from results ring.""" @@ -585,6 +591,7 @@ class AppWindowPipelineMixin: break self._pipeline_metrics.record("processing", int(collection.processing_duration_ns)) record_result_history(self._result_history, collection) + self._record_collection("results", collection) latest = collection return latest diff --git a/python_app/gui/controllers/app_window_recording_mixin.py b/python_app/gui/controllers/app_window_recording_mixin.py new file mode 100644 index 0000000..232b668 --- /dev/null +++ b/python_app/gui/controllers/app_window_recording_mixin.py @@ -0,0 +1,334 @@ +"""Record the next N runtime measurements to disk, in chunks, on a background thread. + +The manual "Save Dataset" button persists what is *already* in history (capped at the +GUI history limit). This mixin adds the complementary "Start + Record" action: arm a +recording, then stream the measurements that arrive *after* arming — up to the +configured count — to disk without ever blocking acquisition. + +Key properties: + +* **Flat memory.** Measurements are flushed in chunks of :data:`_RECORDING_CHUNK_SIZE` + and freed, and at most :data:`_RECORDING_QUEUE_CHUNKS` chunks are ever in flight, so + RAM does not grow with the recording length. +* **Off the GUI thread.** A single background writer thread does all disk I/O; the GUI + poll tick only hands it chunks. The writes never stall acquisition or the web UI. If + the disk genuinely cannot keep up, the bounded queue applies brief backpressure + rather than growing memory without limit. +* **Aligned at capture.** Each result is paired with its raw/preprocessed collection by + ``collection_id`` as it arrives, so no cross-stage buffering is needed. +* **Only new sweeps.** Collections produced before the arm instant are ignored (gated + on ``monotonic_ns``), so a pre-existing ring backlog is not recorded. +* **No double-arm.** Arming is refused while a recording is still in progress. + +The on-disk layout is identical to "Save Dataset" (one streaming dataset directory). +""" + +from __future__ import annotations + +from contextlib import suppress +import logging +from pathlib import Path +import queue +import threading +import time + +logger = logging.getLogger(__name__) + +# Flush to disk every this many results, then drop them from memory. +_RECORDING_CHUNK_SIZE = 200 +# Max chunks queued for the writer thread before the producer applies backpressure; +# bounds in-flight memory to roughly this many chunks of measurements. +_RECORDING_QUEUE_CHUNKS = 8 +# Safety bound on raw/preprocessed collections still waiting for their result (normally +# a handful). If the processor stalls, the oldest are dropped so the maps stay bounded. +_RECORDING_MAX_PENDING = 512 + + +class AppWindowRecordingMixin: + """Stream the next N arriving measurements to a snapshot dataset on a worker thread.""" + + def _init_recording_state(self) -> None: + """Initialise the (idle) disk-recording state. Called once from ``__init__``.""" + self._recording_active = False + self._recording_collecting = False + self._recording_target = 0 + self._recording_since_ns = 0 + self._recording_enqueued = 0 + self._recording_writer = None + self._recording_queue: queue.Queue | None = None + self._recording_thread: threading.Thread | None = None + self._recording_stop = threading.Event() + self._recording_write_error: str | None = None + self._recording_pending_raw: dict[int, object] = {} + self._recording_pending_pre: dict[int, object] = {} + self._recording_chunk_raw: list = [] + self._recording_chunk_pre: list = [] + self._recording_chunk_results: list = [] + + # -- arming --------------------------------------------------------------- + + def _start_run_with_recording(self) -> None: + """Arm a streamed recording of the next N measurements, starting the run if stopped. + + ``N`` comes from the record-count field; the destination is the shared save + path/name. Refuses to start a second recording while one is in progress, and + reports a name clash up front (before any data is collected). + """ + if self._recording_active: + self._show_error( + "A recording is already in progress; wait for it to finish before starting another", + details=f"progress={self._recording_collected_count()}/{self._recording_target}", + ) + return + + target = int(self._record_count.value()) + + destination = self._snapshot_destination_dir() + if destination.exists(): + self._show_error( + "Recording destination already exists; choose a new name or path", + details=f"destination={destination}", + ) + return + + try: + writer = self._store.create_snapshot_stream( + Path(self._save_path_input.text().strip()).expanduser(), + self._save_name_input.text().strip(), + name_prefix=self._radar_config_name_prefix(), + ) + # Adjacent config profile, mirroring the manual save (fail if it clashes). + self._write_gui_profile_to_path( + self._snapshot_config_profile_path(writer.directory), allow_overwrite=False + ) + except Exception as exc: # noqa: BLE001 + self._show_exception("Failed to start disk recording", exc) + return + + self._recording_writer = writer + self._recording_target = target + self._recording_since_ns = time.monotonic_ns() + self._recording_enqueued = 0 + self._recording_write_error = None + self._recording_pending_raw = {} + self._recording_pending_pre = {} + self._recording_chunk_raw = [] + self._recording_chunk_pre = [] + self._recording_chunk_results = [] + self._recording_queue = queue.Queue(maxsize=_RECORDING_QUEUE_CHUNKS) + self._recording_stop = threading.Event() + self._recording_thread = threading.Thread( + target=self._recording_writer_loop, + args=(writer, self._recording_queue, self._recording_stop), + name="disk-recorder", + daemon=True, + ) + self._recording_thread.start() + self._recording_active = True + self._recording_collecting = True + + if not self._supervisor.is_running(): + self._start_run() + + self._log(f"Disk recording armed: streaming the next {target} measurement(s) to {writer.directory}") + + # -- background writer ---------------------------------------------------- + + def _recording_writer_loop( + self, writer, work_queue: queue.Queue, stop: threading.Event + ) -> None: + """Write queued chunks to disk until a sentinel, a stop request, or an error. + + Runs on a daemon thread; touches only the writer (file I/O) and logging — never + Qt. A write failure is recorded for the GUI thread to surface and ends the loop. + """ + while not stop.is_set(): + try: + chunk = work_queue.get(timeout=0.2) + except queue.Empty: + continue + if chunk is None: # sentinel: target reached, all chunks drained + return + raw, preprocessed, results = chunk + try: + writer.append(raw, preprocessed, results) + except Exception as exc: # noqa: BLE001 - reported to the GUI thread + logger.exception("Disk recording chunk write failed") + self._recording_write_error = f"{type(exc).__name__}: {exc}" + return + + # -- capture (called from the ring-read hooks on the poll tick) ----------- + + def _record_collection(self, stage: str, collection) -> None: + """Stream one arriving collection while recording; hand chunks to the writer thread. + + ``stage`` is ``"raw"``, ``"preprocessed"`` or ``"results"``. Raw/preprocessed + collections are held by ``collection_id`` until their result arrives; the result + pairs them and appends an aligned measurement to the current chunk. + """ + if not self._recording_collecting: + return + + produced_ns = int(getattr(collection, "monotonic_ns", 0) or 0) + if produced_ns and produced_ns < self._recording_since_ns: + return + + if stage == "raw": + self._stash_pending(self._recording_pending_raw, collection) + elif stage == "preprocessed": + self._stash_pending(self._recording_pending_pre, collection) + elif stage == "results": + self._record_result(collection) + + @staticmethod + def _stash_pending(pending: dict, collection) -> None: + """Hold a raw/preprocessed collection by id until its result arrives (bounded).""" + pending[int(collection.collection_id)] = collection + while len(pending) > _RECORDING_MAX_PENDING: + pending.pop(next(iter(pending))) # drop oldest; its result never came + + def _record_result(self, result) -> None: + """Pair a result with its raw/preprocessed, buffer it, and flush/finish as needed.""" + if self._recording_collected_count() >= self._recording_target: + return # already have the full target accepted + + collection_id = int(result.collection_id) + raw = self._recording_pending_raw.pop(collection_id, None) + preprocessed = self._recording_pending_pre.pop(collection_id, None) + if raw is not None: + self._recording_chunk_raw.append(raw) + if preprocessed is not None: + self._recording_chunk_pre.append(preprocessed) + self._recording_chunk_results.append(result) + + if len(self._recording_chunk_results) >= _RECORDING_CHUNK_SIZE: + if not self._flush_recording_chunk(): + return + if self._recording_collected_count() >= self._recording_target: + if not self._flush_recording_chunk(): + return + self._stop_collecting() + + def _flush_recording_chunk(self) -> bool: + """Hand the buffered chunk to the writer thread. Returns ``False`` if it aborted.""" + if not self._recording_chunk_results: + return True + if self._recording_write_error is not None or not self._recording_thread.is_alive(): + self._abort_recording_after_write_error() # never block on a dead consumer + return False + chunk = (self._recording_chunk_raw, self._recording_chunk_pre, self._recording_chunk_results) + self._recording_enqueued += len(self._recording_chunk_results) + self._recording_chunk_raw = [] + self._recording_chunk_pre = [] + self._recording_chunk_results = [] + self._recording_queue.put(chunk) # brief backpressure only if the disk lags + return True + + def _stop_collecting(self) -> None: + """Stop accepting measurements and tell the writer to drain and exit.""" + self._recording_collecting = False + with suppress(Exception): + self._recording_queue.put_nowait(None) # sentinel after the last chunk + + # -- completion / teardown (GUI thread) ----------------------------------- + + def _poll_recording_writer(self) -> None: + """Finalize a recording once the writer thread has drained (or failed). + + Called every GUI poll tick. Cheap no-op while idle or still writing. + """ + if not self._recording_active: + return + if self._recording_write_error is not None: + self._abort_recording_after_write_error() + return + if not self._recording_collecting and not self._recording_thread.is_alive(): + directory = self._recording_writer.directory if self._recording_writer is not None else "?" + written = self._recording_enqueued + self._reset_recording_state() + self._log(f"Disk recording complete: {written} measurement(s) written to {directory}") + + def _finalize_recording_on_stop(self) -> None: + """Flush the partial chunk and finish the recording when the run is stopped. + + Pressing Stop mid-recording writes everything collected so far — including a + not-yet-full chunk — to disk and ends the recording, so no buffered measurement + is lost. The dataset then holds exactly what was seen before Stop. Called from + ``_stop_run`` after its ring drains, so the last in-flight measurements are + already captured. The writer is joined briefly (bounded, like the stop drains) + so completion is deterministic without waiting on a later poll tick. + """ + if not self._recording_active or not self._recording_collecting: + return + if not self._flush_recording_chunk(): + return # writer already failed and was surfaced/reset + self._stop_collecting() + + thread = self._recording_thread + directory = self._recording_writer.directory if self._recording_writer is not None else "?" + if thread is not None: + thread.join(timeout=3.0) + write_error = self._recording_write_error + written = self._recording_enqueued + self._reset_recording_state() + if write_error is not None: + self._show_error( + "Disk recording stopped with a write error", + details=f"dataset={directory}\nerror={write_error}", + ) + else: + self._log(f"Disk recording stopped: {written} measurement(s) written to {directory}") + + def _abort_recording_after_write_error(self) -> None: + """Surface a writer-thread failure on the GUI thread and disarm.""" + message = self._recording_write_error or "unknown error" + directory = self._recording_writer.directory if self._recording_writer is not None else "?" + self._reset_recording_state() + self._show_error( + "Disk recording failed and was stopped", + details=f"dataset={directory}\nerror={message}", + ) + + def _reset_recording_state(self) -> None: + """Disarm recording, stop the writer thread, and release all buffers.""" + self._recording_stop.set() + if self._recording_queue is not None: + with suppress(Exception): + self._recording_queue.put_nowait(None) + self._recording_active = False + self._recording_collecting = False + self._recording_writer = None + self._recording_queue = None + self._recording_thread = None + self._recording_write_error = None + self._recording_enqueued = 0 + self._recording_pending_raw = {} + self._recording_pending_pre = {} + self._recording_chunk_raw = [] + self._recording_chunk_pre = [] + self._recording_chunk_results = [] + + def _shutdown_recording(self) -> None: + """Stop the writer thread on app teardown, briefly joining so it is not orphaned.""" + thread = self._recording_thread + self._recording_stop.set() + if self._recording_queue is not None: + with suppress(Exception): + self._recording_queue.put_nowait(None) + if thread is not None and thread.is_alive(): + thread.join(timeout=2.0) + self._reset_recording_state() + + # -- status --------------------------------------------------------------- + + def _recording_collected_count(self) -> int: + """Measurements accepted so far (handed to the writer + still buffered).""" + return self._recording_enqueued + len(self._recording_chunk_results) + + def _recording_status(self) -> dict: + """A compact snapshot of recording progress for the status feed / web UI.""" + return { + "active": self._recording_active, + "collected": self._recording_collected_count(), + "target": self._recording_target, + } diff --git a/python_app/gui/controllers/app_window_snapshot_mixin.py b/python_app/gui/controllers/app_window_snapshot_mixin.py index 271b02f..b13e84f 100644 --- a/python_app/gui/controllers/app_window_snapshot_mixin.py +++ b/python_app/gui/controllers/app_window_snapshot_mixin.py @@ -39,23 +39,55 @@ class AppWindowSnapshotMixin: return output_dir / "config_profile.json" def _save_snapshot(self) -> None: - """Save runtime snapshot in numpy-directory format.""" + """Save the last-N runtime measurements as a numpy snapshot (the "Save Dataset" button).""" self._drain_runtime_rings_for_snapshot() if not self._raw_history and not self._pre_history and not self._result_history: self._show_error("No runtime data is available for save", details=self._runtime_history_details()) return + self._write_snapshot_dataset( + list(self._raw_history), + list(self._pre_history), + list(self._result_history), + int(self._save_count.value()), + ) + + def _snapshot_destination_dir(self) -> Path: + """The directory a save/record would create now, from the current path/name fields. + + Used to surface a name clash *before* doing work (manual save and the disk + recorder both create this directory and fail if it already exists). + """ + output_root = Path(self._save_path_input.text().strip()).expanduser() + return self._store.snapshot_directory( + output_root, + self._save_name_input.text().strip(), + name_prefix=self._radar_config_name_prefix(), + ) + + def _write_snapshot_dataset( + self, + raw_history: list, + preprocessed_history: list, + result_history: list, + last_n: int, + ) -> None: + """Save the given aligned histories as a numpy snapshot + adjacent config profile. + + Shared by the manual "Save Dataset" button (which passes the runtime history) + and the on-the-fly disk recorder (which passes the freshly recorded buffers), + so both produce byte-identical dataset layouts and identical error reporting. + """ try: - last_n = int(self._save_count.value()) output_root = Path(self._save_path_input.text().strip()).expanduser() snapshot_name = self._save_name_input.text().strip() snapshot_dir, summary = self._store.save_runtime_snapshot_numpy( output_root, snapshot_name, - list(self._raw_history), - list(self._pre_history), - list(self._result_history), + raw_history, + preprocessed_history, + result_history, last_n, name_prefix=self._radar_config_name_prefix(), ) diff --git a/python_app/gui/controllers/app_window_web_mixin.py b/python_app/gui/controllers/app_window_web_mixin.py index 2dc8b58..5bbb6fc 100644 --- a/python_app/gui/controllers/app_window_web_mixin.py +++ b/python_app/gui/controllers/app_window_web_mixin.py @@ -18,15 +18,37 @@ from __future__ import annotations import base64 import contextlib import os +import threading import time from pathlib import Path from PyQt6.QtCore import QBuffer, QIODevice, QObject, pyqtSignal from python_app.gui.controllers.app_window_config.live_processing_mixin import web_apply_field_names +from python_app.webui.controller import WebActionError _WEBUI_PORT_ENV = "RADAR_SYSTEM_WEBUI_PORT" _DEFAULT_PORT = 8080 +# Upper bound on how long a browser control call waits for the GUI thread to run and +# report the action. Comfortably above a real save/start, but bounded so a wedged GUI +# thread surfaces as an error instead of hanging the HTTP worker forever. +_WEB_ACTION_TIMEOUT_S = 30.0 + + +class _WebActionCall: + """One synchronous web action: the GUI thread fills the result, the web thread waits. + + The web (uvicorn) thread emits a control signal carrying this object and blocks on + :attr:`done`; the GUI thread runs the desktop action, records any surfaced error in + :attr:`error`, and sets the event. This turns the fire-and-forget signal bridge into + a request/response so failures reach the browser. + """ + + __slots__ = ("done", "error") + + def __init__(self) -> None: + self.done = threading.Event() + self.error: str | None = None # Headless has no shown window, so give the offscreen window a usable size for the # grabbed plot. In GUI mode the user's real (shown) window size is used as-is. _HEADLESS_PLOT_SIZE = (1600, 900) @@ -66,14 +88,18 @@ class AppWindowWebController(QObject): place, so the web thread reads a consistent value without locking. """ - start_requested = pyqtSignal() - stop_requested = pyqtSignal() - remove_last_requested = pyqtSignal() - single_capture_requested = pyqtSignal() - capture_requested = pyqtSignal() + # Control signals carry a trailing _WebActionCall the GUI slot fills in, so the web + # thread can block on the real outcome. apply_settings stays fire-and-forget: it is + # validated up front and returns the live schema, not a pass/fail. + start_requested = pyqtSignal(object) + stop_requested = pyqtSignal(object) + remove_last_requested = pyqtSignal(object) + single_capture_requested = pyqtSignal(object) + capture_requested = pyqtSignal(object) + start_recording_requested = pyqtSignal(str, str, int, object) + load_config_requested = pyqtSignal(str, object) + save_dataset_requested = pyqtSignal(str, str, object) apply_settings_requested = pyqtSignal(dict) - load_config_requested = pyqtSignal(str) - save_dataset_requested = pyqtSignal(str, str) def __init__(self, run_configs_dir: Path, parent: QObject | None = None) -> None: super().__init__(parent) @@ -109,20 +135,38 @@ class AppWindowWebController(QObject): # -- WebController controls (web thread -> Qt main thread) --------------- + def _dispatch(self, signal, *args) -> None: + """Emit a control signal and block until the GUI thread reports the outcome. + + Runs on the web worker thread (the routes call this via a thread pool, so the + event loop is never blocked). Raises :class:`WebActionError` if the desktop + action surfaced an error or did not finish within the timeout. + """ + call = _WebActionCall() + signal.emit(*args, call) + if not call.done.wait(_WEB_ACTION_TIMEOUT_S): + raise WebActionError("The desktop did not complete the action in time") + if call.error is not None: + raise WebActionError(call.error) + def start(self) -> None: - self.start_requested.emit() + self._dispatch(self.start_requested) def stop(self) -> None: - self.stop_requested.emit() + self._dispatch(self.stop_requested) def single_capture(self) -> None: - self.single_capture_requested.emit() + self._dispatch(self.single_capture_requested) def capture_tmp_reference(self) -> None: - self.capture_requested.emit() + self._dispatch(self.capture_requested) def remove_last_measurement(self) -> None: - self.remove_last_requested.emit() + self._dispatch(self.remove_last_requested) + + def start_recording(self, path: str, name: str, count: int) -> None: + """Arm a run + disk recording of the next ``count`` measurements (the desktop button).""" + self._dispatch(self.start_recording_requested, path, name, int(count)) def apply_live_settings(self, fields: dict) -> dict: unknown = set(fields) - _LIVE_FIELD_NAMES @@ -135,16 +179,16 @@ class AppWindowWebController(QObject): """Request loading the run-config named ``name`` (the desktop "Load Config" action). Validates the name against the directory here — on the web thread — so an invalid - or unsafe name fails the HTTP request immediately instead of silently doing nothing - on the Qt side; the actual load runs through the queued signal. + or unsafe name fails the HTTP request immediately; the load itself then runs + synchronously on the Qt side and any load error is surfaced too. """ if _safe_run_config_path(self._run_configs_dir, name) is None: raise ValueError(f"Unknown run config: {name}") - self.load_config_requested.emit(name) + self._dispatch(self.load_config_requested, name) def save_dataset(self, path: str, name: str) -> None: """Save the runtime dataset to ``path``/``name`` (the desktop "Save Dataset" button).""" - self.save_dataset_requested.emit(path, name) + self._dispatch(self.save_dataset_requested, path, name) class AppWindowWebMixin: @@ -168,15 +212,40 @@ class AppWindowWebMixin: # The web picker browses this directory; ensure it exists on fresh deploys. self._run_configs_dir.mkdir(parents=True, exist_ok=True) + # Capture slot for errors a web-triggered action surfaces (None = no web + # action in flight). Read default-safe by `_show_error`/`_show_exception`. + self._web_action_error_capture: list[str] | None = None + controller = AppWindowWebController(self._run_configs_dir, parent=self) - controller.start_requested.connect(self._start_run) - controller.stop_requested.connect(self._stop_run) - controller.single_capture_requested.connect(self._start_single_capture) - controller.capture_requested.connect(self._capture_tmp_reference) - controller.remove_last_requested.connect(self._remove_last_runtime_history) + # Each control signal carries a _WebActionCall the wrapper finalizes, so the + # browser learns whether the desktop action actually succeeded. + controller.start_requested.connect( + lambda call: self._run_web_action(call, self._start_run) + ) + controller.stop_requested.connect( + lambda call: self._run_web_action(call, self._stop_run) + ) + controller.single_capture_requested.connect( + lambda call: self._run_web_action(call, self._start_single_capture) + ) + controller.capture_requested.connect( + lambda call: self._run_web_action(call, self._capture_tmp_reference) + ) + controller.remove_last_requested.connect( + lambda call: self._run_web_action(call, self._remove_last_runtime_history) + ) + controller.start_recording_requested.connect( + lambda path, name, count, call: self._run_web_action( + call, self._start_web_recording, path, name, count + ) + ) + controller.load_config_requested.connect( + lambda name, call: self._run_web_action(call, self._load_web_config, name) + ) + controller.save_dataset_requested.connect( + lambda path, name, call: self._run_web_action(call, self._save_web_dataset, path, name) + ) controller.apply_settings_requested.connect(self._apply_web_live_settings) - controller.load_config_requested.connect(self._load_web_config) - controller.save_dataset_requested.connect(self._save_web_dataset) self._web_controller = controller self._web_update_snapshot() # seed snapshots before the first request @@ -190,6 +259,27 @@ class AppWindowWebMixin: self._web_controller = None self._web_server = None + def _run_web_action(self, call: _WebActionCall, action, *args) -> None: + """Run a web-triggered desktop action on the GUI thread, capturing its outcome. + + Errors the action reports through ``_show_error``/``_show_exception`` are + captured into the call (and still logged/shown on the desktop) instead of + vanishing from the browser's view. Re-entrancy-safe: a nested action — e.g. a + modal error dialog pumping the event loop in GUI mode — saves and restores the + capture slot, so each action only sees its own first error. + """ + previous_capture = self._web_action_error_capture + capture: list[str] = [] + self._web_action_error_capture = capture + try: + action(*args) + call.error = capture[0] if capture else None + except Exception as exc: # noqa: BLE001 - handlers self-report; this is a backstop + call.error = self._exception_summary(exc) + finally: + self._web_action_error_capture = previous_capture + call.done.set() + def _load_web_config(self, name: str) -> None: """Load a run config chosen in the browser through the shared desktop load path. @@ -216,6 +306,20 @@ class AppWindowWebMixin: self._save_name_input.setText(name) self._save_snapshot() + def _start_web_recording(self, path: str, name: str, count: int) -> None: + """Arm disk recording from the browser via the same handler as the desktop button. + + The save path/name fields are mirrored exactly like ``_save_web_dataset`` (blank + path keeps the configured destination), the record count is applied to the shared + spinbox, then the unchanged desktop arming action runs. + """ + if path.strip(): + self._save_path_input.setText(path) + self._save_name_input.setText(name) + if count >= 1: + self._record_count.setValue(min(count, self._record_count.maximum())) + self._start_run_with_recording() + def _web_update_snapshot(self) -> None: """Refresh the snapshots the bridge serves (called on the Qt poll tick). @@ -237,6 +341,9 @@ class AppWindowWebMixin: # Current save path/name, so the web fields can prefill the desktop values. "save_path": self._save_path_input.text(), "save_name": self._save_name_input.text(), + # Default record count + live disk-recording progress for the web UI. + "record_count": int(self._record_count.value()), + "recording": self._recording_status(), # Per-stage capture counts, identical to the desktop history # label (raw -> preprocessed -> results), mirrored to the browser. "raw_count": len(self._raw_history), diff --git a/python_app/gui/controllers/sections/data_actions_section.py b/python_app/gui/controllers/sections/data_actions_section.py index d8da602..fdb49a7 100644 --- a/python_app/gui/controllers/sections/data_actions_section.py +++ b/python_app/gui/controllers/sections/data_actions_section.py @@ -37,10 +37,20 @@ def build_data_actions_group(owner) -> QGroupBox: capture_tmp_reference_button = QPushButton("Capture Tmp Reference") capture_tmp_reference_button.clicked.connect(owner._capture_tmp_reference) capture_tmp_reference_button.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) + record_button = QPushButton("Start + Record to Disk") + record_button.setToolTip( + "Start the run (if stopped) and save the next N measurements to the path/name below." + ) + record_button.clicked.connect(owner._start_run_with_recording) + record_button.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) owner._save_count = QSpinBox() owner._save_count.setMinimum(1) owner._save_count.setMaximum(10_000) owner._save_count.setValue(int(data_defaults.save_count)) + owner._record_count = QSpinBox() + owner._record_count.setMinimum(1) + owner._record_count.setMaximum(1_000_000) + owner._record_count.setValue(int(data_defaults.record_count)) button_grid = QGridLayout() button_grid.setHorizontalSpacing(8) @@ -49,6 +59,7 @@ def build_data_actions_group(owner) -> QGroupBox: button_grid.addWidget(save_vna_json_button, 0, 1) button_grid.addWidget(remove_last_button, 1, 0) button_grid.addWidget(capture_tmp_reference_button, 1, 1) + button_grid.addWidget(record_button, 2, 0, 1, 2) button_grid.setColumnStretch(0, 1) button_grid.setColumnStretch(1, 1) layout.addLayout(button_grid) @@ -60,6 +71,13 @@ def build_data_actions_group(owner) -> QGroupBox: count_row.addStretch(1) layout.addLayout(count_row) + record_count_row = QHBoxLayout() + record_count_row.setSpacing(8) + record_count_row.addWidget(QLabel("Number of Sweeps to Record")) + record_count_row.addWidget(owner._record_count) + record_count_row.addStretch(1) + layout.addLayout(record_count_row) + path_row = QHBoxLayout() path_row.setSpacing(8) owner._save_path_input = QLineEdit(str(data_defaults.save_path)) diff --git a/python_app/hardware_full/kamil_adc/processing.py b/python_app/hardware_full/kamil_adc/processing.py index 2180fe4..b0e5492 100644 --- a/python_app/hardware_full/kamil_adc/processing.py +++ b/python_app/hardware_full/kamil_adc/processing.py @@ -13,7 +13,17 @@ comparable S21 trace is a fixed three-stage pipeline: f(phase) = freq0 + (phase - phase0) * (freq1 - freq0) / (phase1 - phase0) Trigger jitter shifts every sample's absolute phase together, so the measured - band floats from sweep to sweep around the fixed calibration. + band floats from sweep to sweep around the fixed calibration. ``np.unwrap`` + anchors each sweep's ramp to ``np.angle(reference[0])`` on the (-pi, pi] branch, + so once that float carries the anchor across the +/-pi cut the whole ramp jumps + one 2*pi turn (~117 MHz on the rig) even though nothing physical moved. The + genuine float is slow (well under pi between consecutive sweeps) while the wrap + is a discrete 2*pi step, so the processor *unwraps the anchor across sweeps*: + each sweep's anchor is snapped onto the branch nearest the previous accepted + sweep (the first sweep onto the calibration ``phase0`` branch). Validated on + 10 000 live sweeps: 506 branch wraps, yet the largest cross-sweep anchor step + stayed at 1.6 rad (< pi), and the correction cut badly-distorted pass-through + sweeps from 580 to 11 while leaving clean sweeps untouched. 2. **Amplitude normalization.** ``S = main / |reference|`` divides out the stimulus amplitude. Only the magnitude is removed; the reference phase is used @@ -47,6 +57,12 @@ _REFERENCE_AMPLITUDE_FLOOR = 1e-9 # a sweep yielding fewer usable points is malformed and rejected. _MIN_USABLE_POINTS = 2 +# One full turn of reference phase. ``np.unwrap`` anchors each sweep's phase ramp +# to the raw angle of the first sample on the (-pi, pi] branch, so a stray sweep +# whose anchor crossed the +/-pi cut is offset by exactly this; the cross-sweep +# anchor tracking snaps it back (see ``KamilAdcSweepProcessor._align_phase_branch``). +_PHASE_BRANCH_PERIOD_RAD = 2.0 * np.pi + @dataclass(frozen=True, slots=True) class KamilAdcProcessingParams: @@ -110,13 +126,18 @@ class KamilAdcSweepProcessor: sweep, so all traces this processor emits share one identical frequency axis. """ - __slots__ = ("_params", "_grid_hz") + __slots__ = ("_params", "_grid_hz", "_previous_anchor_rad") def __init__(self, params: KamilAdcProcessingParams) -> None: self._params = params self._grid_hz = np.linspace( params.band_start_hz, params.band_stop_hz, params.band_points, dtype=np.float64 ) + # Absolute (branch-tracked) reference phase of the last ACCEPTED sweep's + # first sample, carried across sweeps so a +/-pi anchor wrap can be undone. + # ``None`` until the first sweep is accepted; reset by building a new + # processor (i.e. on reconfigure). + self._previous_anchor_rad: float | None = None @property def params(self) -> KamilAdcProcessingParams: @@ -130,12 +151,44 @@ class KamilAdcSweepProcessor: def reference_frequency_axis(self, reference: np.ndarray) -> np.ndarray: """Map a reference signal's absolute unwrapped phase to frequency (Hz). - Returns frequencies in *step order* (not sorted); see the module docstring - for the calibration law. + Stateless and *uncorrected* (no cross-sweep branch tracking), so it shows + the raw per-sweep axis — used by diagnostics. The live path + (:meth:`process`) applies the branch correction. Returns frequencies in + *step order* (not sorted); see the module docstring for the calibration law. """ - phase = np.unwrap(np.angle(np.asarray(reference))) + return self._phase_to_frequency(np.unwrap(np.angle(np.asarray(reference)))) + + def _phase_to_frequency(self, phase: np.ndarray) -> np.ndarray: + """Apply the affine ``phase -> frequency`` calibration law.""" return self._params.freq0_hz + (phase - self._params.phase0_rad) * self._params.hz_per_rad + def _align_phase_branch(self, phase: np.ndarray) -> tuple[np.ndarray, float]: + """Undo a stray +/-pi anchor wrap by unwrapping the anchor across sweeps. + + ``np.unwrap`` pins the whole ramp to ``phase[0]`` on the (-pi, pi] branch, + so a slow physical float that drags the anchor over the +/-pi cut flips the + entire sweep by one :data:`_PHASE_BRANCH_PERIOD_RAD`. We snap this sweep's + anchor onto the branch nearest the previous accepted sweep's anchor (the + first sweep onto the calibration ``phase0``), then shift the whole ramp by + the same whole number of turns. + + Returns ``(branch_aligned_phase, anchor_to_commit)``. The caller commits the + anchor only once the sweep is accepted, so a rejected/corrupt sweep can + never latch the tracker onto a wrong branch. Genuine sub-pi sweep-to-sweep + float rounds to zero turns and is preserved untouched. + """ + anchor = float(phase[0]) + reference_anchor = ( + self._previous_anchor_rad + if self._previous_anchor_rad is not None + else self._params.phase0_rad + ) + turns = round((reference_anchor - anchor) / _PHASE_BRANCH_PERIOD_RAD) + if turns: + shift = turns * _PHASE_BRANCH_PERIOD_RAD + return phase + shift, anchor + shift + return phase, anchor + def process(self, main: np.ndarray, reference: np.ndarray) -> np.ndarray | None: """Return the S21 trace resampled onto the fixed grid, or ``None`` to reject. @@ -148,8 +201,11 @@ class KamilAdcSweepProcessor: if main.size < _MIN_USABLE_POINTS or main.size != reference.size: return None - # Frequency axis from the absolute unwrapped reference phase (step order). - freqs = self.reference_frequency_axis(reference) + # Frequency axis from the absolute unwrapped reference phase (step order), + # with a stray +/-pi anchor wrap undone relative to the last accepted sweep. + # The aligned anchor is committed only if this sweep is accepted (below). + phase, candidate_anchor = self._align_phase_branch(np.unwrap(np.angle(reference))) + freqs = self._phase_to_frequency(phase) # Amplitude-only normalization; drop points where the reference vanished. reference_amplitude = np.abs(reference) @@ -178,6 +234,11 @@ class KamilAdcSweepProcessor: if freqs[0] > self._params.band_start_hz or freqs[-1] < self._params.band_stop_hz: return None + # The sweep is accepted: commit its branch-aligned anchor so the next + # sweep is tracked relative to it (and a wrap is measured against a real, + # in-band reference rather than a rejected one). + self._previous_anchor_rad = candidate_anchor + # Linear interpolation on real/imaginary parts. Because the band lies # within [freqs[0], freqs[-1]], np.interp never extrapolates here. real = np.interp(self._grid_hz, freqs, s21.real) diff --git a/python_app/models/gui_profile_codec.py b/python_app/models/gui_profile_codec.py index afe0c53..a25368f 100644 --- a/python_app/models/gui_profile_codec.py +++ b/python_app/models/gui_profile_codec.py @@ -504,6 +504,12 @@ def gui_profile_from_dict(payload: dict[str, Any]) -> GuiProfileModel: gui.data_actions.save_name, "gui.data_actions", ), + record_count=_optional_int( + data_actions_object, + "record_count", + gui.data_actions.record_count, + "gui.data_actions", + ), ) preprocess_dialog_object = _as_dict(gui_object.get("preprocess_dialog"), "gui.preprocess_dialog") @@ -632,6 +638,7 @@ def gui_profile_to_dict(model: GuiProfileModel) -> dict[str, Any]: "save_count": gui.data_actions.save_count, "save_path": gui.data_actions.save_path, "save_name": gui.data_actions.save_name, + "record_count": gui.data_actions.record_count, }, "preprocess_dialog": { "set_name": gui.preprocess_dialog.set_name, diff --git a/python_app/models/gui_profile_schema.py b/python_app/models/gui_profile_schema.py index cf328dc..185ec09 100644 --- a/python_app/models/gui_profile_schema.py +++ b/python_app/models/gui_profile_schema.py @@ -132,6 +132,9 @@ class GuiDataActionsStateModel: save_count: int = 10 save_path: str = "" save_name: str = "snapshot_manual" + # How many freshly acquired measurements the "Start + Record" action writes to + # disk before it stops recording (acquisition keeps running). + record_count: int = 100 @dataclass(slots=True) diff --git a/python_app/storage/npz/snapshot_numpy.py b/python_app/storage/npz/snapshot_numpy.py index 41f7aef..29df9c3 100644 --- a/python_app/storage/npz/snapshot_numpy.py +++ b/python_app/storage/npz/snapshot_numpy.py @@ -153,11 +153,18 @@ def save_result_history_binary(stage_dir: Path, history: list[ResultCollection]) ) -def save_trace_history_numpy(stage_dir: Path, history: list[SweepCollection]) -> None: - """Write raw/preprocessed collections as NumPy directory tree.""" +def save_trace_history_numpy( + stage_dir: Path, history: list[SweepCollection], *, index_offset: int = 0 +) -> None: + """Write raw/preprocessed collections as a NumPy directory tree. + + ``index_offset`` continues the per-collection directory numbering across calls so a + streaming recorder can append successive chunks into the same stage directory + without colliding or restarting the index. + """ stage_dir.mkdir(parents=True, exist_ok=True) logger.debug("Writing %d NumPy trace collection(s) to %s", len(history), stage_dir) - for index, collection in enumerate(history): + for index, collection in enumerate(history, start=index_offset): collection_dir = stage_dir / collection_dir_name(index, collection.collection_id, collection.monotonic_ns) collection_dir.mkdir(parents=True, exist_ok=False) @@ -197,11 +204,17 @@ def save_trace_history_numpy(stage_dir: Path, history: list[SweepCollection]) -> ) -def save_result_history_numpy(stage_dir: Path, history: list[ResultCollection]) -> None: - """Write processed result collections as NumPy directory tree.""" +def save_result_history_numpy( + stage_dir: Path, history: list[ResultCollection], *, index_offset: int = 0 +) -> None: + """Write processed result collections as a NumPy directory tree. + + ``index_offset`` continues the per-collection directory numbering across calls (see + :func:`save_trace_history_numpy`) so streamed chunks append cleanly. + """ stage_dir.mkdir(parents=True, exist_ok=True) logger.debug("Writing %d NumPy result collection(s) to %s", len(history), stage_dir) - for index, collection in enumerate(history): + for index, collection in enumerate(history, start=index_offset): collection_dir = stage_dir / collection_dir_name(index, collection.collection_id, collection.monotonic_ns) collection_dir.mkdir(parents=True, exist_ok=False) diff --git a/python_app/storage/npz/store.py b/python_app/storage/npz/store.py index 344560d..f0bf7ce 100644 --- a/python_app/storage/npz/store.py +++ b/python_app/storage/npz/store.py @@ -36,6 +36,50 @@ def _compose_snapshot_stem(name: str, name_prefix: str) -> str: return sanitize_path_component(f"{name_prefix}_{base}" if name_prefix else base) +class SnapshotStreamWriter: + """Append aligned raw/preprocessed/result collections to a snapshot dir in chunks. + + A streaming alternative to :meth:`NpzStore.save_runtime_snapshot_numpy` for long + recordings: the caller flushes small chunks as measurements arrive and frees them, + so memory stays flat instead of buffering every measurement. The on-disk layout is + identical (``raw``/``preprocessed``/``results`` subtrees), and per-collection + directory indices continue across chunks via a running offset per stage. + """ + + __slots__ = ("_dir", "_raw_written", "_preprocessed_written", "_result_written") + + def __init__(self, snapshot_dir: Path) -> None: + self._dir = snapshot_dir + self._raw_written = 0 + self._preprocessed_written = 0 + self._result_written = 0 + + @property + def directory(self) -> Path: + return self._dir + + @property + def result_count(self) -> int: + """Number of result collections written to disk so far.""" + return self._result_written + + def append( + self, + raw: list[SweepCollection], + preprocessed: list[SweepCollection], + results: list[ResultCollection], + ) -> None: + """Write one chunk of (already aligned) collections, continuing each stage's index.""" + save_trace_history_numpy(self._dir / "raw", raw, index_offset=self._raw_written) + save_trace_history_numpy( + self._dir / "preprocessed", preprocessed, index_offset=self._preprocessed_written + ) + save_result_history_numpy(self._dir / "results", results, index_offset=self._result_written) + self._raw_written += len(raw) + self._preprocessed_written += len(preprocessed) + self._result_written += len(results) + + class NpzStore(StoreApi): """Persist preprocess sets and runtime snapshots using NumPy files.""" @@ -194,6 +238,35 @@ class NpzStore(StoreApi): logger.info("Saved binary runtime snapshot (last_n=%d) to %s", last_n, snapshot_dir) return snapshot_dir + def snapshot_directory( + self, output_root_dir: Path, snapshot_name: str, *, name_prefix: str = "" + ) -> Path: + """Return the directory a snapshot save would create for these inputs. + + Lets callers check ``.exists()`` *before* acquiring data (e.g. the disk + recorder arms a run only once the destination is free), so a name clash is + reported up front instead of after the measurements are collected. Note a + blank ``snapshot_name`` resolves to a fresh timestamp each call, so this is + meaningful only for explicit names — exactly the case that can collide. + """ + return output_root_dir / _compose_snapshot_stem(snapshot_name, name_prefix) + + def create_snapshot_stream( + self, output_root_dir: Path, snapshot_name: str, *, name_prefix: str = "" + ) -> SnapshotStreamWriter: + """Create an empty snapshot directory and return a chunked stream writer for it. + + Raises ``FileExistsError`` if the directory already exists (same guard as + :meth:`save_runtime_snapshot_numpy`), so a name clash is reported at arm time. + """ + snapshot_dir = self.snapshot_directory(output_root_dir, snapshot_name, name_prefix=name_prefix) + output_root_dir.mkdir(parents=True, exist_ok=True) + if snapshot_dir.exists(): + raise FileExistsError(f"Snapshot directory already exists: {snapshot_dir}") + snapshot_dir.mkdir(parents=True, exist_ok=False) + logger.info("Opened streaming snapshot directory %s", snapshot_dir) + return SnapshotStreamWriter(snapshot_dir) + def save_runtime_snapshot_numpy( self, output_root_dir: Path, diff --git a/python_app/tests/test_kamil_adc_processing.py b/python_app/tests/test_kamil_adc_processing.py index 288885a..6059cc1 100644 --- a/python_app/tests/test_kamil_adc_processing.py +++ b/python_app/tests/test_kamil_adc_processing.py @@ -167,6 +167,55 @@ class SweepProcessorTest(unittest.TestCase): self.assertTrue(np.all(np.isfinite(result.real))) self.assertTrue(np.all(np.isfinite(result.imag))) + # -- cross-sweep branch tracking ------------------------------------------ + + def test_align_phase_branch_anchors_first_sweep_to_calibration(self) -> None: + # No previous anchor yet -> snap onto the branch nearest phase0 (=0 here). + processor = self._processor() + phase = np.array([0.05, 1.0, 2.0]) + 2.0 * np.pi # one turn above phase0 + aligned, anchor = processor._align_phase_branch(phase) + np.testing.assert_allclose(aligned, np.array([0.05, 1.0, 2.0]), atol=1e-9) + self.assertAlmostEqual(anchor, 0.05, places=6) + + def test_align_phase_branch_snaps_to_previous_anchor(self) -> None: + # A genuine sub-pi float is preserved; a full-turn anchor wrap is undone. + processor = self._processor() + processor._previous_anchor_rad = 0.05 + kept, kept_anchor = processor._align_phase_branch(np.array([0.40, 1.4, 2.4])) + np.testing.assert_allclose(kept, np.array([0.40, 1.4, 2.4]), atol=1e-9) # None: + # The anchor is committed only on accepted sweeps, so a rejected sweep + # cannot latch the tracker onto a wrong branch. + processor = self._processor() + covering = _reference(np.linspace(0.0, 100.0, 401)) + self.assertIsNotNone(processor.process(np.abs(covering).astype(np.complex128), covering)) + anchor_after_accept = processor._previous_anchor_rad + self.assertIsNotNone(anchor_after_accept) + short = _reference(np.linspace(0.0, 40.0, 201)) # does not span the band + self.assertIsNone(processor.process(np.ones(201, dtype=np.complex128), short)) + self.assertEqual(processor._previous_anchor_rad, anchor_after_accept) + + def test_cross_sweep_unwrap_recovers_continuous_anchor_across_a_wrap(self) -> None: + # Two physically adjacent sweeps whose anchor straddles +pi: np.angle wraps + # the second's anchor by ~2*pi, but the cross-sweep tracking must recover + # the continuous value (~3.3), not the wrapped one (~-2.98). + processor = self._processor() + ramp_home = np.linspace(3.0, 80.0, 401) # anchor 3.0 (< pi), covers band + ramp_drift = np.linspace(3.3, 80.3, 401) # anchor 3.3 (> pi) -> angle wraps + ref_home = _reference(ramp_home) + ref_drift = _reference(ramp_drift) + self.assertIsNotNone(processor.process(np.abs(ref_home).astype(np.complex128), ref_home)) + self.assertAlmostEqual(processor._previous_anchor_rad, 3.0, places=2) + self.assertIsNotNone(processor.process(np.abs(ref_drift).astype(np.complex128), ref_drift)) + # Without correction this would be ~-2.98 (one turn below); corrected it + # continues smoothly from 3.0 to ~3.3. + self.assertAlmostEqual(processor._previous_anchor_rad, 3.3, places=2) + if __name__ == "__main__": unittest.main() diff --git a/python_app/tests/test_recording_mixin.py b/python_app/tests/test_recording_mixin.py new file mode 100644 index 0000000..72dd4fd --- /dev/null +++ b/python_app/tests/test_recording_mixin.py @@ -0,0 +1,265 @@ +"""Tests for the streaming disk-recording mixin. + +The recorder pairs each result with its raw/preprocessed collection by id, buffers a +small chunk, and flushes it to a stream writer — so memory stays flat regardless of how +many measurements are recorded. These tests pin that behaviour (chunking, finish-at-N, +arm guards, only-new gating) against light stubs of the AppWindow collaborators — no Qt, +hardware, or disk needed. +""" + +from __future__ import annotations + +from pathlib import Path +import unittest +from unittest import mock + +from python_app.gui.controllers import app_window_recording_mixin as rec +from python_app.gui.controllers.app_window_recording_mixin import AppWindowRecordingMixin + + +class _Collection: + def __init__(self, collection_id: int, monotonic_ns: int = 0) -> None: + self.collection_id = collection_id + self.monotonic_ns = monotonic_ns + + +class _Spin: + def __init__(self, value: int) -> None: + self._value = value + + def value(self) -> int: + return self._value + + +class _Text: + def __init__(self, value: str) -> None: + self._value = value + + def text(self) -> str: + return self._value + + +class _Supervisor: + def __init__(self, running: bool) -> None: + self._running = running + + def is_running(self) -> bool: + return self._running + + +class _FakePath: + def __init__(self, exists: bool) -> None: + self._exists = exists + + def exists(self) -> bool: + return self._exists + + def __str__(self) -> str: + return "/tmp/dataset" + + +class _FakeWriter: + def __init__(self) -> None: + self.directory = Path("/tmp/dataset") + self.appends: list[tuple[int, int, int]] = [] + self.result_count = 0 + + def append(self, raw, preprocessed, results) -> None: + self.appends.append((len(raw), len(preprocessed), len(results))) + self.result_count += len(results) + + +class _FakeStore: + def __init__(self, *, exists: bool = False) -> None: + self._exists = exists + self.created: list[tuple] = [] + self.writer: _FakeWriter | None = None + + def create_snapshot_stream(self, root, name, *, name_prefix=""): + if self._exists: + raise FileExistsError(f"Snapshot directory already exists: {root}/{name}") + self.created.append((root, name, name_prefix)) + self.writer = _FakeWriter() + return self.writer + + +class _Harness(AppWindowRecordingMixin): + def __init__(self, *, count, running, dest_exists=False, store_exists=False) -> None: + self._init_recording_state() + self._record_count = _Spin(count) + self._supervisor = _Supervisor(running) + self._store = _FakeStore(exists=store_exists) + self._save_path_input = _Text("/tmp") + self._save_name_input = _Text("run") + self._dest_exists = dest_exists + self.started = False + self.errors: list[str] = [] + self.exceptions: list[tuple] = [] + self.logs: list[str] = [] + self.profiles: list[tuple] = [] + + def _snapshot_destination_dir(self): + return _FakePath(self._dest_exists) + + def _radar_config_name_prefix(self) -> str: + return "pref" + + def _snapshot_config_profile_path(self, directory): + return Path(directory) / "config_profile.json" + + def _write_gui_profile_to_path(self, path, allow_overwrite) -> None: + self.profiles.append((path, allow_overwrite)) + + def _start_run(self) -> None: + self.started = True + + def _show_error(self, message, *, details=None) -> None: + self.errors.append(message) + + def _show_exception(self, context, exc) -> None: + self.exceptions.append((context, exc)) + + def _log(self, message) -> None: + self.logs.append(message) + + # convenience for tests: stamp collections just after the arm instant so the + # "only record sweeps produced after arming" gate accepts them. + def feed(self, collection_id: int, *, ns: int | None = None) -> None: + if ns is None: + ns = self._recording_since_ns + 1 + self._record_collection("raw", _Collection(collection_id, ns)) + self._record_collection("preprocessed", _Collection(collection_id, ns)) + self._record_collection("results", _Collection(collection_id, ns)) + + +class _RecordingTestBase(unittest.TestCase): + def _make(self, **kwargs) -> _Harness: + harness = _Harness(**kwargs) + self.addCleanup(harness._shutdown_recording) # never leak the writer thread + return harness + + def _drain_and_finalize(self, harness: _Harness) -> None: + """Wait for the writer thread to flush all chunks, then finalize on the GUI side.""" + thread = harness._recording_thread + self.assertIsNotNone(thread) + thread.join(timeout=2.0) + self.assertFalse(thread.is_alive(), "writer thread did not drain") + harness._poll_recording_writer() + + +class RecordingArmTest(_RecordingTestBase): + def test_arm_starts_a_stopped_run_and_opens_a_stream(self) -> None: + h = self._make(count=3, running=False) + h._start_run_with_recording() + self.assertTrue(h.started) + self.assertTrue(h._recording_active) + self.assertEqual(h._recording_target, 3) + self.assertEqual(len(h._store.created), 1) # one stream opened + self.assertEqual(h.profiles[0][1], False) # config profile written, no overwrite + + def test_arm_does_not_restart_a_running_pipeline(self) -> None: + h = self._make(count=3, running=True) + h._start_run_with_recording() + self.assertFalse(h.started) + self.assertTrue(h._recording_active) + + def test_arm_refuses_when_destination_exists(self) -> None: + h = self._make(count=3, running=True, dest_exists=True) + h._start_run_with_recording() + self.assertFalse(h._recording_active) + self.assertEqual(len(h._store.created), 0) + self.assertEqual(len(h.errors), 1) + + def test_second_arm_while_recording_is_refused(self) -> None: + h = self._make(count=5, running=True) + h._start_run_with_recording() + h._start_run_with_recording() # already in progress + self.assertEqual(len(h._store.created), 1) # no second stream + self.assertEqual(len(h.errors), 1) + self.assertTrue(h._recording_active) + + +class RecordingStreamTest(_RecordingTestBase): + def test_ignores_collections_from_before_arming(self) -> None: + h = self._make(count=2, running=True) + h._start_run_with_recording() + arm = h._recording_since_ns + h._record_collection("results", _Collection(1, monotonic_ns=arm - 1)) + self.assertEqual(h._recording_status()["collected"], 0) + self.assertIsNotNone(h._store.writer) + self.assertEqual(h._store.writer.result_count, 0) + + def test_streams_one_chunk_and_finishes_at_target(self) -> None: + h = self._make(count=3, running=True) + h._start_run_with_recording() + writer = h._store.writer + for cid in (1, 2, 3): + h.feed(cid) + self._drain_and_finalize(h) + self.assertEqual(writer.appends, [(3, 3, 3)]) # one flush at target + self.assertEqual(writer.result_count, 3) + self.assertFalse(h._recording_active) # disarmed after the writer drained + + def test_flushes_in_chunks_so_memory_stays_flat(self) -> None: + with mock.patch.object(rec, "_RECORDING_CHUNK_SIZE", 2): + h = self._make(count=5, running=True) + h._start_run_with_recording() + writer = h._store.writer + for cid in range(1, 6): + h.feed(cid) + self._drain_and_finalize(h) + # 2 + 2 + 1: flushed at the chunk boundaries and the final remainder. + self.assertEqual(writer.appends, [(2, 2, 2), (2, 2, 2), (1, 1, 1)]) + self.assertEqual(writer.result_count, 5) + self.assertFalse(h._recording_active) + + def test_does_not_record_beyond_target(self) -> None: + h = self._make(count=2, running=True) + h._start_run_with_recording() + writer = h._store.writer + for cid in (1, 2, 3, 4): # 4 results, target 2 + h.feed(cid) + self._drain_and_finalize(h) + self.assertEqual(writer.result_count, 2) + + def test_records_results_with_missing_raw_or_pre(self) -> None: + h = self._make(count=1, running=True) + h._start_run_with_recording() + writer = h._store.writer + # result with no matching raw/pre buffered (e.g. a dropped raw frame) + h._record_collection("results", _Collection(9, monotonic_ns=h._recording_since_ns + 1)) + self._drain_and_finalize(h) + self.assertEqual(writer.appends, [(0, 0, 1)]) + self.assertEqual(writer.result_count, 1) + + def test_stop_flushes_the_partial_chunk(self) -> None: + # Stop pressed before a chunk fills (and before the target): the buffered + # measurements must be written, not lost. + h = self._make(count=1000, running=True) # target large -> never reached here + h._start_run_with_recording() + writer = h._store.writer + for cid in (1, 2, 3): # 3 < chunk size and < target -> buffered, not yet flushed + h.feed(cid) + self.assertEqual(writer.appends, []) # nothing flushed yet + h._finalize_recording_on_stop() # Stop pressed + self.assertEqual(writer.appends, [(3, 3, 3)]) # partial chunk written + self.assertEqual(writer.result_count, 3) + self.assertFalse(h._recording_active) # recording ended + + def test_stop_when_idle_is_a_noop(self) -> None: + h = self._make(count=5, running=True) + h._finalize_recording_on_stop() # nothing armed + self.assertEqual(len(h._store.created), 0) + self.assertFalse(h._recording_active) + + def test_pending_maps_are_bounded(self) -> None: + with mock.patch.object(rec, "_RECORDING_MAX_PENDING", 4): + h = self._make(count=1000, running=True) + h._start_run_with_recording() + for cid in range(50): # raws whose results never arrive + h._record_collection("raw", _Collection(cid, monotonic_ns=h._recording_since_ns + 1)) + self.assertLessEqual(len(h._recording_pending_raw), 4) + + +if __name__ == "__main__": + unittest.main() diff --git a/python_app/tests/test_storage_webui.py b/python_app/tests/test_storage_webui.py index 8561602..bc75c29 100644 --- a/python_app/tests/test_storage_webui.py +++ b/python_app/tests/test_storage_webui.py @@ -34,6 +34,7 @@ from python_app.storage.npz.vna_history_json import ( # noqa: E402 _normalize_channel, build_vna_history_payload, ) +from python_app.webui.controller import WebActionError # noqa: E402 from python_app.webui.streaming import RingBroadcaster # noqa: E402 @@ -178,21 +179,48 @@ class WebControllerTest(unittest.TestCase): self.assertEqual(self.controller.peek_frame(), {"seq": 1}) # keeps the last frame def test_controls_emit_signals(self) -> None: + # Controls are synchronous now: each emits a call the GUI slot must finalize. + # In-thread, emit() runs the slot directly, so finalizing it returns at once. fired: list[str] = [] - self.controller.start_requested.connect(lambda: fired.append("start")) - self.controller.stop_requested.connect(lambda: fired.append("stop")) - self.controller.single_capture_requested.connect(lambda: fired.append("single")) - self.controller.capture_requested.connect(lambda: fired.append("capture")) + + def handler(label): + def slot(call): + fired.append(label) + call.done.set() + return slot + + self.controller.start_requested.connect(handler("start")) + self.controller.stop_requested.connect(handler("stop")) + self.controller.single_capture_requested.connect(handler("single")) + self.controller.capture_requested.connect(handler("capture")) self.controller.start() self.controller.stop() self.controller.single_capture() self.controller.capture_tmp_reference() self.assertEqual(fired, ["start", "stop", "single", "capture"]) + def test_action_error_is_raised_to_the_caller(self) -> None: + # An error the GUI slot records on the call surfaces as WebActionError (HTTP 400). + self.controller.start_requested.connect( + lambda call: (setattr(call, "error", "destination already exists"), call.done.set()) + ) + with self.assertRaisesRegex(WebActionError, "destination already exists"): + self.controller.start() + + def test_start_recording_forwards_path_name_count(self) -> None: + received: list[tuple[str, str, int]] = [] + self.controller.start_recording_requested.connect( + lambda p, n, c, call: (received.append((p, n, c)), call.done.set()) + ) + self.controller.start_recording("/tmp/out", "run1", 250) + self.assertEqual(received, [("/tmp/out", "run1", 250)]) + def test_lists_configs_sorted_and_load_emits_signal(self) -> None: self.assertEqual(self.controller.list_configs(), ["alpha.json", "beta.json"]) requested: list[str] = [] - self.controller.load_config_requested.connect(requested.append) + self.controller.load_config_requested.connect( + lambda name, call: (requested.append(name), call.done.set()) + ) self.controller.load_config("beta.json") self.assertEqual(requested, ["beta.json"]) @@ -204,7 +232,9 @@ class WebControllerTest(unittest.TestCase): def test_save_dataset_forwards_path_and_name(self) -> None: received: list[tuple[str, str]] = [] - self.controller.save_dataset_requested.connect(lambda p, n: received.append((p, n))) + self.controller.save_dataset_requested.connect( + lambda p, n, call: (received.append((p, n)), call.done.set()) + ) self.controller.save_dataset("/tmp/out", "run1") self.assertEqual(received, [("/tmp/out", "run1")]) diff --git a/python_app/webui/controller.py b/python_app/webui/controller.py index 34d18c5..e85b4c9 100644 --- a/python_app/webui/controller.py +++ b/python_app/webui/controller.py @@ -13,9 +13,26 @@ from __future__ import annotations from typing import Protocol, runtime_checkable +class WebActionError(Exception): + """A web-triggered desktop action failed, carrying the operator-facing reason. + + Control methods run the matching desktop action *synchronously* and raise this + when that action reports an error (the same message the desktop would show), so + the HTTP layer can return it instead of a misleading "ok". Distinct from a plain + ``ValueError`` (rejected by web-side validation before the action even runs). + """ + + @runtime_checkable class WebController(Protocol): - """Control + read surface the web layer needs; implemented by the Qt bridge.""" + """Control + read surface the web layer needs; implemented by the Qt bridge. + + Control methods are *synchronous*: each runs the corresponding desktop action on + the GUI thread and only returns once it has completed, raising + :class:`WebActionError` if the action surfaced an error. This is what lets the + browser show real failures (e.g. a save into an existing directory) instead of a + blind success. + """ def start(self) -> None: """Start a continuous run (the desktop "Start" button).""" @@ -26,6 +43,14 @@ class WebController(Protocol): def stop(self) -> None: """Stop the running pipeline (the desktop "Stop" button).""" + def start_recording(self, path: str, name: str, count: int) -> None: + """Start a run (if stopped) and record the next ``count`` measurements to disk. + + ``path``/``name`` mirror the shared save destination fields (blank ``path`` + keeps the configured one); ``count`` sets how many measurements are written. + Raises :class:`WebActionError` if the destination already exists. + """ + def capture_tmp_reference(self) -> None: """Capture and select a temporary reference (the desktop button).""" diff --git a/python_app/webui/routes.py b/python_app/webui/routes.py index 629d020..a3dc1d3 100644 --- a/python_app/webui/routes.py +++ b/python_app/webui/routes.py @@ -14,7 +14,7 @@ import logging from fastapi import APIRouter, Body, HTTPException, Request, WebSocket, WebSocketDisconnect -from python_app.webui.controller import WebController +from python_app.webui.controller import WebActionError, WebController from python_app.webui.streaming import RingBroadcaster logger = logging.getLogger(__name__) @@ -26,6 +26,21 @@ def _controller(request: Request) -> WebController: return request.app.state.controller +async def _run_action(func, *args) -> None: + """Run a (blocking) controller control call off the event loop, mapping failures to 400. + + The control methods run the desktop action synchronously and raise ``ValueError`` + (rejected input) or ``WebActionError`` (the action itself failed) — both become a + 400 the browser shows, instead of the old silent "ok". Running in a worker thread + keeps the async event loop free while the GUI thread does the work. + """ + try: + await asyncio.to_thread(func, *args) + except (ValueError, WebActionError) as exc: + logger.warning("Web UI action failed: %s", exc) + raise HTTPException(status_code=400, detail=str(exc)) from exc + + @router.get("/api/status") async def get_status(request: Request) -> dict: return _controller(request).status() @@ -35,7 +50,7 @@ async def get_status(request: Request) -> dict: async def post_start(request: Request) -> dict: logger.info("Web UI request: start") controller = _controller(request) - controller.start() + await _run_action(controller.start) return controller.status() @@ -43,7 +58,7 @@ async def post_start(request: Request) -> dict: async def post_single_capture(request: Request) -> dict: logger.info("Web UI request: single capture") controller = _controller(request) - controller.single_capture() + await _run_action(controller.single_capture) return controller.status() @@ -51,7 +66,20 @@ async def post_single_capture(request: Request) -> dict: async def post_stop(request: Request) -> dict: logger.info("Web UI request: stop") controller = _controller(request) - controller.stop() + await _run_action(controller.stop) + return controller.status() + + +@router.post("/api/start_recording") +async def post_start_recording( + request: Request, + path: str = Body("", embed=True), + name: str = Body("", embed=True), + count: int = Body(..., embed=True), +) -> dict: + logger.info("Web UI request: start with disk recording (count=%s)", count) + controller = _controller(request) + await _run_action(controller.start_recording, path, name, count) return controller.status() @@ -59,7 +87,7 @@ async def post_stop(request: Request) -> dict: async def post_tmp_reference(request: Request) -> dict: logger.info("Web UI request: capture temporary reference") controller = _controller(request) - controller.capture_tmp_reference() + await _run_action(controller.capture_tmp_reference) return controller.status() @@ -67,7 +95,7 @@ async def post_tmp_reference(request: Request) -> dict: async def post_remove_last(request: Request) -> dict: logger.info("Web UI request: remove last measurement") controller = _controller(request) - controller.remove_last_measurement() + await _run_action(controller.remove_last_measurement) return controller.status() @@ -80,11 +108,7 @@ async def get_configs(request: Request) -> dict: async def post_load_config(request: Request, name: str = Body(..., embed=True)) -> dict: logger.info("Web UI request: load config %r", name) controller = _controller(request) - try: - controller.load_config(name) - except ValueError as exc: - logger.warning("Web UI rejected config load: %s", exc) - raise HTTPException(status_code=400, detail=str(exc)) from exc + await _run_action(controller.load_config, name) return controller.status() @@ -96,7 +120,7 @@ async def post_save_dataset( ) -> dict: logger.info("Web UI request: save dataset") controller = _controller(request) - controller.save_dataset(path, name) + await _run_action(controller.save_dataset, path, name) return controller.status() diff --git a/python_app/webui/static/app.js b/python_app/webui/static/app.js index 863da04..2bbfb59 100644 --- a/python_app/webui/static/app.js +++ b/python_app/webui/static/app.js @@ -22,6 +22,9 @@ const configActiveEl = document.getElementById("config-active"); const savePathInput = document.getElementById("save-path"); const saveNameInput = document.getElementById("save-name"); const btnSaveDataset = document.getElementById("btn-save-dataset"); +const recordCountInput = document.getElementById("record-count"); +const btnStartRecording = document.getElementById("btn-start-recording"); +const recordingStatusEl = document.getElementById("recording-status"); const settingsToggle = document.getElementById("settings-toggle"); const sidePanel = document.querySelector(".side-panel"); @@ -40,6 +43,7 @@ let pendingPng = null; // newest PNG (base64), shown on the next animation let lastFrameTs = 0; // performance.now() of the last received frame let pipelineRunning = false; // gates the config loader (loading requires a stopped pipeline) let saveFieldsPrefilled = false; // seed the save path/name fields once, then leave the operator's edits +let recordCountPrefilled = false; // seed the record-count field once from the desktop default /* ---- helpers ----------------------------------------------------- */ function toast(message, isError) { @@ -305,14 +309,44 @@ btnSaveDataset.addEventListener("click", async () => { path: savePathInput.value.trim(), name: saveNameInput.value.trim(), }); - toast("Save requested"); // a radar-config prefix is prepended to the name server-side + toast("Dataset saved"); // a radar-config prefix is prepended to the name server-side } catch (err) { - toast(err.message, true); + toast(err.message, true); // e.g. the destination directory already exists } finally { btnSaveDataset.disabled = false; } }); +// Start (if stopped) and record the next N measurements to disk, then stop writing. +// While a recording is in progress the button stays disabled (driven by status), so a +// second recording can't be armed over the first. +let recordingActive = false; + +function updateRecordButtonState() { + btnStartRecording.disabled = recordingActive; +} + +btnStartRecording.addEventListener("click", async () => { + const count = parseInt(recordCountInput.value, 10); + if (!Number.isFinite(count) || count < 1) { + toast("Enter how many sweeps to record (>= 1)", true); + return; + } + btnStartRecording.disabled = true; // optimistic; status keeps it disabled while recording + try { + await api("/api/start_recording", { + path: savePathInput.value.trim(), + name: saveNameInput.value.trim(), + count, + }); + toast(`Recording armed: next ${count} sweep(s)`); + } catch (err) { + toast(err.message, true); // e.g. the destination directory already exists + } finally { + updateRecordButtonState(); // re-enable only if not actually recording + } +}); + /* ---- status ------------------------------------------------------ */ function setStat(el, label, value, cls) { el.className = "stat" + (cls ? " " + cls : ""); @@ -333,6 +367,18 @@ function applyStatus(s) { saveNameInput.value = s.save_name || ""; saveFieldsPrefilled = true; } + if ("record_count" in s && !recordCountPrefilled && document.activeElement !== recordCountInput) { + recordCountInput.value = s.record_count; // seed once; don't clobber later edits + recordCountPrefilled = true; + } + if (s.recording) { + const r = s.recording; + recordingActive = !!r.active; + updateRecordButtonState(); + recordingStatusEl.textContent = r.active + ? `recording ${r.collected} / ${r.target} sweep(s)…` + : ""; + } if ("processor_running" in s) setStat(processorEl, "processor", s.processor_running ? "yes" : "no", s.processor_running ? "ok" : "off"); diff --git a/python_app/webui/static/index.html b/python_app/webui/static/index.html index 4323fa4..62da8f9 100644 --- a/python_app/webui/static/index.html +++ b/python_app/webui/static/index.html @@ -39,7 +39,7 @@
-
Save dataset
+
Save / record dataset
@@ -47,6 +47,12 @@
+
+ + +
+