285 lines
9.3 KiB
Python
285 lines
9.3 KiB
Python
"""Dataclass schema for runtime configuration used by Python pipeline tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ComboModel:
|
|
"""One switch combination used for an acquisition sweep."""
|
|
|
|
input: int
|
|
output: int
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RadarSweepModel:
|
|
"""Sweep settings for LibreVNA acquisition."""
|
|
|
|
# Keep schema defaults minimal/safe; operational values come from run_config.json.
|
|
start_hz: float = 0.0
|
|
stop_hz: float = 0.0
|
|
points: int = 1
|
|
if_bandwidth_hz: float = 1.0
|
|
power_dbm: float = -30.0
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RadarMultiDeviceModel:
|
|
"""Multi-device LibreVNA topology settings."""
|
|
|
|
slave_serials: list[str] = field(default_factory=list)
|
|
force_external_reference: bool = True
|
|
recovery_attempts: int = 3
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RadarModel:
|
|
"""Radar section of run configuration."""
|
|
|
|
model: str = "librevna"
|
|
serial: str = ""
|
|
driver_mode: str = "mock"
|
|
mock_signal_hz: float = 1_000_000.0
|
|
sweep: RadarSweepModel = field(default_factory=RadarSweepModel)
|
|
multi_device: RadarMultiDeviceModel = field(default_factory=RadarMultiDeviceModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SwitchModel:
|
|
"""Generic switch section of run configuration."""
|
|
|
|
name: str
|
|
driver_mode: str = "mock"
|
|
driver: str = ""
|
|
radar_port: int = 0
|
|
positions: int = 1
|
|
default_position: int = 0
|
|
gpio_chip: str = ""
|
|
pin_a: int = -1
|
|
pin_b: int = -1
|
|
invert_logic: bool = False
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RingEndpointModel:
|
|
"""Shared-memory ring endpoint description."""
|
|
|
|
name: str
|
|
capacity: int = 1
|
|
slot_size_bytes: int = 4096
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RingsModel:
|
|
"""Ring endpoints used by orchestration pipeline."""
|
|
|
|
raw: RingEndpointModel = field(default_factory=lambda: RingEndpointModel(name=""))
|
|
raw_tap: RingEndpointModel = field(default_factory=lambda: RingEndpointModel(name=""))
|
|
preprocessed: RingEndpointModel = field(default_factory=lambda: RingEndpointModel(name=""))
|
|
preprocessed_tap: RingEndpointModel = field(default_factory=lambda: RingEndpointModel(name=""))
|
|
results: RingEndpointModel = field(default_factory=lambda: RingEndpointModel(name=""))
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LocatorServerRuntimeModel:
|
|
"""Embedded locator TCP server configuration stored in run config."""
|
|
|
|
device_id: int = 3
|
|
protocol_version: int = 1
|
|
host: str = "0.0.0.0"
|
|
port: int = 8888
|
|
max_payload_bytes: int = 64 * 1024
|
|
client_queue_size: int = 32
|
|
logger_name: str = "locator_runtime"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RuntimeModel:
|
|
"""Runtime process behavior and paths."""
|
|
|
|
settling_ms: int = 0
|
|
idle_sleep_ms: int = 2
|
|
continuous: bool = False
|
|
processing_live_config_path: str = ""
|
|
locator_server: LocatorServerRuntimeModel = field(default_factory=LocatorServerRuntimeModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class PreprocessAssetModel:
|
|
"""One preprocessing asset selected for live acquisition."""
|
|
|
|
set_name: str = ""
|
|
bundle_path: str = ""
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class S21PreprocessModel:
|
|
"""Two-port S21 preprocessing assets."""
|
|
|
|
calibration: PreprocessAssetModel = field(default_factory=PreprocessAssetModel)
|
|
reference: PreprocessAssetModel = field(default_factory=PreprocessAssetModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class S11CalibrationModel:
|
|
"""One-port S11 OSL calibration assets."""
|
|
|
|
open: PreprocessAssetModel = field(default_factory=PreprocessAssetModel)
|
|
short: PreprocessAssetModel = field(default_factory=PreprocessAssetModel)
|
|
load: PreprocessAssetModel = field(default_factory=PreprocessAssetModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class S11PreprocessModel:
|
|
"""One-port S11 preprocessing assets."""
|
|
|
|
calibration: S11CalibrationModel = field(default_factory=S11CalibrationModel)
|
|
reference: PreprocessAssetModel = field(default_factory=PreprocessAssetModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class PreprocessNotchModel:
|
|
"""Optional frequency-domain notch filter applied after calibration and reference subtraction."""
|
|
|
|
enabled: bool = False
|
|
bands_hz: list[tuple[float, float]] = field(default_factory=list)
|
|
taper_width_hz: float = 40_000_000.0
|
|
taper_type: str = "cosine"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class PreprocessModel:
|
|
"""Selected preprocessing artifacts for live acquisition."""
|
|
|
|
s21: S21PreprocessModel = field(default_factory=S21PreprocessModel)
|
|
s11: S11PreprocessModel = field(default_factory=S11PreprocessModel)
|
|
notch: PreprocessNotchModel = field(default_factory=PreprocessNotchModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class GprTxGeometryModel:
|
|
"""One transmitter geometry record keyed by output switch position."""
|
|
|
|
output_pos: int = 0
|
|
x_m: float = 0.0
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class GprRxGeometryModel:
|
|
"""One receiver geometry record keyed by input switch position."""
|
|
|
|
input_pos: int = 0
|
|
x_m: float = 0.0
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class GprModel:
|
|
"""Stable GPR configuration saved in run_config.json."""
|
|
|
|
mode: str = "point"
|
|
relative_permittivity: float = 1.0
|
|
tx_geometry: list[GprTxGeometryModel] = field(default_factory=list)
|
|
rx_geometry: list[GprRxGeometryModel] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RunConfigModel:
|
|
"""Top-level runtime config model consumed by C++ processes and GUI."""
|
|
|
|
radar: RadarModel = field(default_factory=RadarModel)
|
|
input_switch: SwitchModel = field(default_factory=lambda: SwitchModel(name=""))
|
|
output_switch: SwitchModel = field(default_factory=lambda: SwitchModel(name=""))
|
|
rings: RingsModel = field(default_factory=RingsModel)
|
|
runtime: RuntimeModel = field(default_factory=RuntimeModel)
|
|
preprocess: PreprocessModel = field(default_factory=PreprocessModel)
|
|
gpr: GprModel = field(default_factory=GprModel)
|
|
combos: list[ComboModel] = field(default_factory=list)
|
|
|
|
LIBREVNA_MODEL = "librevna"
|
|
LIBREVNA_MULTI_MODEL = "librevna_multi"
|
|
COMPACT_M_K209_MODEL = "compact_m_k209"
|
|
MULTI_DEVICE_INPUT_POSITIONS = 4
|
|
MULTI_DEVICE_OUTPUT_POSITIONS = 2
|
|
|
|
@staticmethod
|
|
def build_full_combos(input_positions: int, output_positions: int) -> list[ComboModel]:
|
|
"""Build full Cartesian product of input/output switch positions."""
|
|
return [
|
|
ComboModel(input=input_pos, output=output_pos)
|
|
for output_pos in range(output_positions)
|
|
for input_pos in range(input_positions)
|
|
]
|
|
|
|
@classmethod
|
|
def build_multi_device_virtual_combos(cls) -> list[ComboModel]:
|
|
"""Build fixed virtual combo matrix for one master and two slave devices."""
|
|
return cls.build_full_combos(
|
|
cls.MULTI_DEVICE_INPUT_POSITIONS,
|
|
cls.MULTI_DEVICE_OUTPUT_POSITIONS,
|
|
)
|
|
|
|
@property
|
|
def is_multi_device(self) -> bool:
|
|
"""Return whether this config targets synchronized multi-device acquisition."""
|
|
return self.radar.model == self.LIBREVNA_MULTI_MODEL
|
|
|
|
def apply_device_model_constraints(self) -> None:
|
|
"""Apply only required wire-format constraints for the selected device model."""
|
|
if not self.is_multi_device:
|
|
return
|
|
|
|
self.radar.model = self.LIBREVNA_MULTI_MODEL
|
|
self.output_switch.name = self.output_switch.name or "virtual_output"
|
|
self.output_switch.driver_mode = "mock"
|
|
self.output_switch.driver = self.output_switch.driver or "h7992"
|
|
self.output_switch.radar_port = 1
|
|
self.output_switch.positions = self.MULTI_DEVICE_OUTPUT_POSITIONS
|
|
self.output_switch.default_position = 0
|
|
|
|
self.input_switch.name = self.input_switch.name or "virtual_input"
|
|
self.input_switch.driver_mode = "mock"
|
|
self.input_switch.driver = self.input_switch.driver or "h7992"
|
|
self.input_switch.radar_port = 2
|
|
self.input_switch.positions = self.MULTI_DEVICE_INPUT_POSITIONS
|
|
self.input_switch.default_position = 0
|
|
self.combos = self.build_multi_device_virtual_combos()
|
|
|
|
def ensure_combos(self) -> None:
|
|
"""Populate combos with full matrix when no explicit run combos are set."""
|
|
if self.is_multi_device:
|
|
self.apply_device_model_constraints()
|
|
return
|
|
if self.combos:
|
|
return
|
|
self.combos = self.build_full_combos(self.input_switch.positions, self.output_switch.positions)
|
|
|
|
@classmethod
|
|
def from_dict(cls, payload: dict[str, Any]) -> RunConfigModel:
|
|
"""Build model from JSON-like payload using codec layer."""
|
|
from python_app.models.run_config_codec import run_config_from_dict
|
|
|
|
return run_config_from_dict(payload)
|
|
|
|
@classmethod
|
|
def load_from_path(cls, path: Path) -> RunConfigModel:
|
|
"""Load JSON file from disk and decode into model."""
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(payload, dict):
|
|
raise ValueError(f"Config root must be JSON object: {path}")
|
|
return cls.from_dict(payload)
|
|
|
|
def clone(self) -> RunConfigModel:
|
|
"""Create deep copy through codec round-trip."""
|
|
return RunConfigModel.from_dict(self.to_dict())
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Encode model into JSON-serializable dictionary."""
|
|
from python_app.models.run_config_codec import run_config_to_dict
|
|
|
|
return run_config_to_dict(self)
|