awe/radar_system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added multidevice support

Browse Source
This commit is contained in:
Ayzen
2026-04-28 17:29:21 +03:00
parent 93c705a33b
commit 1ea2aabf87
37 changed files with 1860 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python_app/workflows/calibration_workflow.py
+7
View File
@@ -17,6 +17,13 @@ def capture_calibration_set(
store: NpzStore,
) -> tuple[str, SweepCollection]:
"""Capture all switch combinations and persist them as calibration set."""
if config.is_multi_device:
raise RuntimeError(
"LibreVNA multi-device S21 through calibration is not supported by this one-shot full-set helper. "
"Use the sequential preprocess capture flow so each virtual combo can be connected through "
"and captured explicitly."
)
combos = RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions)
radar = LibreVnaService(serial=config.radar.serial or None)
python_app/workflows/multi_radar_capture_workflow.py
+104 -30
View File
@@ -9,12 +9,17 @@ import time
import numpy as np
from python_app.hardware_full.librevna_service import LibreVnaService
from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService
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 ComboModel, RunConfigModel
from python_app.storage.npz_store import NpzStore
from python_app.workflows.radar_config_variants import RadarConfigVariant
from python_app.workflows.sequential_capture_workflow import SequentialCaptureState
from python_app.workflows.sequential_capture_workflow import (
MULTI_DEVICE_MANUAL_CAPTURE_KINDS,
SequentialCaptureState,
select_trace_for_combo,
)
@dataclass(frozen=True, slots=True)
@@ -63,9 +68,15 @@ class MultiRadarSequentialCaptureSession:
self._kind = kind
self._set_name = set_name
self._radar_variants = list(radar_variants)
self._combos = RunConfigModel.build_full_combos(
base_config.input_switch.positions,
base_config.output_switch.positions,
self._is_multi_device = base_config.is_multi_device
self._manual_multi_device_capture = self._is_multi_device and kind in MULTI_DEVICE_MANUAL_CAPTURE_KINDS
self._combos = (
RunConfigModel.build_multi_device_virtual_combos()
if self._is_multi_device
else RunConfigModel.build_full_combos(
base_config.input_switch.positions,
base_config.output_switch.positions,
)
)
if not self._combos:
raise RuntimeError("No switch combinations available for capture")
@@ -78,27 +89,38 @@ class MultiRadarSequentialCaptureSession:
self._next_index = 0
self._opened = False
self._radar = LibreVnaService(serial=base_config.radar.serial or None)
self._input_switch = SwitchService(
name=base_config.input_switch.name,
positions=base_config.input_switch.positions,
mode=base_config.input_switch.driver_mode,
driver=base_config.input_switch.driver,
gpio_chip=base_config.input_switch.gpio_chip,
pin_a=base_config.input_switch.pin_a,
pin_b=base_config.input_switch.pin_b,
invert_logic=base_config.input_switch.invert_logic,
)
self._output_switch = SwitchService(
name=base_config.output_switch.name,
positions=base_config.output_switch.positions,
mode=base_config.output_switch.driver_mode,
driver=base_config.output_switch.driver,
gpio_chip=base_config.output_switch.gpio_chip,
pin_a=base_config.output_switch.pin_a,
pin_b=base_config.output_switch.pin_b,
invert_logic=base_config.output_switch.invert_logic,
)
if self._is_multi_device:
self._radar = MultiDeviceLibreVnaService(
master_serial=base_config.radar.serial,
slave_serials=list(base_config.radar.multi_device.slave_serials),
force_external_reference=base_config.radar.multi_device.force_external_reference,
recovery_attempts=base_config.radar.multi_device.recovery_attempts,
backend_mode=base_config.radar.driver_mode,
)
self._input_switch = None
self._output_switch = None
else:
self._radar = LibreVnaService(serial=base_config.radar.serial or None)
self._input_switch = SwitchService(
name=base_config.input_switch.name,
positions=base_config.input_switch.positions,
mode=base_config.input_switch.driver_mode,
driver=base_config.input_switch.driver,
gpio_chip=base_config.input_switch.gpio_chip,
pin_a=base_config.input_switch.pin_a,
pin_b=base_config.input_switch.pin_b,
invert_logic=base_config.input_switch.invert_logic,
)
self._output_switch = SwitchService(
name=base_config.output_switch.name,
positions=base_config.output_switch.positions,
mode=base_config.output_switch.driver_mode,
driver=base_config.output_switch.driver,
gpio_chip=base_config.output_switch.gpio_chip,
pin_a=base_config.output_switch.pin_a,
pin_b=base_config.output_switch.pin_b,
invert_logic=base_config.output_switch.invert_logic,
)
@property
def kind(self) -> str:
@@ -118,8 +140,10 @@ class MultiRadarSequentialCaptureSession:
try:
self._radar.open()
self._radar.configure(self._base_config.radar.sweep)
self._input_switch.open()
self._output_switch.open()
if self._input_switch is not None:
self._input_switch.open()
if self._output_switch is not None:
self._output_switch.open()
except Exception:
self.close()
raise
@@ -127,9 +151,11 @@ class MultiRadarSequentialCaptureSession:
def close(self) -> None:
"""Close all opened hardware resources."""
with suppress(Exception):
self._output_switch.close()
if self._output_switch is not None:
self._output_switch.close()
with suppress(Exception):
self._input_switch.close()
if self._input_switch is not None:
self._input_switch.close()
with suppress(Exception):
self._radar.close()
self._opened = False
@@ -140,7 +166,11 @@ class MultiRadarSequentialCaptureSession:
return SequentialCaptureState(
kind=self._kind,
set_name=self._set_name,
captured_count=len(self._captured_batches),
captured_count=(
self._next_index
if self._is_multi_device and not self._manual_multi_device_capture
else len(self._captured_batches)
),
total_count=len(self._combos),
current_combo=current_combo,
can_undo=bool(self._captured_batches),
@@ -156,6 +186,39 @@ class MultiRadarSequentialCaptureSession:
if combo is None:
raise RuntimeError("Capture session is already complete")
if self._is_multi_device:
traces: list[TraceData] = []
variant_labels: list[str] = []
for variant in self._radar_variants:
self._radar.configure(variant.config.radar.sweep)
if self._base_config.runtime.settling_ms > 0:
time.sleep(self._base_config.runtime.settling_ms / 1000.0)
collection = self._radar.acquire_collection(collection_id=1)
if not collection.traces:
raise RuntimeError(f"Multi-device variant {variant.display_name} returned no traces")
if self._manual_multi_device_capture:
trace = select_trace_for_combo(collection, combo)
self._traces_by_radar_key[variant.radar_key].append(trace)
traces.append(trace)
else:
self._traces_by_radar_key[variant.radar_key].extend(collection.traces)
traces.append(collection.traces[-1])
variant_labels.append(variant.display_name)
batch = MultiRadarCaptureBatch(
combo=combo,
traces=tuple(traces),
variant_labels=tuple(variant_labels),
)
self._captured_batches.append(batch)
if self._manual_multi_device_capture:
self._next_index += 1
else:
self._next_index = len(self._combos)
return batch
assert self._input_switch is not None
assert self._output_switch is not None
self._output_switch.switch_to(combo.output)
self._input_switch.switch_to(combo.input)
if self._base_config.runtime.settling_ms > 0:
@@ -194,6 +257,17 @@ class MultiRadarSequentialCaptureSession:
if not self._captured_batches or self._next_index <= 0:
raise RuntimeError("No captured combo is available to undo")
if self._is_multi_device and not self._manual_multi_device_capture:
removed_batch = self._captured_batches[-1]
for variant in self._radar_variants:
traces = self._traces_by_radar_key[variant.radar_key]
if len(traces) < len(self._combos):
raise RuntimeError("Capture session state is inconsistent; missing multi-device traces")
del traces[-len(self._combos) :]
self._next_index = 0
self._captured_batches.pop()
return removed_batch
expected_combo = self._combos[self._next_index - 1]
removed_batch = self._captured_batches[-1]
if (
python_app/workflows/radar_config_variants.py
+5
View File
@@ -145,6 +145,11 @@ def _load_radar_config_variant(path: Path, *, base_config: RunConfigModel) -> Ra
sweep_points=config.radar.sweep.points,
ifbw_hz=config.radar.sweep.if_bandwidth_hz,
power_dbm=config.radar.sweep.power_dbm,
extra_serials=(
config.radar.multi_device.slave_serials
if config.is_multi_device
else None
),
)
return RadarConfigVariant(
source_path=path,
python_app/workflows/reference_workflow.py
+29
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import time
from python_app.hardware_full.librevna_service import LibreVnaService
from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService
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 RunConfigModel
@@ -17,6 +18,34 @@ def capture_reference_set(
store: NpzStore,
) -> tuple[str, SweepCollection]:
"""Capture all switch combinations and persist them as reference set."""
if config.is_multi_device:
radar = MultiDeviceLibreVnaService(
master_serial=config.radar.serial,
slave_serials=list(config.radar.multi_device.slave_serials),
force_external_reference=config.radar.multi_device.force_external_reference,
recovery_attempts=config.radar.multi_device.recovery_attempts,
backend_mode=config.radar.driver_mode,
)
try:
radar.open()
radar.configure(config.radar.sweep)
collection = radar.acquire_collection(collection_id=1)
finally:
radar.close()
radar_key = radar_key_from_config(
model_name=config.radar.model,
serial=config.radar.serial,
sweep_start_hz=config.radar.sweep.start_hz,
sweep_stop_hz=config.radar.sweep.stop_hz,
sweep_points=config.radar.sweep.points,
ifbw_hz=config.radar.sweep.if_bandwidth_hz,
power_dbm=config.radar.sweep.power_dbm,
extra_serials=config.radar.multi_device.slave_serials,
)
store.save_set("s21_reference", radar_key, set_name, collection)
return radar_key, collection
combos = RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions)
radar = LibreVnaService(serial=config.radar.serial or None)
python_app/workflows/sequential_capture_workflow.py
+90 -26
View File
@@ -9,11 +9,14 @@ import time
import numpy as np
from python_app.hardware_full.librevna_service import LibreVnaService
from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService
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 ComboModel, RunConfigModel
from python_app.storage.npz_store import NpzStore, radar_key_from_config
MULTI_DEVICE_MANUAL_CAPTURE_KINDS = frozenset({"s21_calibration", "s11_open", "s11_short", "s11_load"})
@dataclass(slots=True)
class SequentialCaptureState:
@@ -42,7 +45,13 @@ class SequentialCaptureSession:
self._config = config
self._kind = kind
self._set_name = set_name
self._combos = RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions)
self._is_multi_device = config.is_multi_device
self._manual_multi_device_capture = self._is_multi_device and kind in MULTI_DEVICE_MANUAL_CAPTURE_KINDS
self._combos = (
RunConfigModel.build_multi_device_virtual_combos()
if self._is_multi_device
else RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions)
)
if not self._combos:
raise RuntimeError("No switch combinations available for capture")
@@ -50,27 +59,38 @@ class SequentialCaptureSession:
self._next_index = 0
self._opened = False
self._radar = LibreVnaService(serial=config.radar.serial or None)
self._input_switch = SwitchService(
name=config.input_switch.name,
positions=config.input_switch.positions,
mode=config.input_switch.driver_mode,
driver=config.input_switch.driver,
gpio_chip=config.input_switch.gpio_chip,
pin_a=config.input_switch.pin_a,
pin_b=config.input_switch.pin_b,
invert_logic=config.input_switch.invert_logic,
)
self._output_switch = SwitchService(
name=config.output_switch.name,
positions=config.output_switch.positions,
mode=config.output_switch.driver_mode,
driver=config.output_switch.driver,
gpio_chip=config.output_switch.gpio_chip,
pin_a=config.output_switch.pin_a,
pin_b=config.output_switch.pin_b,
invert_logic=config.output_switch.invert_logic,
)
if self._is_multi_device:
self._radar = MultiDeviceLibreVnaService(
master_serial=config.radar.serial,
slave_serials=list(config.radar.multi_device.slave_serials),
force_external_reference=config.radar.multi_device.force_external_reference,
recovery_attempts=config.radar.multi_device.recovery_attempts,
backend_mode=config.radar.driver_mode,
)
self._input_switch = None
self._output_switch = None
else:
self._radar = LibreVnaService(serial=config.radar.serial or None)
self._input_switch = SwitchService(
name=config.input_switch.name,
positions=config.input_switch.positions,
mode=config.input_switch.driver_mode,
driver=config.input_switch.driver,
gpio_chip=config.input_switch.gpio_chip,
pin_a=config.input_switch.pin_a,
pin_b=config.input_switch.pin_b,
invert_logic=config.input_switch.invert_logic,
)
self._output_switch = SwitchService(
name=config.output_switch.name,
positions=config.output_switch.positions,
mode=config.output_switch.driver_mode,
driver=config.output_switch.driver,
gpio_chip=config.output_switch.gpio_chip,
pin_a=config.output_switch.pin_a,
pin_b=config.output_switch.pin_b,
invert_logic=config.output_switch.invert_logic,
)
@property
def kind(self) -> str:
@@ -90,8 +110,10 @@ class SequentialCaptureSession:
try:
self._radar.open()
self._radar.configure(self._config.radar.sweep)
self._input_switch.open()
self._output_switch.open()
if self._input_switch is not None:
self._input_switch.open()
if self._output_switch is not None:
self._output_switch.open()
except Exception:
self.close()
raise
@@ -99,9 +121,11 @@ class SequentialCaptureSession:
def close(self) -> None:
"""Close all opened hardware resources."""
with suppress(Exception):
self._output_switch.close()
if self._output_switch is not None:
self._output_switch.close()
with suppress(Exception):
self._input_switch.close()
if self._input_switch is not None:
self._input_switch.close()
with suppress(Exception):
self._radar.close()
self._opened = False
@@ -127,6 +151,22 @@ class SequentialCaptureSession:
if combo is None:
raise RuntimeError("Capture session is already complete")
if self._is_multi_device:
collection = self._radar.acquire_collection(collection_id=1)
if self._manual_multi_device_capture:
trace = select_trace_for_combo(collection, combo)
self._traces.append(trace)
self._next_index += 1
return trace
self._traces.extend(collection.traces)
self._next_index = len(self._combos)
if not collection.traces:
raise RuntimeError("Multi-device capture returned no traces")
return collection.traces[-1]
assert self._input_switch is not None
assert self._output_switch is not None
self._output_switch.switch_to(combo.output)
self._input_switch.switch_to(combo.input)
if self._config.runtime.settling_ms > 0:
@@ -150,6 +190,14 @@ class SequentialCaptureSession:
if not self._traces or self._next_index <= 0:
raise RuntimeError("No captured combo is available to undo")
if self._is_multi_device and not self._manual_multi_device_capture:
if len(self._traces) != len(self._combos):
raise RuntimeError("Capture session state is inconsistent; multi-device trace matrix is incomplete")
removed_trace = self._traces[-1]
self._traces.clear()
self._next_index = 0
return removed_trace
expected_combo = self._combos[self._next_index - 1]
removed_trace = self._traces[-1]
if (
@@ -193,6 +241,11 @@ class SequentialCaptureSession:
sweep_points=self._config.radar.sweep.points,
ifbw_hz=self._config.radar.sweep.if_bandwidth_hz,
power_dbm=self._config.radar.sweep.power_dbm,
extra_serials=(
self._config.radar.multi_device.slave_serials
if self._config.is_multi_device
else None
),
)
store.save_set(self._kind, radar_key, self._set_name, collection)
return radar_key, collection
@@ -202,3 +255,14 @@ class SequentialCaptureSession:
if self._next_index >= len(self._combos):
return None
return self._combos[self._next_index]
def select_trace_for_combo(collection: SweepCollection, combo: ComboModel) -> TraceData:
"""Return the trace matching a virtual combo from a full multi-device capture."""
for trace in collection.traces:
if (
int(trace.combo.input_pos) == int(combo.input)
and int(trace.combo.output_pos) == int(combo.output)
):
return trace
raise RuntimeError(f"Multi-device capture is missing trace for input={combo.input}, output={combo.output}")