Add real switch support for multi-device matrix radar

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

Combo-matrix construction was centralized into
RunConfigModel.build_runtime_combos() so the GUI, workflows, and codec
all derive the same widened matrix instead of each computing its own
version of the virtual 2x4 layout.
This commit is contained in:
2026-07-29 17:44:28 +03:00
parent f967da7f53
commit 68bec25f17
11 changed files with 341 additions and 32 deletions
@@ -0,0 +1,130 @@
"""Matrix radar behind real GPIO switches on the stimulus and/or receiver path."""
from __future__ import annotations
from dataclasses import dataclass, replace
import logging
import time
from python_app.hardware_full.matrix_radar_service import MatrixRadarService
from python_app.hardware_full.switch_service import SwitchService
from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData
from python_app.models.run_config_model import RadarSweepModel, SwitchModel
logger = logging.getLogger(__name__)
@dataclass(slots=True)
class SwitchedMatrixRadarService:
"""Widen a matrix radar's combo matrix with real switch positions.
Implements the ``MatrixRadarService`` protocol, so the producer and the
capture workflows treat it as an ordinary matrix radar that simply reports
more positions. The hardware sweep is never stopped: switches are only ever
driven BETWEEN ``acquire_collection`` calls, and the inner service's
free-running collection discards any partially swept cycle.
"""
inner: MatrixRadarService
output_switch: SwitchService | None
input_switch: SwitchService | None
inner_output_positions: int
inner_input_positions: int
settling_ms: int = 0
def open(self) -> None:
"""Open the inner radar and both switches."""
self.inner.open()
if self.output_switch is not None:
self.output_switch.open()
if self.input_switch is not None:
self.input_switch.open()
def close(self) -> None:
"""Close switches first, then the inner radar; never raises."""
for switch in (self.input_switch, self.output_switch):
if switch is not None:
try:
switch.close()
except Exception as exc: # noqa: BLE001 — shutdown path
logger.warning("Switch close ignored error: %s", exc)
self.inner.close()
def configure(self, sweep: RadarSweepModel) -> None:
"""Apply sweep settings to the inner radar."""
self.inner.configure(sweep)
def recover(self) -> None:
"""Reconnect the inner radar; switches are not on the USB transport."""
self.inner.recover()
def acquire_collection(self, collection_id: int = 1) -> SweepCollection:
"""Acquire the full widened matrix, one inner collection per switch step.
A partial failure raises instead of returning a short collection: the
preprocessor requires every runtime combo to be present, so half a matrix
is worse than a dropped frame.
"""
capture_start_ns = time.monotonic_ns()
out_steps = self.output_switch.position_count() if self.output_switch is not None else 1
in_steps = self.input_switch.position_count() if self.input_switch is not None else 1
total_inputs = in_steps * self.inner_input_positions
total_outputs = out_steps * self.inner_output_positions
# Place each trace at its canonical index rather than appending. The GPR stage
# rejects a collection whose trace order differs from run.combos, and run.combos
# is built output-major (`build_full_combos`) while these loops run switch-major.
# Appending happens to agree for an output switch and to disagree for an input one.
slots: list[TraceData | None] = [None] * (total_inputs * total_outputs)
for out_k in range(out_steps):
if self.output_switch is not None:
self.output_switch.switch_to(out_k)
for in_k in range(in_steps):
if self.input_switch is not None:
self.input_switch.switch_to(in_k)
# Settle AFTER the last switch change and BEFORE collecting, so the
# cycle we anchor on starts with the RF path already stable.
if self.settling_ms > 0:
time.sleep(self.settling_ms / 1000.0)
sub = self.inner.acquire_collection(collection_id)
for trace in sub.traces:
input_pos = in_k * self.inner_input_positions + int(trace.combo.input)
output_pos = out_k * self.inner_output_positions + int(trace.combo.output)
slots[output_pos * total_inputs + input_pos] = replace(
trace, combo=ComboKey(input=input_pos, output=output_pos)
)
if any(trace is None for trace in slots):
missing = sum(1 for trace in slots if trace is None)
raise RuntimeError(
f"Switched matrix collection is incomplete: {missing} of {len(slots)} combos missing"
)
return SweepCollection(
collection_id=int(collection_id),
monotonic_ns=time.monotonic_ns(),
traces=[trace for trace in slots if trace is not None],
capture_start_ns=capture_start_ns,
capture_end_ns=time.monotonic_ns(),
)
def build_physical_switch(
model: SwitchModel,
physical_positions: int,
radar_driver_mode: str,
) -> SwitchService | None:
"""Build the driver for a real switch described by a virtual switch section.
The config section carries the LOGICAL axis size and a forced "mock" mode so
the C++ loader accepts it; the real driver needs the PHYSICAL position count
and native mode. Mock radar runs keep mock switches so the whole path can be
exercised without GPIO.
"""
if physical_positions <= 1:
return None
driver_mode = "mock" if radar_driver_mode.strip().lower() == "mock" else "native"
return SwitchService.from_model(
replace(model, positions=physical_positions, driver_mode=driver_mode)
)