diff --git a/data_acq_and_processing/processing/locator/include/locator/observation.hpp b/data_acq_and_processing/processing/locator/include/locator/observation.hpp index 44482bc..b05722d 100644 --- a/data_acq_and_processing/processing/locator/include/locator/observation.hpp +++ b/data_acq_and_processing/processing/locator/include/locator/observation.hpp @@ -4,11 +4,11 @@ namespace radar::locator { // One outbound locator observation: object position relative to the radar in metres. // `crs` is the cross-range (X) coordinate; `dst` is the range (Z) coordinate. -// Values are kept as float because they are quantised to two decimal places before -// being serialised on the wire. +// Values use double so that, after quantisation to one decimal place (decimetres), +// they serialise to clean JSON (e.g. "12.3" rather than a float's "12.300000190734863"). struct Observation { - float crs = 0.0F; - float dst = 0.0F; + double crs = 0.0; + double dst = 0.0; }; } // namespace radar::locator diff --git a/data_acq_and_processing/processing/locator/src/payload_builder.cpp b/data_acq_and_processing/processing/locator/src/payload_builder.cpp index 7a4bf77..f3f1bd2 100644 --- a/data_acq_and_processing/processing/locator/src/payload_builder.cpp +++ b/data_acq_and_processing/processing/locator/src/payload_builder.cpp @@ -42,10 +42,11 @@ constexpr std::uint32_t kMinTableColumns = 3U; // [x_m, z_m, score, ...] return fallback; } -// Quantise to two decimal places. Equivalent to Python's `round(value, 2)` -// but explicit so behaviour does not silently depend on the local C library. -[[nodiscard]] auto quantise_to_centimetres(float value) -> float { - return std::round(value * 100.0F) / 100.0F; +// Quantise to one decimal place (0.1 m = decimetres). Equivalent to Python's +// `round(value, 1)` but explicit so behaviour does not silently depend on the +// local C library. +[[nodiscard]] auto quantise_to_decimetres(double value) -> double { + return std::round(value * 10.0) / 10.0; } [[nodiscard]] auto passes_basic_filter( @@ -156,8 +157,8 @@ auto observations_from_collection( quantised.reserve(visible.size()); for (const auto& observation : visible) { quantised.push_back({ - .crs = quantise_to_centimetres(observation.crs), - .dst = quantise_to_centimetres(observation.dst), + .crs = quantise_to_decimetres(observation.crs), + .dst = quantise_to_decimetres(observation.dst), }); } return quantised; @@ -172,8 +173,8 @@ auto observations_from_collection( result.reserve(kept); for (std::size_t index = 0; index < kept; ++index) { result.push_back({ - .crs = quantise_to_centimetres(visible[index].crs), - .dst = quantise_to_centimetres(visible[index].dst), + .crs = quantise_to_decimetres(visible[index].crs), + .dst = quantise_to_decimetres(visible[index].dst), }); } return result; diff --git a/data_acq_and_processing/processing/processors/src/gpr_backprojection_processor.ipp b/data_acq_and_processing/processing/processors/src/gpr_backprojection_processor.ipp index 61fcccb..2c98186 100644 --- a/data_acq_and_processing/processing/processors/src/gpr_backprojection_processor.ipp +++ b/data_acq_and_processing/processing/processors/src/gpr_backprojection_processor.ipp @@ -319,6 +319,38 @@ void normalize_in_place(std::vector& values) { } } +// Run `body(row_begin, row_end)` over a partition of [0, row_count) across the +// available hardware threads. Each call owns a disjoint, contiguous row range, so +// a body that writes only its own rows needs no synchronization. The calling +// thread runs the first chunk while spawned workers handle the rest. Falls back to +// a single serial call when there is one row or no concurrency is reported. +template +void parallel_for_rows(std::size_t row_count, const Body& body) { + if (row_count == 0U) { + return; + } + + const unsigned int detected = std::thread::hardware_concurrency(); + const std::size_t worker_count = std::clamp( + detected == 0U ? 1U : static_cast(detected), 1U, row_count + ); + if (worker_count == 1U) { + body(0U, row_count); + return; + } + + const std::size_t chunk = (row_count + worker_count - 1U) / worker_count; + std::vector workers; + workers.reserve(worker_count - 1U); + for (std::size_t begin = chunk; begin < row_count; begin += chunk) { + workers.emplace_back(body, begin, std::min(begin + chunk, row_count)); + } + body(0U, std::min(chunk, row_count)); + for (auto& worker : workers) { + worker.join(); + } +} + [[nodiscard]] auto build_gaussian_kernel(double sigma) -> std::vector { if (!(sigma > 0.0)) { return {1.0}; @@ -355,27 +387,36 @@ void normalize_in_place(std::vector& values) { std::vector temp(values.size(), 0.0); std::vector output(values.size(), 0.0); - for (std::size_t row = 0U; row < height; ++row) { - for (std::size_t col = 0U; col < width; ++col) { - double sum = 0.0; - for (std::ptrdiff_t offset = -radius; offset <= radius; ++offset) { - const auto sample_col = reflect_index(static_cast(col) + offset, width); - sum += values[(row * width) + sample_col] * kernel[static_cast(offset + radius)]; + // Separable convolution: every output row depends only on its own row (pass 1) + // or only on already-finished `temp` (pass 2), so each pass parallelizes over + // disjoint rows. parallel_for_rows joins between passes — that join is the + // barrier guaranteeing `temp` is complete before pass 2 reads it. Per-cell + // arithmetic is unchanged, so the result is identical to a serial sweep. + parallel_for_rows(height, [&](std::size_t row_begin, std::size_t row_end) { + for (std::size_t row = row_begin; row < row_end; ++row) { + for (std::size_t col = 0U; col < width; ++col) { + double sum = 0.0; + for (std::ptrdiff_t offset = -radius; offset <= radius; ++offset) { + const auto sample_col = reflect_index(static_cast(col) + offset, width); + sum += values[(row * width) + sample_col] * kernel[static_cast(offset + radius)]; + } + temp[(row * width) + col] = sum; } - temp[(row * width) + col] = sum; } - } + }); - for (std::size_t row = 0U; row < height; ++row) { - for (std::size_t col = 0U; col < width; ++col) { - double sum = 0.0; - for (std::ptrdiff_t offset = -radius; offset <= radius; ++offset) { - const auto sample_row = reflect_index(static_cast(row) + offset, height); - sum += temp[(sample_row * width) + col] * kernel[static_cast(offset + radius)]; + parallel_for_rows(height, [&](std::size_t row_begin, std::size_t row_end) { + for (std::size_t row = row_begin; row < row_end; ++row) { + for (std::size_t col = 0U; col < width; ++col) { + double sum = 0.0; + for (std::ptrdiff_t offset = -radius; offset <= radius; ++offset) { + const auto sample_row = reflect_index(static_cast(row) + offset, height); + sum += temp[(sample_row * width) + col] * kernel[static_cast(offset + radius)]; + } + output[(row * width) + col] = sum; } - output[(row * width) + col] = sum; } - } + }); return output; } @@ -901,6 +942,32 @@ void normalize_pair_ascans( return grid; } +// Compact signature that fully determines the imaging grid: the antenna layout +// plus the depth window and imaging plane. Two collections with an equal signature +// produce an identical grid, so the (expensive) grid build can be memoized. +[[nodiscard]] auto grid_signature( + const GeometrySelection& selection, + double max_depth_m, + double min_z_m, + double imaging_plane_y_m +) -> std::vector { + std::vector signature; + signature.reserve((selection.x_tx.size() + selection.x_rx.size()) * 3U + 3U); + const auto append = [&](const std::vector& axis) { + signature.insert(signature.end(), axis.begin(), axis.end()); + }; + append(selection.x_tx); + append(selection.y_tx); + append(selection.z_tx); + append(selection.x_rx); + append(selection.y_rx); + append(selection.z_rx); + signature.push_back(max_depth_m); + signature.push_back(min_z_m); + signature.push_back(imaging_plane_y_m); + return signature; +} + [[nodiscard]] auto interpolate_complex(const AscanResult& ascan, double tau_s) -> std::complex { if (ascan.samples.empty() || !(ascan.dt_s > 0.0) || tau_s < 0.0) { return {0.0, 0.0}; @@ -982,38 +1049,6 @@ void normalize_pair_ascans( return std::clamp(range_weight * angle_weight, 0.0, kTotalWeightMax); } -// Run `body(row_begin, row_end)` over a partition of [0, row_count) across the -// available hardware threads. Each call owns a disjoint, contiguous row range, so -// a body that writes only its own rows needs no synchronization. The calling -// thread runs the first chunk while spawned workers handle the rest. Falls back to -// a single serial call when there is one row or no concurrency is reported. -template -void parallel_for_rows(std::size_t row_count, const Body& body) { - if (row_count == 0U) { - return; - } - - const unsigned int detected = std::thread::hardware_concurrency(); - const std::size_t worker_count = std::clamp( - detected == 0U ? 1U : static_cast(detected), 1U, row_count - ); - if (worker_count == 1U) { - body(0U, row_count); - return; - } - - const std::size_t chunk = (row_count + worker_count - 1U) / worker_count; - std::vector workers; - workers.reserve(worker_count - 1U); - for (std::size_t begin = chunk; begin < row_count; begin += chunk) { - workers.emplace_back(body, begin, std::min(begin + chunk, row_count)); - } - body(0U, std::min(chunk, row_count)); - for (auto& worker : workers) { - worker.join(); - } -} - [[nodiscard]] auto backproject_coherent( const std::vector& selected_traces, const std::unordered_map& ascans_by_pair, @@ -1857,7 +1892,20 @@ void add_bp_score_metrics( normalize_pair_ascans(ascans_by_pair, velocity_mps, min_depth_m, max_depth_m); const double imaging_plane_y_m = static_cast(live_config.gpr_imaging_plane_y_m); - const auto grid = build_grid(selection, max_depth_m, kGridZMinM, imaging_plane_y_m); + + // The imaging grid (90k cells x 6 distance fields) only depends on antenna + // geometry and the depth window, which are constant across a run, so it is + // memoized between collections and rebuilt only when that signature changes. + // process_backprojection_gpr is entered by a single processing thread (the + // backprojection workers join before it returns), so a local static is safe. + static std::vector cached_grid_signature; + static GridDefinition cached_grid; + const auto signature = grid_signature(selection, max_depth_m, kGridZMinM, imaging_plane_y_m); + if (cached_grid.x_grid.empty() || signature != cached_grid_signature) { + cached_grid = build_grid(selection, max_depth_m, kGridZMinM, imaging_plane_y_m); + cached_grid_signature = signature; + } + const GridDefinition& grid = cached_grid; if (grid.x_grid.empty() || grid.z_grid.empty()) { return results; } diff --git a/python_app/gui/controllers/app_window_web_mixin.py b/python_app/gui/controllers/app_window_web_mixin.py index eab1dce..2dc8b58 100644 --- a/python_app/gui/controllers/app_window_web_mixin.py +++ b/python_app/gui/controllers/app_window_web_mixin.py @@ -68,6 +68,7 @@ class AppWindowWebController(QObject): start_requested = pyqtSignal() stop_requested = pyqtSignal() + remove_last_requested = pyqtSignal() single_capture_requested = pyqtSignal() capture_requested = pyqtSignal() apply_settings_requested = pyqtSignal(dict) @@ -120,6 +121,9 @@ class AppWindowWebController(QObject): def capture_tmp_reference(self) -> None: self.capture_requested.emit() + def remove_last_measurement(self) -> None: + self.remove_last_requested.emit() + def apply_live_settings(self, fields: dict) -> dict: unknown = set(fields) - _LIVE_FIELD_NAMES if unknown: @@ -169,6 +173,7 @@ class AppWindowWebMixin: controller.stop_requested.connect(self._stop_run) controller.single_capture_requested.connect(self._start_single_capture) controller.capture_requested.connect(self._capture_tmp_reference) + controller.remove_last_requested.connect(self._remove_last_runtime_history) controller.apply_settings_requested.connect(self._apply_web_live_settings) controller.load_config_requested.connect(self._load_web_config) controller.save_dataset_requested.connect(self._save_web_dataset) diff --git a/python_app/hardware_full/librevna_driver/protocol/frame.py b/python_app/hardware_full/librevna_driver/protocol/frame.py index 7a48999..c015e42 100644 --- a/python_app/hardware_full/librevna_driver/protocol/frame.py +++ b/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(" _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] diff --git a/python_app/hardware_full/librevna_driver/transport/usb.py b/python_app/hardware_full/librevna_driver/transport/usb.py index 209572e..b9f2699 100644 --- a/python_app/hardware_full/librevna_driver/transport/usb.py +++ b/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(): diff --git a/python_app/hardware_full/librevna_multi_device_driver/controller.py b/python_app/hardware_full/librevna_multi_device_driver/controller.py index ddbb0c9..e9e55d0 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/controller.py +++ b/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) diff --git a/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py b/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py index 1ff260c..91f9db8 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py +++ b/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 diff --git a/python_app/hardware_full/librevna_multi_device_driver/exceptions.py b/python_app/hardware_full/librevna_multi_device_driver/exceptions.py new file mode 100644 index 0000000..e949e26 --- /dev/null +++ b/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.""" diff --git a/python_app/hardware_full/librevna_multi_device_driver/models.py b/python_app/hardware_full/librevna_multi_device_driver/models.py index 256bc2e..9341bb1 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/models.py +++ b/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.""" diff --git a/python_app/hardware_full/librevna_multi_device_driver/protocol.py b/python_app/hardware_full/librevna_multi_device_driver/protocol.py index 30464ec..c78cdb8 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/protocol.py +++ b/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(" 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))) diff --git a/python_app/hardware_full/multi_device_service.py b/python_app/hardware_full/multi_device_service.py index e730a61..7d10ba7 100644 --- a/python_app/hardware_full/multi_device_service.py +++ b/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") diff --git a/python_app/tests/test_librevna_multidevice.py b/python_app/tests/test_librevna_multidevice.py new file mode 100644 index 0000000..cfc4055 --- /dev/null +++ b/python_app/tests/test_librevna_multidevice.py @@ -0,0 +1,325 @@ +"""Unit tests for the LibreVNA multi-device acquisition pipeline. + +These cover the host-side logic that does not need real hardware: frame resync +(F2), datapoint parse hardening (N2), the cross-device cycle-0 alignment check +(N1), and the transient-vs-fatal error classification driving the retry tier +(F1/F3). +""" + +from __future__ import annotations + +import struct +import threading +import unittest + +from python_app.hardware_full.librevna_driver.enums import PacketType as NativePacketType +from python_app.hardware_full.librevna_driver.models import Packet +from python_app.hardware_full.librevna_driver.protocol.frame import FrameScanner, encode_frame +import numpy as np + +from python_app.hardware_full.librevna_multi_device_driver.cycle_collection import ( + _CrossDeviceCycleSynchronizer, + _expected_sweep_cycle_seconds, + _is_transient_collection_error, + calculate_last_cycle_s_parameters, +) +from python_app.hardware_full.librevna_multi_device_driver.exceptions import TransientCollectionError +from python_app.hardware_full.librevna_multi_device_driver.models import normalize_master_stimulus_ports +from python_app.hardware_full.librevna_multi_device_driver.protocol import parse_vna_datapoint_payload +from python_app.hardware_full.multi_device_service import ( + _MAX_TRANSIENT_COLLECTION_RETRIES, + MultiDeviceLibreVnaService, +) + + +def _control_frame(packet_type: NativePacketType, payload: bytes) -> bytes: + """Encode a CRC-checked control frame (e.g. ACK).""" + return encode_frame(Packet(packet_type, payload)) + + +def _datapoint_frame(frequency_hz: int, point_index: int, values: list[tuple[int, complex]]) -> bytes: + """Encode a CRC-less VNADatapoint frame from (description_mask, value) pairs.""" + masks = bytes(mask for mask, _ in values) + reals = struct.pack(f"<{len(values)}f", *[value.real for _, value in values]) + imags = struct.pack(f"<{len(values)}f", *[value.imag for _, value in values]) + payload = struct.pack(" None: + scanner = FrameScanner() + stream = _control_frame(NativePacketType.ACK, b"") + _control_frame(NativePacketType.SET_IDLE, b"") + packets = scanner.feed(stream) + self.assertEqual([p.type for p in packets], [NativePacketType.ACK, NativePacketType.SET_IDLE]) + self.assertEqual(scanner.discarded_byte_count, 0) + + def test_resyncs_past_garbage_between_valid_frames(self) -> None: + scanner = FrameScanner() + good_a = _control_frame(NativePacketType.ACK, b"\x01\x02\x03") + good_b = _control_frame(NativePacketType.SET_IDLE, b"") + # Garbage that contains stray 0x5A bytes (false headers) with junk lengths. + garbage = bytes([0x5A, 0x10, 0x00, 0xFF, 0x5A, 0x5A, 0xAB, 0xCD]) + packets = scanner.feed(good_a + garbage + good_b) + self.assertEqual([p.type for p in packets], [NativePacketType.ACK, NativePacketType.SET_IDLE]) + self.assertGreater(scanner.discarded_byte_count, 0) + + def test_resyncs_on_corrupted_crc(self) -> None: + scanner = FrameScanner() + good = _control_frame(NativePacketType.ACK, b"") + corrupt = bytearray(_control_frame(NativePacketType.SET_IDLE, b"\x07\x07")) + corrupt[-1] ^= 0xFF # flip a CRC byte -> CRCError on decode + recovered = _control_frame(NativePacketType.ACK, b"\x09") + packets = scanner.feed(good + bytes(corrupt) + recovered) + # The corrupt frame is dropped; the surrounding valid frames survive. + self.assertEqual([p.type for p in packets], [NativePacketType.ACK, NativePacketType.ACK]) + self.assertGreater(scanner.discarded_byte_count, 0) + + def test_never_raises_on_pure_garbage(self) -> None: + scanner = FrameScanner() + # A blob full of 0x5A false headers must never raise and never hang. + blob = bytes((0x5A, 0x00, 0xFF, 0x5A, 0x08, 0x00, 0x63) * 200) + packets = scanner.feed(blob) + self.assertEqual(packets, []) + + def test_reassembles_frame_split_across_chunks(self) -> None: + scanner = FrameScanner() + frame = _control_frame(NativePacketType.ACK, b"\xaa\xbb\xcc\xdd") + self.assertEqual(scanner.feed(frame[:3]), []) + packets = scanner.feed(frame[3:]) + self.assertEqual([p.type for p in packets], [NativePacketType.ACK]) + self.assertEqual(scanner.discarded_byte_count, 0) + + def test_datapoint_frame_round_trips_without_crc(self) -> None: + scanner = FrameScanner() + frame = _datapoint_frame(3_000_000_000, 0, [(0x00, 1 + 2j), (0x10, 3 - 4j)]) + packets = scanner.feed(frame) + self.assertEqual([p.type for p in packets], [NativePacketType.VNA_DATAPOINT]) + + def test_corrupted_datapoint_resyncs_to_next_good_frame(self) -> None: + scanner = FrameScanner() + # A datapoint frame whose trailing CRC field is non-zero is rejected + # (VNADatapoint must carry zero CRC); the scanner must resync to the ACK. + bad = bytearray(_datapoint_frame(3_000_000_000, 5, [(0x00, 1 + 1j)])) + bad[-1] = 0x01 # non-zero CRC trailer -> CRCError + good = _control_frame(NativePacketType.ACK, b"") + packets = scanner.feed(bytes(bad) + good) + self.assertEqual([p.type for p in packets], [NativePacketType.ACK]) + self.assertGreater(scanner.discarded_byte_count, 0) + + +class CrossDeviceCycleSynchronizerTest(unittest.TestCase): + """N1: cycle-0 anchors are accepted only when all devices saw the same wrap.""" + + @staticmethod + def _run(device_count: int, anchor_times: list[float], tolerance: float, timeout: float = 2.0): + sync = _CrossDeviceCycleSynchronizer( + device_count=device_count, + alignment_tolerance_seconds=tolerance, + rendezvous_timeout_seconds=timeout, + ) + results: dict[int, bool] = {} + lock = threading.Lock() + + def worker(index: int) -> None: + verdict = sync.confirm_aligned_cycle_zero(anchor_times[index]) + with lock: + results[index] = verdict + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(len(anchor_times))] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5.0) + return results + + def test_aligned_anchors_accept(self) -> None: + # Three devices saw point 0 within ~0.4 ms (host jitter) -> aligned. + results = self._run(3, [100.0000, 100.0002, 100.0004], tolerance=2e-3) + self.assertEqual(results, {0: True, 1: True, 2: True}) + + def test_off_by_one_sweep_is_rejected(self) -> None: + # One device anchored ~8 ms (one sweep period) later -> misaligned. + results = self._run(3, [100.000, 100.001, 100.008], tolerance=2e-3) + self.assertEqual(results, {0: False, 1: False, 2: False}) + + def test_single_device_trivially_aligns(self) -> None: + results = self._run(1, [100.0], tolerance=2e-3) + self.assertEqual(results, {0: True}) + + def test_missing_peer_times_out_to_not_aligned(self) -> None: + # Synchronizer expects 3 devices but only 2 arrive: the barrier breaks + # after the rendezvous timeout and the waiters report not-aligned. + results = self._run(3, [100.0, 100.0], tolerance=2e-3, timeout=0.3) + self.assertEqual(results, {0: False, 1: False}) + + +class MasterStimulusPortValidationTest(unittest.TestCase): + """Q1: the single shared validator enforces the ports 1/2, unique rule.""" + + def test_accepts_valid(self) -> None: + self.assertEqual(normalize_master_stimulus_ports([1, 2]), (1, 2)) + self.assertEqual(normalize_master_stimulus_ports((2,)), (2,)) + + def test_rejects_invalid(self) -> None: + for bad in ([], [3], [1, 2, 1], [1, 1], [0]): + with self.assertRaises(ValueError): + normalize_master_stimulus_ports(bad) + + +class DatapointParseHardeningTest(unittest.TestCase): + """N2: reject misframed/misaligned datapoint payloads instead of trusting them.""" + + @staticmethod + def _payload(point_index: int, values: list[tuple[int, complex]]) -> bytes: + masks = bytes(mask for mask, _ in values) + reals = struct.pack(f"<{len(values)}f", *[v.real for _, v in values]) + imags = struct.pack(f"<{len(values)}f", *[v.imag for _, v in values]) + return struct.pack(" None: + parsed = parse_vna_datapoint_payload(self._payload(7, [(0x00, 1 + 2j), (0x10, 3 + 4j)])) + self.assertIsNotNone(parsed) + self.assertEqual(parsed.point_index, 7) + self.assertEqual(parsed.frequency_hz, 3_000_000_000) + self.assertEqual(set(parsed.receiver_values_by_description_mask), {0x00, 0x10}) + + def test_too_short_payload_rejected(self) -> None: + self.assertIsNone(parse_vna_datapoint_payload(b"\x00" * 8)) + + def test_misaligned_body_rejected(self) -> None: + # 12-byte header + a body that is not a multiple of 9 bytes -> misframe. + for extra in (1, 5, 8, 10): + self.assertIsNone( + parse_vna_datapoint_payload(struct.pack(" None: + self.assertIsNone(parse_vna_datapoint_payload(struct.pack(" None: + # Experimental anchor: 751 points x 2 stages at 50 kHz IF -> a real ~150 ms + # sweep. The conservative estimate is 30 ms and the tolerance is half of it. + cycle_seconds = _expected_sweep_cycle_seconds(751, 2, 50_000) + self.assertAlmostEqual(cycle_seconds, 0.03004, places=4) + tolerance = 0.5 * cycle_seconds + self.assertAlmostEqual(tolerance, 0.01502, places=4) + # A true off-by-one is one real sweep (~150 ms): flagged with large margin. + real_sweep_seconds = 0.150 + self.assertGreater(real_sweep_seconds, 5.0 * tolerance) + # Typical host jitter (~1 ms) clears the tolerance with room to spare. + typical_jitter_seconds = 0.001 + self.assertLess(typical_jitter_seconds, 0.25 * tolerance) + + def test_scales_with_stage_count(self) -> None: + single = _expected_sweep_cycle_seconds(751, 1, 50_000) + dual = _expected_sweep_cycle_seconds(751, 2, 50_000) + self.assertAlmostEqual(dual, 2.0 * single, places=6) + + +class LastCycleSParameterTest(unittest.TestCase): + """Reference/measurement division + the corrupt-reference and NaN guards.""" + + def test_divides_measurement_by_reference(self) -> None: + reference = {1: np.array([[2 + 0j, 0 + 2j, 4 + 0j]], dtype=complex)} + measurement = {"S31": np.array([[4 + 0j, 0 + 4j, 2 + 0j]], dtype=complex)} + result = calculate_last_cycle_s_parameters(measurement, reference) + np.testing.assert_allclose(result["s31"], np.array([2 + 0j, 2 + 0j, 0.5 + 0j])) + + def test_nan_reference_is_transient(self) -> None: + reference = {1: np.array([[2 + 0j, np.nan + 1j * np.nan, 4 + 0j]], dtype=complex)} + measurement = {"S31": np.array([[4 + 0j, 4 + 0j, 2 + 0j]], dtype=complex)} + with self.assertRaises(TransientCollectionError): + calculate_last_cycle_s_parameters(measurement, reference) + + def test_nan_measurement_is_transient(self) -> None: + reference = {1: np.array([[2 + 0j, 2 + 0j, 4 + 0j]], dtype=complex)} + measurement = {"S31": np.array([[4 + 0j, np.nan + 1j * np.nan, 2 + 0j]], dtype=complex)} + with self.assertRaises(TransientCollectionError): + calculate_last_cycle_s_parameters(measurement, reference) + + def test_near_zero_reference_is_transient(self) -> None: + reference = {1: np.array([[2 + 0j, 0 + 0j, 4 + 0j]], dtype=complex)} + measurement = {"S31": np.array([[4 + 0j, 4 + 0j, 2 + 0j]], dtype=complex)} + with self.assertRaises(TransientCollectionError): + calculate_last_cycle_s_parameters(measurement, reference) + + +class TransientErrorClassificationTest(unittest.TestCase): + """F1/F3: only device-healthy failures are transient (cheap retry).""" + + def test_transient_types(self) -> None: + self.assertTrue(_is_transient_collection_error(TransientCollectionError("x"))) + self.assertTrue(_is_transient_collection_error(TimeoutError("no progress"))) + + def test_fatal_types(self) -> None: + self.assertFalse(_is_transient_collection_error(RuntimeError("USB transport failed"))) + self.assertFalse(_is_transient_collection_error(ValueError("bad config"))) + + def test_transient_is_runtimeerror_subclass(self) -> None: + # So existing `except Exception`/`except RuntimeError` recovery still catches it. + self.assertIsInstance(TransientCollectionError("x"), RuntimeError) + + +class _StubService(MultiDeviceLibreVnaService): + """Non-slots subclass so a test can inject collection behaviour + count reopens.""" + + def _acquire_native_collection(self, collection_id: int, capture_start_ns: int): # type: ignore[override] + self.collection_calls += 1 + return self.behavior(collection_id, capture_start_ns) + + def recover(self, **_kwargs) -> None: # type: ignore[override] + self.reopen_calls += 1 + + +class TransientRetryTierTest(unittest.TestCase): + """F1/F3: transient failures retry in place (no USB reopen); fatals escalate.""" + + @staticmethod + def _stub(behavior) -> _StubService: + service = _StubService(master_serial="M", slave_serials=["A", "B"], backend_mode="mock") + service.behavior = behavior + service.collection_calls = 0 + service.reopen_calls = 0 + return service + + def test_succeeds_after_transient_failures_without_reopen(self) -> None: + def flaky(_collection_id: int, _capture_start_ns: int) -> str: + if service.collection_calls <= _MAX_TRANSIENT_COLLECTION_RETRIES: + raise TransientCollectionError("dropped datapoint") + return "collection" + + service = self._stub(flaky) + result = service._acquire_native_collection_with_transient_retries(1, 0) + self.assertEqual(result, "collection") + self.assertEqual(service.collection_calls, _MAX_TRANSIENT_COLLECTION_RETRIES + 1) + self.assertEqual(service.reopen_calls, 0) # never reopened USB for transient errors + + def test_exhausted_transient_retries_raise_transient(self) -> None: + def always_transient(_collection_id: int, _capture_start_ns: int) -> str: + raise TransientCollectionError("persistent drop") + + service = self._stub(always_transient) + with self.assertRaises(TransientCollectionError): + service._acquire_native_collection_with_transient_retries(1, 0) + self.assertEqual(service.collection_calls, _MAX_TRANSIENT_COLLECTION_RETRIES + 1) + + def test_fatal_error_is_not_retried_in_place(self) -> None: + def fatal(_collection_id: int, _capture_start_ns: int) -> str: + raise RuntimeError("USB transport failed") + + service = self._stub(fatal) + with self.assertRaises(RuntimeError): + service._acquire_native_collection_with_transient_retries(1, 0) + self.assertEqual(service.collection_calls, 1) # fatal escalates immediately, no in-place retry + + +if __name__ == "__main__": + unittest.main() diff --git a/python_app/webui/controller.py b/python_app/webui/controller.py index 736186e..34d18c5 100644 --- a/python_app/webui/controller.py +++ b/python_app/webui/controller.py @@ -29,6 +29,9 @@ class WebController(Protocol): def capture_tmp_reference(self) -> None: """Capture and select a temporary reference (the desktop button).""" + def remove_last_measurement(self) -> None: + """Remove the most recent measurement (the desktop button).""" + def apply_live_settings(self, fields: dict) -> list: """Apply live processor settings; returns the current settings schema.""" diff --git a/python_app/webui/routes.py b/python_app/webui/routes.py index 5ff69ea..629d020 100644 --- a/python_app/webui/routes.py +++ b/python_app/webui/routes.py @@ -63,6 +63,14 @@ async def post_tmp_reference(request: Request) -> dict: return controller.status() +@router.post("/api/remove_last") +async def post_remove_last(request: Request) -> dict: + logger.info("Web UI request: remove last measurement") + controller = _controller(request) + controller.remove_last_measurement() + return controller.status() + + @router.get("/api/configs") async def get_configs(request: Request) -> dict: return {"names": _controller(request).list_configs()} diff --git a/python_app/webui/static/app.js b/python_app/webui/static/app.js index 90ce412..863da04 100644 --- a/python_app/webui/static/app.js +++ b/python_app/webui/static/app.js @@ -13,6 +13,7 @@ const btnStart = document.getElementById("btn-start"); const btnSingle = document.getElementById("btn-single"); const btnStop = document.getElementById("btn-stop"); const btnTmpRef = document.getElementById("btn-tmp-ref"); +const btnRemoveLast = document.getElementById("btn-remove-last"); const btnApply = document.getElementById("btn-apply"); const btnResetHistory = document.getElementById("btn-reset-history"); const btnLoadConfig = document.getElementById("btn-load-config"); @@ -82,6 +83,7 @@ bindControl(btnStart, "/api/start", {}); bindControl(btnSingle, "/api/single_capture", {}); bindControl(btnStop, "/api/stop", {}); bindControl(btnTmpRef, "/api/tmp_reference", {}); +bindControl(btnRemoveLast, "/api/remove_last", {}); /* ---- settings panel --------------------------------------------- */ settingsToggle.addEventListener("click", () => sidePanel.classList.toggle("collapsed")); diff --git a/python_app/webui/static/index.html b/python_app/webui/static/index.html index 821658d..4323fa4 100644 --- a/python_app/webui/static/index.html +++ b/python_app/webui/static/index.html @@ -14,6 +14,7 @@ + diff --git a/python_app/webui/static/style.css b/python_app/webui/static/style.css index d48a1cf..c54260c 100644 --- a/python_app/webui/static/style.css +++ b/python_app/webui/static/style.css @@ -318,14 +318,35 @@ body { } .toast.error { background: var(--danger); } -/* Narrow screens / phones: stack the plot above the settings, full-width controls. */ +/* Narrow screens / phones: stack the plot above the settings, full-width controls, + and let the whole page scroll instead of clipping the settings panel. */ @media (max-width: 760px) { + /* Let the whole page scroll normally instead of clipping to the viewport. */ + html, body { + height: auto; + overflow: visible; + } + body { min-height: 100vh; } /* Still fill the screen when content is short. */ + .topbar { flex-wrap: wrap; } .controls { width: 100%; } .controls .btn { flex: 1 1 auto; } - .layout { flex-direction: column; padding: 10px; gap: 10px; } + + .layout { + flex: none; /* Size to content instead of stretching to the viewport. */ + flex-direction: column; + padding: 10px; + gap: 10px; + } .plot-panel { flex: none; height: 45vh; } - .side-panel { width: auto; flex: 1; min-height: 0; } + + .side-panel { + width: auto; + flex: none; /* Expand to the full height of its content. */ + } + .panel-body { overflow: visible; } /* Do not clip the settings. */ + .settings-fields { overflow-y: visible; } /* Scroll the page, not an inner box. */ + .field input[type="text"], .field input[type="number"], .field select { width: 130px; } diff --git a/run_config_examples/run_config_librevna_multi.example.json b/run_config_examples/run_config_librevna_multi.example.json index c2adc18..19fecca 100644 --- a/run_config_examples/run_config_librevna_multi.example.json +++ b/run_config_examples/run_config_librevna_multi.example.json @@ -1,7 +1,7 @@ { "radar": { - "model": "librevna_multi", - "serial": "207730885532", + "model": "librevna_multi", + "serial": "209E307A5532", "remote_host": "127.0.0.1", "remote_port": 50209, "driver_mode": "native",