diff --git a/python_app/gui/app_window.py b/python_app/gui/app_window.py
index a27b630..71078a9 100644
--- a/python_app/gui/app_window.py
+++ b/python_app/gui/app_window.py
@@ -15,9 +15,10 @@ import logging
import os
from pathlib import Path
import sys
+import threading
import traceback
-from PyQt6.QtCore import QObject, QTimer, pyqtSignal
+from PyQt6.QtCore import QTimer
from PyQt6.QtGui import QTextCursor
from PyQt6.QtWidgets import QApplication, QMainWindow, QMessageBox
@@ -58,33 +59,60 @@ def _panel_extra(details: str | None, once_key: str | None) -> dict[str, object]
return {"panel_details": details, "panel_once_key": once_key}
-class _PanelLogBridge(QObject):
- """Marshals log records from any thread onto the GUI thread for panel rendering.
+class _PanelLogBuffer:
+ """Thread-safe bounded buffer between logging handlers and the GUI flush timer.
- A :class:`logging.Handler` can fire on a worker thread (readers, broadcaster),
- but the log widget may only be touched on the GUI thread; emitting this queued
- signal hands the record across safely (the GPIO-button pattern).
+ A :class:`logging.Handler` can fire on a worker thread (readers, broadcaster)
+ at a very high rate — e.g. the USB RX threads while the free-running sweep
+ streams. Posting one queued Qt event per record used to flood the GUI event
+ queue and keep the interface frozen long after a blocking operation finished
+ while the backlog rendered. Instead, records land in this bounded buffer and
+ a periodic GUI-side timer drains them in one batch; overflow drops the oldest
+ records and reports how many were lost.
"""
- record = pyqtSignal(str, str, object, object) # display level, message, details, once_key
+ _CAPACITY = 2000
+
+ def __init__(self) -> None:
+ self._lock = threading.Lock()
+ self._entries: deque[tuple[str, str, str | None, str | None]] = deque(maxlen=self._CAPACITY)
+ self._dropped_count = 0
+
+ def append(self, level: str, text: str, details: str | None, once_key: str | None) -> None:
+ """Store one record, evicting the oldest when full (any thread)."""
+ with self._lock:
+ if len(self._entries) == self._CAPACITY:
+ self._dropped_count += 1
+ self._entries.append((level, text, details, once_key))
+
+ def drain(self) -> tuple[list[tuple[str, str, str | None, str | None]], int]:
+ """Return and clear all buffered records plus the overflow-drop count."""
+ with self._lock:
+ entries = list(self._entries)
+ self._entries.clear()
+ dropped_count = self._dropped_count
+ self._dropped_count = 0
+ return entries, dropped_count
class _QtLogPanelHandler(logging.Handler):
"""Logging handler that forwards application log records to the GUI log panel."""
- def __init__(self, bridge: _PanelLogBridge) -> None:
+ def __init__(self, buffer: _PanelLogBuffer) -> None:
super().__init__()
- self._bridge = bridge
+ self._buffer = buffer
def emit(self, record: logging.LogRecord) -> None:
- """Forward one record to the panel bridge, mapping WARNING to the short 'WARN'."""
+ """Buffer one record for the panel, mapping WARNING to the short 'WARN'."""
try:
display_level = "WARN" if record.levelname == "WARNING" else record.levelname
- self._bridge.record.emit(
+ details = getattr(record, "panel_details", None)
+ once_key = getattr(record, "panel_once_key", None)
+ self._buffer.append(
display_level,
record.getMessage(),
- getattr(record, "panel_details", None),
- getattr(record, "panel_once_key", None),
+ details if isinstance(details, str) else None,
+ once_key if isinstance(once_key, str) else None,
)
except Exception: # noqa: BLE001 - logging must never raise into the caller
self.handleError(record)
@@ -145,23 +173,53 @@ class AppWindow(
log_dir = self._project_root / "python_app/runtime/logs"
configure_logging(level=DEFAULT_LOG_LEVEL, log_dir=log_dir, console=True)
self._gui_logger = get_logger("gui")
- self._log_panel_bridge = _PanelLogBridge()
- self._log_panel_bridge.record.connect(self._on_log_record)
+ self._log_panel_buffer = _PanelLogBuffer()
def _attach_log_panel(self) -> None:
"""Route application log records into the on-screen panel (widget now exists)."""
- add_handler(_QtLogPanelHandler(self._log_panel_bridge))
+ add_handler(_QtLogPanelHandler(self._log_panel_buffer))
+ # One bounded flush per tick instead of one queued event per record: the
+ # panel can never flood the GUI event queue, no matter how chatty a
+ # DEBUG-level driver gets.
+ self._log_flush_timer = QTimer(self)
+ self._log_flush_timer.setInterval(100)
+ self._log_flush_timer.timeout.connect(self._flush_log_panel_buffer)
+ self._log_flush_timer.start()
- def _on_log_record(self, level: str, text: str, details: object, once_key: object) -> None:
- """Render one forwarded log record in the panel (always on the GUI thread)."""
- if not hasattr(self, "_log_box"):
+ def _flush_log_panel_buffer(self) -> None:
+ """Render every buffered log record into the panel as one batched insert."""
+ entries, dropped_count = self._log_panel_buffer.drain()
+ if (not entries and not dropped_count) or not hasattr(self, "_log_box"):
return
- self._append_log_entry(
- level,
- text,
- details=details if isinstance(details, str) else None,
- once_key=once_key if isinstance(once_key, str) else None,
- )
+
+ entry_htmls: list[str] = []
+ if dropped_count:
+ entry_htmls.append(
+ self._render_log_entry_html(
+ "WARN",
+ f"Log panel overflow: {dropped_count} record(s) dropped "
+ "(they are still in the log file).",
+ )
+ )
+ error_seen = False
+ for level, text, details, once_key in entries:
+ if once_key is not None:
+ if once_key in self._logged_once_keys:
+ continue
+ self._logged_once_keys.add(once_key)
+ entry_htmls.append(self._render_log_entry_html(level, text, details))
+ error_seen = error_seen or level.upper() == "ERROR"
+
+ if not entry_htmls:
+ return
+ cursor = self._log_box.textCursor()
+ cursor.movePosition(QTextCursor.MoveOperation.End)
+ self._log_box.setTextCursor(cursor)
+ self._log_box.insertHtml("".join(entry_htmls))
+ self._log_box.insertPlainText("\n")
+ self._log_box.ensureCursorVisible()
+ if error_seen and hasattr(self, "_status_label"):
+ self._status_label.setText("Status: error")
def _init_runtime_services(self) -> None:
"""Initialize long-lived service objects used by mixins."""
@@ -266,6 +324,10 @@ class AppWindow(
def _init_capture_state(self) -> None:
"""Initialize one-shot capture and sequence-control flags."""
self._capture_session: SequentialCaptureSession | MultiRadarSequentialCaptureSession | None = None
+ # Guards the blocking per-combo capture against duplicate requests, and keeps
+ # the dialog's action buttons disabled until the post-capture input backlog
+ # is dropped (see AppWindowPreprocessMixin._begin/_end_preprocess_capture).
+ self._preprocess_capture_busy = False
self._resume_pipeline_after_capture = False
self._single_capture_active = False
self._single_capture_start_ns: int | None = None
@@ -551,20 +613,8 @@ class AppWindow(
"""Return full chained traceback for error dialogs and log details."""
return "".join(traceback.TracebackException.from_exception(exc).format(chain=True)).strip()
- def _append_log_entry(
- self,
- level: str,
- text: str,
- *,
- details: str | None = None,
- once_key: str | None = None,
- ) -> None:
- """Append formatted log entry with timestamp and optional details."""
- if once_key is not None:
- if once_key in self._logged_once_keys:
- return
- self._logged_once_keys.add(once_key)
-
+ def _render_log_entry_html(self, level: str, text: str, details: str | None = None) -> str:
+ """Render one log entry as the panel's HTML block."""
level_upper = level.upper()
palette = {
"DEBUG": ("#6c7b8d", "#52627a", "#8a97a8"),
@@ -586,16 +636,30 @@ class AppWindow(
"{html.escape(details)}"
)
+ return "" + "".join(body_parts) + "
"
+
+ def _append_log_entry(
+ self,
+ level: str,
+ text: str,
+ *,
+ details: str | None = None,
+ once_key: str | None = None,
+ ) -> None:
+ """Append formatted log entry with timestamp and optional details."""
+ if once_key is not None:
+ if once_key in self._logged_once_keys:
+ return
+ self._logged_once_keys.add(once_key)
- entry_html = "" + "".join(body_parts) + "
"
cursor = self._log_box.textCursor()
cursor.movePosition(QTextCursor.MoveOperation.End)
self._log_box.setTextCursor(cursor)
- self._log_box.insertHtml(entry_html)
+ self._log_box.insertHtml(self._render_log_entry_html(level, text, details))
self._log_box.insertPlainText("\n")
self._log_box.ensureCursorVisible()
- if level_upper == "ERROR" and hasattr(self, "_status_label"):
+ if level.upper() == "ERROR" and hasattr(self, "_status_label"):
self._status_label.setText("Status: error")
def _on_log_level_selected(self, level_text: str) -> None:
diff --git a/python_app/gui/controllers/app_window_preprocess_mixin.py b/python_app/gui/controllers/app_window_preprocess_mixin.py
index 01fef05..e4b9d42 100644
--- a/python_app/gui/controllers/app_window_preprocess_mixin.py
+++ b/python_app/gui/controllers/app_window_preprocess_mixin.py
@@ -2,6 +2,8 @@
from __future__ import annotations
+from PyQt6.QtCore import QTimer
+
from python_app.gui.preprocess_dialog import PreprocessDialog
from python_app.gui.trace_png_export import export_trace_png
from python_app.orchestration.preprocess_assets import (
@@ -519,13 +521,23 @@ class AppWindowPreprocessMixin:
if session is None:
self._show_error("No active capture sequence")
return
+ # The capture blocks the event loop, so clicks made during it are delivered
+ # only after it finishes. `_begin_preprocess_capture` disables the action
+ # buttons for that whole window (re-enabled via a posted event), so a queued
+ # click lands on a disabled button instead of silently starting — and
+ # advancing the combo cursor of — another capture.
+ if not self._begin_preprocess_capture():
+ return
try:
- capture_result = session.capture_current_combo()
- except Exception as exc: # noqa: BLE001
- self._on_capture_combo_failed(session, exc)
- return
- self._record_preprocess_capture(session, capture_result)
+ try:
+ capture_result = session.capture_current_combo()
+ except Exception as exc: # noqa: BLE001
+ self._on_capture_combo_failed(session, exc)
+ return
+ self._record_preprocess_capture(session, capture_result)
+ finally:
+ self._end_preprocess_capture()
def _capture_all_remaining(self) -> None:
"""Capture all remaining combos for the active preprocess session."""
@@ -539,6 +551,8 @@ class AppWindowPreprocessMixin:
details=self._capture_state_details(),
)
return
+ if not self._begin_preprocess_capture():
+ return
display_name = preprocess_asset_display_name(session.kind)
dialog = self._ensure_preprocess_dialog()
@@ -547,13 +561,40 @@ class AppWindowPreprocessMixin:
f"{display_name} batch capture started: remaining="
f"{session.state().total_count - session.state().captured_count}"
)
- while not session.is_complete():
- try:
- capture_result = session.capture_current_combo()
- except Exception as exc: # noqa: BLE001
- self._on_capture_combo_failed(session, exc)
- return
- self._record_preprocess_capture(session, capture_result)
+ try:
+ while not session.is_complete():
+ try:
+ capture_result = session.capture_current_combo()
+ except Exception as exc: # noqa: BLE001
+ self._on_capture_combo_failed(session, exc)
+ return
+ self._record_preprocess_capture(session, capture_result)
+ finally:
+ self._end_preprocess_capture()
+
+ def _begin_preprocess_capture(self) -> bool:
+ """Mark a blocking combo capture as running; refuse when one already is.
+
+ Returns False for a duplicate request (e.g. a click delivered while an
+ error dialog inside a capture pumps the event loop).
+ """
+ if self._preprocess_capture_busy:
+ self._log("Preprocess combo capture already in progress; ignoring duplicate request.")
+ return False
+ self._preprocess_capture_busy = True
+ # Disable the sequence action buttons for the whole blocked window.
+ self._update_capture_dialog_state()
+ return True
+
+ def _end_preprocess_capture(self) -> None:
+ """Re-enable capture actions after the pending input backlog is discarded.
+
+ The zero-delay timer fires only after Qt has dispatched the window-system
+ events queued while the capture blocked the loop; those clicks hit the
+ still-disabled buttons and are dropped, then the buttons come back.
+ """
+ self._preprocess_capture_busy = False
+ QTimer.singleShot(0, self._update_capture_dialog_state)
def _on_capture_combo_failed(
self,
@@ -817,6 +858,10 @@ class AppWindowPreprocessMixin:
and state.current_combo is not None
),
variant_count=state.variant_count,
+ # While a blocking capture is executing, every action stays disabled no
+ # matter what the session state allows: clicks queued during the freeze
+ # must land on disabled buttons (see `_end_preprocess_capture`).
+ actions_enabled=not self._preprocess_capture_busy,
)
def _cleanup_capture_session(self) -> None:
diff --git a/python_app/gui/preprocess_dialog.py b/python_app/gui/preprocess_dialog.py
index 6e300ab..f4636a2 100644
--- a/python_app/gui/preprocess_dialog.py
+++ b/python_app/gui/preprocess_dialog.py
@@ -354,8 +354,14 @@ class PreprocessDialog(QDialog):
can_finalize: bool,
can_capture_all: bool,
variant_count: int = 1,
+ actions_enabled: bool = True,
) -> None:
- """Update sequence progress/status widgets."""
+ """Update sequence progress/status widgets.
+
+ With ``actions_enabled=False`` the progress labels still update but every
+ sequence action button is kept disabled — used while a blocking capture
+ runs, so input queued during the freeze cannot trigger another action.
+ """
if kind is None:
self._active_kind_label.setText("")
self._progress_label.setText("0 / 0")
@@ -368,13 +374,14 @@ class PreprocessDialog(QDialog):
self._capture_all_button.setText("Capture All Remaining")
return
+ actions_enabled = bool(actions_enabled)
active_label = preprocess_asset_display_name(kind) if kind in PREPROCESS_ASSET_SPECS else kind
self._active_kind_label.setText(active_label)
self._progress_label.setText(f"{captured_count} / {total_count}")
- self._undo_last_button.setEnabled(bool(can_undo))
- self._save_sequence_button.setEnabled(bool(can_finalize))
- self._capture_all_button.setEnabled(bool(can_capture_all))
- self._abort_button.setEnabled(True)
+ self._undo_last_button.setEnabled(bool(can_undo) and actions_enabled)
+ self._save_sequence_button.setEnabled(bool(can_finalize) and actions_enabled)
+ self._capture_all_button.setEnabled(bool(can_capture_all) and actions_enabled)
+ self._abort_button.setEnabled(actions_enabled)
self._capture_all_button.setText("Capture All Remaining")
if next_input is None or next_output is None:
self._combo_label.setText("")
@@ -384,7 +391,7 @@ class PreprocessDialog(QDialog):
if int(variant_count) > 1:
combo_text += f" | radar configs={int(variant_count)}"
self._combo_label.setText(combo_text)
- self._capture_next_button.setEnabled(True)
+ self._capture_next_button.setEnabled(actions_enabled)
def set_available_sets(self, available_sets: dict[str, list[str]]) -> None:
"""Replace combo-box choices for all preprocess assets."""
diff --git a/python_app/hardware_full/librevna_driver/transport/usb.py b/python_app/hardware_full/librevna_driver/transport/usb.py
index b9f2699..de82a1b 100644
--- a/python_app/hardware_full/librevna_driver/transport/usb.py
+++ b/python_app/hardware_full/librevna_driver/transport/usb.py
@@ -5,6 +5,7 @@ from __future__ import annotations
from contextlib import suppress
import logging
import threading
+import time
from typing import Callable
from ..exceptions import DeviceDisconnectedError, TimeoutError
@@ -44,6 +45,10 @@ class USBTransport:
self._rx_thread: threading.Thread | None = None
self._stop_event = threading.Event()
self._tx_lock = threading.Lock()
+ # Aggregation window for the RX debug trace (see `_rx_loop`).
+ self._rx_debug_bytes = 0
+ self._rx_debug_chunks = 0
+ self._rx_debug_window_start = 0.0
self.connected_serial: str | None = None
@@ -270,7 +275,27 @@ class USBTransport:
if data:
if logger.isEnabledFor(logging.DEBUG):
- logger.debug("USB RX %d bytes", len(data))
+ # Aggregate: the free-running datapoint stream completes bulk
+ # reads hundreds of times per second, and a log record per chunk
+ # floods every handler (file, stderr, and the GUI panel, which
+ # marshals each record onto the GUI thread). One summary per
+ # second keeps the throughput trace without the flood.
+ self._rx_debug_bytes += len(data)
+ self._rx_debug_chunks += 1
+ now = time.monotonic()
+ if self._rx_debug_window_start == 0.0:
+ self._rx_debug_window_start = now
+ elif now - self._rx_debug_window_start >= 1.0:
+ logger.debug(
+ "USB RX %d bytes in %d chunks over %.2f s (serial=%s)",
+ self._rx_debug_bytes,
+ self._rx_debug_chunks,
+ now - self._rx_debug_window_start,
+ self.connected_serial,
+ )
+ self._rx_debug_bytes = 0
+ self._rx_debug_chunks = 0
+ self._rx_debug_window_start = now
self._on_data(bytes(data))
logger.debug("USB RX thread stopped")
diff --git a/python_app/hardware_full/librevna_multi_device_driver/transport.py b/python_app/hardware_full/librevna_multi_device_driver/transport.py
index 16c5dc5..1d81b13 100644
--- a/python_app/hardware_full/librevna_multi_device_driver/transport.py
+++ b/python_app/hardware_full/librevna_multi_device_driver/transport.py
@@ -15,6 +15,17 @@ from python_app.hardware_full.librevna_multi_device_driver.protocol import Packe
logger = logging.getLogger(__name__)
+# The sweep free-runs by design, so devices stream datapoints continuously even
+# while no acquisition is consuming them (e.g. an operator pausing between manual
+# combo captures). An unbounded queue then grows without limit — hundreds of MB
+# over a few minutes — and the next acquisition's drain spends seconds discarding
+# the backlog on the GUI thread. Bound the queue and drop the OLDEST packet on
+# overflow: every acquisition drains stale packets before collecting anyway, and
+# whenever packets actually matter (ACK waits, cycle collection) a consumer is
+# already pulling, so the queue never approaches the bound. Sized to hold many
+# full sweeps of datapoints with a wide margin.
+_RECEIVED_PACKET_QUEUE_MAX = 32768
+
class LibreVnaUsbBulkConnection:
"""Minimal packet transport for one LibreVNA device."""
@@ -29,7 +40,9 @@ class LibreVnaUsbBulkConnection:
raise ValueError("serial_number is required for multi-device acquisition")
self.serial_number = serial_number
self._scanner = FrameScanner()
- self._received_packets: queue.Queue[tuple[int, bytes]] = queue.Queue()
+ self._received_packets: queue.Queue[tuple[int, bytes]] = queue.Queue(
+ maxsize=_RECEIVED_PACKET_QUEUE_MAX
+ )
self._fatal_error: Exception | None = None
self._fatal_lock = threading.Lock()
self._transport = USBTransport(
@@ -108,7 +121,20 @@ class LibreVnaUsbBulkConnection:
logger.warning("Dropping unparseable USB chunk from %s: %s", self.serial_number, exc)
return
for packet in packets:
- self._received_packets.put((int(packet.type), bytes(packet.payload)))
+ entry = (int(packet.type), bytes(packet.payload))
+ while True:
+ try:
+ self._received_packets.put_nowait(entry)
+ break
+ except queue.Full:
+ # Blocking here would stall the USB read thread; discard the
+ # oldest packet instead — stale data is what the pre-collect
+ # drain throws away anyway. Racing a concurrent consumer just
+ # means the queue already has room again.
+ try:
+ self._received_packets.get_nowait()
+ except queue.Empty:
+ pass
def _on_disconnect(self, exc: Exception) -> None:
"""Record an asynchronous transport disconnect as the fatal error."""
diff --git a/python_app/hardware_full/switched_matrix_radar_service.py b/python_app/hardware_full/switched_matrix_radar_service.py
index 0968233..5b4437c 100644
--- a/python_app/hardware_full/switched_matrix_radar_service.py
+++ b/python_app/hardware_full/switched_matrix_radar_service.py
@@ -83,42 +83,8 @@ class SwitchedMatrixRadarService:
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)
- )
+ for trace in self._acquire_step_traces(out_k, in_k, collection_id):
+ slots[trace.combo.output * total_inputs + trace.combo.input] = trace
if any(trace is None for trace in slots):
missing = sum(1 for trace in slots if trace is None)
@@ -134,6 +100,89 @@ class SwitchedMatrixRadarService:
capture_end_ns=time.monotonic_ns(),
)
+ def acquire_combo_collection(
+ self,
+ *,
+ input_pos: int,
+ output_pos: int,
+ collection_id: int = 1,
+ ) -> SweepCollection:
+ """Acquire only the physical switch step that carries one widened combo.
+
+ The per-combo capture workflows need a single trace at a time; sweeping
+ every switch position for that (a full ``acquire_collection``) multiplies
+ the capture time by the number of physical steps and freezes the caller
+ for the whole sweep. One widened combo lives entirely inside one
+ (out_k, in_k) step, so acquiring just that step is sufficient. The result
+ contains that step's traces with widened combo keys, including the
+ requested combo.
+ """
+ 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
+ if not (0 <= int(input_pos) < total_inputs and 0 <= int(output_pos) < total_outputs):
+ raise ValueError(
+ f"Widened combo out of range: input={input_pos} (of {total_inputs}), "
+ f"output={output_pos} (of {total_outputs})"
+ )
+
+ capture_start_ns = time.monotonic_ns()
+ out_k = int(output_pos) // self.inner_output_positions
+ in_k = int(input_pos) // self.inner_input_positions
+ traces = self._acquire_step_traces(out_k, in_k, collection_id)
+ return SweepCollection(
+ collection_id=int(collection_id),
+ monotonic_ns=time.monotonic_ns(),
+ traces=traces,
+ capture_start_ns=capture_start_ns,
+ capture_end_ns=time.monotonic_ns(),
+ )
+
+ def _acquire_step_traces(self, out_k: int, in_k: int, collection_id: int) -> list[TraceData]:
+ """Drive both switches to one step, settle, and collect its widened traces."""
+ 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
+ return [
+ replace(
+ trace,
+ combo=ComboKey(
+ input=in_k * self.inner_input_positions + int(trace.combo.input),
+ output=out_k * self.inner_output_positions + int(trace.combo.output),
+ ),
+ )
+ for trace in sub.traces
+ ]
+
+
def build_physical_switch(
model: SwitchModel,
physical_positions: int,
diff --git a/python_app/tests/test_log_panel_buffer.py b/python_app/tests/test_log_panel_buffer.py
new file mode 100644
index 0000000..86d887d
--- /dev/null
+++ b/python_app/tests/test_log_panel_buffer.py
@@ -0,0 +1,48 @@
+"""Unit tests for the bounded GUI log-panel buffer.
+
+The buffer decouples logging handlers (any thread, potentially very chatty at
+DEBUG) from the GUI: records are batched by a flush timer instead of posting one
+queued Qt event per record, and overflow drops the oldest records with a count.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from python_app.gui.app_window import _PanelLogBuffer
+
+
+class PanelLogBufferTest(unittest.TestCase):
+ """Bounded capacity, oldest-first eviction, and accurate drop accounting."""
+
+ def test_drain_returns_entries_in_order_and_clears(self) -> None:
+ buffer = _PanelLogBuffer()
+ buffer.append("INFO", "first", None, None)
+ buffer.append("WARN", "second", "details", "key")
+
+ entries, dropped_count = buffer.drain()
+
+ self.assertEqual(dropped_count, 0)
+ self.assertEqual(
+ entries,
+ [("INFO", "first", None, None), ("WARN", "second", "details", "key")],
+ )
+ self.assertEqual(buffer.drain(), ([], 0))
+
+ def test_overflow_drops_oldest_and_counts(self) -> None:
+ buffer = _PanelLogBuffer()
+ overflow = 100
+ total = _PanelLogBuffer._CAPACITY + overflow
+ for index in range(total):
+ buffer.append("DEBUG", f"m{index}", None, None)
+
+ entries, dropped_count = buffer.drain()
+
+ self.assertEqual(dropped_count, overflow)
+ self.assertEqual(len(entries), _PanelLogBuffer._CAPACITY)
+ self.assertEqual(entries[0][1], f"m{overflow}")
+ self.assertEqual(entries[-1][1], f"m{total - 1}")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/python_app/tests/test_switched_matrix_capture.py b/python_app/tests/test_switched_matrix_capture.py
new file mode 100644
index 0000000..49c030f
--- /dev/null
+++ b/python_app/tests/test_switched_matrix_capture.py
@@ -0,0 +1,187 @@
+"""Unit tests for switch-widened matrix capture.
+
+Cover the targeted single-step acquisition on ``SwitchedMatrixRadarService`` and
+verify the manual per-combo capture workflow uses it instead of sweeping the full
+widened matrix (the regression that froze the GUI for the whole matrix per click).
+"""
+
+from __future__ import annotations
+
+import time
+import unittest
+from unittest import mock
+
+import numpy as np
+
+from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService
+from python_app.hardware_full.switched_matrix_radar_service import SwitchedMatrixRadarService
+from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData
+from python_app.models.run_config_model import RunConfigModel
+from python_app.workflows.sequential_capture_workflow import SequentialCaptureSession
+
+_INNER_INPUTS = 4
+_INNER_OUTPUTS = 2
+_POINTS = 8
+
+
+class _FakeInnerMatrixRadar:
+ """Matrix radar stub emitting the canonical 2x4 combo set per acquisition."""
+
+ def __init__(self) -> None:
+ self.acquire_count = 0
+
+ def open(self) -> None:
+ pass
+
+ def close(self) -> None:
+ pass
+
+ def configure(self, sweep) -> None:
+ pass
+
+ def recover(self) -> None:
+ pass
+
+ def acquire_collection(self, collection_id: int = 1) -> SweepCollection:
+ self.acquire_count += 1
+ frequency_hz = np.linspace(1e6, 2e6, _POINTS, dtype=np.float32)
+ traces = [
+ TraceData(
+ combo=ComboKey(input=input_pos, output=output_pos),
+ frequency_hz=frequency_hz,
+ s11=np.full(_POINTS, complex(self.acquire_count, 0), dtype=np.complex64),
+ s21=np.full(_POINTS, complex(input_pos, output_pos), dtype=np.complex64),
+ )
+ for output_pos in range(_INNER_OUTPUTS)
+ for input_pos in range(_INNER_INPUTS)
+ ]
+ return SweepCollection(
+ collection_id=int(collection_id),
+ monotonic_ns=time.monotonic_ns(),
+ traces=traces,
+ )
+
+
+class _FakeSwitch:
+ """Switch stub recording every position it is driven to."""
+
+ def __init__(self, positions: int) -> None:
+ self.positions = positions
+ self.switched_to: list[int] = []
+
+ def open(self) -> None:
+ pass
+
+ def close(self) -> None:
+ pass
+
+ def position_count(self) -> int:
+ return self.positions
+
+ def switch_to(self, position: int) -> None:
+ self.switched_to.append(int(position))
+
+
+def _switched_service(input_steps: int = 3) -> tuple[SwitchedMatrixRadarService, _FakeInnerMatrixRadar, _FakeSwitch]:
+ inner = _FakeInnerMatrixRadar()
+ input_switch = _FakeSwitch(input_steps)
+ service = SwitchedMatrixRadarService(
+ inner=inner,
+ output_switch=None,
+ input_switch=input_switch,
+ inner_output_positions=_INNER_OUTPUTS,
+ inner_input_positions=_INNER_INPUTS,
+ settling_ms=0,
+ )
+ return service, inner, input_switch
+
+
+class SwitchedMatrixComboAcquisitionTest(unittest.TestCase):
+ """acquire_combo_collection must acquire exactly one physical switch step."""
+
+ def test_acquires_only_the_step_containing_the_combo(self) -> None:
+ service, inner, input_switch = _switched_service(input_steps=3)
+
+ # Widened input 9 lives in physical step 9 // 4 = 2.
+ collection = service.acquire_combo_collection(input_pos=9, output_pos=1)
+
+ self.assertEqual(inner.acquire_count, 1)
+ self.assertEqual(input_switch.switched_to, [2])
+ self.assertEqual(len(collection.traces), _INNER_INPUTS * _INNER_OUTPUTS)
+ combos = {(trace.combo.input, trace.combo.output) for trace in collection.traces}
+ self.assertIn((9, 1), combos)
+ # Every trace of the step is remapped into the widened axis of that step.
+ self.assertEqual(
+ combos,
+ {(2 * _INNER_INPUTS + i, o) for i in range(_INNER_INPUTS) for o in range(_INNER_OUTPUTS)},
+ )
+
+ def test_rejects_out_of_range_combo(self) -> None:
+ service, _inner, _input_switch = _switched_service(input_steps=3)
+ with self.assertRaises(ValueError):
+ service.acquire_combo_collection(input_pos=12, output_pos=0)
+ with self.assertRaises(ValueError):
+ service.acquire_combo_collection(input_pos=0, output_pos=2)
+
+ def test_full_collection_still_covers_widened_matrix_in_canonical_order(self) -> None:
+ service, inner, input_switch = _switched_service(input_steps=3)
+
+ collection = service.acquire_collection(collection_id=7)
+
+ self.assertEqual(inner.acquire_count, 3)
+ self.assertEqual(input_switch.switched_to, [0, 1, 2])
+ expected_combos = [
+ (input_pos, output_pos)
+ for output_pos in range(_INNER_OUTPUTS)
+ for input_pos in range(3 * _INNER_INPUTS)
+ ]
+ self.assertEqual(
+ [(trace.combo.input, trace.combo.output) for trace in collection.traces],
+ expected_combos,
+ )
+
+
+class ManualComboCaptureUsesTargetedAcquisitionTest(unittest.TestCase):
+ """The per-combo capture session must not sweep the full widened matrix."""
+
+ @staticmethod
+ def _switched_mock_config() -> RunConfigModel:
+ config = RunConfigModel()
+ config.radar.model = RunConfigModel.LIBREVNA_MULTI_MODEL
+ config.radar.driver_mode = "mock"
+ config.radar.multi_device.slave_serials = ["SLAVE_A", "SLAVE_B"]
+ config.radar.multi_device.input_switch_positions = 3
+ config.apply_device_model_constraints()
+ return config
+
+ def test_manual_capture_runs_one_inner_collection_per_median_sweep(self) -> None:
+ config = self._switched_mock_config()
+ session = SequentialCaptureSession(
+ config=config,
+ kind="s21_calibration",
+ set_name="targeted_test",
+ median_sweep_count=2,
+ )
+ with mock.patch.object(
+ MultiDeviceLibreVnaService,
+ "acquire_collection",
+ autospec=True,
+ side_effect=MultiDeviceLibreVnaService.acquire_collection,
+ ) as inner_acquire:
+ session.open()
+ try:
+ trace = session.capture_current_combo()
+ finally:
+ session.close()
+
+ first_combo = config.combos[0]
+ self.assertEqual(
+ (trace.combo.input, trace.combo.output),
+ (first_combo.input, first_combo.output),
+ )
+ # 2 median sweeps of ONE physical step — not 2 x 3 full-matrix steps.
+ self.assertEqual(inner_acquire.call_count, 2)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/python_app/workflows/multi_radar_capture_workflow.py b/python_app/workflows/multi_radar_capture_workflow.py
index 25bca17..3a9cec5 100644
--- a/python_app/workflows/multi_radar_capture_workflow.py
+++ b/python_app/workflows/multi_radar_capture_workflow.py
@@ -19,6 +19,7 @@ from python_app.workflows.radar_config_variants import RadarConfigVariant
from python_app.workflows.sequential_capture_workflow import (
MATRIX_RADAR_MANUAL_CAPTURE_KINDS,
SequentialCaptureState,
+ acquire_matrix_combo_collection,
combine_collections_via_median,
combine_traces_via_median,
select_trace_for_combo,
@@ -192,20 +193,27 @@ class MultiRadarSequentialCaptureSession:
self._radar.configure(variant.config.radar.sweep)
if self._base_config.runtime.settling_ms > 0:
time.sleep(self._base_config.runtime.settling_ms / 1000.0)
- collections: list[SweepCollection] = []
- for _ in range(self._median_sweep_count):
- collection = self._radar.acquire_collection(collection_id=1)
- if not collection.traces:
- raise RuntimeError(
- f"Matrix radar variant {variant.display_name} returned no traces"
- )
- collections.append(collection)
if self._manual_matrix_radar_capture:
- per_sweep_traces = [select_trace_for_combo(collection, combo) for collection in collections]
+ # Only this combo is kept, so acquire the smallest collection
+ # that contains it instead of the full (switch-widened) matrix.
+ per_sweep_traces = [
+ select_trace_for_combo(
+ acquire_matrix_combo_collection(self._radar, combo), combo
+ )
+ for _ in range(self._median_sweep_count)
+ ]
trace = combine_traces_via_median(per_sweep_traces)
pending_traces_by_radar_key[variant.radar_key] = [trace]
display_traces.append(trace)
else:
+ collections: list[SweepCollection] = []
+ for _ in range(self._median_sweep_count):
+ collection = self._radar.acquire_collection(collection_id=1)
+ if not collection.traces:
+ raise RuntimeError(
+ f"Matrix radar variant {variant.display_name} returned no traces"
+ )
+ collections.append(collection)
combined_collection = combine_collections_via_median(collections)
pending_traces_by_radar_key[variant.radar_key] = list(combined_collection.traces)
display_traces.append(combined_collection.traces[-1])
diff --git a/python_app/workflows/sequential_capture_workflow.py b/python_app/workflows/sequential_capture_workflow.py
index 5e1038a..ad3f53a 100644
--- a/python_app/workflows/sequential_capture_workflow.py
+++ b/python_app/workflows/sequential_capture_workflow.py
@@ -156,20 +156,27 @@ class SequentialCaptureSession:
raise RuntimeError("Capture session is already complete")
if self._is_matrix_radar:
- collections: list[SweepCollection] = []
- for _ in range(self._median_sweep_count):
- collection = self._radar.acquire_collection(collection_id=1)
- if not collection.traces:
- raise RuntimeError("Matrix radar capture returned no traces")
- collections.append(collection)
if self._manual_matrix_radar_capture:
- per_sweep_traces = [select_trace_for_combo(collection, combo) for collection in collections]
+ # Only this combo is kept, so acquire the smallest collection that
+ # contains it instead of the full (switch-widened) matrix.
+ per_sweep_traces = [
+ select_trace_for_combo(
+ acquire_matrix_combo_collection(self._radar, combo), combo
+ )
+ for _ in range(self._median_sweep_count)
+ ]
trace = combine_traces_via_median(per_sweep_traces)
self._traces.append(trace)
self._next_index += 1
logger.debug("Captured matrix combo input=%d output=%d", combo.input, combo.output)
return trace
+ collections: list[SweepCollection] = []
+ for _ in range(self._median_sweep_count):
+ collection = self._radar.acquire_collection(collection_id=1)
+ if not collection.traces:
+ raise RuntimeError("Matrix radar capture returned no traces")
+ collections.append(collection)
combined_collection = combine_collections_via_median(collections)
self._traces.extend(combined_collection.traces)
self._next_index = len(self._combos)
@@ -301,6 +308,32 @@ class SequentialCaptureSession:
return self._combos[self._next_index]
+def acquire_matrix_combo_collection(
+ radar: MatrixRadarService,
+ combo: ComboModel,
+ collection_id: int = 1,
+) -> SweepCollection:
+ """Acquire the smallest matrix collection that contains one combo.
+
+ A switch-widened matrix radar (``SwitchedMatrixRadarService``) can acquire just
+ the physical switch step carrying the combo, which is several times faster than
+ the full matrix and keeps the per-combo capture UI responsive. Plain matrix
+ radars expose only full-matrix acquisition, so they fall back to it.
+ """
+ acquire_combo = getattr(radar, "acquire_combo_collection", None)
+ if callable(acquire_combo):
+ collection = acquire_combo(
+ input_pos=int(combo.input),
+ output_pos=int(combo.output),
+ collection_id=collection_id,
+ )
+ else:
+ collection = radar.acquire_collection(collection_id=collection_id)
+ if not collection.traces:
+ raise RuntimeError("Matrix radar capture returned no traces")
+ return collection
+
+
def select_trace_for_combo(collection: SweepCollection, combo: ComboModel) -> TraceData:
"""Return the trace matching a virtual combo from a full multi-device capture."""
for trace in collection.traces: