From 6ada811c2f5752aa438aaa55abc029e4ecb918ca Mon Sep 17 00:00:00 2001 From: Ayzen Date: Wed, 26 Aug 2026 15:02:21 +0300 Subject: [PATCH] some fixes --- python_app/gui/app_window.py | 148 ++++++++++---- .../app_window_preprocess_mixin.py | 69 +++++-- python_app/gui/preprocess_dialog.py | 19 +- .../librevna_driver/transport/usb.py | 27 ++- .../librevna_multi_device_driver/transport.py | 30 ++- .../switched_matrix_radar_service.py | 121 ++++++++---- python_app/tests/test_log_panel_buffer.py | 48 +++++ .../tests/test_switched_matrix_capture.py | 187 ++++++++++++++++++ .../workflows/multi_radar_capture_workflow.py | 26 ++- .../workflows/sequential_capture_workflow.py | 47 ++++- 10 files changed, 607 insertions(+), 115 deletions(-) create mode 100644 python_app/tests/test_log_panel_buffer.py create mode 100644 python_app/tests/test_switched_matrix_capture.py 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: