new gpr

python_app/hardware_full/librevna_driver/protocol/frame.py

@@ -79,11 +79,21 @@ def decode_frame(frame: bytes) -> Packet:
 
 
 class FrameScanner:
-    """Incremental frame scanner for raw USB byte streams."""
+    """Incremental frame scanner for raw USB byte streams.
+
+    Tolerant of corruption: a ``0x5A`` that opens a frame which does not decode
+    (bad CRC, unknown packet type, length mismatch) is treated as a false header.
+    Such a byte is skipped one at a time and scanning resyncs on the next
+    candidate header, so a malformed or partially-lost frame costs only a brief
+    resync — ``feed`` never raises and never permanently desynchronizes the
+    stream. ``discarded_byte_count`` exposes how many bytes were skipped this way
+    for diagnostics.
+    """
 
     def __init__(self) -> None:
-        """Initialize internal undecoded byte buffer."""
+        """Initialize internal undecoded byte buffer and resync counter."""
         self._buffer = bytearray()
+        self.discarded_byte_count = 0
 
     def clear(self) -> None:
         """Drop all buffered undecoded bytes."""
@@ -109,15 +119,31 @@ class FrameScanner:
 
             (length,) = struct.unpack_from("<H", self._buffer, 1)
             if length < _FRAME_OVERHEAD or length > _MAX_FRAME_LENGTH:
-                logger.debug("Discarding byte due to invalid frame length=%d", length)
-                del self._buffer[0]
+                self._discard_false_header(f"invalid frame length={length}")
                 continue
 
             if len(self._buffer) < length:
                 break
 
             frame = bytes(self._buffer[:length])
+            try:
+                packet = decode_frame(frame)
+            except (ParseError, CRCError) as exc:
+                # The 0x5A was a false header — e.g. a byte inside a CRC-less
+                # VNADatapoint payload, or a frame mangled by USB byte loss. Skip
+                # one byte and resync on the next candidate header rather than
+                # propagating (which would otherwise kill the whole transport).
+                self._discard_false_header(str(exc))
+                continue
+
             del self._buffer[:length]
-            decoded.append(decode_frame(frame))
+            decoded.append(packet)
 
         return decoded
+
+    def _discard_false_header(self, reason: str) -> None:
+        """Drop one buffered byte past a false frame header and count the resync."""
+        self.discarded_byte_count += 1
+        if logger.isEnabledFor(logging.DEBUG):
+            logger.debug("Resyncing frame stream past false header: %s", reason)
+        del self._buffer[0]

python_app/hardware_full/librevna_driver/transport/usb.py

@@ -130,7 +130,7 @@ class USBTransport:
             raise DeviceDisconnectedError(f"Failed to prepare USB kernel driver state: {exc}") from exc
 
         try:
-            selected_handle.claimInterface(self.INTERFACE)
+            self._claim_interface_with_busy_recovery(selected_handle)
         except usb1.USBError as exc:
             selected_handle.close()
             if self._ctx is not None:
@@ -152,6 +152,29 @@ class USBTransport:
         self._rx_thread = threading.Thread(target=self._rx_loop, name="librevna-usb-rx", daemon=True)
         self._rx_thread.start()
 
+    def _claim_interface_with_busy_recovery(self, handle: "usb1.USBDeviceHandle") -> None:  # type: ignore[name-defined]
+        """Claim the data interface, clearing a stale BUSY claim once if needed.
+
+        When a previous session's RX thread wedged inside libusb, ``disconnect()``
+        deliberately leaks that handle with the interface still claimed to avoid a
+        use-after-free. A fresh ``connect()`` then fails with
+        ``LIBUSB_ERROR_BUSY``. Resetting the device clears the orphaned claim so
+        the retry succeeds, instead of the device being unusable until it is
+        physically replugged. If the reset forces a re-enumeration the retry raises
+        and the caller falls back to its normal reopen backoff.
+        """
+        try:
+            handle.claimInterface(self.INTERFACE)
+            return
+        except usb1.USBErrorBusy as exc:
+            logger.warning(
+                "USB interface %d busy on claim; resetting device to clear a stale claim: %s",
+                self.INTERFACE,
+                exc,
+            )
+        handle.resetDevice()
+        handle.claimInterface(self.INTERFACE)
+
     def disconnect(self) -> None:
         """Stop RX thread and close USB resources."""
         logger.debug("USB disconnect requested")
@@ -234,8 +257,8 @@ class USBTransport:
                 continue
             except usb1.USBErrorNoDevice as exc:
                 if self._on_disconnect is not None:
-                    self._on_disconnect(DeviceDisconnectedError("USB device disconnected"))
-                logger.warning("USB RX stopped: device disconnected")
+                    self._on_disconnect(DeviceDisconnectedError(f"USB device disconnected: {exc}"))
+                logger.warning("USB RX stopped: device disconnected: %s", exc)
                 return
             except usb1.USBError as exc:
                 if self._stop_event.is_set():

python_app/hardware_full/librevna_multi_device_driver/controller.py

@@ -16,6 +16,7 @@ from python_app.hardware_full.librevna_multi_device_driver.cycle_collection impo
 from python_app.hardware_full.librevna_multi_device_driver.models import (
     SweepConfiguration,
     SweepMeasurementResult,
+    normalize_master_stimulus_ports,
 )
 from python_app.hardware_full.librevna_multi_device_driver.protocol import (
     PacketType,
@@ -367,12 +368,5 @@ class MultiDeviceVnaController:
 
     @staticmethod
     def _normalize_master_stimulus_ports(master_stimulus_ports: Sequence[int]) -> tuple[int, ...]:
-        """Validate and return master stimulus ports as a tuple of ints (ports 1/2 only)."""
-        stimulus_ports = tuple(int(port) for port in master_stimulus_ports)
-        if not stimulus_ports:
-            raise ValueError("master_stimulus_ports must not be empty")
-        if len(stimulus_ports) > 2 or set(stimulus_ports) - {1, 2}:
-            raise ValueError("master_stimulus_ports may contain only ports 1 and 2")
-        if len(set(stimulus_ports)) != len(stimulus_ports):
-            raise ValueError("master_stimulus_ports must not contain duplicates")
-        return stimulus_ports
+        """Validate and return master stimulus ports (ports 1/2 only, unique)."""
+        return normalize_master_stimulus_ports(master_stimulus_ports)

python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py

@@ -10,9 +10,13 @@ import time
 
 import numpy as np
 
+from python_app.hardware_full.librevna_multi_device_driver.exceptions import (
+    TransientCollectionError,
+)
 from python_app.hardware_full.librevna_multi_device_driver.models import (
     SweepConfiguration,
     SweepMeasurementResult,
+    normalize_master_stimulus_ports,
 )
 from python_app.hardware_full.librevna_multi_device_driver.protocol import (
     PacketType,
@@ -39,6 +43,102 @@ _MAX_FULL_CYCLE_SECONDS = 8.0
 # and let the orphan thread die when the producer process exits.
 _THREAD_JOIN_TIMEOUT_SECONDS = 2.0
 
+# Floor for the cross-device cycle-0 alignment tolerance (see
+# _CrossDeviceCycleSynchronizer). Keeps the tolerance sane for very fast sweeps
+# where points/IF-bandwidth would otherwise estimate a sub-millisecond cycle.
+_MIN_CYCLE_ALIGNMENT_TOLERANCE_SECONDS = 0.5e-3
+
+# Below this magnitude a master reference value is treated as a corrupt incident
+# signal rather than a real measurement: dividing by it would explode the
+# S-parameter into noise. Mirrors the single-device C++ driver's reference guard.
+_MIN_REFERENCE_MAGNITUDE = 1e-12
+
+
+class _CrossDeviceCycleSynchronizer:
+    """Confirm every device thread anchors cycle 0 on the SAME physical sweep.
+
+    The devices are hardware-trigger synchronized, so they emit ``point_index==0``
+    at the same physical instant. After a host-side queue drain, however, a sweep
+    wrap can slip between the per-device drains, leaving one collector anchored to
+    physical sweep N and another to N+1. Their reference and measurement would then
+    be divided across different sweeps, producing a whole-trace random-phase result
+    (the reported "noise sweep").
+
+    Each collector calls :meth:`confirm_aligned_cycle_zero` the moment it observes
+    its first ``point_index==0``, passing the host monotonic time it saw that wrap,
+    and the calls rendezvous at a :class:`threading.Barrier`. Because the wrap is
+    physically simultaneous, aligned threads observe it within host-latency jitter
+    of one another, while an off-by-one thread is ~one sweep period away. If the
+    spread of anchor times exceeds ``alignment_tolerance_seconds`` the alignment is
+    rejected, so the caller re-drains and re-collects instead of emitting noise.
+
+    A single device (no slaves) trivially aligns. If a peer never reaches the
+    barrier within ``rendezvous_timeout_seconds`` the barrier breaks and every
+    waiting thread reports "not aligned", which the caller also treats as a
+    transient failure to retry.
+    """
+
+    def __init__(
+        self,
+        *,
+        device_count: int,
+        alignment_tolerance_seconds: float,
+        rendezvous_timeout_seconds: float,
+    ) -> None:
+        """Create a one-shot synchronizer for ``device_count`` collector threads."""
+        self._barrier = threading.Barrier(max(1, int(device_count)))
+        self._alignment_tolerance_seconds = float(alignment_tolerance_seconds)
+        self._rendezvous_timeout_seconds = float(rendezvous_timeout_seconds)
+        self._lock = threading.Lock()
+        self._anchor_times: list[float] = []
+        self._is_aligned = False
+
+    def confirm_aligned_cycle_zero(self, anchor_monotonic_seconds: float) -> bool:
+        """Block until every device reported cycle 0; return whether all agree.
+
+        Returns ``True`` only when all devices anchored within the tolerance (same
+        physical sweep). Returns ``False`` on misalignment or if the rendezvous
+        times out / breaks.
+        """
+        with self._lock:
+            self._anchor_times.append(anchor_monotonic_seconds)
+        try:
+            arrival_index = self._barrier.wait(timeout=self._rendezvous_timeout_seconds)
+            # Exactly one thread computes the verdict; the second rendezvous
+            # publishes it to every thread before any of them reads it.
+            if arrival_index == 0:
+                spread_seconds = max(self._anchor_times) - min(self._anchor_times)
+                self._is_aligned = spread_seconds <= self._alignment_tolerance_seconds
+            self._barrier.wait(timeout=self._rendezvous_timeout_seconds)
+        except threading.BrokenBarrierError:
+            return False
+        return self._is_aligned
+
+
+def _expected_sweep_cycle_seconds(point_count: int, stage_count: int, if_bandwidth_hz: float) -> float:
+    """Return a conservative LOWER bound on one sweep cycle's wall-clock duration.
+
+    Every frequency point is measured once per excitation stage (one stage per
+    active master port), each taking at least one IF integration period
+    (1 / IF-bandwidth). Real sweeps are longer because of per-point settling/LO
+    overhead, so this underestimates — which is exactly what the cycle-alignment
+    tolerance wants (a tolerance safely below a true one-sweep off-by-one).
+    Measured anchor: 751 points x 2 stages at 50 kHz -> 30 ms here vs a real
+    ~150 ms sweep.
+    """
+    return (int(point_count) * max(1, int(stage_count))) / max(1.0, float(if_bandwidth_hz))
+
+
+def _is_transient_collection_error(error: BaseException) -> bool:
+    """Return whether a collection error is recoverable without a USB reopen.
+
+    Timeouts (no-progress / cycle / cycle-start guards) and explicit transient
+    errors mean the devices are still alive and only this cycle was lost. A
+    genuine transport failure (surfaced as a plain RuntimeError from the USB
+    layer) is fatal and must fall through to the heavy recovery path.
+    """
+    return isinstance(error, (TransientCollectionError, TimeoutError))
+
 
 def collect_complete_running_sweep_cycles(
     *,
@@ -53,13 +153,7 @@ def collect_complete_running_sweep_cycles(
     cycle_count = int(cycle_count)
     if cycle_count < 1:
         raise ValueError("cycle_count must be >= 1")
-    stimulus_ports = tuple(int(port) for port in master_stimulus_ports)
-    if not stimulus_ports:
-        raise ValueError("master_stimulus_ports must not be empty")
-    if len(stimulus_ports) > 2 or set(stimulus_ports) - {1, 2}:
-        raise ValueError("master_stimulus_ports may contain only ports 1 and 2")
-    if len(set(stimulus_ports)) != len(stimulus_ports):
-        raise ValueError("master_stimulus_ports must not contain duplicates")
+    stimulus_ports = normalize_master_stimulus_ports(master_stimulus_ports)
     stage_by_master_port = {port: stage for stage, port in enumerate(stimulus_ports)}
 
     all_device_connections = [master_device_connection, *slave_device_connections]
@@ -79,6 +173,20 @@ def collect_complete_running_sweep_cycles(
         device_connection.serial_number: 0
         for device_connection in all_device_connections
     }
+    rejected_datapoint_counts_by_device_serial = {
+        device_connection.serial_number: 0
+        for device_connection in all_device_connections
+    }
+
+    # In-band frequency gate. A misframed CRC-less datapoint that still parses
+    # carries an essentially random 64-bit frequency, which almost never lands in
+    # the swept band, so this rejects corrupt points cheaply. One frequency step of
+    # margin keeps every legitimate point (which lies within [start, stop]).
+    band_low_hz = float(min(active_sweep_configuration.start_hz, active_sweep_configuration.stop_hz))
+    band_high_hz = float(max(active_sweep_configuration.start_hz, active_sweep_configuration.stop_hz))
+    frequency_step_hz = (band_high_hz - band_low_hz) / (point_count - 1) if point_count > 1 else band_high_hz
+    band_low_hz -= frequency_step_hz
+    band_high_hz += frequency_step_hz
     stop_collection_requested = threading.Event()
     if datapoint_timeout_seconds is None:
         datapoint_timeout_seconds = LIBREVNA_NATIVE_SWEEP_TIMEOUT_SECONDS
@@ -92,6 +200,57 @@ def collect_complete_running_sweep_cycles(
     # stall capture indefinitely.
     cycle_start_guard_seconds = max(2.0, datapoint_timeout_seconds * 4.0)
 
+    # Cross-device cycle-0 alignment. The off-by-one race (master anchored to a
+    # different physical sweep than a slave) shows up as anchor-time spreads on the
+    # order of one sweep period, whereas aligned anchors differ only by host
+    # latency jitter. The tolerance is half of a conservative LOWER bound on the
+    # sweep period: every frequency point is measured once per excitation stage
+    # (one stage per active master port), and the real period is strictly larger
+    # still because point_count*stage_count/IF-bandwidth ignores per-point settling
+    # overhead. That keeps the tolerance well below a true off-by-one yet well above
+    # host jitter. (Measured anchor: 751 points x 2 stages at 50 kHz IF -> a real
+    # ~150 ms sweep; this estimates 30 ms, tolerance 15 ms, so an off-by-one of
+    # ~150 ms is flagged with ~10x margin and ~1 ms jitter clears it with ~7x.)
+    # A spread above tolerance is rejected and retried, never silently emitted.
+    stage_count = max(1, len(stimulus_ports))
+    expected_cycle_seconds = _expected_sweep_cycle_seconds(
+        point_count, stage_count, active_sweep_configuration.if_bandwidth
+    )
+    cycle_alignment_tolerance_seconds = max(
+        _MIN_CYCLE_ALIGNMENT_TOLERANCE_SECONDS, 0.5 * expected_cycle_seconds
+    )
+    # Keep the rendezvous timeout below the collector join deadline: if a device
+    # stops streaming, the threads waiting at the barrier then break and exit with
+    # a transient error (cheap re-collect) instead of being orphaned into the
+    # "failed to stop" path that forces a USB reopen. It still allows several sweep
+    # periods for the slowest device to reach its first point 0 after a drain.
+    cycle_rendezvous_timeout_seconds = min(
+        0.9 * _THREAD_JOIN_TIMEOUT_SECONDS, max(1.0, 5.0 * expected_cycle_seconds)
+    )
+    cycle_synchronizer = _CrossDeviceCycleSynchronizer(
+        device_count=len(all_device_connections),
+        alignment_tolerance_seconds=cycle_alignment_tolerance_seconds,
+        rendezvous_timeout_seconds=cycle_rendezvous_timeout_seconds,
+    )
+    cycle_misalignment_reported = threading.Event()
+
+    def report_cycle_misalignment() -> None:
+        """Record a transient misalignment once and stop all collector threads."""
+        if not cycle_misalignment_reported.is_set():
+            cycle_misalignment_reported.set()
+            logger.warning(
+                "Cross-device cycle-0 misalignment detected (tolerance %.3f ms); "
+                "rejecting collection for retry",
+                cycle_alignment_tolerance_seconds * 1e3,
+            )
+            collection_errors.append(
+                TransientCollectionError(
+                    "Cross-device cycle-0 misalignment: master reference and slave "
+                    "measurement would span different sweeps"
+                )
+            )
+        stop_collection_requested.set()
+
     def collect_datapoints_from_device(
         device_connection: LibreVnaUsbBulkConnection,
         handle_datapoint: Callable[[ParsedVnaDatapoint], bool],
@@ -211,15 +370,25 @@ def collect_complete_running_sweep_cycles(
                 continue
 
             parsed_datapoint = parse_vna_datapoint_payload(payload)
-            if parsed_datapoint and 0 <= parsed_datapoint.point_index < point_count:
-                datapoint_was_consumed = handle_datapoint(parsed_datapoint)
-                if datapoint_was_consumed:
-                    # Only refreshed on accepted datapoints so the no-progress
-                    # timeout above stays honest about real cycle progress.
-                    last_consumed_timestamp = time.monotonic()
-                    has_consumed_any_datapoint = True
-                    datapoint_counts_by_device_serial[device_connection.serial_number] += 1
-                    datapoints_received += 1
+            if (
+                parsed_datapoint is None
+                or not (0 <= parsed_datapoint.point_index < point_count)
+                or not (band_low_hz <= parsed_datapoint.frequency_hz <= band_high_hz)
+            ):
+                # Unparseable, out-of-range index, or out-of-band frequency: almost
+                # certainly a corrupt/misframed point. Drop it without refreshing the
+                # progress timer so a stream of garbage still trips the timeout.
+                rejected_datapoint_counts_by_device_serial[device_connection.serial_number] += 1
+                continue
+
+            datapoint_was_consumed = handle_datapoint(parsed_datapoint)
+            if datapoint_was_consumed:
+                # Only refreshed on accepted datapoints so the no-progress
+                # timeout above stays honest about real cycle progress.
+                last_consumed_timestamp = time.monotonic()
+                has_consumed_any_datapoint = True
+                datapoint_counts_by_device_serial[device_connection.serial_number] += 1
+                datapoints_received += 1
 
     def build_cycle_tracking_handler(
         cycle_aware_handler: Callable[[ParsedVnaDatapoint, int], None],
@@ -231,18 +400,18 @@ def collect_complete_running_sweep_cycles(
         wrap, drops datapoints past ``cycle_count``, and reports whether each
         datapoint was consumed.
         """
-        # The controller restarts the sweep before every collection, so the
-        # first packet each device emits is point 0 of a brand-new cycle 0.
-        # Anchoring cycle 0 on the first observed point_index==0 — instead of
-        # synthesising it from a wrap — pins master and slave threads to the
-        # same physical cycle even if a stale straggler from the just-stopped
-        # sweep escaped the post-idle drain: such a straggler always carries
-        # a non-zero point_index and is discarded until the genuine cycle 0
-        # arrives. From that anchor, each subsequent wrap advances the cycle
-        # counter normally.
+        # In steady state the controller leaves the hardware sweep FREE-RUNNING
+        # between collections (the unchanged-config fast path only drains the host
+        # queues), so the first datapoint each device emits after a drain is some
+        # mid-sweep point, not a fresh point 0. Each thread therefore discards
+        # until the next observed point_index==0 and anchors cycle 0 there. Because
+        # the drain can let a wrap slip between devices, that first point 0 is only
+        # accepted once `_CrossDeviceCycleSynchronizer` confirms every device saw
+        # point 0 of the SAME physical sweep; a mismatch rejects the collection for
+        # a transient retry. From the agreed anchor, each subsequent wrap advances
+        # the cycle counter normally.
         cycle_tracking_state = {
             "current_cycle_index": 0,
-            "previous_point_index": -1,
             "synchronized": False,
         }
 
@@ -258,14 +427,23 @@ def collect_complete_running_sweep_cycles(
             if not cycle_tracking_state["synchronized"]:
                 if current_point_index != 0:
                     return False
+                # Candidate cycle 0. Commit it only once every device confirms it
+                # observed point 0 of the SAME physical sweep; otherwise reject the
+                # whole collection so the caller re-drains and re-collects rather
+                # than pairing a reference and measurement from different sweeps.
+                if not cycle_synchronizer.confirm_aligned_cycle_zero(time.monotonic()):
+                    report_cycle_misalignment()
+                    return False
                 cycle_tracking_state["synchronized"] = True
-                cycle_tracking_state["previous_point_index"] = current_point_index
                 cycle_aware_handler(parsed_datapoint, 0)
                 return True
 
-            if current_point_index < cycle_tracking_state["previous_point_index"]:
+            # A new sweep cycle begins exactly when the point index wraps to 0.
+            # Keying on ==0 (rather than "the index decreased") means a corrupt
+            # mid-cycle index that slipped past the in-band gate cannot fabricate a
+            # spurious wrap and desynchronize the cycle counter.
+            if current_point_index == 0:
                 cycle_tracking_state["current_cycle_index"] += 1
-            cycle_tracking_state["previous_point_index"] = current_point_index
             current_cycle_index = cycle_tracking_state["current_cycle_index"]
             if current_cycle_index >= cycle_count:
                 return False
@@ -419,7 +597,15 @@ def collect_complete_running_sweep_cycles(
             )
 
     if collection_errors:
-        raise RuntimeError(f"Sweep collection failed: {collection_errors[0]}") from collection_errors[0]
+        first_error = collection_errors[0]
+        # Timeouts / no-progress / misalignment leave the devices healthy: surface
+        # them as transient so the caller re-arms and re-collects instead of doing a
+        # full USB reopen. A genuine transport failure (e.g. USB death) is fatal.
+        if _is_transient_collection_error(first_error):
+            raise TransientCollectionError(
+                f"Sweep collection failed transiently: {first_error}"
+            ) from first_error
+        raise RuntimeError(f"Sweep collection failed: {first_error}") from first_error
 
     if slave_device_connections and min(datapoint_counts_by_device_serial.values(), default=0) == 0:
         logger.error(
@@ -427,11 +613,16 @@ def collect_complete_running_sweep_cycles(
             "(per-device counts: %s)",
             datapoint_counts_by_device_serial,
         )
-        raise RuntimeError(
+        raise TransientCollectionError(
             "No datapoints received from at least one device; hardware trigger sync did not start. "
             "Check Trigger Out/In loop and 10 MHz reference wiring."
         )
 
+    if any(rejected_datapoint_counts_by_device_serial.values()):
+        logger.debug(
+            "Sweep cycle collection rejected corrupt datapoints (per-device: %s)",
+            rejected_datapoint_counts_by_device_serial,
+        )
     logger.debug("Sweep cycle collection complete (per-device counts: %s)", datapoint_counts_by_device_serial)
     return SweepMeasurementResult(
         frequencies_hz=frequencies_hz,
@@ -476,13 +667,23 @@ def calculate_last_cycle_s_parameters(
         master_reference_measurements = master_reference_measurements_by_port[master_stimulus_port]
         if np.isnan(master_reference_measurements).any():
             missing_reference_count = int(np.isnan(master_reference_measurements).sum())
-            raise RuntimeError(
+            raise TransientCollectionError(
                 f"Master port {master_stimulus_port} reference missing for {missing_reference_count} datapoints"
             )
         if np.isnan(raw_receiver_measurements).any():
             missing_measurement_count = int(np.isnan(raw_receiver_measurements).sum())
-            raise RuntimeError(
+            raise TransientCollectionError(
                 f"Measurement {s_parameter_name} missing for {missing_measurement_count} datapoints"
             )
-        s_parameters[s_parameter_name.lower()] = raw_receiver_measurements[-1] / master_reference_measurements[-1]
+
+        last_cycle_reference = master_reference_measurements[-1]
+        near_zero_reference_mask = np.abs(last_cycle_reference) <= _MIN_REFERENCE_MAGNITUDE
+        if near_zero_reference_mask.any():
+            # A ~zero incident reference is a corrupt point: dividing by it would
+            # explode this S-parameter into noise. Reject the cycle for retry.
+            raise TransientCollectionError(
+                f"Master port {master_stimulus_port} reference is ~zero for "
+                f"{int(near_zero_reference_mask.sum())} datapoints (corrupt incident signal)"
+            )
+        s_parameters[s_parameter_name.lower()] = raw_receiver_measurements[-1] / last_cycle_reference
     return s_parameters

python_app/hardware_full/librevna_multi_device_driver/exceptions.py

@@ -0,0 +1,15 @@
+"""Exception types for multi-device LibreVNA acquisition.
+
+The distinction matters for recovery policy: a ``TransientCollectionError`` means
+the devices are still healthy and the sweep merely produced an unusable cycle
+(a dropped/incomplete datapoint, a no-progress timeout, or a cross-device cycle
+misalignment). Such errors are cheaply recoverable by re-arming and re-collecting
+— they must NOT trigger the heavy USB close()/reopen path, which is reserved for
+genuine transport death.
+"""
+
+from __future__ import annotations
+
+
+class TransientCollectionError(RuntimeError):
+    """A recoverable single-cycle collection failure; retry without reopening USB."""

python_app/hardware_full/librevna_multi_device_driver/models.py

@@ -2,11 +2,29 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from dataclasses import dataclass
 
 import numpy as np
 
 
+def normalize_master_stimulus_ports(master_stimulus_ports: Sequence[int]) -> tuple[int, ...]:
+    """Validate master stimulus ports and return them as a tuple of ints.
+
+    Ports may only be 1 and/or 2, with no duplicates and at least one entry. This
+    is the single source of truth shared by the protocol builder, the controller,
+    and the cycle collector so the rule cannot drift between call sites.
+    """
+    stimulus_ports = tuple(int(port) for port in master_stimulus_ports)
+    if not stimulus_ports:
+        raise ValueError("master_stimulus_ports must not be empty")
+    if len(stimulus_ports) > 2 or set(stimulus_ports) - {1, 2}:
+        raise ValueError("master_stimulus_ports may contain only ports 1 and 2")
+    if len(set(stimulus_ports)) != len(stimulus_ports):
+        raise ValueError("master_stimulus_ports must not contain duplicates")
+    return stimulus_ports
+
+
 @dataclass(frozen=True, slots=True)
 class SweepConfiguration:
     """User-facing configuration for one VNA sweep."""

python_app/hardware_full/librevna_multi_device_driver/protocol.py

@@ -6,7 +6,10 @@ import struct
 from collections.abc import Sequence
 from dataclasses import dataclass
 
-from python_app.hardware_full.librevna_multi_device_driver.models import SweepConfiguration
+from python_app.hardware_full.librevna_multi_device_driver.models import (
+    SweepConfiguration,
+    normalize_master_stimulus_ports,
+)
 
 
 class PacketType:
@@ -51,13 +54,7 @@ def build_sweep_settings_payload(
     master_stimulus_ports: Sequence[int],
 ) -> bytes:
     """Build protocol-v14 SweepSettings for staged master outputs or synchronized receivers."""
-    stimulus_ports = tuple(int(port) for port in master_stimulus_ports)
-    if not stimulus_ports:
-        raise ValueError("master_stimulus_ports must not be empty")
-    if len(stimulus_ports) > 2 or set(stimulus_ports) - {1, 2}:
-        raise ValueError("master_stimulus_ports may contain only ports 1 and 2")
-    if len(set(stimulus_ports)) != len(stimulus_ports):
-        raise ValueError("master_stimulus_ports must not contain duplicates")
+    stimulus_ports = normalize_master_stimulus_ports(master_stimulus_ports)
 
     excitation_power_centidecibels_milliwatt = int(round(sweep_configuration.power_dbm * 100.0))
     dwell_time_microseconds = max(0, min(int(sweep_configuration.dwell_us), 0xFFFF))
@@ -109,10 +106,17 @@ def parse_vna_datapoint_payload(payload: bytes) -> ParsedVnaDatapoint | None:
     if len(payload) < fixed_header_length_bytes:
         return None
 
-    frequency_hz, _power_level_centidecibels_milliwatt, point_index = struct.unpack_from("<QhH", payload, 0)
-    value_count = (len(payload) - fixed_header_length_bytes) // per_value_storage_length_bytes
-    if value_count == 0:
+    # A genuine VNADatapoint payload is a 12-byte header followed by exactly
+    # `value_count` blocks of (float32 real, float32 imag, uint8 mask) = 9 bytes.
+    # Since these frames carry no CRC, a misframe (resync onto a 0x5A inside a
+    # payload) can still pass framing — reject any payload whose body is not a
+    # whole number of value blocks, which a random misframe almost never is.
+    value_bytes = len(payload) - fixed_header_length_bytes
+    if value_bytes <= 0 or value_bytes % per_value_storage_length_bytes != 0:
         return None
+    value_count = value_bytes // per_value_storage_length_bytes
+
+    frequency_hz, _power_level_centidecibels_milliwatt, point_index = struct.unpack_from("<QhH", payload, 0)
 
     real_components = struct.unpack_from(f"<{value_count}f", payload, 12)
     imaginary_components = struct.unpack_from(f"<{value_count}f", payload, 12 + 4 * value_count)

python_app/hardware_full/librevna_multi_device_driver/transport.py

@@ -35,7 +35,10 @@ class LibreVnaUsbBulkConnection:
         self._transport = USBTransport(
             on_data=self._on_data,
             on_disconnect=self._on_disconnect,
-            read_chunk_size=4096,
+            # Larger bulk reads amortize the per-transfer overhead during the
+            # high-rate datapoint stream; FrameScanner reassembles frames across
+            # chunks, so chunk size does not affect decoding.
+            read_chunk_size=65536,
         )
         logger.debug("Opening LibreVNA USB connection (serial=%s)", serial_number)
         self._transport.connect(serial=serial_number, timeout_s=2.0)
@@ -93,13 +96,16 @@ class LibreVnaUsbBulkConnection:
     def _on_data(self, chunk: bytes) -> None:
         """Decode a received USB chunk into frames and queue (type, payload) tuples.
 
-        Any decode failure is recorded as the fatal transport error so the next
-        send/receive call surfaces it to the caller.
+        ``FrameScanner.feed`` already resyncs past malformed/corrupted frames
+        without raising, so a single bad chunk no longer poisons the device. Any
+        residual unexpected scanner error is logged and the chunk dropped — genuine
+        transport death is reported separately through ``_on_disconnect`` — so a
+        recoverable parsing glitch can never escalate to a full device teardown.
         """
         try:
             packets = self._scanner.feed(chunk)
-        except Exception as exc:  # noqa: BLE001
-            self._set_fatal_error(exc)
+        except Exception as exc:  # noqa: BLE001 — never let a parse glitch kill the device
+            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)))

python_app/hardware_full/multi_device_service.py

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING
 import numpy as np
 
 from python_app.hardware_full.librevna_multi_device_driver.cycle_collection import LIBREVNA_NATIVE_SWEEP_TIMEOUT_SECONDS
+from python_app.hardware_full.librevna_multi_device_driver.exceptions import TransientCollectionError
 from python_app.hardware_full.librevna_multi_device_driver.models import SweepConfiguration
 from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData
 from python_app.models.run_config_model import RadarSweepModel, RunConfigModel
@@ -21,10 +22,11 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-# Delays applied between successive USB reopen attempts inside recover(). Picked
-# to give libusb time to re-enumerate a stuck device while staying short enough
-# that a healthy reconnect feels instant. The total worst-case wait is the sum
-# of all entries (1.75 s today) plus the cost of close()/open() themselves.
+# Backoff delays applied only BEFORE each reopen RETRY inside recover(); the first
+# attempt runs immediately (no upfront sleep) so a transient stall recovers at
+# once. Picked to give libusb time to re-enumerate a stuck device while staying
+# short enough that a healthy reconnect feels instant. The total worst-case extra
+# wait across all retries is the sum of these entries (1.75 s today).
 _REOPEN_BACKOFF_SECONDS: tuple[float, ...] = (0.25, 0.5, 1.0)
 
 # Hard ceiling on the wall-clock time a single acquire_collection() may spend in
@@ -33,6 +35,14 @@ _REOPEN_BACKOFF_SECONDS: tuple[float, ...] = (0.25, 0.5, 1.0)
 # systemd TimeoutStopSec so the unit is never SIGKILLed for hanging on exit.
 _MAX_RECOVERY_WALL_SECONDS: float = 10.0
 
+# How many times a TRANSIENT collection failure (dropped/incomplete datapoint,
+# no-progress timeout, or cross-device cycle misalignment) is retried in place —
+# re-arming the sweep and re-collecting WITHOUT a USB reopen — before escalating
+# to the heavy close()/reopen recovery. Each retry costs ~one sweep period, so a
+# handful absorbs ordinary glitches without the multi-second re-enumeration that
+# previously caused the periodic freeze.
+_MAX_TRANSIENT_COLLECTION_RETRIES: int = 2
+
 _INPUT_S_PARAMETERS_BY_OUTPUT: dict[int, tuple[str, ...]] = {
     0: ("s31", "s41", "s51", "s61"),
     1: ("s32", "s42", "s52", "s62"),
@@ -133,8 +143,12 @@ class MultiDeviceLibreVnaService:
             return
         self.close()
 
+        # Try to reopen immediately first, then back off only after a failure: a
+        # transient USB stall usually clears at once, so the common recovery should
+        # not pay an upfront sleep. The backoff delays apply only between retries.
+        reopen_delays = (0.0, *_REOPEN_BACKOFF_SECONDS)
         last_error: Exception | None = None
-        for attempt_index, delay_s in enumerate(_REOPEN_BACKOFF_SECONDS, start=1):
+        for attempt_index, delay_s in enumerate(reopen_delays, start=1):
             # Bail out the instant a stop is requested or the recovery budget is
             # spent, rather than committing to another (re)open attempt.
             if stop_event is not None and stop_event.is_set():
@@ -143,20 +157,21 @@ class MultiDeviceLibreVnaService:
             if deadline_monotonic is not None and time.monotonic() >= deadline_monotonic:
                 logger.warning("Multi-device recover() aborted: recovery time budget exhausted")
                 break
-            # Interruptible backoff: wait() returns early the moment stop is set.
-            if stop_event is not None:
-                if stop_event.wait(delay_s):
-                    logger.info("Multi-device recover() aborted: stop requested")
-                    return
-            else:
-                time.sleep(delay_s)
+            # Interruptible backoff before retries; the first attempt has no delay.
+            if delay_s > 0.0:
+                if stop_event is not None:
+                    if stop_event.wait(delay_s):
+                        logger.info("Multi-device recover() aborted: stop requested")
+                        return
+                else:
+                    time.sleep(delay_s)
             try:
                 self.open()
                 if self._controller is not None:
                     logger.info(
                         "Multi-device reopen succeeded on attempt %d/%d (after %.2fs)",
                         attempt_index,
-                        len(_REOPEN_BACKOFF_SECONDS),
+                        len(reopen_delays),
                         delay_s,
                     )
                     return
@@ -165,7 +180,7 @@ class MultiDeviceLibreVnaService:
                 logger.warning(
                     "Multi-device reopen attempt %d/%d failed after %.2fs: %s",
                     attempt_index,
-                    len(_REOPEN_BACKOFF_SECONDS),
+                    len(reopen_delays),
                     delay_s,
                     exc,
                 )
@@ -230,7 +245,9 @@ class MultiDeviceLibreVnaService:
         recovery_deadline = time.monotonic() + _MAX_RECOVERY_WALL_SECONDS
         for attempt_index in range(self.recovery_attempts + 1):
             try:
-                return self._acquire_native_collection(collection_id, capture_start_ns)
+                return self._acquire_native_collection_with_transient_retries(
+                    collection_id, capture_start_ns, stop_event=stop_event
+                )
             except Exception as exc:  # noqa: BLE001
                 last_error = exc
                 if attempt_index >= self.recovery_attempts:
@@ -272,6 +289,43 @@ class MultiDeviceLibreVnaService:
         assert last_error is not None
         raise last_error
 
+    def _acquire_native_collection_with_transient_retries(
+        self,
+        collection_id: int,
+        capture_start_ns: int,
+        *,
+        stop_event: threading.Event | None = None,
+    ) -> SweepCollection:
+        """Acquire one collection, retrying device-healthy failures in place.
+
+        A ``TransientCollectionError`` (dropped/incomplete datapoint, no-progress
+        timeout, or cross-device cycle misalignment) leaves the USB transports
+        alive, so it is recovered by simply re-collecting: the controller auto-idled
+        on failure, so the next ``configure_continuous_sweep`` re-sends
+        ``SWEEP_SETTINGS`` (re-arm). This costs ~one sweep period instead of the
+        multi-second ``close()``/reopen that genuine transport death requires.
+        Non-transient errors propagate immediately to the reopen-based recovery.
+        """
+        transient_error: TransientCollectionError | None = None
+        for transient_attempt in range(_MAX_TRANSIENT_COLLECTION_RETRIES + 1):
+            try:
+                return self._acquire_native_collection(collection_id, capture_start_ns)
+            except TransientCollectionError as exc:
+                transient_error = exc
+                if transient_attempt >= _MAX_TRANSIENT_COLLECTION_RETRIES:
+                    raise
+                if stop_event is not None and stop_event.is_set():
+                    raise
+                logger.debug(
+                    "transient multi-device collection error (%d/%d); re-arming and re-collecting "
+                    "without USB reopen: %s",
+                    transient_attempt + 1,
+                    _MAX_TRANSIENT_COLLECTION_RETRIES,
+                    exc,
+                )
+        assert transient_error is not None  # loop either returns or raises
+        raise transient_error
+
     def _acquire_native_collection(self, collection_id: int, capture_start_ns: int) -> SweepCollection:
         if self._controller is None:
             raise RuntimeError("Multi-device controller is not open")