79 lines
3.5 KiB
Python
79 lines
3.5 KiB
Python
"""LibreVNA acquisition factory for single devices, matrices, and mock scenes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Protocol
|
|
|
|
from python_app.models.dataset_model import ComboKey, SweepCollection
|
|
from python_app.models.run_config_schema import RadarSweepModel, RunConfigModel
|
|
|
|
|
|
class MatrixRadarService(Protocol):
|
|
"""Common API for radars that emit a full combo matrix per acquisition."""
|
|
|
|
def open(self) -> None:
|
|
"""Open hardware connections."""
|
|
|
|
def close(self) -> None:
|
|
"""Close hardware connections."""
|
|
|
|
def configure(self, sweep: RadarSweepModel) -> None:
|
|
"""Apply sweep settings to the radar."""
|
|
|
|
def acquire_collection(self, collection_id: int = 1, *, combo: ComboKey | None = None) -> SweepCollection:
|
|
"""Acquire the matrix, or the smallest physical sweep containing combo."""
|
|
...
|
|
|
|
def create_matrix_radar_service(config: RunConfigModel) -> MatrixRadarService:
|
|
"""Create the configured LibreVNA collection source and its physical switches."""
|
|
if config.radar.driver_mode == "mock" and config.radar.model in {
|
|
RunConfigModel.LIBREVNA_MODEL, RunConfigModel.LIBREVNA_MULTI_MODEL,
|
|
}:
|
|
from python_app.hardware_full.point_target_mock_service import PointTargetMockService
|
|
|
|
return PointTargetMockService(config)
|
|
if config.radar.model == RunConfigModel.LIBREVNA_MODEL:
|
|
from python_app.hardware_full.librevna_service import LibreVnaService
|
|
from python_app.hardware_full.switch_service import SwitchService
|
|
from python_app.hardware_full.switched_matrix_radar_service import SwitchedMatrixRadarService
|
|
|
|
return SwitchedMatrixRadarService(
|
|
inner=LibreVnaService(serial=config.radar.serial or None),
|
|
output_switch=SwitchService.from_model(config.output_switch),
|
|
input_switch=SwitchService.from_model(config.input_switch),
|
|
inner_output_positions=1,
|
|
inner_input_positions=1,
|
|
settling_ms=config.runtime.settling_ms,
|
|
combos=[ComboKey(c.input, c.output) for c in (config.combos or config.build_runtime_combos())],
|
|
)
|
|
if config.radar.model != RunConfigModel.LIBREVNA_MULTI_MODEL:
|
|
raise RuntimeError("matrix radar service factory requires a matrix-mode radar.model")
|
|
|
|
from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService
|
|
|
|
out_physical = config.matrix_output_switch_positions
|
|
in_physical = config.matrix_input_switch_positions
|
|
inner = 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,
|
|
switch_substage_count=config.matrix_stimulus_switch_positions,
|
|
)
|
|
if out_physical <= 1 and in_physical <= 1:
|
|
return inner
|
|
|
|
from python_app.hardware_full.switched_matrix_radar_service import (
|
|
SwitchedMatrixRadarService,
|
|
build_physical_switch,
|
|
)
|
|
|
|
return SwitchedMatrixRadarService(
|
|
inner=inner,
|
|
output_switch=build_physical_switch(config.output_switch, out_physical, config.radar.driver_mode),
|
|
input_switch=build_physical_switch(config.input_switch, in_physical, config.radar.driver_mode),
|
|
inner_output_positions=config.matrix_virtual_output_positions,
|
|
inner_input_positions=RunConfigModel.MULTI_DEVICE_INPUT_POSITIONS,
|
|
settling_ms=config.runtime.settling_ms,
|
|
)
|