386 lines
20 KiB
Python
386 lines
20 KiB
Python
"""Bind configuration models to settings widgets without owning runtime state."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from contextlib import ExitStack
|
|
from copy import deepcopy
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtCore import QSignalBlocker
|
|
from PyQt6.QtWidgets import QCheckBox, QComboBox, QDoubleSpinBox, QFormLayout, QLabel, QSpinBox
|
|
|
|
from python_app.storage.npz.paths import radar_config_filename_prefix
|
|
from python_app.gui.antenna_geometry_editor import AntennaGeometryEditor
|
|
from python_app.gui.controllers.sections.views import ScalarWidget, SettingWidget, read_setting, write_setting
|
|
from python_app.models.gui_profile_schema import GuiPreprocessDialogStateModel, GuiProcessingStateModel, GuiStateModel, Scalar
|
|
from python_app.models.run_config_schema import ComboModel, GprModel, GprRxGeometryModel, GprTxGeometryModel, RunConfigModel
|
|
from python_app.models.run_config_validation import parse_combos_from_text, parse_position_csv, validate_gpr_model
|
|
from python_app.models.setting_fields import GUI_FIELDS, active_live_fields, active_web_fields
|
|
from python_app.orchestration.live_processing_config import ProcessingLiveConfig
|
|
from python_app.orchestration.preprocess_assets import PREPROCESS_ASSET_NAMES, preprocess_asset_model
|
|
|
|
# Stable fields have one mapping for both applying a profile and reading edits.
|
|
_SWEEP_FIELDS = (
|
|
("radar.sweep.start_hz", "_start_hz_input"),
|
|
("radar.sweep.stop_hz", "_stop_hz_input"),
|
|
("radar.sweep.points", "_points_input"),
|
|
("radar.sweep.if_bandwidth_hz", "_ifbw_input"),
|
|
("radar.sweep.power_dbm", "_power_input"),
|
|
)
|
|
_ADC_FIELDS = (
|
|
("radar.kamil_adc.project_dir", "_adc_project_dir_input"),
|
|
("radar.kamil_adc.executable_path", "_adc_executable_path_input"),
|
|
("radar.kamil_adc.tty_path", "_adc_tty_path_input"),
|
|
("radar.kamil_adc.args", "_adc_args_input"),
|
|
("radar.kamil_adc.env", "_adc_env_input"),
|
|
("radar.kamil_adc.startup_timeout_s", "_adc_startup_timeout_s_input"),
|
|
("radar.kamil_adc.sweep_timeout_s", "_adc_sweep_timeout_s_input"),
|
|
("radar.kamil_adc.stop_timeout_s", "_adc_stop_timeout_s_input"),
|
|
("radar.laser_control.enabled", "_optical_enabled_checkbox"),
|
|
("radar.laser_control.port", "_optical_port_input"),
|
|
("radar.laser_control.mode", "_optical_mode_combo"),
|
|
("radar.laser_control.pi_coeff1_p", "_optical_pi_coeff1_p_input"),
|
|
("radar.laser_control.pi_coeff1_i", "_optical_pi_coeff1_i_input"),
|
|
("radar.laser_control.pi_coeff2_p", "_optical_pi_coeff2_p_input"),
|
|
("radar.laser_control.pi_coeff2_i", "_optical_pi_coeff2_i_input"),
|
|
("radar.laser_control.manual.temp1", "_optical_manual_temp1_input"),
|
|
("radar.laser_control.manual.temp2", "_optical_manual_temp2_input"),
|
|
("radar.laser_control.manual.current1", "_optical_manual_current1_input"),
|
|
("radar.laser_control.manual.current2", "_optical_manual_current2_input"),
|
|
("radar.laser_control.variation.variation_type", "_optical_variation_type_combo"),
|
|
("radar.laser_control.variation.static_temp1", "_optical_static_temp1_input"),
|
|
("radar.laser_control.variation.static_temp2", "_optical_static_temp2_input"),
|
|
("radar.laser_control.variation.static_current1", "_optical_static_current1_input"),
|
|
("radar.laser_control.variation.static_current2", "_optical_static_current2_input"),
|
|
("radar.laser_control.variation.min_value", "_optical_min_value_input"),
|
|
("radar.laser_control.variation.max_value", "_optical_max_value_input"),
|
|
("radar.laser_control.variation.step", "_optical_step_input"),
|
|
("radar.laser_control.variation.time_step", "_optical_time_step_input"),
|
|
("radar.laser_control.variation.delay_time", "_optical_delay_time_input"),
|
|
)
|
|
_GEOMETRY_FIELDS = {
|
|
"relative_permittivity": "_gpr_relative_permittivity",
|
|
"tx_geometry": "_gpr_tx_geometry_input",
|
|
"rx_geometry": "_gpr_rx_geometry_input",
|
|
}
|
|
|
|
|
|
def web_apply_field_names() -> frozenset[str]:
|
|
return frozenset(binding.spec.web for binding in GUI_FIELDS if binding.spec.web) | frozenset(_GEOMETRY_FIELDS)
|
|
|
|
|
|
def _model_field(config: RunConfigModel, path: str) -> tuple[object, str]:
|
|
owner: object = config
|
|
*parents, name = path.split(".")
|
|
for parent in parents:
|
|
owner = getattr(owner, parent)
|
|
return owner, name
|
|
|
|
|
|
def _scalar_widget(widget: SettingWidget) -> ScalarWidget:
|
|
if isinstance(widget, AntennaGeometryEditor):
|
|
raise TypeError("Geometry is not a scalar setting")
|
|
return widget
|
|
|
|
|
|
def _read_widget(widget: SettingWidget) -> Scalar | list[dict]:
|
|
return widget.records() if isinstance(widget, AntennaGeometryEditor) else read_setting(widget)
|
|
|
|
|
|
def _set_web_widget(widget: SettingWidget, value: object) -> None:
|
|
"""Validate untrusted web input before Qt can coerce or clamp it."""
|
|
if isinstance(widget, AntennaGeometryEditor):
|
|
widget.set_records(value)
|
|
return
|
|
if isinstance(widget, QCheckBox):
|
|
if not isinstance(value, bool):
|
|
raise ValueError("Checkbox value must be true or false")
|
|
elif isinstance(widget, (QSpinBox, QDoubleSpinBox)):
|
|
if isinstance(value, bool) or not isinstance(value, (int, float, str)):
|
|
raise ValueError("Numeric setting requires a number")
|
|
numeric = float(value)
|
|
if not math.isfinite(numeric) or not widget.minimum() <= numeric <= widget.maximum():
|
|
raise ValueError("Numeric setting is outside its allowed range")
|
|
if isinstance(widget, QSpinBox) and numeric != int(numeric):
|
|
raise ValueError("Setting requires an integer")
|
|
value = int(numeric) if isinstance(widget, QSpinBox) else numeric
|
|
elif isinstance(widget, QComboBox):
|
|
if not isinstance(value, str) or widget.findText(value) < 0:
|
|
raise ValueError(f"Unknown setting option: {value}")
|
|
elif isinstance(value, list):
|
|
if any(isinstance(item, bool) or not isinstance(item, int) for item in value):
|
|
raise ValueError("Position selection requires integer values")
|
|
value = ",".join(str(item) for item in value)
|
|
if not isinstance(value, (str, int, float, bool)):
|
|
raise ValueError("Setting requires a scalar value")
|
|
write_setting(widget, value)
|
|
|
|
|
|
class ConfigController:
|
|
"""Read and apply editor values; callers own persistence and runtime effects."""
|
|
|
|
def __init__(self, fields: Mapping[str, SettingWidget], single_combo_selected: Callable[[], bool]) -> None:
|
|
self.fields = fields
|
|
self._single_combo_selected = single_combo_selected
|
|
|
|
def _scalar(self, name: str) -> Scalar:
|
|
value = read_setting(_scalar_widget(self.fields[name]))
|
|
return value.strip() if isinstance(value, str) else value
|
|
|
|
def dataset_destination(self, config: RunConfigModel) -> tuple[Path, str, str]:
|
|
radar = config.radar
|
|
prefix = radar_config_filename_prefix(radar.model, radar.sweep.start_hz, radar.sweep.stop_hz,
|
|
radar.sweep.points, radar.sweep.power_dbm)
|
|
return Path(str(self._scalar("_save_path_input"))).expanduser(), str(self._scalar("_save_name_input")), prefix
|
|
|
|
def _read_radar(self, config: RunConfigModel) -> None:
|
|
for path, widget_id in (*_SWEEP_FIELDS, *(_ADC_FIELDS if config.is_kamil_adc else ())):
|
|
owner, name = _model_field(config, path)
|
|
original = getattr(owner, name)
|
|
value = self._scalar(widget_id)
|
|
if isinstance(original, bool):
|
|
if not isinstance(value, bool):
|
|
raise ValueError(f"{path} must be a boolean")
|
|
parsed = value
|
|
elif isinstance(original, int):
|
|
parsed = int(str(value))
|
|
elif isinstance(original, float):
|
|
parsed = float(value)
|
|
elif isinstance(original, list):
|
|
parsed = [line.strip() for line in str(value).splitlines() if line.strip()]
|
|
elif isinstance(original, dict):
|
|
parsed = json.loads(str(value) or "{}")
|
|
if not isinstance(parsed, dict) or any(
|
|
not isinstance(key, str) or not isinstance(item, str) for key, item in parsed.items()
|
|
):
|
|
raise ValueError("Environment JSON must contain only string keys and values")
|
|
else:
|
|
parsed = str(value)
|
|
setattr(owner, name, parsed)
|
|
|
|
def build_gui_state(self, defaults: RunConfigModel, preprocess: GuiPreprocessDialogStateModel) -> GuiStateModel:
|
|
gui = GuiStateModel(preprocess_dialog=deepcopy(preprocess))
|
|
for binding in GUI_FIELDS:
|
|
if binding.spec.widget:
|
|
binding.write(gui, self._scalar(binding.spec.widget))
|
|
gui.switches.combo_mode = "single" if not defaults.is_multi_device and self._single_combo_selected() else "text"
|
|
if defaults.is_multi_device:
|
|
gui.switches.combos_text = ",".join(f"{combo.input}:{combo.output}" for combo in defaults.build_runtime_combos())
|
|
return gui
|
|
|
|
def build_config(
|
|
self, defaults: RunConfigModel, selected_sets: Mapping[str, str], live_path: Path, log_level: str,
|
|
) -> RunConfigModel:
|
|
config = defaults.clone()
|
|
self._read_radar(config)
|
|
config.runtime.settling_ms = int(str(self._scalar("_settling_ms")))
|
|
config.runtime.processing_live_config_path = str(live_path)
|
|
config.logging.level = log_level.strip().lower()
|
|
if config.is_multi_device:
|
|
if len(config.radar.multi_device.slave_serials) != 2:
|
|
raise ValueError("LibreVNA multi-device mode requires exactly two slave serials")
|
|
config.apply_device_model_constraints()
|
|
else:
|
|
config.combos = (
|
|
[ComboModel(int(str(self._scalar("_single_combo_input"))), int(str(self._scalar("_single_combo_output"))))]
|
|
if self._single_combo_selected() else parse_combos_from_text(str(self._scalar("_combos_text")))
|
|
)
|
|
config.ensure_combos()
|
|
static = (
|
|
config.input_switch.positions <= 1 and config.output_switch.positions <= 1
|
|
or config.input_switch.driver_mode == config.output_switch.driver_mode == "mock"
|
|
)
|
|
if self._scalar("_processing_mode") not in {"gpr", "legacy_gpr", "sar3d"} and static:
|
|
config.combos = [ComboModel(0, 0)]
|
|
for key in PREPROCESS_ASSET_NAMES:
|
|
asset = preprocess_asset_model(config, key)
|
|
asset.bundle_path = ""
|
|
asset.set_name = selected_sets.get(key, "")
|
|
config.gpr = self.geometry()
|
|
validate_gpr_model(config.gpr, input_switch_positions=config.input_switch.positions,
|
|
output_switch_positions=config.output_switch.positions, sweep=config.radar.sweep)
|
|
return config
|
|
|
|
def radar_key(self, defaults: RunConfigModel) -> str:
|
|
config = defaults.clone()
|
|
self._read_radar(config)
|
|
return config.radar_key()
|
|
|
|
def _geometry(self, name: str) -> AntennaGeometryEditor:
|
|
widget = self.fields[_GEOMETRY_FIELDS[name]]
|
|
if not isinstance(widget, AntennaGeometryEditor):
|
|
raise TypeError(f"{name} requires an antenna geometry editor")
|
|
return widget
|
|
|
|
def processing_settings(self) -> GuiProcessingStateModel:
|
|
gui = GuiStateModel()
|
|
gui.processing.selected_mode = str(self._scalar("_processing_mode"))
|
|
for binding in GUI_FIELDS:
|
|
if binding.mode in {gui.processing.selected_mode, "pass_through"} and binding.spec.widget:
|
|
binding.write(gui, self._scalar(binding.spec.widget))
|
|
return gui.processing
|
|
|
|
def geometry(self) -> GprModel:
|
|
return GprModel(
|
|
relative_permittivity=float(self._scalar("_gpr_relative_permittivity")),
|
|
tx_geometry=[GprTxGeometryModel(**entry) for entry in self._geometry("tx_geometry").records()],
|
|
rx_geometry=[GprRxGeometryModel(**entry) for entry in self._geometry("rx_geometry").records()],
|
|
)
|
|
|
|
def build_live_config(
|
|
self, *, reprocess_current_result: bool = True, history_clear_before_ns: int = 0,
|
|
history_removed: Sequence[dict[str, int]] = (), settings_revision: int = 0,
|
|
) -> ProcessingLiveConfig:
|
|
config = ProcessingLiveConfig(
|
|
reprocess_current_result=reprocess_current_result, history_clear_before_ns=history_clear_before_ns,
|
|
history_removed=deepcopy(list(history_removed)), settings_revision=settings_revision,
|
|
)
|
|
for binding in active_live_fields(str(self._scalar("_processing_mode"))):
|
|
value = self._scalar(binding.spec.widget)
|
|
if binding.spec.live in {"gpr_input_positions", "gpr_output_positions"}:
|
|
value = parse_position_csv(str(value))
|
|
setattr(config, binding.spec.live, value)
|
|
config.relative_permittivity = float(self._scalar("_gpr_relative_permittivity"))
|
|
config.tx_geometry = self._geometry("tx_geometry").records()
|
|
config.rx_geometry = self._geometry("rx_geometry").records()
|
|
for axis in ("x", "z"):
|
|
lower_name, upper_name = f"gpr_visible_{axis}_min_m", f"gpr_visible_{axis}_max_m"
|
|
lower, upper = sorted((getattr(config, lower_name), getattr(config, upper_name)))
|
|
if upper - lower < .1:
|
|
center = .5 * (lower + upper)
|
|
lower, upper = center - .05, center + .05
|
|
setattr(config, lower_name, lower)
|
|
setattr(config, upper_name, upper)
|
|
return config
|
|
|
|
def apply_profile_widgets(self, config: RunConfigModel, gui: GuiStateModel) -> None:
|
|
with ExitStack() as blockers:
|
|
for widget in self.fields.values():
|
|
blockers.enter_context(QSignalBlocker(widget))
|
|
for path, widget_id in (*_SWEEP_FIELDS, *_ADC_FIELDS, ("runtime.settling_ms", "_settling_ms")):
|
|
owner, name = _model_field(config, path)
|
|
value = getattr(owner, name)
|
|
if isinstance(value, dict):
|
|
value = json.dumps(value, indent=2, sort_keys=True)
|
|
elif isinstance(value, list):
|
|
value = "\n".join(value)
|
|
elif isinstance(value, float):
|
|
value = f"{value:g}"
|
|
write_setting(_scalar_widget(self.fields[widget_id]), value)
|
|
# Set frequency ranges before values so a previously narrow profile
|
|
# cannot silently clamp the newly loaded processing band.
|
|
for binding in GUI_FIELDS:
|
|
if binding.path[-1] in {"start_freq_mhz", "stop_freq_mhz"}:
|
|
widget = self.fields[binding.spec.widget]
|
|
if isinstance(widget, QDoubleSpinBox):
|
|
widget.setRange(config.radar.sweep.start_hz / 1e6, config.radar.sweep.stop_hz / 1e6)
|
|
if binding.spec.widget:
|
|
write_setting(_scalar_widget(self.fields[binding.spec.widget]), binding.read(gui))
|
|
write_setting(_scalar_widget(self.fields["_log_level_combo"]), config.logging.level.capitalize())
|
|
write_setting(_scalar_widget(self.fields["_gpr_relative_permittivity"]), config.gpr.relative_permittivity)
|
|
for name, count, records in (
|
|
("tx_geometry", config.output_switch.positions, config.gpr.tx_geometry),
|
|
("rx_geometry", config.input_switch.positions, config.gpr.rx_geometry),
|
|
):
|
|
editor = self._geometry(name)
|
|
editor.set_position_count(count)
|
|
editor.set_records(records)
|
|
|
|
def _web_widgets(self, mode: str) -> dict[str, tuple[str, SettingWidget]]:
|
|
widgets = {binding.spec.web: (binding.spec.group, self.fields[binding.spec.widget])
|
|
for binding in active_web_fields(mode)}
|
|
for name, widget_id in _GEOMETRY_FIELDS.items():
|
|
if mode in ({"legacy_gpr"} if name == "relative_permittivity" else {"gpr", "legacy_gpr", "sar3d"}):
|
|
widgets[name] = ("Geometry & medium", self.fields[widget_id])
|
|
return widgets
|
|
|
|
def web_schema(self) -> list[dict[str, object]]:
|
|
schema = []
|
|
for name, (group, widget) in self._web_widgets(str(self._scalar("_processing_mode"))).items():
|
|
if isinstance(widget, QCheckBox):
|
|
label = widget.text()
|
|
else:
|
|
parent = widget.parentWidget()
|
|
assert parent is not None
|
|
form = parent.layout()
|
|
assert isinstance(form, QFormLayout)
|
|
caption = form.labelForField(widget)
|
|
assert isinstance(caption, QLabel)
|
|
label = caption.text()
|
|
entry: dict[str, object] = {"name": name, "label": label, "group": group,
|
|
"enabled": widget.isEnabled(), "value": _read_widget(widget)}
|
|
if isinstance(widget, QComboBox):
|
|
entry.update(kind="select", options=[widget.itemText(i) for i in range(widget.count())])
|
|
elif isinstance(widget, QCheckBox):
|
|
entry.update(kind="bool")
|
|
elif isinstance(widget, (QSpinBox, QDoubleSpinBox)):
|
|
entry.update(kind="int" if isinstance(widget, QSpinBox) else "float",
|
|
min=widget.minimum(), max=widget.maximum(), step=widget.singleStep())
|
|
if isinstance(widget, QDoubleSpinBox):
|
|
entry["decimals"] = widget.decimals()
|
|
elif isinstance(widget, AntennaGeometryEditor):
|
|
entry.update(kind="geometry", position_field=widget.position_field, position_count=widget.position_count)
|
|
else:
|
|
entry.update(kind="text")
|
|
schema.append(entry)
|
|
return schema
|
|
|
|
def restore_live_config(self, config: Mapping[str, object]) -> None:
|
|
"""Restore desktop edits from the last successfully published revision."""
|
|
with ExitStack() as blockers:
|
|
for widget in self.fields.values():
|
|
blockers.enter_context(QSignalBlocker(widget))
|
|
for binding in active_live_fields(str(config["processor_mode"])):
|
|
value = config[binding.spec.live]
|
|
if isinstance(value, list):
|
|
value = ",".join(map(str, value))
|
|
if not isinstance(value, (str, int, float, bool)):
|
|
raise TypeError(f"Invalid published setting: {binding.spec.live}")
|
|
write_setting(_scalar_widget(self.fields[binding.spec.widget]), value)
|
|
for name, widget_id in _GEOMETRY_FIELDS.items():
|
|
_set_web_widget(self.fields[widget_id], config[name])
|
|
|
|
def apply_web_fields(self, fields: Mapping[str, object], commit: Callable[[], None]) -> None:
|
|
"""Apply one batch without signals; restore editors if validation or publication fails."""
|
|
with ExitStack() as blockers:
|
|
scalar_snapshots = []
|
|
geometry_snapshots = []
|
|
combo_snapshots = []
|
|
ranges = []
|
|
for widget in self.fields.values():
|
|
blockers.enter_context(QSignalBlocker(widget))
|
|
if isinstance(widget, QComboBox):
|
|
combo_snapshots.append((widget, widget.currentIndex()))
|
|
elif isinstance(widget, AntennaGeometryEditor):
|
|
geometry_snapshots.append((widget, widget.records()))
|
|
else:
|
|
scalar_snapshots.append((widget, read_setting(widget)))
|
|
if isinstance(widget, (QSpinBox, QDoubleSpinBox)):
|
|
ranges.append((widget, widget.minimum(), widget.maximum()))
|
|
try:
|
|
if "processor_mode" in fields:
|
|
_set_web_widget(self.fields["_processing_mode"], fields["processor_mode"])
|
|
bindings = self._web_widgets(str(self._scalar("_processing_mode")))
|
|
for name, value in fields.items():
|
|
if name not in bindings:
|
|
raise ValueError(f"Unknown setting for current mode: {name}")
|
|
_set_web_widget(bindings[name][1], value)
|
|
commit()
|
|
except Exception:
|
|
for widget, lower, upper in ranges:
|
|
widget.setRange(lower, upper)
|
|
for widget, index in combo_snapshots:
|
|
widget.setCurrentIndex(index)
|
|
for widget, records in geometry_snapshots:
|
|
widget.set_records(records)
|
|
for widget, value in scalar_snapshots:
|
|
write_setting(widget, value)
|
|
raise
|
|
|
|
def mode_summary(self, mode: str) -> str:
|
|
return ", ".join(f"{binding.spec.web}={self._scalar(binding.spec.widget)}"
|
|
for binding in active_web_fields(mode) if binding.spec.web != "processor_mode")
|