Merge branch 'laser-temp-monitoring' into feature/switched-matrix-radar

MultiDeviceLibreVnaService only exposes the 2x4 virtual combo matrix on
its own USB transport. When a physical GPIO switch sits on the master
stimulus and/or slave receiver path, the effective matrix is wider than
that. SwitchedMatrixRadarService wraps the inner service and drives the
extra switch(es) between acquire_collection calls, widening the combo
matrix by the physical position counts (matrix_output_switch_positions /
matrix_input_switch_positions in RunConfigModel).

Combo-matrix construction was centralized into
RunConfigModel.build_runtime_combos() so the GUI, workflows, and codec
all derive the same widened matrix instead of each computing its own
version of the virtual 2x4 layout.

.gitignore

@@ -227,4 +227,5 @@ python_app/runtime
 SHARE_INTERNET_TO_PI.md
 
 CLAUDE.md
-./docs
+docs/
+test_end_2/

python_app/gui/app_window.py

@@ -274,7 +274,7 @@ class AppWindow(
 
     def _init_history_state(self) -> None:
         """Initialize runtime history buffers and render-cache state."""
-        bscan_history_limit = self._history_limit_from_config()
+        bscan_cpp_replay_window = self._cpp_bscan_replay_window_from_config()
         save_history_limit = self._save_history_limit_from_config()
         self._raw_history: deque[SweepCollection] = deque(maxlen=save_history_limit)
         self._pre_history: deque[SweepCollection] = deque(maxlen=save_history_limit)
@@ -282,11 +282,21 @@ class AppWindow(
 
         # Sequence id must survive GUI restarts so history commands stay monotonic.
         self._history_command_seq = self._load_history_command_seq(self._live_config_writer.path)
-        self._bscan_history_limit = bscan_history_limit
+        self._bscan_cpp_replay_window = bscan_cpp_replay_window
         self._bscan_history_by_combo = {}
         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
@@ -305,9 +315,9 @@ class AppWindow(
         self._active_processing_mode = "pass_through"
         self._radar_limits: dict[str, float | int] | None = None
 
-    def _history_limit_from_config(self) -> int:
-        """Return B-scan render history limit derived from configured ring capacities."""
-        return self._history_limit_for_config(self._defaults_config)
+    def _cpp_bscan_replay_window_from_config(self) -> int:
+        """Return the C++ B-scan replay window for the active config."""
+        return self._cpp_bscan_replay_window_for_config(self._defaults_config)
 
     def _save_history_limit_from_config(self) -> int:
         """Return maxlen for GUI snapshot-save deques (independent of ring capacities)."""
@@ -745,6 +755,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.

python_app/gui/controllers/app_window_config/live_processing_mixin.py

@@ -360,6 +360,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
@@ -508,7 +512,6 @@ class AppWindowLiveProcessingMixin:
 
     def _clear_history_mode_caches(self) -> None:
         """Drop cached render state for pass-through, B-scan, and GPR views."""
-        self._bscan_history_floor_collection_id = 0
         self._clear_bscan_plot_history()
         if hasattr(self, "_bscan_plot"):
             self._bscan_plot.clear()

python_app/gui/controllers/app_window_config/profile_io_mixin.py

@@ -61,7 +61,7 @@ class AppWindowConfigProfileIOMixin:
         """Return the canonical virtual combo matrix shown for matrix-mode radars."""
         return ",".join(
             f"{int(combo.input)}:{int(combo.output)}"
-            for combo in RunConfigModel.build_matrix_radar_virtual_combos()
+            for combo in self._defaults_config.build_runtime_combos()
         )
 
     def _sync_pass_through_y_controls(self) -> None:
@@ -121,15 +121,15 @@ class AppWindowConfigProfileIOMixin:
 
         Save-side deques use a config-independent limit so that processing-side
         ring capacities can stay small without truncating the save buffer. The
-        B-scan render limit still follows ring capacities to keep plot updates
-        responsive.
+        C++ replay window still follows ring capacities, since it bounds how much
+        of the history the processor can re-publish coherently.
         """
         save_history_limit = self._save_history_limit_for_config(config)
-        bscan_history_limit = self._history_limit_for_config(config)
+        bscan_cpp_replay_window = self._cpp_bscan_replay_window_for_config(config)
         self._raw_history = deque(self._raw_history, maxlen=save_history_limit)
         self._pre_history = deque(self._pre_history, maxlen=save_history_limit)
         self._result_history = deque(self._result_history, maxlen=save_history_limit)
-        self._bscan_history_limit = bscan_history_limit
+        self._bscan_cpp_replay_window = bscan_cpp_replay_window
         self._clear_bscan_plot_history()
 
     def _save_current_config(self) -> None:
@@ -283,6 +283,7 @@ class AppWindowConfigProfileIOMixin:
             self._bscan_start_freq_mhz,
             self._bscan_stop_freq_mhz,
             self._bscan_subtract_mean_ascan,
+            self._bscan_history_window,
             self._gpr_relative_permittivity,
             self._gpr_tx_geometry_input,
             self._gpr_rx_geometry_input,
@@ -438,6 +439,7 @@ class AppWindowConfigProfileIOMixin:
             self._bscan_start_freq_mhz.setValue(float(gui_state.processing.bscan.start_freq_mhz))
             self._bscan_stop_freq_mhz.setValue(float(gui_state.processing.bscan.stop_freq_mhz))
             self._bscan_subtract_mean_ascan.setChecked(bool(gui_state.processing.bscan.subtract_mean_ascan))
+            self._bscan_history_window.setValue(int(gui_state.processing.bscan.history_window_scans))
 
             self._gpr_relative_permittivity.setValue(float(config.gpr.relative_permittivity))
             self._gpr_tx_geometry_input.setPlainText(

python_app/gui/controllers/app_window_config/state_builders.py

@@ -34,6 +34,13 @@ from python_app.storage.npz_store import radar_key_from_config
 # history without touching the processing-side ring sizes.
 GUI_SAVE_HISTORY_LIMIT: int = 1000
 
+# Mirror of `kBscanReplayWindow` in
+# data_acq_and_processing/processing/data_processor/src/data_processor.cpp. When a
+# live B-scan setting changes, the C++ processor re-processes and re-publishes only
+# this many of the newest collections; anything older keeps the payload it was first
+# computed with. Keep the two constants in sync.
+CPP_BSCAN_REPLAY_WINDOW: int = 50
+
 
 class AppWindowConfigStateBuildersMixin:
     """Build stable and GUI-only config models from current widget state."""
@@ -125,7 +132,7 @@ class AppWindowConfigStateBuildersMixin:
         if config.is_matrix_radar:
             return ",".join(
                 f"{int(combo.input)}:{int(combo.output)}"
-                for combo in RunConfigModel.build_matrix_radar_virtual_combos()
+                for combo in config.build_runtime_combos()
             )
         combos = list(config.combos)
         full_combos = config.build_full_combos(config.input_switch.positions, config.output_switch.positions)
@@ -163,14 +170,24 @@ class AppWindowConfigStateBuildersMixin:
         return (min(x_values) - margin_m, max(x_values) + margin_m)
 
     @staticmethod
-    def _history_limit_for_config(config: RunConfigModel) -> int:
-        """Return B-scan render history limit derived from config ring capacities."""
+    def _cpp_bscan_replay_window_for_config(config: RunConfigModel) -> int:
+        """Return how many newest collections the C++ processor re-processes on a
+        live B-scan settings change.
+
+        Deliberately reproduces `replay_history_limit()` in `data_processor.cpp`
+        formula-for-formula. The ring capacities matter because `ShmRing` overwrites
+        the oldest unread slot on overflow, so a replay burst must fit in the results
+        ring for the GUI to receive all of it.
+
+        This is the single place to change if the replay window ever becomes
+        configurable on the C++ side.
+        """
         return max(
             1,
             min(
-                int(config.rings.raw_tap.capacity),
-                int(config.rings.preprocessed_tap.capacity),
+                int(config.rings.preprocessed.capacity),
                 int(config.rings.results.capacity),
+                CPP_BSCAN_REPLAY_WINDOW,
             ),
         )
 
@@ -180,7 +197,7 @@ class AppWindowConfigStateBuildersMixin:
 
         Independent of ring capacities — see :data:`GUI_SAVE_HISTORY_LIMIT`.
         The `config` argument is kept for symmetry with
-        :meth:`_history_limit_for_config` and possible future per-profile
+        :meth:`_cpp_bscan_replay_window_for_config` and possible future per-profile
         overrides.
         """
         del config
@@ -344,6 +361,7 @@ class AppWindowConfigStateBuildersMixin:
                     start_freq_mhz=float(self._bscan_start_freq_mhz.value()),
                     stop_freq_mhz=float(self._bscan_stop_freq_mhz.value()),
                     subtract_mean_ascan=bool(self._bscan_subtract_mean_ascan.isChecked()),
+                    history_window_scans=int(self._bscan_history_window.value()),
                 ),
                 gpr=GuiGprStateModel(
                     input_positions=self._gpr_input_positions_input.text().strip(),

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)
@@ -682,7 +689,7 @@ class AppWindowPipelineMixin:
     def _reset_runtime_history(self) -> None:
         """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()
 

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",
 ]

python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py

@@ -16,17 +16,19 @@ def _result_tail(
     *,
     result_history: list[ResultCollection],
     history_limit: int,
-    floor_collection_id: int,
 ) -> list[ResultCollection]:
-    """Return filtered and de-duplicated result-history tail for B-scan usage."""
-    filtered = [
-        collection
-        for collection in result_history[-history_limit:]
-        if int(collection.collection_id) > int(floor_collection_id)
-    ]
+    """Return the de-duplicated newest `history_limit` entries for B-scan usage.
+
+    Selection is purely positional. An earlier version also dropped entries below a
+    `collection_id` floor, which cannot work here: ids are neither dense (the results
+    ring overwrites unread slots) nor monotonic across a run boundary (the C++ side
+    numbers from 1 again). Given the floor was derived from the first entry of this
+    very slice, the comparison provably removed nothing when ids ascend, and removed
+    exactly the newest frames when they do not.
+    """
     unique_reversed_tail: list[ResultCollection] = []
     seen_keys: set[tuple[int, int]] = set()
-    for collection in reversed(filtered):
+    for collection in reversed(result_history[-history_limit:]):
         key = (int(collection.collection_id), int(collection.monotonic_ns))
         if key in seen_keys:
             continue
@@ -43,15 +45,14 @@ def build_bscan_signature(
     subtract_mean_ascan_enabled: bool,
     result_history: list[ResultCollection],
     history_limit: int,
-    floor_collection_id: int,
 ) -> tuple[object, ...]:
     """Build deterministic signature used to detect B-scan cache invalidation."""
     result_tail = _result_tail(
         result_history=result_history,
         history_limit=history_limit,
-        floor_collection_id=floor_collection_id,
     )
     return (
+        int(history_limit),
         str(live_config.bscan_axis),
         str(live_config.bscan_channel),
         float(live_config.bscan_cut_m),
@@ -60,7 +61,6 @@ def build_bscan_signature(
         float(live_config.bscan_start_freq_mhz),
         float(live_config.bscan_stop_freq_mhz),
         bool(subtract_mean_ascan_enabled),
-        int(floor_collection_id),
         tuple((int(collection.collection_id), int(collection.monotonic_ns), len(collection.blocks)) for collection in result_tail),
     )
 
@@ -81,19 +81,27 @@ def apply_mean_ascan_subtraction(
 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] = {}
 
     result_tail = _result_tail(
         result_history=result_history,
         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:
@@ -109,6 +117,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 (
@@ -117,11 +126,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
 
@@ -213,6 +236,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_window_underfilled()
         sweep_width = float(max(sweep_count, 1))
         x_min = 0.5
         x_max = x_min + sweep_width
@@ -236,9 +260,48 @@ class AppWindowBscanPlotMixin:
         )
         return True
 
+    def _warn_if_bscan_window_underfilled(self) -> None:
+        """Warn once when there is less retained raw material than the operator asked for.
+
+        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.
+        """
+        window = self._bscan_display_window_scans()
+        retained = len(self._pre_history)
+        if retained >= window:
+            return
+
+        self._log_warning(
+            f"B-scan is set to show {window} sweeps but only {retained} are retained; "
+            "the image shows what history there is.",
+            details=(
+                "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:
+        """Return how many past sweeps the B-scan should render right now.
+
+        While acquisition runs the window stays at the C++ replay window: results
+        arrive continuously, the whole history is rebuilt on every new one, and a
+        1000-wide rebuild on the live path would cost ~20x per frame.
+
+        Once stopped, the operator reviews a frozen history, so the user-configured
+        window applies and may reach back over the whole GUI result deque.
+        """
+        if self._supervisor.is_running():
+            return int(self._bscan_cpp_replay_window)
+        return max(1, int(self._bscan_history_window.value()))
+
     def _sync_bscan_history_from_results(self) -> None:
         """Rebuild B-scan history cache when live params or inputs changed."""
-        self._advance_bscan_floor_to_cpp_window()
         signature = self._bscan_signature()
         if signature == self._bscan_render_signature:
             return
@@ -248,25 +311,84 @@ 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_history_limit,
-            floor_collection_id=self._bscan_history_floor_collection_id,
+            history_limit=self._bscan_display_window_scans(),
         )
 
+    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, leaving the deque unsorted for the rest of the session.
+        """
+        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()
+        stats: dict[str, object] = {}
         history_by_combo, depth_axis_by_combo = rebuild_bscan_history_from_results(
             result_history=result_history,
-            history_limit=self._bscan_history_limit,
-            floor_collection_id=self._bscan_history_floor_collection_id,
+            history_limit=window,
+            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,
+            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,
+        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.
+
+        Three independent causes produce the same symptom, so each is reported as its
+        own number rather than a single verdict:
+          * `tail` — the run simply produced fewer sweeps than were asked for;
+          * `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"newest-{window} tail={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."""
@@ -351,30 +473,6 @@ class AppWindowBscanPlotMixin:
         self._bscan_depth_axis_by_combo.clear()
         self._bscan_render_signature = None
 
-    def _advance_bscan_floor_to_cpp_window(self) -> None:
-        """Clamp B-scan source history to C++ available replay window."""
-        if not self._result_history:
-            return
-
-        cpp_window_limit = min(
-            int(self._defaults_config.rings.preprocessed.capacity),
-            int(self._defaults_config.rings.results.capacity),
-        )
-        cpp_window_limit = max(1, cpp_window_limit)
-        latest_collection_id = int(self._result_history[-1].collection_id)
-        current_floor = int(self._bscan_history_floor_collection_id)
-
-        # Collection ids restart from 1 on new C++ run; release floor only while
-        # acquisition is running, so manual "remove last" behavior in stopped mode
-        # remains deterministic.
-        if latest_collection_id < current_floor and self._supervisor.is_running():
-            self._bscan_history_floor_collection_id = 0
-            current_floor = 0
-
-        floor_candidate = max(0, latest_collection_id - cpp_window_limit)
-        if floor_candidate > current_floor:
-            self._bscan_history_floor_collection_id = floor_candidate
-
     def _ensure_phase_view_box(self) -> pg.ViewBox:
         """Create or return secondary right-axis ViewBox for phase curves."""
         plot_item = self._bscan_plot.getPlotItem()

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,
+        # so its newest `window` entries are a mix of freshly and stale-processed
+        # frames. 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)

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."""

python_app/gui/controllers/app_window_preprocess_mixin.py

@@ -10,7 +10,10 @@ from python_app.orchestration.preprocess_assets import (
     preprocess_asset_channel,
     preprocess_asset_display_name,
 )
-from python_app.workflows.kamil_adc_neutral_preprocess import build_kamil_adc_neutral_s21_sets
+from python_app.workflows.kamil_adc_neutral_preprocess import (
+    build_neutral_s21_sets,
+    supports_neutral_preprocess_sets,
+)
 from python_app.workflows.multi_radar_capture_workflow import (
     MultiRadarCaptureBatch,
     MultiRadarSequentialCaptureSession,
@@ -216,8 +219,8 @@ class AppWindowPreprocessMixin:
         dialog.undo_last_requested.connect(self._undo_last_capture)
         dialog.finalize_sequence_requested.connect(self._finalize_capture_sequence)
         dialog.abort_sequence_requested.connect(self._abort_capture_sequence)
-        dialog.create_kamil_adc_neutral_sets_requested.connect(self._create_kamil_adc_neutral_sets)
-        dialog.set_kamil_adc_neutral_sets_visible(self._defaults_config.is_kamil_adc)
+        dialog.create_neutral_sets_requested.connect(self._create_neutral_sets)
+        dialog.set_neutral_sets_visible(supports_neutral_preprocess_sets(self._defaults_config))
         dialog.set_radar_config_summary(
             directory_path=self._preprocess_radar_scan_summary.directory_path,
             json_file_count=self._preprocess_radar_scan_summary.json_file_count,
@@ -292,7 +295,7 @@ class AppWindowPreprocessMixin:
             f"{preprocess_asset_display_name(key)}={len(names)}"
             for key, names in available_sets.items()
         )
-        dialog.set_kamil_adc_neutral_sets_visible(self._defaults_config.is_kamil_adc)
+        dialog.set_neutral_sets_visible(supports_neutral_preprocess_sets(self._defaults_config))
         self._log(f"Preprocess set lists refreshed: radar_key={radar_key}, {available_counts}")
         if unavailable_selections:
             self._log_warning(
@@ -391,8 +394,8 @@ class AppWindowPreprocessMixin:
             self._show_exception(f"Failed to start {kind} sequence", exc)
             self._resume_pipeline_if_needed()
 
-    def _create_kamil_adc_neutral_sets(self) -> None:
-        """Save neutral S21 calibration/reference sets for the current Kamil ADC settings."""
+    def _create_neutral_sets(self) -> None:
+        """Save neutral S21 calibration/reference sets for the current radar settings."""
         if self._capture_session is not None:
             self._show_error(
                 "Cannot create neutral sets during active capture sequence",
@@ -409,8 +412,10 @@ class AppWindowPreprocessMixin:
         pipeline_was_paused = False
         try:
             config = self._build_config()
-            if not config.is_kamil_adc:
-                self._show_error("Neutral S21 sets are available only for kamil_adc")
+            if not supports_neutral_preprocess_sets(config):
+                self._show_error(
+                    "Neutral S21 sets are available only for kamil_adc and librevna_multi"
+                )
                 return
 
             radar_key = self._radar_key(config)
@@ -425,12 +430,12 @@ class AppWindowPreprocessMixin:
                 )
 
             if self._supervisor.is_running():
-                self._log("Pipeline paused for Kamil ADC neutral-set creation")
+                self._log("Pipeline paused for neutral-set creation")
                 self._stop_run()
                 pipeline_was_paused = True
 
-            calibration, reference = build_kamil_adc_neutral_s21_sets(config)
-            point_count = config.radar.kamil_adc.band.points
+            calibration, reference = build_neutral_s21_sets(config)
+            point_count = int(calibration.traces[0].frequency_hz.size)
             self._store.save_set("s21_calibration", radar_key, set_name, calibration)
             self._store.save_set("s21_reference", radar_key, set_name, reference)
 
@@ -445,11 +450,11 @@ class AppWindowPreprocessMixin:
                 f"Neutral S21 sets saved: {set_name} ({len(calibration.traces)} combos, {point_count} points)"
             )
             self._log(
-                "Kamil ADC neutral S21 sets saved: "
+                "Neutral S21 sets saved: "
                 f"set={set_name}, radar_key={radar_key}, combos={len(calibration.traces)}, points={point_count}"
             )
         except Exception as exc:  # noqa: BLE001
-            self._show_exception("Failed to create Kamil ADC neutral sets", exc)
+            self._show_exception("Failed to create neutral S21 sets", exc)
         finally:
             if pipeline_was_paused:
                 self._start_run()

python_app/gui/controllers/app_window_snapshot_mixin.py

@@ -241,7 +241,7 @@ class AppWindowSnapshotMixin:
                 retained_pre=retained_pre,
                 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)

python_app/gui/controllers/sections/processing_section.py

@@ -18,6 +18,7 @@ from PyQt6.QtWidgets import (
     QWidget,
 )
 
+from python_app.gui.controllers.app_window_config.state_builders import GUI_SAVE_HISTORY_LIMIT
 from python_app.gui.controllers.sections.layout_helpers import FormRow, build_two_column_form_widget
 
 
@@ -160,6 +161,15 @@ def build_processing_group(owner) -> QGroupBox:
     owner._bscan_subtract_mean_ascan = QCheckBox("Subtract mean A-scan")
     owner._bscan_subtract_mean_ascan.setChecked(bool(bscan_defaults.subtract_mean_ascan))
 
+    owner._bscan_history_window = QSpinBox()
+    owner._bscan_history_window.setMinimum(1)
+    owner._bscan_history_window.setMaximum(GUI_SAVE_HISTORY_LIMIT)
+    owner._bscan_history_window.setValue(int(bscan_defaults.history_window_scans))
+    owner._bscan_history_window.setToolTip(
+        "How many past sweeps the B-scan shows once acquisition is stopped. While "
+        "running, the window stays clamped to the C++ ring capacity."
+    )
+
     bscan_page = _build_processing_mode_page(
         owner._processing_mode_pages,
         [
@@ -169,6 +179,7 @@ def build_processing_group(owner) -> QGroupBox:
             ("Gain", owner._bscan_gain),
             ("Start MHz", owner._bscan_start_freq_mhz),
             ("Stop MHz", owner._bscan_stop_freq_mhz),
+            ("Scans to show (stopped)", owner._bscan_history_window),
             owner._bscan_subtract_mean_ascan,
         ],
         split_index=4,
@@ -579,6 +590,7 @@ def build_processing_group(owner) -> QGroupBox:
     owner._bscan_start_freq_mhz.valueChanged.connect(owner._on_processing_live_settings_changed)
     owner._bscan_stop_freq_mhz.valueChanged.connect(owner._on_processing_live_settings_changed)
     owner._bscan_subtract_mean_ascan.toggled.connect(owner._on_processing_live_settings_changed)
+    owner._bscan_history_window.valueChanged.connect(owner._on_processing_live_settings_changed)
     owner._gpr_input_positions_input.editingFinished.connect(owner._on_processing_live_settings_changed)
     owner._gpr_output_positions_input.editingFinished.connect(owner._on_processing_live_settings_changed)
     owner._gpr_min_depth_m.valueChanged.connect(owner._on_processing_live_settings_changed)

python_app/gui/preprocess_dialog.py

@@ -46,7 +46,7 @@ class PreprocessDialog(QDialog):
     undo_last_requested = pyqtSignal()
     finalize_sequence_requested = pyqtSignal()
     abort_sequence_requested = pyqtSignal()
-    create_kamil_adc_neutral_sets_requested = pyqtSignal()
+    create_neutral_sets_requested = pyqtSignal()
 
     def __init__(self, parent=None) -> None:
         """Initialize window metadata and compose dialog UI."""
@@ -92,17 +92,18 @@ class PreprocessDialog(QDialog):
         self._set_name_input = QLineEdit("set_001", group)
         refresh_button = QPushButton("Refresh Sets", group)
         refresh_button.clicked.connect(self.refresh_requested.emit)
-        self._kamil_adc_neutral_sets_button = QPushButton("Create Neutral S21 Sets", group)
-        self._kamil_adc_neutral_sets_button.setToolTip(
-            "Save S21 calibration=1 and S21 reference=0 for the current Kamil ADC settings."
+        self._neutral_sets_button = QPushButton("Create Neutral S21 Sets", group)
+        self._neutral_sets_button.setToolTip(
+            "Save S21 calibration=1 and S21 reference=0 for the current radar settings, "
+            "so the pipeline can run before any real calibration exists."
         )
-        self._kamil_adc_neutral_sets_button.clicked.connect(
-            self.create_kamil_adc_neutral_sets_requested.emit
+        self._neutral_sets_button.clicked.connect(
+            self.create_neutral_sets_requested.emit
         )
-        self._kamil_adc_neutral_sets_button.setVisible(False)
+        self._neutral_sets_button.setVisible(False)
         header_row.addWidget(QLabel("Set name"))
         header_row.addWidget(self._set_name_input, stretch=1)
-        header_row.addWidget(self._kamil_adc_neutral_sets_button)
+        header_row.addWidget(self._neutral_sets_button)
         header_row.addWidget(refresh_button)
         layout.addLayout(header_row)
         layout.addLayout(self._build_median_sweep_row(group))
@@ -413,10 +414,10 @@ class PreprocessDialog(QDialog):
         """Set short human-readable status line."""
         self._status_label.setText(message)
 
-    def set_kamil_adc_neutral_sets_visible(self, visible: bool) -> None:
-        """Show Kamil ADC neutral-set shortcut only in the matching radar mode."""
-        self._kamil_adc_neutral_sets_button.setVisible(bool(visible))
-        self._kamil_adc_neutral_sets_button.setEnabled(bool(visible))
+    def set_neutral_sets_visible(self, visible: bool) -> None:
+        """Show the neutral-set shortcut only for radar models that support it."""
+        self._neutral_sets_button.setVisible(bool(visible))
+        self._neutral_sets_button.setEnabled(bool(visible))
 
     def reset_preview(self) -> None:
         """Clear preview surfaces and restore default empty-state text when possible."""

python_app/hardware_full/librevna_multi_device_driver/controller.py

@@ -131,7 +131,14 @@ class MultiDeviceVnaController:
         if not self._reference_configuration_applied:
             self._configure_reference_clocks()
 
-        self._drain_all_received_packets()
+        drain_started_seconds = time.monotonic()
+        drained_packet_count = self._drain_all_received_packets()
+        logger.debug(
+            "timing: drain discarded %d stale packet(s) in %.2f ms (t=%.1f ms)",
+            drained_packet_count,
+            (time.monotonic() - drain_started_seconds) * 1e3,
+            time.monotonic() * 1e3,
+        )
 
         if (
             self._sweep_is_running
@@ -334,12 +341,16 @@ class MultiDeviceVnaController:
         self._sweep_is_running = True
         logger.debug("Sweep settings applied to all devices; sweep running")
 
-    def _drain_all_received_packets(self) -> None:
+    def _drain_all_received_packets(self) -> int:
         """Empty every device's received-packet queue, in parallel for 2+ devices.
 
         Concurrent draining keeps cross-device timing skew small so a hardware
         cycle wrap cannot slip between per-device drains and desynchronize the
         cycle counters.
+
+        Returns the total number of discarded packets, which the caller logs: a large
+        count means the host was far behind the free-running stream, a near-zero count
+        means the drain landed right after a sweep boundary.
         """
         # Drain every device queue in parallel rather than one after another:
         # serial drain leaves up to a few hundred microseconds of skew between
@@ -349,22 +360,31 @@ class MultiDeviceVnaController:
         # so concurrent get_nowait calls do not contend. A single device case
         # just runs inline to avoid the thread-spawn overhead.
         if len(self._all_devices) < 2:
-            for device_connection in self._all_devices:
-                device_connection.drain_received_packets()
-            return
+            return sum(
+                len(device_connection.drain_received_packets())
+                for device_connection in self._all_devices
+            )
+
+        drained_counts = [0] * len(self._all_devices)
+
+        def drain_one_device(device_index: int, device_connection: LibreVnaUsbBulkConnection) -> None:
+            """Drain one device's queue and record how many packets it held."""
+            drained_counts[device_index] = len(device_connection.drain_received_packets())
 
         drain_threads = [
             threading.Thread(
-                target=device_connection.drain_received_packets,
+                target=drain_one_device,
+                args=(device_index, device_connection),
                 name=f"drain-{device_connection.serial_number}",
                 daemon=True,
             )
-            for device_connection in self._all_devices
+            for device_index, device_connection in enumerate(self._all_devices)
         ]
         for drain_thread in drain_threads:
             drain_thread.start()
         for drain_thread in drain_threads:
             drain_thread.join()
+        return sum(drained_counts)
 
     @staticmethod
     def _normalize_master_stimulus_ports(master_stimulus_ports: Sequence[int]) -> tuple[int, ...]:

python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py

@@ -441,6 +441,7 @@ def collect_complete_running_sweep_cycles(
     def build_cycle_tracking_handler(
         cycle_aware_handler: Callable[[ParsedVnaDatapoint, int], None],
         device_state: _DeviceCollectionState,
+        device_label: str = "device",
     ) -> Callable[[ParsedVnaDatapoint], bool]:
         """Wrap a cycle-aware handler with cross-device cycle tracking.
 
@@ -462,6 +463,11 @@ def collect_complete_running_sweep_cycles(
         cycle_tracking_state = {
             "current_cycle_index": 0,
             "synchronized": False,
+            # How many mid-sweep points were thrown away before the anchor was found.
+            # Near zero means the drain landed on a sweep boundary — the case where a
+            # stale point 0 could still have been in flight; a large count means the
+            # remainder of the in-progress sweep was safely skipped.
+            "pre_anchor_skipped": 0,
         }
 
         def handle_datapoint(parsed_datapoint: ParsedVnaDatapoint) -> bool:
@@ -475,6 +481,7 @@ def collect_complete_running_sweep_cycles(
 
             if not cycle_tracking_state["synchronized"]:
                 if current_point_index != 0:
+                    cycle_tracking_state["pre_anchor_skipped"] += 1
                     return False
                 # Candidate cycle 0. Commit it only once every device confirms it
                 # observed point 0 of the SAME physical sweep; otherwise reject the
@@ -484,6 +491,14 @@ def collect_complete_running_sweep_cycles(
                     report_cycle_misalignment()
                     return False
                 cycle_tracking_state["synchronized"] = True
+                logger.debug(
+                    "timing: %s anchored cycle 0 after skipping %d mid-sweep point(s) of %d "
+                    "(t=%.1f ms)",
+                    device_label,
+                    cycle_tracking_state["pre_anchor_skipped"],
+                    point_count,
+                    time.monotonic() * 1e3,
+                )
                 cycle_aware_handler(parsed_datapoint, 0)
                 return True
 
@@ -493,6 +508,12 @@ def collect_complete_running_sweep_cycles(
             # spurious wrap and desynchronize the cycle counter.
             if current_point_index == 0:
                 cycle_tracking_state["current_cycle_index"] += 1
+                logger.debug(
+                    "timing: %s first point of NEXT sweep arrived (cycle -> %d, t=%.1f ms)",
+                    device_label,
+                    cycle_tracking_state["current_cycle_index"],
+                    time.monotonic() * 1e3,
+                )
             current_cycle_index = cycle_tracking_state["current_cycle_index"]
             if current_cycle_index >= cycle_count:
                 # The sweep just wrapped past the final requested cycle, closing its
@@ -505,6 +526,14 @@ def collect_complete_running_sweep_cycles(
                 return False
 
             cycle_aware_handler(parsed_datapoint, current_cycle_index)
+            if current_point_index == point_count - 1:
+                logger.debug(
+                    "timing: %s last point of cycle %d arrived (index=%d, t=%.1f ms)",
+                    device_label,
+                    current_cycle_index,
+                    current_point_index,
+                    time.monotonic() * 1e3,
+                )
             return True
 
         return handle_datapoint
@@ -587,7 +616,9 @@ def collect_complete_running_sweep_cycles(
                             point_index,
                         ] = port_receiver_value
 
-        return build_cycle_tracking_handler(handle_slave_datapoint, device_state)
+        return build_cycle_tracking_handler(
+            handle_slave_datapoint, device_state, device_label=f"slave{slave_index}"
+        )
 
     master_device_state = _DeviceCollectionState()
     collection_threads = [
@@ -595,7 +626,9 @@ def collect_complete_running_sweep_cycles(
             target=collect_datapoints_from_device,
             args=(
                 master_device_connection,
-                build_cycle_tracking_handler(handle_master_datapoint, master_device_state),
+                build_cycle_tracking_handler(
+                    handle_master_datapoint, master_device_state, device_label="master"
+                ),
                 master_device_state,
             ),
             daemon=True,

python_app/hardware_full/matrix_radar_service.py

@@ -46,13 +46,31 @@ def create_matrix_radar_service(config: RunConfigModel) -> MatrixRadarService:
     if model == RunConfigModel.LIBREVNA_MULTI_MODEL:
         from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService
 
-        return MultiDeviceLibreVnaService(
+        inner = MultiDeviceLibreVnaService(
             master_serial=config.radar.serial,
             slave_serials=list(config.radar.multi_device.slave_serials),
             force_external_reference=config.radar.multi_device.force_external_reference,
             recovery_attempts=config.radar.multi_device.recovery_attempts,
             backend_mode=config.radar.driver_mode,
         )
+        out_physical = config.matrix_output_switch_positions
+        in_physical = config.matrix_input_switch_positions
+        if out_physical <= 1 and in_physical <= 1:
+            return inner
+
+        from python_app.hardware_full.switched_matrix_radar_service import (
+            SwitchedMatrixRadarService,
+            build_physical_switch,
+        )
+
+        return SwitchedMatrixRadarService(
+            inner=inner,
+            output_switch=build_physical_switch(config.output_switch, out_physical, config.radar.driver_mode),
+            input_switch=build_physical_switch(config.input_switch, in_physical, config.radar.driver_mode),
+            inner_output_positions=RunConfigModel.MULTI_DEVICE_OUTPUT_POSITIONS,
+            inner_input_positions=RunConfigModel.MULTI_DEVICE_INPUT_POSITIONS,
+            settling_ms=config.runtime.settling_ms,
+        )
 
     if model == RunConfigModel.SN9000_MODEL:
         if config.radar.driver_mode != "native":

python_app/hardware_full/switch_service.py

@@ -66,6 +66,10 @@ class SwitchService:
         """Switch to requested position."""
         self._driver.switch_to(position)
 
+    def position_count(self) -> int:
+        """Return number of positions supported by the backend driver."""
+        return self._driver.position_count()
+
     @property
     def current_position(self) -> int:
         """Return current switch position reported by backend driver."""

python_app/hardware_full/switched_matrix_radar_service.py

@@ -0,0 +1,154 @@
+"""Matrix radar behind real GPIO switches on the stimulus and/or receiver path."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field, replace
+import logging
+import time
+
+from python_app.hardware_full.matrix_radar_service import MatrixRadarService
+from python_app.hardware_full.switch_service import SwitchService
+from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData
+from python_app.models.run_config_model import RadarSweepModel, SwitchModel
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(slots=True)
+class SwitchedMatrixRadarService:
+    """Widen a matrix radar's combo matrix with real switch positions.
+
+    Implements the ``MatrixRadarService`` protocol, so the producer and the
+    capture workflows treat it as an ordinary matrix radar that simply reports
+    more positions. The hardware sweep is never stopped: switches are only ever
+    driven BETWEEN ``acquire_collection`` calls, and the inner service's
+    free-running collection discards any partially swept cycle.
+    """
+
+    inner: MatrixRadarService
+    output_switch: SwitchService | None
+    input_switch: SwitchService | None
+    inner_output_positions: int
+    inner_input_positions: int
+    settling_ms: int = 0
+    # Monotonic end of the previous inner collection, so the DEBUG timing trace can
+    # report how long the gap between "sweep collected" and "switch driven" really is
+    # — that gap is where a stale in-flight point 0 can still slip past the drain.
+    _last_inner_end_ns: int = field(init=False, default=0, repr=False)
+
+    def open(self) -> None:
+        """Open the inner radar and both switches."""
+        self.inner.open()
+        if self.output_switch is not None:
+            self.output_switch.open()
+        if self.input_switch is not None:
+            self.input_switch.open()
+
+    def close(self) -> None:
+        """Close switches first, then the inner radar; never raises."""
+        for switch in (self.input_switch, self.output_switch):
+            if switch is not None:
+                try:
+                    switch.close()
+                except Exception as exc:  # noqa: BLE001 — shutdown path
+                    logger.warning("Switch close ignored error: %s", exc)
+        self.inner.close()
+
+    def configure(self, sweep: RadarSweepModel) -> None:
+        """Apply sweep settings to the inner radar."""
+        self.inner.configure(sweep)
+
+    def recover(self) -> None:
+        """Reconnect the inner radar; switches are not on the USB transport."""
+        self.inner.recover()
+
+    def acquire_collection(self, collection_id: int = 1) -> SweepCollection:
+        """Acquire the full widened matrix, one inner collection per switch step.
+
+        A partial failure raises instead of returning a short collection: the
+        preprocessor requires every runtime combo to be present, so half a matrix
+        is worse than a dropped frame.
+        """
+        capture_start_ns = time.monotonic_ns()
+        out_steps = self.output_switch.position_count() if self.output_switch is not None else 1
+        in_steps = self.input_switch.position_count() if self.input_switch is not None else 1
+        total_inputs = in_steps * self.inner_input_positions
+        total_outputs = out_steps * self.inner_output_positions
+
+        # Place each trace at its canonical index rather than appending. The GPR stage
+        # rejects a collection whose trace order differs from run.combos, and run.combos
+        # is built output-major (`build_full_combos`) while these loops run switch-major.
+        # Appending happens to agree for an output switch and to disagree for an input one.
+        slots: list[TraceData | None] = [None] * (total_inputs * total_outputs)
+
+        for out_k in range(out_steps):
+            for in_k in range(in_steps):
+                step_start_ns = time.monotonic_ns()
+                if self.output_switch is not None:
+                    self.output_switch.switch_to(out_k)
+                if self.input_switch is not None:
+                    self.input_switch.switch_to(in_k)
+                switched_ns = time.monotonic_ns()
+                # Settle AFTER the last switch change and BEFORE collecting, so the
+                # cycle we anchor on starts with the RF path already stable.
+                if self.settling_ms > 0:
+                    time.sleep(self.settling_ms / 1000.0)
+                settled_ns = time.monotonic_ns()
+
+                sub = self.inner.acquire_collection(collection_id)
+                inner_end_ns = time.monotonic_ns()
+                logger.debug(
+                    "timing: collection %d step out=%d in=%d | gap_prev_collect_to_switch=%s ms, "
+                    "switch=%.3f ms, settle=%.2f ms, inner_collect=%.2f ms",
+                    collection_id,
+                    out_k,
+                    in_k,
+                    (
+                        f"{(step_start_ns - self._last_inner_end_ns) / 1e6:.2f}"
+                        if self._last_inner_end_ns
+                        else "n/a"
+                    ),
+                    (switched_ns - step_start_ns) / 1e6,
+                    (settled_ns - switched_ns) / 1e6,
+                    (inner_end_ns - settled_ns) / 1e6,
+                )
+                self._last_inner_end_ns = inner_end_ns
+                for trace in sub.traces:
+                    input_pos = in_k * self.inner_input_positions + int(trace.combo.input)
+                    output_pos = out_k * self.inner_output_positions + int(trace.combo.output)
+                    slots[output_pos * total_inputs + input_pos] = replace(
+                        trace, combo=ComboKey(input=input_pos, output=output_pos)
+                    )
+
+        if any(trace is None for trace in slots):
+            missing = sum(1 for trace in slots if trace is None)
+            raise RuntimeError(
+                f"Switched matrix collection is incomplete: {missing} of {len(slots)} combos missing"
+            )
+
+        return SweepCollection(
+            collection_id=int(collection_id),
+            monotonic_ns=time.monotonic_ns(),
+            traces=[trace for trace in slots if trace is not None],
+            capture_start_ns=capture_start_ns,
+            capture_end_ns=time.monotonic_ns(),
+        )
+
+def build_physical_switch(
+    model: SwitchModel,
+    physical_positions: int,
+    radar_driver_mode: str,
+) -> SwitchService | None:
+    """Build the driver for a real switch described by a virtual switch section.
+
+    The config section carries the LOGICAL axis size and a forced "mock" mode so
+    the C++ loader accepts it; the real driver needs the PHYSICAL position count
+    and native mode. Mock radar runs keep mock switches so the whole path can be
+    exercised without GPIO.
+    """
+    if physical_positions <= 1:
+        return None
+    driver_mode = "mock" if radar_driver_mode.strip().lower() == "mock" else "native"
+    return SwitchService.from_model(
+        replace(model, positions=physical_positions, driver_mode=driver_mode)
+    )

python_app/models/gui_profile_codec.py

@@ -244,6 +244,12 @@ def gui_profile_from_dict(payload: dict[str, Any]) -> GuiProfileModel:
                 gui.processing.bscan.subtract_mean_ascan,
                 "gui.processing.bscan",
             ),
+            history_window_scans=_optional_int(
+                bscan_object,
+                "history_window_scans",
+                gui.processing.bscan.history_window_scans,
+                "gui.processing.bscan",
+            ),
         ),
         gpr=GuiGprStateModel(
             input_positions=_optional_string(
@@ -579,6 +585,7 @@ def gui_profile_to_dict(model: GuiProfileModel) -> dict[str, Any]:
                 "start_freq_mhz": gui.processing.bscan.start_freq_mhz,
                 "stop_freq_mhz": gui.processing.bscan.stop_freq_mhz,
                 "subtract_mean_ascan": gui.processing.bscan.subtract_mean_ascan,
+                "history_window_scans": gui.processing.bscan.history_window_scans,
             },
             "gpr": {
                 "input_positions": gui.processing.gpr.input_positions,

python_app/models/gui_profile_schema.py

@@ -48,6 +48,10 @@ class GuiBscanStateModel:
     start_freq_mhz: float = 100.0
     stop_freq_mhz: float = 8800.0
     subtract_mean_ascan: bool = False
+    # How many past sweeps the B-scan heatmap renders once acquisition is stopped.
+    # While running the window stays at the C++ replay window (see
+    # `_cpp_bscan_replay_window_for_config`); this only widens the stopped-mode view.
+    history_window_scans: int = 50
 
 
 @dataclass(slots=True)

python_app/models/run_config_codec.py

@@ -219,6 +219,16 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel:
         "recovery_attempts",
         model.radar.multi_device.recovery_attempts,
     )
+    model.radar.multi_device.output_switch_positions = _read_int(
+       multi_device_payload,
+       "output_switch_positions",
+       model.radar.multi_device.output_switch_positions,
+    )
+    model.radar.multi_device.input_switch_positions = _read_int(
+           multi_device_payload,
+           "input_switch_positions",
+           model.radar.multi_device.input_switch_positions,
+        )
     model.radar.kamil_adc.project_dir = _read_str(
         kamil_adc_payload, "project_dir", model.radar.kamil_adc.project_dir
     )
@@ -498,6 +508,8 @@ def run_config_to_dict(model: RunConfigModel) -> dict[str, Any]:
                 "slave_serials": list(model.radar.multi_device.slave_serials),
                 "force_external_reference": model.radar.multi_device.force_external_reference,
                 "recovery_attempts": model.radar.multi_device.recovery_attempts,
+                "output_switch_positions": model.radar.multi_device.output_switch_positions,
+                "input_switch_positions": model.radar.multi_device.input_switch_positions
             },
             "kamil_adc": {
                 "project_dir": model.radar.kamil_adc.project_dir,

python_app/models/run_config_schema.py

@@ -40,6 +40,8 @@ class RadarMultiDeviceModel:
     slave_serials: list[str] = field(default_factory=list)
     force_external_reference: bool = True
     recovery_attempts: int = 3
+    output_switch_positions: int = 1   # 1 = свитча нет
+    input_switch_positions: int = 1    # 1 = свитча нет
 
 
 @dataclass(slots=True)
@@ -390,6 +392,24 @@ class RunConfigModel:
         """Return whether this config acquires the full virtual switch matrix per sweep."""
         return self.is_multi_device or self.is_sn9000
 
+    @property
+    def matrix_output_switch_positions(self) -> int:
+        """Physical positions of the real switch on the master stimulus path."""
+        if not self.is_multi_device:
+            return 1
+        return max(1, int(self.radar.multi_device.output_switch_positions))
+
+    @property
+    def matrix_input_switch_positions(self) -> int:
+        """Physical positions of the real switch on the slave receiver path."""
+        if not self.is_multi_device:
+            return 1
+        return max(1, int(self.radar.multi_device.input_switch_positions))
+
+    def build_runtime_combos(self) -> list[ComboModel]:
+        """Build the combo matrix from the effective switch axis sizes."""
+        return self.build_full_combos(self.input_switch.positions, self.output_switch.positions)
+ 
     @property
     def is_kamil_adc(self) -> bool:
         """Return whether this config targets the external Kamil ADC acquisition path."""
@@ -446,22 +466,25 @@ class RunConfigModel:
         if not self.is_matrix_radar:
             return
         self._apply_matrix_virtual_switches()
-        self.combos = self.build_matrix_radar_virtual_combos()
+        self.combos = self.build_runtime_combos()
 
     def _apply_matrix_virtual_switches(self) -> None:
-        """Pin the canonical 2x4 virtual switch matrix used by all matrix-mode radars."""
+        """Pin the virtual switch matrix, widened by any real switch on the path."""
+        out_physical = self.matrix_output_switch_positions
+        in_physical = self.matrix_input_switch_positions
+
         self.output_switch.name = self.output_switch.name or "virtual_output"
         self.output_switch.driver_mode = "mock"
         self.output_switch.driver = self.output_switch.driver or "h7992"
         self.output_switch.radar_port = 1
-        self.output_switch.positions = self.MULTI_DEVICE_OUTPUT_POSITIONS
+        self.output_switch.positions = out_physical * self.MULTI_DEVICE_OUTPUT_POSITIONS
         self.output_switch.default_position = 0
 
         self.input_switch.name = self.input_switch.name or "virtual_input"
         self.input_switch.driver_mode = "mock"
         self.input_switch.driver = self.input_switch.driver or "h7992"
         self.input_switch.radar_port = 2
-        self.input_switch.positions = self.MULTI_DEVICE_INPUT_POSITIONS
+        self.input_switch.positions = in_physical * self.MULTI_DEVICE_INPUT_POSITIONS
         self.input_switch.default_position = 0
 
     def ensure_combos(self) -> None:

python_app/orchestration/process_supervisor.py

@@ -105,7 +105,10 @@ def _read_log_tail(path: Path, max_bytes: int = 16384) -> str:
             data = handle.read()
     except OSError:
         return ""
-    return data.decode("utf-8", errors="replace").strip()
+    # Drop NULs: logs written by an older supervisor can carry a sparse hole from
+    # the pre-O_APPEND truncate bug, and a tail landing in it would otherwise turn
+    # an exit report (or a rolled `.prev`) into megabytes of NUL padding.
+    return data.replace(b"\0", b"").decode("utf-8", errors="replace").strip()
 
 
 class ProcessSupervisor:
@@ -232,8 +235,17 @@ class ProcessSupervisor:
         self._roll_log_to_prev(stdout_path)
         self._roll_log_to_prev(stderr_path)
 
-        stdout_file = open(stdout_path, "wb")
-        stderr_file = open(stderr_path, "wb")
+        # O_APPEND ("ab"), not "wb": the child inherits these fds and keeps its own
+        # file offset. Without O_APPEND, the in-place truncate in
+        # `_roll_log_if_oversized` leaves that offset far past the new end of file,
+        # so the next write lands there and the kernel fills everything before it
+        # with a hole of NUL bytes — the log becomes unreadable and the size cap
+        # stops working entirely. O_APPEND makes the kernel seek to EOF atomically
+        # on every write, so a truncate genuinely restarts the file at offset 0.
+        # `_roll_log_to_prev` above already renamed any previous log away, so not
+        # truncating on open costs nothing.
+        stdout_file = open(stdout_path, "ab")
+        stderr_file = open(stderr_path, "ab")
         try:
             handle = subprocess.Popen(
                 command,
@@ -499,6 +511,11 @@ class ProcessSupervisor:
         The child holds an open fd to this inode, so a rename would not redirect
         its writes. Instead keep one rolled generation via copy-to-`.prev` and
         truncate the live inode in place, freeing the allocated disk blocks.
+
+        This relies on the child's fd being opened with O_APPEND (see `_spawn`):
+        only then does the child resume writing at offset 0 after the truncate.
+        With a plain write fd it would keep writing at its stale offset, punching
+        a multi-hundred-megabyte NUL hole and defeating the cap.
         """
         try:
             if path.stat().st_size <= _LOG_MAX_BYTES:

python_app/scripts/matrix_raw_producer.py

@@ -11,6 +11,7 @@ import threading
 import time
 
 from python_app.hardware_full.matrix_radar_service import MatrixRadarService, create_matrix_radar_service
+from python_app.logging_setup import coerce_level
 from python_app.models.run_config_model import RunConfigModel
 from python_app.orchestration.shm import ShmRingWriter
 from python_app.storage.npz.serialize import RAW_MAGIC, serialize_trace_collection
@@ -95,6 +96,10 @@ def main() -> int:
 
     config = RunConfigModel.load_from_path(args.config)
     config.apply_device_model_constraints()
+    # Honor the configured verbosity so the DEBUG switch/sweep timing trace can be
+    # turned on from the profile instead of requiring a code edit. basicConfig above
+    # only installed the handler; the package logger owns the level.
+    logging.getLogger("python_app").setLevel(coerce_level(config.logging.level))
     if not config.is_matrix_radar:
         raise RuntimeError(
             "matrix_raw_producer requires a matrix-mode radar.model "

python_app/tests/test_bscan_history_window.py

@@ -0,0 +1,171 @@
+"""Configurable B-scan display window.
+
+The B-scan used to be pinned to the C++ replay window (~50 sweeps) by two separate
+mechanisms: the render-side history limit and a monotonically rising
+`collection_id` floor. Widening only the first would have changed nothing, because
+the floor kept filtering older collections out for good.
+
+The floor is gone: selection is positional, which is the only criterion that holds
+when ids are sparse (the results ring drops) or restart from 1 (a new C++ run).
+These tests pin the observable consequences — how many columns end up on screen —
+rather than any internal counter.
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+from PyQt6.QtWidgets import QApplication  # noqa: E402
+
+from python_app.gui.app_window import AppWindow  # noqa: E402
+from python_app.gui.runtime.history import record_result_history  # noqa: E402
+from python_app.models.dataset_model import (  # noqa: E402
+    ComboKey,
+    ResultBlock,
+    ResultCollection,
+    ResultPayload,
+)
+
+_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 _bscan_result(collection_id: int) -> ResultCollection:
+    payload = ResultPayload(
+        processing_name="bscan",
+        kind=1,
+        frequency_hz=np.array([0.5, 1.0], dtype=np.float32),
+        trace=np.array([collection_id + 0j, collection_id + 0j], dtype=np.complex64),
+    )
+    block = ResultBlock(combo=ComboKey(input=0, output=0), payloads=[payload])
+    return ResultCollection(collection_id=collection_id, monotonic_ns=collection_id, blocks=[block])
+
+
+class BscanDisplayWindowTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.w = _window
+        self._original_is_running = self.w._supervisor.is_running
+        self.w._result_history.clear()
+        self.w._bscan_render_signature = None
+
+    def tearDown(self) -> None:
+        self.w._supervisor.is_running = self._original_is_running
+        self.w._result_history.clear()
+        self.w._bscan_render_signature = None
+
+    def _set_running(self, running: bool) -> None:
+        self.w._supervisor.is_running = lambda: running
+
+    def _fill_history(self, count: int, *, id_step: int = 1) -> None:
+        for index in range(count):
+            self.w._result_history.append(_bscan_result(1 + index * id_step))
+
+    def test_running_acquisition_ignores_the_user_window(self) -> None:
+        # A live rebuild runs on every incoming result, so the live path stays pinned
+        # to the replay window no matter what the operator typed for stopped review.
+        self._set_running(True)
+        self.w._bscan_history_window.setValue(300)
+        self.assertEqual(self.w._bscan_display_window_scans(), self.w._bscan_cpp_replay_window)
+
+    def test_stopped_acquisition_uses_the_user_window(self) -> None:
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(300)
+        self.assertEqual(self.w._bscan_display_window_scans(), 300)
+
+    def test_widening_the_window_brings_older_frames_back(self) -> None:
+        # The regression this whole change exists for: a live run renders 50 columns,
+        # and widening the window after Stop must reach back over the retained history
+        # rather than stay pinned to whatever the live path last drew.
+        self._fill_history(300)
+        self._set_running(True)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(
+            len(self.w._bscan_history_by_combo[(0, 0)]), self.w._bscan_cpp_replay_window
+        )
+
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(300)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 300)
+
+    def test_restarted_collection_ids_do_not_hide_the_fresh_run(self) -> None:
+        # Regression: Start after a stopped review made the image melt to 0 columns and
+        # then snap back to 50. A new C++ run numbers from 1, so entries that are newest
+        # by position carry the smallest ids; any id-based cut-off derived from the old
+        # run rejected exactly them.
+        self._fill_history(300)
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(300)
+        self.w._sync_bscan_history_from_results()
+
+        self._set_running(True)
+        window = self.w._bscan_cpp_replay_window
+        for fresh in range(1, window + 1):
+            self.w._result_history.append(_bscan_result(fresh))
+            self.w._sync_bscan_history_from_results()
+            self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), window)
+
+    def test_window_counts_entries_not_collection_ids(self) -> None:
+        # The results ring overwrites unread slots when the producer outruns the GUI
+        # poll loop, so retained collection ids are sparse. Counting in ids rather than
+        # in entries is exactly the case where asking for 150 sweeps rendered only 87.
+        self._fill_history(300, id_step=3)
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(150)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 150)
+
+    def test_reprocessed_results_replace_rather_than_duplicate_columns(self) -> None:
+        # The processor's recompute is triggered by feeding sweeps back through it, and
+        # it republishes them under their ORIGINAL ids. Two independent mechanisms keep
+        # that from doubling every column, and neither may be confused with the removed
+        # `collection_id` floor: a threshold cannot tell a duplicate from its original,
+        # since they share the id. Deduplication is by key equality.
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(300)
+        self._fill_history(200)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 200)
+
+        # Intake: a recomputed collection replaces the entry holding the same key.
+        for collection in list(self.w._result_history):
+            record_result_history(self.w._result_history, _bscan_result(collection.collection_id))
+        self.assertEqual(len(self.w._result_history), 200)
+
+        # Render: a duplicate that reached the deque by another path is still collapsed,
+        # keeping the newer of the two.
+        self.w._result_history.append(_bscan_result(200))
+        self.w._bscan_render_signature = None
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 200)
+
+    def test_rebuild_renders_the_full_widened_window(self) -> None:
+        self._fill_history(300)
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(300)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 300)
+
+        self.w._bscan_history_window.setValue(50)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 50)
+
+
+if __name__ == "__main__":
+    unittest.main()

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, leaving its
+        newest `window` entries a mix of freshly and stale-processed frames. 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()

python_app/tests/test_processing.py

@@ -219,7 +219,7 @@ class RebuildBscanHistoryTest(unittest.TestCase):
             self._bscan_collection(1, (0, 0), [1.0, 2.0], [10.0, 20.0]),
             self._bscan_collection(2, (0, 0), [1.0, 2.0], [11.0, 21.0]),
         ]
-        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=0)
+        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10)
         self.assertEqual(len(by_combo[(0, 0)]), 2)
         self.assertTrue(np.array_equal(axes[(0, 0)], np.array([1.0, 2.0], dtype=np.float32)))
 
@@ -229,7 +229,7 @@ class RebuildBscanHistoryTest(unittest.TestCase):
             self._bscan_collection(2, (0, 0), [1.0, 2.0], [1.0, 2.0], kind=2),         # wrong kind
             self._bscan_collection(3, (0, 0), [1.0, 2.0, 3.0], [1.0, 2.0]),            # size mismatch
         ]
-        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=0)
+        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=10)
         self.assertEqual(by_combo, {})
 
     def test_depth_axis_change_resets_history(self) -> None:
@@ -237,17 +237,21 @@ class RebuildBscanHistoryTest(unittest.TestCase):
             self._bscan_collection(1, (0, 0), [1.0, 2.0], [10.0, 20.0]),
             self._bscan_collection(2, (0, 0), [1.0, 2.0, 3.0], [11.0, 21.0, 31.0]),  # new depth axis
         ]
-        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=0)
+        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10)
         self.assertEqual(len(by_combo[(0, 0)]), 1)  # reset on axis change; only the latest sweep remains
         self.assertEqual(axes[(0, 0)].shape, (3,))
 
-    def test_floor_collection_id_excludes_older(self) -> None:
+    def test_selection_is_positional_not_by_collection_id(self) -> None:
+        # Ids are neither dense nor monotonic across a run boundary, so the tail is
+        # taken by position only. Here the newest two entries carry the SMALLEST ids;
+        # an id-based cut-off would have dropped exactly them.
         history = [
-            self._bscan_collection(1, (0, 0), [1.0], [10.0]),
-            self._bscan_collection(2, (0, 0), [1.0], [20.0]),
+            self._bscan_collection(cid, (0, 0), [1.0], [float(cid)])
+            for cid in (98, 99, 100, 1, 2)
         ]
-        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=1)
-        self.assertEqual(len(by_combo[(0, 0)]), 1)  # only collection_id > 1
+        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=3)
+        self.assertEqual(len(by_combo[(0, 0)]), 3)
+        self.assertEqual([sweep[0] for sweep in by_combo[(0, 0)]], [100.0, 1.0, 2.0])
 
 
 # --------------------------------------------------------------------------- #

python_app/workflows/kamil_adc_neutral_preprocess.py

@@ -1,4 +1,15 @@
-"""Neutral preprocessing-set helpers for Kamil ADC acquisition."""
+"""Neutral preprocessing-set helpers — the "run without calibration" path.
+
+A neutral pair is a calibration set carrying unit S21 (1+0j) and a reference set
+carrying zero S21. The C++ through-calibrator divides measured/calibration and the
+reference is subtracted, so applying both leaves the measured S21 untouched. That
+lets an operator start the pipeline before any real calibration exists, which the
+required-asset check in `_start_run` would otherwise refuse.
+
+Supported models: Kamil ADC (axis from the ADC processing grid) and every
+VNA-style model, including synchronized multi-device LibreVNA (axis from the
+configured linear sweep grid).
+"""
 
 from __future__ import annotations
 
@@ -17,31 +28,65 @@ from python_app.models.run_config_model import ComboModel, RunConfigModel
 logger = logging.getLogger(__name__)
 
 
-def build_kamil_adc_neutral_s21_sets(
+def supports_neutral_preprocess_sets(config: RunConfigModel) -> bool:
+    """Return whether neutral S21 sets can be generated for this radar model.
+
+    Enabled for the Kamil ADC and for synchronized multi-device LibreVNA, the two
+    models whose emitted frequency axis is fully derivable from the config alone.
+    Other models still work through `build_neutral_s21_sets`, but are kept out of the
+    UI shortcut until their axis has been verified against real hardware.
+    """
+    return bool(config.is_kamil_adc or config.is_multi_device)
+
+
+def neutral_frequency_grid_hz(config: RunConfigModel) -> np.ndarray:
+    """Return the exact per-trace frequency axis the configured radar emits.
+
+    Neutral sets must line up sample-for-sample with live sweeps, so the axis comes
+    from the same source the acquisition path uses: the ADC processing grid for Kamil
+    ADC, and the configured linear sweep grid for every VNA-style model (LibreVNA
+    single and multi-device, SN9000, Compact-M). The C++ preprocessor re-checks this
+    axis against the measured one within a tolerance, so a mismatch fails loudly
+    instead of silently corrupting the correction.
+    """
+    if config.is_kamil_adc:
+        # Single source of truth for the axis: the same grid the processor emits.
+        processor = KamilAdcSweepProcessor(
+            KamilAdcProcessingParams.from_kamil_model(config.radar.kamil_adc)
+        )
+        return processor.grid_hz
+
+    points = int(config.radar.sweep.points)
+    if points < 1:
+        raise ValueError("Neutral sets require radar.sweep.points >= 1")
+    if points == 1:
+        return np.array([float(config.radar.sweep.start_hz)], dtype=np.float32)
+    # Mirrors both acquisition paths: the native collector seeds this same linspace
+    # and the mock backend generates it outright.
+    return np.linspace(
+        float(config.radar.sweep.start_hz),
+        float(config.radar.sweep.stop_hz),
+        points,
+        dtype=np.float32,
+    )
+
+
+def build_neutral_s21_sets(
     config: RunConfigModel,
 ) -> tuple[SweepCollection, SweepCollection]:
-    """Build neutral S21 calibration/reference collections for the Kamil ADC radar.
+    """Build neutral S21 calibration/reference collections for the active radar.
 
-    The calibration uses unit S21 (1+0j) and the reference uses zero S21 across
-    every configured combo, so applying them in the preprocessing pipeline leaves
-    the input S21 unchanged. The frequency axis is the exact acquisition grid
-    (``radar.kamil_adc.band``), so neutral sets line up sample-for-sample with
-    live sweeps. Returns the ``(calibration, reference)`` collections.
+    Covers every combo in the effective matrix, so a matrix radar widened by real
+    switches gets a neutral pair for all of its positions and the preprocessor's
+    ``validate_combos()`` is satisfied. Returns ``(calibration, reference)``.
     """
-    if not config.is_kamil_adc:
-        raise ValueError("Neutral Kamil ADC sets require radar.model='kamil_adc'")
-
     combos = list(config.combos)
     if not combos:
-        combos = RunConfigModel.build_full_combos(
-            config.input_switch.positions, config.output_switch.positions
-        )
+        combos = config.build_runtime_combos()
     if not combos:
-        raise ValueError("Kamil ADC neutral sets require at least one switch combo")
+        raise ValueError("Neutral sets require at least one switch combo")
 
-    # Single source of truth for the axis: the same grid the processor emits.
-    processor = KamilAdcSweepProcessor(KamilAdcProcessingParams.from_kamil_model(config.radar.kamil_adc))
-    frequency_hz = processor.grid_hz
+    frequency_hz = neutral_frequency_grid_hz(config)
 
     now_ns = time.monotonic_ns()
     calibration = _neutral_collection(
@@ -57,11 +102,27 @@ def build_kamil_adc_neutral_s21_sets(
         monotonic_ns=now_ns,
     )
     logger.info(
-        "Built neutral Kamil ADC S21 sets: combos=%d points=%d", len(combos), int(frequency_hz.size)
+        "Built neutral S21 sets: model=%s combos=%d points=%d",
+        config.radar.model,
+        len(combos),
+        int(frequency_hz.size),
     )
     return calibration, reference
 
 
+def build_kamil_adc_neutral_s21_sets(
+    config: RunConfigModel,
+) -> tuple[SweepCollection, SweepCollection]:
+    """Build neutral S21 sets, rejecting anything but the Kamil ADC radar.
+
+    Kept as the model-checked entry point for the ADC path; new callers that must
+    work for several radar models should use `build_neutral_s21_sets` instead.
+    """
+    if not config.is_kamil_adc:
+        raise ValueError("Neutral Kamil ADC sets require radar.model='kamil_adc'")
+    return build_neutral_s21_sets(config)
+
+
 def _neutral_collection(
     *,
     combos: list[ComboModel],

python_app/workflows/multi_radar_capture_workflow.py

@@ -81,14 +81,7 @@ class MultiRadarSequentialCaptureSession:
         self._manual_matrix_radar_capture = (
             self._is_matrix_radar and kind in MATRIX_RADAR_MANUAL_CAPTURE_KINDS
         )
-        self._combos = (
-            RunConfigModel.build_matrix_radar_virtual_combos()
-            if self._is_matrix_radar
-            else RunConfigModel.build_full_combos(
-                base_config.input_switch.positions,
-                base_config.output_switch.positions,
-            )
-        )
+        self._combos = base_config.build_runtime_combos()
         if not self._combos:
             raise RuntimeError("No switch combinations available for capture")
 

python_app/workflows/sequential_capture_workflow.py

@@ -64,11 +64,7 @@ class SequentialCaptureSession:
         self._manual_matrix_radar_capture = (
             self._is_matrix_radar and kind in MATRIX_RADAR_MANUAL_CAPTURE_KINDS
         )
-        self._combos = (
-            RunConfigModel.build_matrix_radar_virtual_combos()
-            if self._is_matrix_radar
-            else RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions)
-        )
+        self._combos = config.build_runtime_combos()
         if not self._combos:
             raise RuntimeError("No switch combinations available for capture")
 

run_config.json

@@ -1,6 +1,6 @@
 {
   "radar": {
-    "model": "librevna",
+    "model": "librevna_multi",
     "serial": "",
     "remote_host": "127.0.0.1",
     "remote_port": 50209,
@@ -8,9 +8,14 @@
     "mock_signal_hz": 5000000.0,
     "visa_library": "",
     "multi_device": {
-      "slave_serials": [],
+      "slave_serials": [
+        "20A1307D5532",
+        "2072306C5532"
+      ],
       "force_external_reference": false,
-      "recovery_attempts": 3
+      "recovery_attempts": 3,
+      "output_switch_positions": 1,
+      "input_switch_positions": 3
     },
     "kamil_adc": {
       "project_dir": "",
@@ -20,7 +25,18 @@
       "env": {},
       "startup_timeout_s": 5.0,
       "sweep_timeout_s": 5.0,
-      "stop_timeout_s": 2.0
+      "stop_timeout_s": 2.0,
+      "phase_calibration": {
+        "phase0_rad": 0.0,
+        "freq0_hz": 2046000000.0,
+        "phase1_rad": 300.0,
+        "freq1_hz": 5612000000.0
+      },
+      "band": {
+        "start_hz": 2100000000.0,
+        "stop_hz": 5500000000.0,
+        "points": 2048
+      }
     },
     "laser_control": {
       "enabled": false,
@@ -75,7 +91,7 @@
       "driver_mode": "mock",
       "driver": "h7992",
       "radar_port": 2,
-      "positions": 4,
+      "positions": 12,
       "default_position": 0,
       "gpio_chip": "/dev/gpiochip0",
       "pin_a": 22,
@@ -92,11 +108,14 @@
     "debounce_ms": 50,
     "action": "capture_tmp_reference"
   },
+  "logging": {
+    "level": "debug"
+  },
   "run": {
     "settling_ms": 0,
     "idle_sleep_ms": 2,
     "continuous": true,
-    "processing_live_config_path": "python_app/runtime/processing_live.json",
+    "processing_live_config_path": "/home/guriy/Documents/radar_system/python_app/runtime/processing_live.json",
     "locator_server": {
       "device_id": 3,
       "protocol_version": 1,
@@ -123,6 +142,38 @@
         "input": 3,
         "output": 0
       },
+      {
+        "input": 4,
+        "output": 0
+      },
+      {
+        "input": 5,
+        "output": 0
+      },
+      {
+        "input": 6,
+        "output": 0
+      },
+      {
+        "input": 7,
+        "output": 0
+      },
+      {
+        "input": 8,
+        "output": 0
+      },
+      {
+        "input": 9,
+        "output": 0
+      },
+      {
+        "input": 10,
+        "output": 0
+      },
+      {
+        "input": 11,
+        "output": 0
+      },
       {
         "input": 0,
         "output": 1
@@ -138,17 +189,49 @@
       {
         "input": 3,
         "output": 1
+      },
+      {
+        "input": 4,
+        "output": 1
+      },
+      {
+        "input": 5,
+        "output": 1
+      },
+      {
+        "input": 6,
+        "output": 1
+      },
+      {
+        "input": 7,
+        "output": 1
+      },
+      {
+        "input": 8,
+        "output": 1
+      },
+      {
+        "input": 9,
+        "output": 1
+      },
+      {
+        "input": 10,
+        "output": 1
+      },
+      {
+        "input": 11,
+        "output": 1
       }
     ]
   },
   "preprocess": {
     "s21": {
       "calibration": {
-        "set_name": "smoke_cal",
+        "set_name": "smoke_cal3",
         "bundle_path": ""
       },
       "reference": {
-        "set_name": "smoke_ref",
+        "set_name": "smoke_cal3",
         "bundle_path": ""
       }
     },
@@ -219,6 +302,54 @@
         "x_m": 0.185,
         "y_m": 0.0,
         "z_m": 0.0
+      },
+      {
+        "input_pos": 4,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
+      },
+      {
+        "input_pos": 5,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
+      },
+      {
+        "input_pos": 6,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
+      },
+      {
+        "input_pos": 7,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
+      },
+      {
+        "input_pos": 8,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
+      },
+      {
+        "input_pos": 9,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
+      },
+      {
+        "input_pos": 10,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
+      },
+      {
+        "input_pos": 11,
+        "x_m": 0.0,
+        "y_m": 0.0,
+        "z_m": 0.0
       }
     ]
   },
@@ -253,7 +384,7 @@
     "version": 1,
     "switches": {
       "combo_mode": "text",
-      "combos_text": "0:0,1:0,2:0,3:0,0:1,1:1,2:1,3:1",
+      "combos_text": "0:0,1:0,2:0,3:0,4:0,5:0,6:0,7:0,8:0,9:0,10:0,11:0,0:1,1:1,2:1,3:1,4:1,5:1,6:1,7:1,8:1,9:1,10:1,11:1",
       "single_input": "0",
       "single_output": "0"
     },
@@ -262,6 +393,7 @@
       "pass_through": {
         "show_magnitude": true,
         "show_phase": false,
+        "unwrap_phase": false,
         "combo_filter": "",
         "fixed_y_enabled": false,
         "y_min_db": -100.0,
@@ -333,7 +465,8 @@
     "data_actions": {
       "save_count": 10,
       "save_path": "python_app/data/snapshots",
-      "save_name": "snapshot_simulator"
+      "save_name": "snapshot_simulator",
+      "record_count": 100
     },
     "preprocess_dialog": {
       "set_name": "smoke_cal",
@@ -342,4 +475,4 @@
       "median_sweep_count": 5
     }
   }
-}
+}