510 lines
19 KiB
Python
510 lines
19 KiB
Python
"""Dataclass schema for runtime configuration used by Python pipeline tools."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from python_app.file_io import sanitize_path_component
|
|
|
|
logger = logging.getLogger(__name__)
|
|
MAX_COMBOS = 4096
|
|
|
|
|
|
@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
|
|
# Extra per-stage settling requested from the device, on top of its own PLL
|
|
# settling delay (60 us by default). With per-point stimulus switching this is the
|
|
# window the switch controller has to move the switch between stages.
|
|
dwell_us: int = 0
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RadarMultiDeviceModel:
|
|
"""Multi-device LibreVNA topology settings."""
|
|
|
|
slave_serials: list[str] = field(default_factory=list)
|
|
force_external_reference: bool = False
|
|
recovery_attempts: int = 3
|
|
output_switch_positions: int = 1 # 1 = свитча нет
|
|
input_switch_positions: int = 1 # 1 = свитча нет
|
|
# Positions of the stimulus switch stepped through WITHIN one frequency point,
|
|
# driven by the controller in the hardware trigger loop rather than by this host.
|
|
# 1 = выключено (стоковое поведение). Требует доработанного битстрима мастера.
|
|
stimulus_switch_positions: int = 1
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RphRadarPhaseCalibrationModel:
|
|
"""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 RphRadarBandModel:
|
|
"""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 RphRadarModel:
|
|
"""RadioPhotonic radar (RPH) acquisition: E-502 collector process or RFG board UDP stream."""
|
|
|
|
# "e502": the C++ collector streams over tty_path (project_dir..stop_timeout_s are its
|
|
# settings); "rfg_udp": the RFG ADC board sends averaged sweep points to udp_bind.
|
|
source: str = "e502"
|
|
udp_bind: str = "0.0.0.0:5000"
|
|
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: RphRadarPhaseCalibrationModel = field(
|
|
default_factory=RphRadarPhaseCalibrationModel
|
|
)
|
|
band: RphRadarBandModel = field(default_factory=RphRadarBandModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LaserManualModeModel:
|
|
"""Manual laser-control setpoints (defaults: the lab GUI, RadioPhotonic_PCB_PC_software)."""
|
|
|
|
temp1: float = 24.0
|
|
temp2: float = 24.9
|
|
current1: float = 97.3
|
|
current2: float = 102.0
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LaserVariationModeModel:
|
|
"""Laser-control variation task parameters."""
|
|
|
|
variation_type: str = "CHANGE_CURRENT_LD1"
|
|
static_temp1: float = 24.0
|
|
static_temp2: float = 24.9
|
|
static_current1: float = 97.3
|
|
static_current2: float = 102.0
|
|
min_value: float = 97.3
|
|
max_value: float = 107.0
|
|
step: float = 0.05
|
|
time_step: int = 50
|
|
delay_time: int = 10
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LaserControlModel:
|
|
"""Laser-control board settings applied before RadioPhotonic radar (RPH) acquisition."""
|
|
|
|
enabled: bool = False
|
|
port: str = ""
|
|
mode: str = "manual"
|
|
pi_coeff1_p: int = 1280
|
|
pi_coeff1_i: int = 25
|
|
pi_coeff2_p: int = 1280
|
|
pi_coeff2_i: int = 179
|
|
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 = ""
|
|
driver_mode: str = "mock"
|
|
sweep: RadarSweepModel = field(default_factory=RadarSweepModel)
|
|
multi_device: RadarMultiDeviceModel = field(default_factory=RadarMultiDeviceModel)
|
|
rph_radar: RphRadarModel = field(default_factory=RphRadarModel)
|
|
laser_control: LaserControlModel = field(default_factory=LaserControlModel)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SwitchModel:
|
|
"""Generic switch section of run configuration."""
|
|
|
|
name: str
|
|
driver_mode: str = "mock"
|
|
radar_port: int = 0
|
|
positions: int = 1
|
|
default_position: int = 0
|
|
gpio_chip: str = ""
|
|
pin_a: int = -1
|
|
pin_b: int = -1
|
|
|
|
|
|
@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)
|
|
object_calibration: 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)
|
|
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."""
|
|
|
|
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"
|
|
RPH_RADAR_MODEL = "rph_radar"
|
|
MULTI_DEVICE_INPUT_POSITIONS = 4
|
|
MULTI_DEVICE_OUTPUT_POSITIONS = 2
|
|
# Stimulus ports the master itself owns. The virtual output axis is this times the
|
|
# per-point stimulus switch positions, so it equals MULTI_DEVICE_OUTPUT_POSITIONS
|
|
# unless per-point switching is enabled.
|
|
MULTI_DEVICE_MASTER_PORTS = 2
|
|
|
|
@staticmethod
|
|
def build_full_combos(input_positions: int, output_positions: int) -> list[ComboModel]:
|
|
"""Build full Cartesian product of input/output switch positions."""
|
|
if any(isinstance(value, bool) or not isinstance(value, int) or value <= 0
|
|
for value in (input_positions, output_positions)):
|
|
raise ValueError("Switch positions must be positive integers")
|
|
if input_positions * output_positions > MAX_COMBOS:
|
|
raise ValueError(f"Too many combos: limit is {MAX_COMBOS}")
|
|
return [
|
|
ComboModel(input=input_pos, output=output_pos)
|
|
for output_pos in range(output_positions)
|
|
for input_pos in range(input_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
|
|
|
|
@property
|
|
def matrix_output_switch_positions(self) -> int:
|
|
"""Physical positions of the real switch on the master stimulus path."""
|
|
if not self.is_multi_device:
|
|
return 1
|
|
return max(1, int(self.radar.multi_device.output_switch_positions))
|
|
|
|
@property
|
|
def matrix_input_switch_positions(self) -> int:
|
|
"""Physical positions of the real switch on the slave receiver path."""
|
|
if not self.is_multi_device:
|
|
return 1
|
|
return max(1, int(self.radar.multi_device.input_switch_positions))
|
|
|
|
@property
|
|
def matrix_stimulus_switch_positions(self) -> int:
|
|
"""Positions the stimulus switch steps through inside one frequency point."""
|
|
if not self.is_multi_device:
|
|
return 1
|
|
return max(1, int(self.radar.multi_device.stimulus_switch_positions))
|
|
|
|
@property
|
|
def matrix_virtual_output_positions(self) -> int:
|
|
"""Virtual output axis size produced by one acquisition.
|
|
|
|
Per-point stimulus switching multiplies the master's own ports, because each
|
|
port is measured once per switch position at every frequency — so those extra
|
|
outputs cost no extra sweep and are NOT driven by this host.
|
|
"""
|
|
if not self.is_multi_device:
|
|
return self.MULTI_DEVICE_OUTPUT_POSITIONS
|
|
return self.MULTI_DEVICE_MASTER_PORTS * self.matrix_stimulus_switch_positions
|
|
|
|
def build_runtime_combos(self) -> list[ComboModel]:
|
|
"""Build the combo matrix from the effective switch axis sizes."""
|
|
return self.build_full_combos(self.input_switch.positions, self.output_switch.positions)
|
|
|
|
@property
|
|
def is_rph_radar(self) -> bool:
|
|
"""Return whether this config targets the external RadioPhotonic radar (RPH) acquisition path."""
|
|
return self.radar.model == self.RPH_RADAR_MODEL
|
|
|
|
def radar_key(self) -> str:
|
|
"""Stable hardware/sweep identity used to find saved preprocess sets."""
|
|
radar, sweep = self.radar, self.radar.sweep
|
|
if self.is_rph_radar:
|
|
laser = radar.laser_control
|
|
mode = laser.mode.strip().lower()
|
|
optical = asdict(laser.variation if mode == "variation" else laser.manual)
|
|
rph = radar.rph_radar
|
|
# The RFG source always runs the profile's sweep (its length and step timing), even when
|
|
# the laser is started elsewhere; E-502 keys stay as they were before the source field.
|
|
rfg = rph.source == "rfg_udp"
|
|
physical = (asdict(rph.band), asdict(rph.phase_calibration), rph.source if rfg else rph.args,
|
|
(self.input_switch.positions, self.output_switch.positions),
|
|
(mode, optical, laser.pi_coeff1_p, laser.pi_coeff1_i,
|
|
laser.pi_coeff2_p, laser.pi_coeff2_i) if laser.enabled or rfg else None)
|
|
digest = hashlib.sha256(json.dumps(physical, sort_keys=True).encode()).hexdigest()[:16]
|
|
return f"{radar.model}_{digest}"
|
|
serials = [radar.serial or "no_serial"]
|
|
if self.is_multi_device:
|
|
serials.extend(value.strip() or "no_serial" for value in radar.multi_device.slave_serials)
|
|
serial = "_".join(sanitize_path_component(value) for value in serials)
|
|
points = str(int(sweep.points))
|
|
# Switch positions decide what each combo index means, so sets from another matrix never match.
|
|
matrix = (f"_out{self.matrix_output_switch_positions}_in{self.matrix_input_switch_positions}"
|
|
f"_stim{self.matrix_stimulus_switch_positions}") if self.is_multi_device else ""
|
|
return (
|
|
f"{radar.model}_{serial}"
|
|
f"_st{_format_float_for_key(sweep.start_hz)}_sp{_format_float_for_key(sweep.stop_hz)}"
|
|
f"_p{points}_if{_format_float_for_key(sweep.if_bandwidth_hz)}"
|
|
f"_pw{_format_float_for_key(sweep.power_dbm)}{matrix}"
|
|
)
|
|
|
|
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._apply_matrix_virtual_switches()
|
|
self.combos = self.build_runtime_combos()
|
|
|
|
def _apply_matrix_virtual_switches(self) -> None:
|
|
"""Pin the virtual switch matrix, widened by any real switch on the path."""
|
|
out_physical = self.matrix_output_switch_positions
|
|
in_physical = self.matrix_input_switch_positions
|
|
|
|
self.output_switch.name = self.output_switch.name or "virtual_output"
|
|
self.output_switch.driver_mode = "mock"
|
|
self.output_switch.radar_port = 1
|
|
self.output_switch.positions = out_physical * self.matrix_virtual_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.radar_port = 2
|
|
self.input_switch.positions = in_physical * 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_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: object) -> 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)
|
|
|
|
|
|
def _format_float_for_key(value: float) -> str:
|
|
"""Format float compactly for deterministic key serialization."""
|
|
integer = int(round(value))
|
|
if abs(value - float(integer)) < 1e-6:
|
|
return str(integer)
|
|
return f"{value:.6f}".rstrip("0").rstrip(".")
|