diff --git a/python_app/gui/app_window.py b/python_app/gui/app_window.py index 05934d2..92422e3 100644 --- a/python_app/gui/app_window.py +++ b/python_app/gui/app_window.py @@ -287,6 +287,17 @@ class AppWindow( self._bscan_depth_axis_by_combo = {} self._bscan_history_floor_collection_id = 0 self._bscan_render_signature = None + # Writer into the processor's input ring, used to re-feed retained sweeps so the + # whole visible B-scan is recomputed. Opened lazily, only while acquisition is + # stopped (see AppWindowBscanReplayMixin). + self._replay_ring_writer = None + self._bscan_replay_active = False + # Results from the last completed replay. While non-empty they, not + # `_result_history`, are what the B-scan renders (see AppWindowBscanReplayMixin). + self._bscan_replay_results = [] + self._bscan_reprocess_timer = QTimer(self) + self._bscan_reprocess_timer.setSingleShot(True) + self._bscan_reprocess_timer.timeout.connect(self._reprocess_history_through_processor) self._gpr_lookup_table = None self._gpr_image_item = None self._gpr_tx_item = None @@ -745,6 +756,10 @@ class AppWindow( # 0) Stop the GPIO button watcher so a late press cannot start work. self._stop_control_button_watcher() self._resume_pipeline_after_capture = False + # 0) Drop the B-scan replay writer so a pending debounce cannot push into a + # ring we are about to tear down. + self._bscan_reprocess_timer.stop() + self._close_replay_ring_writer() # 1) Abort active capture first (releases exclusive hardware resources). self._abort_capture_sequence(resume_pipeline=False) # 2) Stop all managed processes/readers. diff --git a/python_app/gui/controllers/app_window_config/live_processing_mixin.py b/python_app/gui/controllers/app_window_config/live_processing_mixin.py index 2d4c3b3..41e0148 100644 --- a/python_app/gui/controllers/app_window_config/live_processing_mixin.py +++ b/python_app/gui/controllers/app_window_config/live_processing_mixin.py @@ -358,6 +358,10 @@ class AppWindowLiveProcessingMixin: self._drain_results_until_quiet(timeout_s=0.25, poll_s=0.01) self._sync_bscan_history_from_results() self._draw_bscan_heatmap_from_history() + # The processor only refreshed its own newest sweeps; queue a full + # recompute of everything on screen. Debounced, so dragging a spin box + # sends one burst instead of one per step. + self._schedule_bscan_history_reprocess() elif self._is_gpr_processing_mode(current_mode): latest = self._drain_results_until_quiet(timeout_s=0.8, poll_s=0.02) collection = latest diff --git a/python_app/gui/controllers/app_window_pipeline_mixin.py b/python_app/gui/controllers/app_window_pipeline_mixin.py index aa82914..5cbd2c7 100644 --- a/python_app/gui/controllers/app_window_pipeline_mixin.py +++ b/python_app/gui/controllers/app_window_pipeline_mixin.py @@ -150,6 +150,13 @@ class AppWindowPipelineMixin: # completed and the GUI hung. single_capture_start_ns = time.monotonic_ns() if single_capture else None + # `data_preprocessor` is about to become the owner of the preprocessed ring + # again, so the GUI must stop holding a writer into it. A queued replay would + # otherwise interleave with the real producer. + self._bscan_reprocess_timer.stop() + self._close_replay_ring_writer() + self._discard_bscan_replay_results() + self._supervisor.start(config_path, allow_clean_orchestrator_exit=single_capture) self._close_readers() self._raw_reader = ShmRingReader(config.rings.raw_tap.name) @@ -683,6 +690,7 @@ class AppWindowPipelineMixin: """Reset runtime history and B-scan caches.""" self._replace_runtime_history(retained_raw=[], retained_pre=[], retained_result=[]) self._bscan_history_floor_collection_id = 0 + self._discard_bscan_replay_results() self._clear_history_mode_caches() self._update_history_indicator() diff --git a/python_app/gui/controllers/app_window_plot/__init__.py b/python_app/gui/controllers/app_window_plot/__init__.py index 091f3de..54cd519 100644 --- a/python_app/gui/controllers/app_window_plot/__init__.py +++ b/python_app/gui/controllers/app_window_plot/__init__.py @@ -1,11 +1,13 @@ """Plot-rendering mixins split by rendering mode.""" from python_app.gui.controllers.app_window_plot.bscan_plot_mixin import AppWindowBscanPlotMixin +from python_app.gui.controllers.app_window_plot.bscan_replay_mixin import AppWindowBscanReplayMixin from python_app.gui.controllers.app_window_plot.gpr_plot_mixin import AppWindowGprPlotMixin from python_app.gui.controllers.app_window_plot.trace_plot_mixin import AppWindowTracePlotMixin __all__ = [ "AppWindowBscanPlotMixin", + "AppWindowBscanReplayMixin", "AppWindowGprPlotMixin", "AppWindowTracePlotMixin", ] diff --git a/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py b/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py index 6524a9c..de2eb11 100644 --- a/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py +++ b/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py @@ -83,8 +83,15 @@ def rebuild_bscan_history_from_results( result_history: list[ResultCollection], history_limit: int, floor_collection_id: int, + stats: dict[str, object] | None = None, ) -> tuple[dict[tuple[int, int], deque[np.ndarray]], dict[tuple[int, int], np.ndarray]]: - """Rebuild B-scan history and depth axes from processed result payloads.""" + """Rebuild B-scan history and depth axes from processed result payloads. + + Pass `stats` to receive a breakdown of why the rebuilt image may hold fewer + columns than `history_limit`. The three causes are independent and only + distinguishable here: too little history, frames carrying no `bscan` payload, and + depth-axis changes that reset the accumulated deque (see below). + """ history_by_combo: dict[tuple[int, int], deque[np.ndarray]] = {} depth_axis_by_combo: dict[tuple[int, int], np.ndarray] = {} @@ -93,8 +100,11 @@ def rebuild_bscan_history_from_results( history_limit=history_limit, floor_collection_id=floor_collection_id, ) + axis_resets = 0 + without_bscan = 0 for collection in result_tail: + carried_bscan = False for block in collection.blocks: key = (block.combo.input, block.combo.output) for payload in block.payloads: @@ -110,6 +120,7 @@ def rebuild_bscan_history_from_results( if depth_axis.size == 0 or amplitudes.size == 0: continue + carried_bscan = True history = history_by_combo.get(key) stored_axis = depth_axis_by_combo.get(key) if ( @@ -118,11 +129,25 @@ def rebuild_bscan_history_from_results( or stored_axis.shape != depth_axis.shape or not np.allclose(stored_axis, depth_axis, rtol=1e-4, atol=1e-6) ): + # A changed depth axis makes previously accumulated columns + # un-stackable, so the deque restarts and everything gathered so far + # for this combo is dropped. Frames computed with different + # bscan_max_depth_m / frequency bounds land here — which is exactly + # what a partially replayed history looks like. + if history is not None: + axis_resets += 1 history = deque(maxlen=history_limit) history_by_combo[key] = history depth_axis_by_combo[key] = depth_axis.copy() history.append(amplitudes.copy()) + if not carried_bscan: + without_bscan += 1 + + if stats is not None: + stats["tail"] = len(result_tail) + stats["without_bscan"] = without_bscan + stats["axis_resets"] = axis_resets return history_by_combo, depth_axis_by_combo @@ -214,7 +239,7 @@ class AppWindowBscanPlotMixin: depth_max = float(np.max(depth_axis)) depth_span = max(depth_max - depth_min, 1e-6) sweep_count = sweeps.shape[0] - self._warn_if_bscan_exceeds_replay_window(sweep_count) + self._warn_if_bscan_window_underfilled() sweep_width = float(max(sweep_count, 1)) x_min = 0.5 x_max = x_min + sweep_width @@ -238,33 +263,30 @@ class AppWindowBscanPlotMixin: ) return True - def _warn_if_bscan_exceeds_replay_window(self, sweep_count: int) -> None: - """Warn once when the image reaches past what the C++ processor can replay. + def _warn_if_bscan_window_underfilled(self) -> None: + """Warn once when there is less retained raw material than the operator asked for. - Beyond that window a frame keeps the payload it was first computed with, so - editing Gain / Cut / Max depth / Start-Stop MHz silently leaves the older - columns on their previous settings — the image mixes two parameter sets. + Settings edits are recomputed across the whole image by re-feeding + `_pre_history` to the processor (see :class:`AppWindowBscanReplayMixin`), so a + wide window is no longer a coherency problem. What it can still be is an empty + promise: asking for more sweeps than were ever captured simply shows fewer. """ - replay_window = int(self._bscan_cpp_replay_window) - if sweep_count <= replay_window: + window = self._bscan_display_window_scans() + retained = len(self._pre_history) + if retained >= window: return - stale_count = sweep_count - replay_window self._log_warning( - f"B-scan shows {sweep_count} sweeps but the processor replays only the newest " - f"{replay_window}; the older {stale_count} keep the settings they were captured with.", + f"B-scan is set to show {window} sweeps but only {retained} are retained; " + "the image shows what history there is.", details=( - "Changing Gain / Cut m / Max depth m / Start MHz / Stop MHz re-processes " - f"only the newest {replay_window} sweeps.\n" - f"Reduce 'Scans to show (stopped)' to {replay_window} for an image that is " - "coherent across every column." - ), - # Keyed on the operator-controlled window, not the live column count, so - # that repeated "Remove Last" in stopped mode does not re-warn every click. - once_key=( - f"bscan_window_exceeds_replay_{replay_window}_" - f"{self._bscan_display_window_scans()}" + "The GUI keeps a bounded history of preprocessed sweeps, so a window " + "wider than the run itself cannot be filled.\n" + f"Capture more sweeps, or set 'Scans to show (stopped)' to {retained} or less." ), + # Keyed on the operator-controlled window rather than the live retained count, + # so that repeated "Remove Last" does not re-warn on every click. + once_key=f"bscan_window_underfilled_{window}", ) def _bscan_display_window_scans(self) -> int: @@ -293,25 +315,90 @@ class AppWindowBscanPlotMixin: def _bscan_signature(self) -> tuple[object, ...]: """Build state signature for B-scan history cache invalidation.""" live_config = self._live_processing_config() - result_history = list(self._result_history) + # Must read the same source the rebuild will, or the cache decides nothing + # changed while the image would in fact be built from different collections. + result_history, from_replay = self._bscan_source_collections() return build_bscan_signature( live_config=live_config, subtract_mean_ascan_enabled=bool(self._bscan_subtract_mean_ascan.isChecked()), result_history=result_history, history_limit=self._bscan_display_window_scans(), - floor_collection_id=self._bscan_history_floor_collection_id, + floor_collection_id=0 if from_replay else self._bscan_history_floor_collection_id, ) + def _bscan_source_collections(self) -> tuple[list[ResultCollection], bool]: + """Return the collections the image is built from, and whether they are replayed. + + A completed replay is the better source: it is exactly `window` long and every + entry went through the processor with the same settings. The runtime history is + not usable right after one, because results whose ids it never held are appended + out of order and the positional floor then cuts them away again. + """ + replayed = getattr(self, "_bscan_replay_results", None) + if replayed: + return list(replayed), True + return list(self._result_history), False + def _rebuild_bscan_history_from_results(self) -> None: """Recompute B-scan history cache from results history buffer.""" - result_history = list(self._result_history) + result_history, from_replay = self._bscan_source_collections() + window = self._bscan_display_window_scans() + # A replay already selected the window; re-applying the floor would cut it again. + floor_collection_id = 0 if from_replay else int(self._bscan_history_floor_collection_id) + stats: dict[str, object] = {} history_by_combo, depth_axis_by_combo = rebuild_bscan_history_from_results( result_history=result_history, - history_limit=self._bscan_display_window_scans(), - floor_collection_id=self._bscan_history_floor_collection_id, + history_limit=window, + floor_collection_id=floor_collection_id, + stats=stats, ) self._bscan_history_by_combo = history_by_combo self._bscan_depth_axis_by_combo = depth_axis_by_combo + self._log_bscan_window_shortfall( + window=window, + floor_collection_id=floor_collection_id, + result_history_len=len(result_history), + from_replay=from_replay, + history_by_combo=history_by_combo, + stats=stats, + ) + + def _log_bscan_window_shortfall( + self, + *, + window: int, + floor_collection_id: int, + result_history_len: int, + from_replay: bool, + history_by_combo: dict[tuple[int, int], deque[np.ndarray]], + stats: dict[str, object], + ) -> None: + """Explain at DEBUG why the image holds fewer columns than were requested. + + Four independent causes produce the same symptom, so each is reported as its + own number rather than a single verdict: + * `result history` / `tail` — the run simply produced fewer sweeps; + * `without_bscan` — frames processed in another mode carry no bscan payload; + * `axis_resets` — a changed depth axis restarted the deque, dropping every + column gathered before it (the usual cause after a partial replay); + * per-combo counts — the image renders one combo at a time. + """ + rendered = max((len(history) for history in history_by_combo.values()), default=0) + if rendered >= window: + return + + per_combo = ", ".join( + f"in{input_pos}/out{output_pos}={len(history)}" + for (input_pos, output_pos), history in sorted(history_by_combo.items()) + ) + self._log_debug( + f"B-scan window not filled: rendered={rendered} of requested={window}. " + f"source={'replay' if from_replay else 'result history'} len={result_history_len}, " + f"above floor(id>{floor_collection_id})={stats.get('tail')}, " + f"of those without a bscan payload={stats.get('without_bscan')}, " + f"depth-axis resets={stats.get('axis_resets')}. " + f"Per combo: {per_combo or 'none'}." + ) def _pick_bscan_display_key(self) -> tuple[int, int] | None: """Choose combo history key to render.""" diff --git a/python_app/gui/controllers/app_window_plot/bscan_replay_mixin.py b/python_app/gui/controllers/app_window_plot/bscan_replay_mixin.py new file mode 100644 index 0000000..691ed64 --- /dev/null +++ b/python_app/gui/controllers/app_window_plot/bscan_replay_mixin.py @@ -0,0 +1,253 @@ +"""Re-feed retained sweeps to the processor so it recomputes the whole B-scan. + +The processor keeps only ~50 preprocessed sweeps of its own (`kBscanReplayWindow` in +`data_processor.cpp`), so changing a live B-scan setting used to refresh just the newest +50 columns while everything older kept the parameters it was captured with — one image +stitched from two parameter sets. + +The GUI already holds up to 1000 preprocessed sweeps in `_pre_history`, byte-identical +to what the processor consumes: `data_preprocessor` pushes the very same serialized +buffer into both the working ring and the tap the GUI reads. So instead of making the +processor hoard sweeps, we hand its own raw material back to it and let the ordinary +`pop -> process -> publish` path recompute every column with the current settings. +""" + +from __future__ import annotations + +import time + +from python_app.gui.runtime.history import record_result_history +from python_app.orchestration.shm import ShmRingWriter +from python_app.storage.npz.serialize import PREPROC_MAGIC, serialize_trace_collection + +# The results ring overwrites unread slots, so a burst must never exceed what the GUI +# drains between chunks. Comfortably below the (typically 50-slot) ring capacity. +_REPLAY_CHUNK_SIZE = 12 + +# Per-chunk budget. Generous: the processor may be busy, and overshooting only costs +# a redraw that reflects fewer columns. +_REPLAY_CHUNK_TIMEOUT_S = 2.0 + +# Restarting the timer on every edit collapses a spin-box drag into one replay. +_BSCAN_REPROCESS_DEBOUNCE_MS = 300 + + +class AppWindowBscanReplayMixin: + """Recomputes the full visible B-scan by re-feeding sweeps to the processor.""" + + def _reprocess_history_through_processor(self) -> bool: + """Re-feed the visible tail of `_pre_history` so every column is recomputed. + + Returns True when a replay actually ran. Refuses (returning False) whenever the + preconditions for safely writing into the processor's input ring do not hold. + """ + # `_pump_events_during_drain` runs the event loop, so a settings edit made while + # a long replay is in flight can re-arm the debounce and fire this method inside + # itself — two nested bursts sharing one ring and one result count. + if getattr(self, "_bscan_replay_active", False): + return False + if not self._can_reprocess_history(): + return False + + window = self._bscan_display_window_scans() + tail = list(self._pre_history)[-window:] + # `_pre_history` and `_result_history` are fed by two different rings, each + # dropping independently when the producer outruns the GUI. A low overlap means + # the replayed results arrive under ids the result history never held, so they + # are appended rather than replacing the columns already on screen. + result_keys = { + (int(c.collection_id), int(c.monotonic_ns)) for c in self._result_history + } + overlap = sum( + 1 for c in tail if (int(c.collection_id), int(c.monotonic_ns)) in result_keys + ) + self._log_debug( + f"B-scan replay starting: window={window}, preprocessed history=" + f"{len(self._pre_history)}, result history={len(self._result_history)}, " + f"to re-send={len(tail)}, of those already in result history={overlap}." + ) + if not tail: + return False + + writer = self._ensure_replay_ring_writer() + if writer is None: + return False + + # The processor replays its own retained sweeps on every live-config revision + # bump. We are about to send the same collections (and more), so suppress it + # rather than let it publish the newest 50 twice. + self._write_live_processing_config(reprocess_current_result=False) + + # Our own event pumping would otherwise let the poll timer fire and consume the + # replayed results through `_read_all_results`, which records them to disk and + # feeds the pipeline metrics. Restart in `finally`: losing the ring poll on an + # exception would leave the GUI permanently blind. + self._timer.stop() + self._bscan_replay_active = True + sent = 0 + received = 0 + replayed: list = [] + try: + for start in range(0, len(tail), _REPLAY_CHUNK_SIZE): + chunk = tail[start : start + _REPLAY_CHUNK_SIZE] + pushed = 0 + for collection in chunk: + payload = serialize_trace_collection(collection, PREPROC_MAGIC) + if not writer.push(payload): + # Only fails when the payload exceeds the slot size, which is a + # config problem rather than a transient one: stop the burst. + self._log_warning( + "B-scan replay stopped: a preprocessed sweep does not fit the ring slot.", + details=( + f"payload={len(payload)} bytes, " + f"slot={writer.slot_size_bytes} bytes" + ), + once_key="bscan_replay_payload_too_large", + ) + break + pushed += 1 + sent += pushed + got = self._collect_replayed_results( + expected=pushed, timeout_s=_REPLAY_CHUNK_TIMEOUT_S, into=replayed + ) + received += got + if got < pushed: + # A short chunk means the processor did not answer in time; keep + # going, but say which one so a systematic stall is visible. + self._log_debug( + f"B-scan replay chunk at offset {start}: pushed={pushed}, recovered={got}." + ) + if pushed != len(chunk): + break + finally: + self._bscan_replay_active = False + self._timer.start() + + # Render straight from what came back rather than from `_result_history`. + # + # The two histories are fed by different rings that drop independently, so the + # sweeps we re-sent only partly overlap the results already on record. The + # non-overlapping ones get appended to the deque even though their ids are old, + # which breaks its ordering — and the positional floor then cuts exactly those + # off again. Observed: 300 re-sent, 300 recovered, 145 drawn. What came back is + # by construction the right count and uniformly processed, so use it directly. + self._bscan_replay_results = replayed + + # Replayed collections keep their original ids and timestamps, so the render + # signature is unchanged even though the payload values are not. Drop it or the + # cache would decide nothing needs rebuilding. + self._bscan_render_signature = None + self._sync_bscan_history_from_results() + self._draw_bscan_heatmap_from_history() + + if received < sent: + self._log_warning( + f"B-scan replay recovered {received} of {sent} re-sent sweeps; " + "some columns may still show their captured settings.", + once_key=f"bscan_replay_incomplete_{sent}_{received}", + ) + self._log_debug(f"B-scan replay finished: sent={sent}, recovered={received}.") + return True + + def _can_reprocess_history(self) -> bool: + """Return whether re-feeding the processor's input ring is safe right now.""" + if self._processing_mode.currentText() != "bscan": + return False + # `data_preprocessor` owns the preprocessed ring while acquisition runs; writing + # into it concurrently would corrupt the sequence counters. + if self._supervisor.is_running(): + return False + if not self._supervisor.is_processor_running(): + return False + return self._result_reader is not None + + def _collect_replayed_results( + self, *, expected: int, timeout_s: float, into: list | None = None + ) -> int: + """Pop `expected` replayed results, recording them into runtime history. + + Deliberately bypasses `_read_all_results`: that path also feeds the pipeline + metrics and the disk recorder, and a replay is neither new acquisition nor + something that should be written to disk a second time. + + `into` also receives them in arrival order, which is what the B-scan renders — + see `_reprocess_history_through_processor` for why the runtime history alone is + not a usable source afterwards. + """ + if expected <= 0 or self._result_reader is None: + return 0 + + deadline = time.monotonic() + timeout_s + received = 0 + while received < expected and time.monotonic() < deadline: + collection = self._result_reader.pop_result_collection() + if collection is None: + self._pump_events_during_drain(0.005) + continue + record_result_history(self._result_history, collection) + if into is not None: + into.append(collection) + received += 1 + return received + + def _ensure_replay_ring_writer(self) -> ShmRingWriter | None: + """Return a writer attached to the processor's input ring, creating it lazily. + + Geometry comes from `_active_run_config` — the config the C++ side actually + started with — never from the editable `_defaults_config`: `ShmRingWriter` owns + the rings it opens and *recreates* a segment whose geometry disagrees, which + would destroy the ring under a live processor. + """ + existing = getattr(self, "_replay_ring_writer", None) + if existing is not None: + return existing + + config = getattr(self, "_active_run_config", None) + if config is None: + # Processor outlived the GUI that started it: its ring geometry is unknown, + # and guessing risks recreating the segment underneath it. + self._log_warning( + "Cannot recompute the full B-scan: this GUI session did not start the " + "pipeline, so the processor's ring geometry is unknown.", + once_key="bscan_replay_no_active_run_config", + ) + return None + + ring = config.rings.preprocessed + try: + self._replay_ring_writer = ShmRingWriter( + ring.name, int(ring.capacity), int(ring.slot_size_bytes) + ) + except Exception as exc: # noqa: BLE001 + self._log_exception("Failed to open the B-scan replay ring writer", exc, level="WARN") + self._replay_ring_writer = None + return self._replay_ring_writer + + def _discard_bscan_replay_results(self) -> None: + """Fall back to the runtime history as the render source. + + Called whenever the replayed set stops describing what should be on screen: + acquisition resuming (live frames must win) or the history being edited. + """ + if not getattr(self, "_bscan_replay_results", None): + return + self._bscan_replay_results = [] + self._bscan_render_signature = None + + def _close_replay_ring_writer(self) -> None: + """Detach from the processor's input ring (safe to call repeatedly).""" + writer = getattr(self, "_replay_ring_writer", None) + if writer is None: + return + try: + writer.close() + except Exception as exc: # noqa: BLE001 + self._log_exception("Failed to close the B-scan replay ring writer", exc, level="WARN") + finally: + self._replay_ring_writer = None + + def _schedule_bscan_history_reprocess(self) -> None: + """Debounce a full recompute so dragging a spin box does not send hundreds of sweeps.""" + if not self._can_reprocess_history(): + return + self._bscan_reprocess_timer.start(_BSCAN_REPROCESS_DEBOUNCE_MS) diff --git a/python_app/gui/controllers/app_window_plot_mixin.py b/python_app/gui/controllers/app_window_plot_mixin.py index 5d89196..8f27bf9 100644 --- a/python_app/gui/controllers/app_window_plot_mixin.py +++ b/python_app/gui/controllers/app_window_plot_mixin.py @@ -4,6 +4,7 @@ from __future__ import annotations from python_app.gui.controllers.app_window_plot import ( AppWindowBscanPlotMixin, + AppWindowBscanReplayMixin, AppWindowGprPlotMixin, AppWindowTracePlotMixin, ) @@ -13,6 +14,7 @@ from python_app.models.dataset_model import ResultCollection class AppWindowPlotMixin( AppWindowTracePlotMixin, AppWindowBscanPlotMixin, + AppWindowBscanReplayMixin, AppWindowGprPlotMixin, ): """Routes plotting to trace, B-scan, or GPR-specific mixins.""" diff --git a/python_app/gui/controllers/app_window_snapshot_mixin.py b/python_app/gui/controllers/app_window_snapshot_mixin.py index b13e84f..f18f123 100644 --- a/python_app/gui/controllers/app_window_snapshot_mixin.py +++ b/python_app/gui/controllers/app_window_snapshot_mixin.py @@ -242,6 +242,7 @@ class AppWindowSnapshotMixin: retained_result=retained_result, ) self._bscan_history_floor_collection_id = 0 + self._discard_bscan_replay_results() self._clear_history_mode_caches() self._write_live_processing_config(history_command=history_command, bump_history_seq=True) diff --git a/python_app/tests/test_bscan_replay.py b/python_app/tests/test_bscan_replay.py new file mode 100644 index 0000000..ef18c5c --- /dev/null +++ b/python_app/tests/test_bscan_replay.py @@ -0,0 +1,282 @@ +"""Re-feeding retained sweeps to the processor to recompute the whole B-scan. + +The processor keeps only ~50 preprocessed sweeps of its own, so a live settings edit +used to refresh just the newest 50 columns. The GUI holds up to 1000 of them and hands +them back through the processor's input ring, which only works if the bytes the GUI +writes are exactly the ones the C++ side expects — that wire-format contract is what +the first test pins, without needing the pipeline running. + +The guard tests pin the other half: the GUI must never write into that ring while +`data_preprocessor` owns it. +""" + +from __future__ import annotations + +from contextlib import suppress +import os +from pathlib import Path +import unittest + +import numpy as np + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") +from PyQt6.QtCore import QSignalBlocker # noqa: E402 +from PyQt6.QtWidgets import QApplication # noqa: E402 + +from python_app.gui.app_window import AppWindow # noqa: E402 +from python_app.models.dataset_model import ( # noqa: E402 + ComboKey, + ResultBlock, + ResultCollection, + ResultPayload, + SweepCollection, + TraceData, +) +from python_app.orchestration.shm import ShmRingWriter # noqa: E402 +from python_app.orchestration.shm.ring_reader import ShmRingReader # noqa: E402 +from python_app.storage.npz.serialize import PREPROC_MAGIC, serialize_trace_collection # noqa: E402 + +# Mirrors kPreprocessedCollectionMagic in common_cpp/ipc/src/shared_types.cpp. If this +# ever drifts, the processor silently rejects everything the GUI re-feeds. +_CPP_PREPROCESSED_MAGIC = 0x32525050 + +_app: QApplication | None = None +_window: AppWindow | None = None + + +def setUpModule() -> None: + global _app, _window + _app = QApplication.instance() or QApplication([]) + _window = AppWindow(Path(".")) + + +def tearDownModule() -> None: + if _window is not None: + _window.close() + + +def _sweep(collection_id: int, *, combos: int = 3, points: int = 64) -> SweepCollection: + traces = [ + TraceData( + combo=ComboKey(input=index, output=index + 1), + frequency_hz=np.linspace(1e9, 8e9, points, dtype=np.float32), + s11=(np.arange(points) + index).astype(np.complex64) + 0.5j, + s21=(np.arange(points) * 2 + index).astype(np.complex64) - 0.25j, + ) + for index in range(combos) + ] + return SweepCollection( + collection_id=collection_id, + monotonic_ns=collection_id * 1_000_000 + 7, + traces=traces, + capture_start_ns=collection_id * 10, + capture_end_ns=collection_id * 10 + 5, + ) + + +def _bscan_result(collection_id: int, *, points: int = 32) -> ResultCollection: + payload = ResultPayload( + processing_name="bscan", + kind=1, + frequency_hz=np.linspace(0.0, 1.0, points, dtype=np.float32), + trace=(np.arange(points) + collection_id).astype(np.complex64), + ) + return ResultCollection( + collection_id=collection_id, + monotonic_ns=collection_id, + blocks=[ResultBlock(combo=ComboKey(input=0, output=0), payloads=[payload])], + ) + + +class PreprocessedWireFormatTest(unittest.TestCase): + """The bytes the GUI re-feeds must be the ones the C++ processor decodes.""" + + def setUp(self) -> None: + self.ring_name = f"/radar_test_{self._testMethodName}" + with suppress(OSError): + (Path("/dev/shm") / self.ring_name[1:]).unlink() + self.writer = ShmRingWriter(self.ring_name, 8, 1 << 20) + self.reader = ShmRingReader(self.ring_name) + self.addCleanup(self._cleanup) + + def _cleanup(self) -> None: + with suppress(Exception): + self.reader.close() + with suppress(Exception): + self.writer.close() + with suppress(OSError): + (Path("/dev/shm") / self.ring_name[1:]).unlink() + + def _assert_same(self, original: SweepCollection, decoded: SweepCollection) -> None: + self.assertEqual(decoded.collection_id, original.collection_id) + self.assertEqual(decoded.monotonic_ns, original.monotonic_ns) + self.assertEqual(decoded.capture_start_ns, original.capture_start_ns) + self.assertEqual(decoded.capture_end_ns, original.capture_end_ns) + self.assertEqual(len(decoded.traces), len(original.traces)) + for left, right in zip(original.traces, decoded.traces, strict=True): + self.assertEqual((left.combo.input, left.combo.output), (right.combo.input, right.combo.output)) + np.testing.assert_array_equal(left.frequency_hz, right.frequency_hz) + np.testing.assert_array_equal(left.s11, right.s11) + np.testing.assert_array_equal(left.s21, right.s21) + + def test_magic_matches_the_cpp_decoder(self) -> None: + payload = serialize_trace_collection(_sweep(1), PREPROC_MAGIC) + self.assertEqual(int.from_bytes(payload[:4], "little"), _CPP_PREPROCESSED_MAGIC) + + def test_round_trip_preserves_every_field(self) -> None: + original = _sweep(42) + self.assertTrue(self.writer.push(serialize_trace_collection(original, PREPROC_MAGIC))) + + decoded = self.reader.pop_preprocessed_collection() + self.assertIsNotNone(decoded) + assert decoded is not None + self._assert_same(original, decoded) + + def test_batch_keeps_order(self) -> None: + originals = [_sweep(cid) for cid in range(100, 105)] + for collection in originals: + self.assertTrue(self.writer.push(serialize_trace_collection(collection, PREPROC_MAGIC))) + + decoded = [] + while (collection := self.reader.pop_preprocessed_collection()) is not None: + decoded.append(collection) + + self.assertEqual([c.collection_id for c in decoded], [c.collection_id for c in originals]) + for left, right in zip(originals, decoded, strict=True): + self._assert_same(left, right) + + def test_overflow_drops_oldest(self) -> None: + # Why the replay pushes in chunks instead of one burst: an undrained ring + # silently overwrites, which would punch holes into the very image we are + # trying to make coherent. + capacity = 8 + for cid in range(200, 200 + capacity + 4): + self.writer.push(serialize_trace_collection(_sweep(cid), PREPROC_MAGIC)) + + survived = [] + while (collection := self.reader.pop_preprocessed_collection()) is not None: + survived.append(collection.collection_id) + + self.assertEqual(len(survived), capacity) + self.assertEqual(survived[-1], 200 + capacity + 3) + + +class _IdleResultReader: + """Stands in for a connected results reader that simply has nothing to hand out.""" + + def pop_result_collection(self): + return None + + def close(self) -> None: + return None + + +class ReplayGuardTest(unittest.TestCase): + """The GUI must refuse to write into a ring `data_preprocessor` still owns.""" + + def setUp(self) -> None: + self.w = _window + # The 50 ms ring poll and the debounce would both run against this half-faked + # window while the test drives it by hand; park them for the duration. + self.w._timer.stop() + self.w._bscan_reprocess_timer.stop() + self.addCleanup(self.w._timer.start) + self.addCleanup(self.w._bscan_reprocess_timer.stop) + + self._original_is_running = self.w._supervisor.is_running + self._original_is_processor_running = self.w._supervisor.is_processor_running + self._original_mode = self.w._processing_mode.currentText() + # Changing the mode fires the live-settings handler, which would kick off the + # very replay these tests are inspecting. + with QSignalBlocker(self.w._processing_mode): + self.w._processing_mode.setCurrentText("bscan") + self.w._supervisor.is_running = lambda: False + self.w._supervisor.is_processor_running = lambda: True + self.w._result_reader = _IdleResultReader() + + def tearDown(self) -> None: + self.w._supervisor.is_running = self._original_is_running + self.w._supervisor.is_processor_running = self._original_is_processor_running + with QSignalBlocker(self.w._processing_mode): + self.w._processing_mode.setCurrentText(self._original_mode) + self.w._result_reader = None + + def test_refuses_while_acquisition_runs(self) -> None: + self.w._supervisor.is_running = lambda: True + self.assertFalse(self.w._can_reprocess_history()) + + def test_refuses_when_processor_is_down(self) -> None: + self.w._supervisor.is_processor_running = lambda: False + self.assertFalse(self.w._can_reprocess_history()) + + def test_refuses_outside_bscan_mode(self) -> None: + self.w._processing_mode.setCurrentText("pass_through") + self.assertFalse(self.w._can_reprocess_history()) + + def test_refuses_without_a_results_reader(self) -> None: + self.w._result_reader = None + self.assertFalse(self.w._can_reprocess_history()) + + def test_allows_when_stopped_with_a_live_processor(self) -> None: + self.assertTrue(self.w._can_reprocess_history()) + + def test_replayed_results_are_rendered_instead_of_runtime_history(self) -> None: + """Regression: 300 re-sent, 300 recovered, only 145 drawn. + + `_pre_history` and `_result_history` come from two rings that drop + independently, so the re-sent sweeps only partly overlap the recorded results. + The non-overlapping ones get appended to the deque under old ids, breaking its + ordering, and the positional floor then cuts exactly those away. Rendering from + the replayed set instead sidesteps the whole problem. + """ + window = 30 + replayed = [_bscan_result(cid) for cid in range(9000, 9000 + window)] + + self.w._result_history.clear() + # Stand-in for a runtime history whose ids barely overlap the replayed ones. + self.w._result_history.extend(_bscan_result(cid) for cid in range(1, 200)) + self.addCleanup(self.w._result_history.clear) + + self.w._bscan_replay_results = replayed + self.addCleanup(self.w._discard_bscan_replay_results) + self.w._bscan_history_window.setValue(window) + self.w._bscan_render_signature = None + self.w._sync_bscan_history_from_results() + + self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), window) + + def test_discarding_replay_falls_back_to_runtime_history(self) -> None: + self.w._bscan_replay_results = [_bscan_result(1)] + self.w._discard_bscan_replay_results() + self.assertEqual(self.w._bscan_replay_results, []) + self.assertIsNone(self.w._bscan_render_signature) + + def test_replay_refuses_to_re_enter_itself(self) -> None: + # Pumping the event loop mid-replay can fire the debounce again; a nested burst + # would share the ring and corrupt the result accounting. + self.w._pre_history.clear() + self.w._pre_history.extend(_sweep(cid) for cid in range(1, 6)) + self.addCleanup(self.w._pre_history.clear) + + self.w._bscan_replay_active = True + self.addCleanup(setattr, self.w, "_bscan_replay_active", False) + + self.assertFalse(self.w._reprocess_history_through_processor()) + + def test_replay_refuses_without_an_active_run_config(self) -> None: + # A GUI restarted against a still-running processor does not know the ring + # geometry, and ShmRingWriter would recreate the segment underneath it. + self.w._pre_history.clear() + self.w._pre_history.extend(_sweep(cid) for cid in range(1, 6)) + self.addCleanup(self.w._pre_history.clear) + + previous = getattr(self.w, "_active_run_config", None) + self.w._active_run_config = None + self.addCleanup(setattr, self.w, "_active_run_config", previous) + + self.assertIsNone(self.w._ensure_replay_ring_writer()) + self.assertFalse(self.w._reprocess_history_through_processor()) + + +if __name__ == "__main__": + unittest.main()