"""Dataclass schema for runtime configuration used by Python pipeline tools.""" from __future__ import annotations from dataclasses import dataclass, field import hashlib import json import logging from pathlib import Path from typing import Any logger = logging.getLogger(__name__) @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.""" # A valid default range (stop > start) so a bare/default config is self-consistent; # operational values come from run_config.json. (1 MHz .. 6 GHz mirrors the real configs.) start_hz: float = 1_000_000.0 stop_hz: float = 6_000_000_000.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 KamilAdcPhaseCalibrationModel: """Affine law mapping the reference signal's unwrapped phase to frequency. Two fixed anchor points ``(phase0_rad, freq0_hz)`` and ``(phase1_rad, freq1_hz)`` define ``f(phase) = freq0_hz + (phase - phase0_rad) * (freq1_hz - freq0_hz) / (phase1_rad - phase0_rad)``, applied to the *absolute* unwrapped phase of every sweep. These are physical constants of the reference arm and must be supplied by config — never derived from a live sweep. """ phase0_rad: float = 0.0 freq0_hz: float = 2_046_000_000.0 phase1_rad: float = 300.0 freq1_hz: float = 5_612_000_000.0 @dataclass(slots=True) class KamilAdcBandModel: """Fixed frequency window every sweep is cropped to and resampled onto. Each sweep is resampled onto ``linspace(start_hz, stop_hz, points)`` so all sweeps share one identical axis and can be averaged/subtracted. The window must lie inside the (floating) range each sweep actually covers; sweeps that fail to cover it are rejected rather than edge-extrapolated. """ start_hz: float = 2_100_000_000.0 stop_hz: float = 5_500_000_000.0 points: int = 2048 @dataclass(slots=True) class KamilAdcModel: """External Kamil ADC acquisition process settings.""" project_dir: str = "" executable_path: str = "" tty_path: str = "" args: list[str] = field(default_factory=list) env: dict[str, str] = field(default_factory=dict) startup_timeout_s: float = 5.0 sweep_timeout_s: float = 5.0 stop_timeout_s: float = 2.0 phase_calibration: KamilAdcPhaseCalibrationModel = field( default_factory=KamilAdcPhaseCalibrationModel ) band: KamilAdcBandModel = field(default_factory=KamilAdcBandModel) @dataclass(slots=True) class LaserManualModeModel: """Manual laser-control setpoints.""" temp1: float = 25.0 temp2: float = 25.0 current1: float = 30.0 current2: float = 30.0 @dataclass(slots=True) class LaserVariationModeModel: """Laser-control variation task parameters.""" variation_type: str = "CHANGE_CURRENT_LD1" static_temp1: float = 25.0 static_temp2: float = 25.0 static_current1: float = 30.0 static_current2: float = 30.0 min_value: float = 30.0 max_value: float = 35.0 step: float = 0.1 time_step: int = 20 delay_time: int = 3 # Max allowed |measured - target| laser temperature before the temperature # checker warns, °C. Applied independently to both lasers (temp1/temp2). temp_tolerance_c: float = 0.03 @dataclass(slots=True) class LaserControlModel: """Laser-control board settings applied before Kamil ADC acquisition.""" enabled: bool = False port: str = "" mode: str = "manual" pi_coeff1_p: int = 2560 pi_coeff1_i: int = 128 pi_coeff2_p: int = 2560 pi_coeff2_i: int = 128 manual: LaserManualModeModel = field(default_factory=LaserManualModeModel) variation: LaserVariationModeModel = field(default_factory=LaserVariationModeModel) @dataclass(slots=True) class RadarModel: """Radar section of run configuration.""" model: str = "librevna" serial: str = "" remote_host: str = "127.0.0.1" remote_port: int = 50209 driver_mode: str = "mock" mock_signal_hz: float = 1_000_000.0 visa_library: str = "" sweep: RadarSweepModel = field(default_factory=RadarSweepModel) multi_device: RadarMultiDeviceModel = field(default_factory=RadarMultiDeviceModel) kamil_adc: KamilAdcModel = field(default_factory=KamilAdcModel) laser_control: LaserControlModel = field(default_factory=LaserControlModel) @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 ControlButtonModel: """Physical GPIO push-button that triggers a runtime action on press. Default wiring: the button sits between the GPIO line and GND with the internal pull-up enabled, so the line idles high and a press drives it low (``active_low``). The watcher reacts to the press edge only, so one push yields one action. Disabled by default so non-Pi hosts never touch GPIO. """ ACTION_CAPTURE_TMP_REFERENCE = "capture_tmp_reference" enabled: bool = False gpio_chip: str = "/dev/gpiochip0" pin: int = -1 active_low: bool = True bias: str = "" debounce_ms: int = 50 action: str = ACTION_CAPTURE_TMP_REFERENCE @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. y_m / z_m default to 0 so 1D antenna layouts keep their pre-3D semantics. """ output_pos: int = 0 x_m: float = 0.0 y_m: float = 0.0 z_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 y_m: float = 0.0 z_m: float = 0.0 @dataclass(slots=True) class GprModel: """Stable GPR configuration saved in run_config.json.""" 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 LoggingModel: """Application logging settings shared by the GUI and headless daemon. ``level`` is the verbosity floor (one of DEBUG/INFO/WARNING/ERROR, case-insensitive); it is chosen from the UI log-level selector, applied to the ``python_app`` logger at startup, and persisted here so the same verbosity is restored on the next run. """ level: str = "info" @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) control_button: ControlButtonModel = field(default_factory=ControlButtonModel) logging: LoggingModel = field(default_factory=LoggingModel) LIBREVNA_MODEL = "librevna" LIBREVNA_MULTI_MODEL = "librevna_multi" COMPACT_M_K209_MODEL = "compact_m_k209" KAMIL_ADC_MODEL = "kamil_adc" SN9000_MODEL = "sn9000" 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, ) @classmethod def build_matrix_radar_virtual_combos(cls) -> list[ComboModel]: """Build the canonical 2x4 virtual combo matrix shared by matrix-mode radars.""" return cls.build_multi_device_virtual_combos() @property def is_multi_device(self) -> bool: """Return whether this config targets synchronized multi-device acquisition.""" return self.radar.model == self.LIBREVNA_MULTI_MODEL @property def is_sn9000(self) -> bool: """Return whether this config targets the SN9000 multi-port analyzer.""" return self.radar.model == self.SN9000_MODEL @property def is_matrix_radar(self) -> bool: """Return whether this config acquires the full virtual switch matrix per sweep.""" return self.is_multi_device or self.is_sn9000 @property def is_kamil_adc(self) -> bool: """Return whether this config targets the external Kamil ADC acquisition path.""" return self.radar.model == self.KAMIL_ADC_MODEL def radar_key_extra_parts(self) -> list[str]: """Return model-specific identity parts that affect captured data.""" if self.is_multi_device: return list(self.radar.multi_device.slave_serials) if self.is_kamil_adc: payload = { "kamil_adc": { "project_dir": self.radar.kamil_adc.project_dir, "executable_path": self.radar.kamil_adc.executable_path, "tty_path": self.radar.kamil_adc.tty_path, "args": list(self.radar.kamil_adc.args), "env": dict(sorted(self.radar.kamil_adc.env.items())), }, "laser_control": { "enabled": self.radar.laser_control.enabled, "port": self.radar.laser_control.port, "mode": self.radar.laser_control.mode, "pi_coeff1_p": self.radar.laser_control.pi_coeff1_p, "pi_coeff1_i": self.radar.laser_control.pi_coeff1_i, "pi_coeff2_p": self.radar.laser_control.pi_coeff2_p, "pi_coeff2_i": self.radar.laser_control.pi_coeff2_i, "manual": { "temp1": self.radar.laser_control.manual.temp1, "temp2": self.radar.laser_control.manual.temp2, "current1": self.radar.laser_control.manual.current1, "current2": self.radar.laser_control.manual.current2, }, "variation": { "variation_type": self.radar.laser_control.variation.variation_type, "static_temp1": self.radar.laser_control.variation.static_temp1, "static_temp2": self.radar.laser_control.variation.static_temp2, "static_current1": self.radar.laser_control.variation.static_current1, "static_current2": self.radar.laser_control.variation.static_current2, "min_value": self.radar.laser_control.variation.min_value, "max_value": self.radar.laser_control.variation.max_value, "step": self.radar.laser_control.variation.step, "time_step": self.radar.laser_control.variation.time_step, "delay_time": self.radar.laser_control.variation.delay_time, }, }, } encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:16] return [f"kamil_{digest}"] return [] def apply_device_model_constraints(self) -> None: """Apply only required wire-format constraints for the selected device model.""" if not self.is_matrix_radar: return self._apply_matrix_virtual_switches() self.combos = self.build_matrix_radar_virtual_combos() def _apply_matrix_virtual_switches(self) -> None: """Pin the canonical 2x4 virtual switch matrix used by all matrix-mode radars.""" 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 def ensure_combos(self) -> None: """Populate combos with full matrix when no explicit run combos are set.""" if self.is_matrix_radar: 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 a JSON file from disk and decode it into a model. Raises ValueError when the file's JSON root is not an object. """ logger.debug("Loading run config from %s", path) 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)