"""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.rph_radar.project_dir", "_adc_project_dir_input"), ("radar.rph_radar.executable_path", "_adc_executable_path_input"), ("radar.rph_radar.tty_path", "_adc_tty_path_input"), ("radar.rph_radar.args", "_adc_args_input"), ("radar.rph_radar.env", "_adc_env_input"), ("radar.rph_radar.startup_timeout_s", "_adc_startup_timeout_s_input"), ("radar.rph_radar.sweep_timeout_s", "_adc_sweep_timeout_s_input"), ("radar.rph_radar.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.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_rph_radar 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")