178 lines
8.4 KiB
Python
178 lines
8.4 KiB
Python
"""Encoding and decoding logic for :mod:`python_app.models.run_config_schema`."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from python_app.models.run_config_schema import ComboModel, RunConfigModel
|
|
from python_app.models.run_config_validation import as_dict, load_ring_payload, load_switch_payload
|
|
|
|
|
|
def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel:
|
|
"""Decode JSON-like payload into :class:`RunConfigModel`."""
|
|
# Schema carries only minimal-safe fallbacks; operational defaults live in run_config.json.
|
|
model = RunConfigModel()
|
|
|
|
radar_payload = as_dict(payload.get("radar"), "radar")
|
|
sweep_payload = as_dict(radar_payload.get("sweep"), "radar.sweep")
|
|
switches_payload = as_dict(payload.get("switches"), "switches")
|
|
port1_payload = as_dict(switches_payload.get("port1"), "switches.port1")
|
|
port2_payload = as_dict(switches_payload.get("port2"), "switches.port2")
|
|
run_payload = as_dict(payload.get("run"), "run")
|
|
preprocess_payload = as_dict(payload.get("preprocess"), "preprocess")
|
|
rings_payload = as_dict(payload.get("rings"), "rings")
|
|
raw_ring_payload = as_dict(rings_payload.get("raw"), "rings.raw")
|
|
raw_tap_ring_payload = as_dict(rings_payload.get("raw_tap"), "rings.raw_tap")
|
|
pre_ring_payload = as_dict(rings_payload.get("preprocessed"), "rings.preprocessed")
|
|
pre_tap_ring_payload = as_dict(rings_payload.get("preprocessed_tap"), "rings.preprocessed_tap")
|
|
result_ring_payload = as_dict(rings_payload.get("results"), "rings.results")
|
|
|
|
model.radar.model = str(radar_payload.get("model", model.radar.model))
|
|
model.radar.serial = str(radar_payload.get("serial", model.radar.serial))
|
|
model.radar.driver_mode = str(radar_payload.get("driver_mode", model.radar.driver_mode))
|
|
model.radar.mock_signal_hz = float(radar_payload.get("mock_signal_hz", model.radar.mock_signal_hz))
|
|
|
|
model.radar.sweep.start_hz = float(sweep_payload.get("start_hz", model.radar.sweep.start_hz))
|
|
model.radar.sweep.stop_hz = float(sweep_payload.get("stop_hz", model.radar.sweep.stop_hz))
|
|
model.radar.sweep.points = int(sweep_payload.get("points", model.radar.sweep.points))
|
|
model.radar.sweep.if_bandwidth_hz = float(
|
|
sweep_payload.get("if_bandwidth_hz", model.radar.sweep.if_bandwidth_hz)
|
|
)
|
|
model.radar.sweep.power_dbm = float(sweep_payload.get("stimulus_power_dbm", model.radar.sweep.power_dbm))
|
|
|
|
load_switch_payload(port1_payload, model.output_switch)
|
|
load_switch_payload(port2_payload, model.input_switch)
|
|
|
|
model.runtime.settling_ms = int(run_payload.get("settling_ms", model.runtime.settling_ms))
|
|
model.runtime.idle_sleep_ms = int(run_payload.get("idle_sleep_ms", model.runtime.idle_sleep_ms))
|
|
model.runtime.continuous = bool(run_payload.get("continuous", model.runtime.continuous))
|
|
model.runtime.processing_live_config_path = str(
|
|
run_payload.get("processing_live_config_path", model.runtime.processing_live_config_path)
|
|
)
|
|
|
|
model.preprocess.calibration_set = str(preprocess_payload.get("calibration_set", model.preprocess.calibration_set))
|
|
model.preprocess.reference_set = str(preprocess_payload.get("reference_set", model.preprocess.reference_set))
|
|
model.preprocess.calibration_bundle_path = str(
|
|
preprocess_payload.get("calibration_bundle_path", model.preprocess.calibration_bundle_path)
|
|
)
|
|
model.preprocess.reference_bundle_path = str(
|
|
preprocess_payload.get("reference_bundle_path", model.preprocess.reference_bundle_path)
|
|
)
|
|
|
|
load_ring_payload(raw_ring_payload, model.rings.raw)
|
|
load_ring_payload(raw_tap_ring_payload, model.rings.raw_tap)
|
|
load_ring_payload(pre_ring_payload, model.rings.preprocessed)
|
|
load_ring_payload(pre_tap_ring_payload, model.rings.preprocessed_tap)
|
|
load_ring_payload(result_ring_payload, model.rings.results)
|
|
|
|
combos_payload = run_payload.get("combos", [])
|
|
model.combos = []
|
|
if isinstance(combos_payload, list):
|
|
for combo in combos_payload:
|
|
combo_payload = as_dict(combo, "run.combos[]")
|
|
model.combos.append(
|
|
ComboModel(
|
|
input=int(combo_payload.get("input", 0)),
|
|
output=int(combo_payload.get("output", 0)),
|
|
)
|
|
)
|
|
|
|
model.ensure_combos()
|
|
return model
|
|
|
|
|
|
def load_run_config(path: Path) -> RunConfigModel:
|
|
"""Load JSON config from path and decode into :class:`RunConfigModel`."""
|
|
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 run_config_from_dict(payload)
|
|
|
|
|
|
def run_config_to_dict(model: RunConfigModel) -> dict[str, Any]:
|
|
"""Encode :class:`RunConfigModel` to C++ pipeline-compatible JSON structure."""
|
|
model.ensure_combos()
|
|
return {
|
|
"radar": {
|
|
"model": model.radar.model,
|
|
"serial": model.radar.serial,
|
|
"driver_mode": model.radar.driver_mode,
|
|
"mock_signal_hz": model.radar.mock_signal_hz,
|
|
"sweep": {
|
|
"start_hz": model.radar.sweep.start_hz,
|
|
"stop_hz": model.radar.sweep.stop_hz,
|
|
"points": model.radar.sweep.points,
|
|
"if_bandwidth_hz": model.radar.sweep.if_bandwidth_hz,
|
|
"stimulus_power_dbm": model.radar.sweep.power_dbm,
|
|
},
|
|
},
|
|
"switches": {
|
|
"port1": {
|
|
"name": model.output_switch.name,
|
|
"driver_mode": model.output_switch.driver_mode,
|
|
"driver": model.output_switch.driver,
|
|
"radar_port": model.output_switch.radar_port,
|
|
"positions": model.output_switch.positions,
|
|
"default_position": model.output_switch.default_position,
|
|
"gpio_chip": model.output_switch.gpio_chip,
|
|
"pin_a": model.output_switch.pin_a,
|
|
"pin_b": model.output_switch.pin_b,
|
|
"invert_logic": model.output_switch.invert_logic,
|
|
},
|
|
"port2": {
|
|
"name": model.input_switch.name,
|
|
"driver_mode": model.input_switch.driver_mode,
|
|
"driver": model.input_switch.driver,
|
|
"radar_port": model.input_switch.radar_port,
|
|
"positions": model.input_switch.positions,
|
|
"default_position": model.input_switch.default_position,
|
|
"gpio_chip": model.input_switch.gpio_chip,
|
|
"pin_a": model.input_switch.pin_a,
|
|
"pin_b": model.input_switch.pin_b,
|
|
"invert_logic": model.input_switch.invert_logic,
|
|
},
|
|
},
|
|
"run": {
|
|
"settling_ms": model.runtime.settling_ms,
|
|
"idle_sleep_ms": model.runtime.idle_sleep_ms,
|
|
"continuous": model.runtime.continuous,
|
|
"processing_live_config_path": model.runtime.processing_live_config_path,
|
|
"combos": [{"input": combo.input, "output": combo.output} for combo in model.combos],
|
|
},
|
|
"preprocess": {
|
|
"calibration_set": model.preprocess.calibration_set,
|
|
"reference_set": model.preprocess.reference_set,
|
|
"calibration_bundle_path": model.preprocess.calibration_bundle_path,
|
|
"reference_bundle_path": model.preprocess.reference_bundle_path,
|
|
},
|
|
"rings": {
|
|
"raw": {
|
|
"name": model.rings.raw.name,
|
|
"capacity": model.rings.raw.capacity,
|
|
"slot_size_bytes": model.rings.raw.slot_size_bytes,
|
|
},
|
|
"raw_tap": {
|
|
"name": model.rings.raw_tap.name,
|
|
"capacity": model.rings.raw_tap.capacity,
|
|
"slot_size_bytes": model.rings.raw_tap.slot_size_bytes,
|
|
},
|
|
"preprocessed": {
|
|
"name": model.rings.preprocessed.name,
|
|
"capacity": model.rings.preprocessed.capacity,
|
|
"slot_size_bytes": model.rings.preprocessed.slot_size_bytes,
|
|
},
|
|
"preprocessed_tap": {
|
|
"name": model.rings.preprocessed_tap.name,
|
|
"capacity": model.rings.preprocessed_tap.capacity,
|
|
"slot_size_bytes": model.rings.preprocessed_tap.slot_size_bytes,
|
|
},
|
|
"results": {
|
|
"name": model.rings.results.name,
|
|
"capacity": model.rings.results.capacity,
|
|
"slot_size_bytes": model.rings.results.slot_size_bytes,
|
|
},
|
|
},
|
|
}
|