"""Configuration and live-processing binding mixin for the main window.""" from __future__ import annotations from collections import deque from contextlib import ExitStack import json from pathlib import Path from PyQt6.QtCore import QSignalBlocker from PyQt6.QtWidgets import QFileDialog from python_app.hardware_full.librevna_service import LibreVnaService from python_app.models.gui_profile_model import ( GuiBscanStateModel, GuiDataActionsStateModel, GuiGprStateModel, GuiPassThroughStateModel, GuiPreprocessDialogStateModel, GuiProcessingStateModel, GuiProfileModel, GuiStateModel, GuiSwitchStateModel, ) from python_app.models.run_config_model import ComboModel, GprRxGeometryModel, GprTxGeometryModel, RunConfigModel from python_app.models.run_config_validation import validate_gpr_model from python_app.orchestration.config_writer import parse_combos_from_text from python_app.orchestration.live_processing_config import ProcessingLiveConfig from python_app.orchestration.preprocess_assets import ( PREPROCESS_ASSET_KEYS, VISIBLE_PREPROCESS_ASSET_KEYS, preprocess_asset_model, ) from python_app.storage.npz_store import radar_key_from_config class AppWindowConfigMixin: """Builds runtime config models from current UI state.""" @staticmethod def _parse_csv_int_list(text: str) -> list[int]: """Parse comma-separated integer selection list.""" cleaned = text.strip() if not cleaned: return [] values: list[int] = [] for part in cleaned.split(","): token = part.strip() if not token: continue values.append(int(token)) return values @staticmethod def _parse_gpr_tx_geometry_text(text: str) -> list[GprTxGeometryModel]: """Parse line-based Tx geometry editor text.""" entries: list[GprTxGeometryModel] = [] for line_number, raw_line in enumerate(text.splitlines(), start=1): line = raw_line.strip() if not line: continue parts = line.split() if len(parts) != 2: raise ValueError(f"Invalid Tx geometry line {line_number}: expected `output_pos x_m`") entries.append( GprTxGeometryModel( output_pos=int(parts[0]), x_m=float(parts[1]), ) ) return entries @staticmethod def _parse_gpr_rx_geometry_text(text: str) -> list[GprRxGeometryModel]: """Parse line-based Rx geometry editor text.""" entries: list[GprRxGeometryModel] = [] for line_number, raw_line in enumerate(text.splitlines(), start=1): line = raw_line.strip() if not line: continue parts = line.split() if len(parts) != 2: raise ValueError(f"Invalid Rx geometry line {line_number}: expected `input_pos x_m`") entries.append( GprRxGeometryModel( input_pos=int(parts[0]), x_m=float(parts[1]), ) ) return entries @staticmethod def _format_combos_text_from_config(config: RunConfigModel) -> str: """Render configured combos for UI text editor, keeping full matrix as empty.""" combos = list(config.combos) full_combos = config.build_full_combos(config.input_switch.positions, config.output_switch.positions) if len(combos) == len(full_combos) and all( int(left.input) == int(right.input) and int(left.output) == int(right.output) for left, right in zip(combos, full_combos, strict=True) ): return "" return ",".join(f"{int(combo.input)}:{int(combo.output)}" for combo in combos) @staticmethod def _default_gpr_input_positions_from_config(config: RunConfigModel) -> str: """Build default live GPR input-position selection from stable config.""" geometry_values = {int(entry.input_pos) for entry in config.gpr.rx_geometry} combo_values = {int(combo.input) for combo in config.combos} values = sorted(geometry_values & combo_values) or sorted(geometry_values) return ",".join(str(value) for value in values) @staticmethod def _default_gpr_output_positions_from_config(config: RunConfigModel) -> str: """Build default live GPR output-position selection from stable config.""" geometry_values = {int(entry.output_pos) for entry in config.gpr.tx_geometry} combo_values = {int(combo.output) for combo in config.combos} values = sorted(geometry_values & combo_values) or sorted(geometry_values) return ",".join(str(value) for value in values) @staticmethod def _history_limit_for_config(config: RunConfigModel) -> int: """Return unified GUI history limit derived from config ring capacities.""" return max( 1, min( int(config.rings.raw_tap.capacity), int(config.rings.preprocessed_tap.capacity), int(config.rings.results.capacity), ), ) def _default_gui_state_for_config(self, config: RunConfigModel) -> GuiStateModel: """Build fallback GUI-only defaults for a stable run config.""" default_combo = config.combos[0] if config.combos else ComboModel(input=0, output=0) default_mode = "single" if len(config.combos) == 1 else "text" return GuiStateModel( switches=GuiSwitchStateModel( combo_mode=default_mode, combos_text=self._format_combos_text_from_config(config), single_input=str(int(default_combo.input)), single_output=str(int(default_combo.output)), ), processing=GuiProcessingStateModel( selected_mode="pass_through", pass_through=GuiPassThroughStateModel( show_magnitude=True, show_phase=True, fixed_y_enabled=False, y_min_db=-100.0, y_max_db=0.0, ), bscan=GuiBscanStateModel( axis="abs", cut_m=0.824, max_depth_m=1.0, gain=1.0, start_freq_mhz=100.0, stop_freq_mhz=8800.0, ), gpr=GuiGprStateModel( input_positions=self._default_gpr_input_positions_from_config(config), output_positions=self._default_gpr_output_positions_from_config(config), min_depth_m=2.0, max_depth_m=14.0, comp_power=0.2, start_freq_mhz=3000.0, stop_freq_mhz=6000.0, background_subtract_enabled=True, background_mean_count=10, ), ), data_actions=GuiDataActionsStateModel( save_count=10, save_path=str(self._project_root / "python_app/data/snapshots"), save_name="snapshot_manual", ), preprocess_dialog=GuiPreprocessDialogStateModel(set_name="set_001"), ) def _current_preprocess_set_name(self) -> str: """Return current preprocess dialog set name, even when dialog is still closed.""" if self._preprocess_dialog is not None: self._preprocess_set_name = self._preprocess_dialog.set_name() return self._preprocess_set_name def _build_gui_state(self) -> GuiStateModel: """Build GUI-only persistent state from current widget values.""" return GuiStateModel( switches=GuiSwitchStateModel( combo_mode="single" if self._single_combo_select_button.isChecked() else "text", combos_text=self._combos_text.text().strip(), single_input=self._single_combo_input.text().strip(), single_output=self._single_combo_output.text().strip(), ), processing=GuiProcessingStateModel( selected_mode=self._processing_mode.currentText(), pass_through=GuiPassThroughStateModel( show_magnitude=bool(self._show_magnitude_checkbox.isChecked()), show_phase=bool(self._show_phase_checkbox.isChecked()), fixed_y_enabled=bool(self._pass_through_fixed_y_enabled.isChecked()), y_min_db=float(self._pass_through_y_min_db.value()), y_max_db=float(self._pass_through_y_max_db.value()), ), bscan=GuiBscanStateModel( axis=self._bscan_axis.currentText(), cut_m=float(self._bscan_cut_m.value()), max_depth_m=float(self._bscan_max_depth_m.value()), gain=float(self._bscan_gain.value()), start_freq_mhz=float(self._bscan_start_freq_mhz.value()), stop_freq_mhz=float(self._bscan_stop_freq_mhz.value()), ), gpr=GuiGprStateModel( input_positions=self._gpr_input_positions_input.text().strip(), output_positions=self._gpr_output_positions_input.text().strip(), min_depth_m=float(self._gpr_min_depth_m.value()), max_depth_m=float(self._gpr_max_depth_m.value()), comp_power=float(self._gpr_comp_power.value()), start_freq_mhz=float(self._gpr_start_freq_mhz.value()), stop_freq_mhz=float(self._gpr_stop_freq_mhz.value()), background_subtract_enabled=bool(self._gpr_background_subtract_enabled.isChecked()), background_mean_count=int(self._gpr_background_mean_count.value()), ), ), data_actions=GuiDataActionsStateModel( save_count=int(self._save_count.value()), save_path=self._save_path_input.text().strip(), save_name=self._save_name_input.text().strip(), ), preprocess_dialog=GuiPreprocessDialogStateModel( set_name=self._current_preprocess_set_name(), ), ) def _build_gui_profile(self) -> GuiProfileModel: """Build full GUI config profile from current window state.""" return GuiProfileModel( run_config=self._build_config(), gui=self._build_gui_state(), ) def _write_gui_profile_to_path(self, output_path: Path, *, allow_overwrite: bool = True) -> GuiProfileModel: """Serialize current full GUI profile to `output_path` and return the persisted model.""" if not allow_overwrite and output_path.exists(): raise FileExistsError(f"Config profile output already exists: {output_path}") profile = self._build_gui_profile() output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(profile.to_dict(), indent=2), encoding="utf-8") return profile def _set_combo_selection_mode(self, mode: str) -> None: """Highlight current combo mode and enable only the relevant editors.""" text_selected = mode != "single" self._run_combos_select_button.setChecked(text_selected) self._single_combo_select_button.setChecked(not text_selected) self._combos_text.setEnabled(text_selected) self._single_combo_output.setEnabled(not text_selected) self._single_combo_input.setEnabled(not text_selected) def _sync_pass_through_y_controls(self) -> None: """Enable Y-range editors only when fixed Y mode is active.""" enabled = bool(self._pass_through_fixed_y_enabled.isChecked()) self._pass_through_y_min_db.setEnabled(enabled) self._pass_through_y_max_db.setEnabled(enabled) def _apply_history_limit_from_config(self, config: RunConfigModel) -> None: """Resize in-memory history buffers to match the loaded config.""" history_limit = self._history_limit_for_config(config) self._raw_history = deque(self._raw_history, maxlen=history_limit) self._pre_history = deque(self._pre_history, maxlen=history_limit) self._result_history = deque(self._result_history, maxlen=history_limit) self._bscan_history_limit = history_limit self._clear_bscan_plot_history() def _save_current_config(self) -> None: """Persist current full GUI profile to a user-selected JSON file.""" try: suggested_path = str(self._active_profile_path) selected_path, _selected_filter = QFileDialog.getSaveFileName( self, "Save Config Profile", suggested_path, "JSON Files (*.json);;All Files (*)", ) if not selected_path: return output_path = self._normalize_profile_path(Path(selected_path)) if not output_path.suffix: output_path = output_path.with_suffix(".json") profile = self._write_gui_profile_to_path(output_path) self._defaults_config = profile.run_config.clone() self._gui_defaults = profile.gui self._remember_active_profile_path(output_path) self._log( f"Config profile saved: path={output_path}, " f"combos={len(profile.run_config.combos)}, " f"sweep={profile.run_config.radar.sweep.start_hz:g}.." f"{profile.run_config.radar.sweep.stop_hz:g} Hz, " f"points={profile.run_config.radar.sweep.points}, " f"ifbw={profile.run_config.radar.sweep.if_bandwidth_hz:g} Hz, " f"power={profile.run_config.radar.sweep.power_dbm:g} dBm, " f"processing_mode={profile.gui.processing.selected_mode}" ) except Exception as exc: # noqa: BLE001 self._show_exception("Failed to save config profile", exc) def _load_config_from_dialog(self) -> None: """Load full GUI profile from a user-selected JSON file.""" if self._capture_session is not None: self._show_error( "Cannot load config during active capture sequence", details=self._capture_state_details(), ) return if self._supervisor.is_running() or self._supervisor.is_processor_running(): self._show_error( "Stop all pipeline processes before loading a config profile", details=self._process_state_details(), ) return selected_path, _selected_filter = QFileDialog.getOpenFileName( self, "Load Config Profile", str(self._active_profile_path), "JSON Files (*.json);;All Files (*)", ) if not selected_path: return try: self._load_config_profile(Path(selected_path)) except Exception as exc: # noqa: BLE001 self._show_exception("Failed to load config profile", exc) def _load_config_profile(self, profile_path: Path) -> None: """Load config profile from `profile_path` and atomically apply it to the UI.""" normalized_path = self._normalize_profile_path(profile_path) profile = GuiProfileModel.load_from_path(normalized_path) self._apply_loaded_profile(profile, normalized_path) profile_kind = "legacy run config" if profile.gui is None else "full GUI profile" self._log( f"Config profile loaded: path={normalized_path}, " f"kind={profile_kind}, " f"combos={len(self._defaults_config.combos)}, " f"processing_mode={self._processing_mode.currentText()}" ) def _apply_loaded_profile(self, profile: GuiProfileModel, profile_path: Path) -> None: """Apply already parsed profile to GUI state without restarting the pipeline.""" config = profile.run_config.clone() gui_state = profile.gui if profile.gui is not None else self._default_gui_state_for_config(config) selected_preprocess_sets = { key: str(preprocess_asset_model(config, key).set_name) for key in VISIBLE_PREPROCESS_ASSET_KEYS } radio_widgets = ( self._serial_input, self._radar_mode, self._start_hz_input, self._stop_hz_input, self._points_input, self._ifbw_input, self._power_input, self._settling_ms, self._combos_text, self._single_combo_output, self._single_combo_input, self._run_combos_select_button, self._single_combo_select_button, self._processing_mode, self._show_magnitude_checkbox, self._show_phase_checkbox, self._pass_through_fixed_y_enabled, self._pass_through_y_min_db, self._pass_through_y_max_db, self._bscan_axis, self._bscan_cut_m, self._bscan_max_depth_m, self._bscan_gain, self._bscan_start_freq_mhz, self._bscan_stop_freq_mhz, self._gpr_config_mode, self._gpr_relative_permittivity, self._gpr_tx_geometry_input, self._gpr_rx_geometry_input, self._gpr_input_positions_input, self._gpr_output_positions_input, self._gpr_min_depth_m, self._gpr_max_depth_m, self._gpr_comp_power, self._gpr_start_freq_mhz, self._gpr_stop_freq_mhz, self._gpr_background_subtract_enabled, self._gpr_background_mean_count, self._save_count, self._save_path_input, self._save_name_input, ) with ExitStack() as blockers: for widget in radio_widgets: blockers.enter_context(QSignalBlocker(widget)) self._serial_input.setText(str(config.radar.serial)) self._set_combo_current_text(self._radar_mode, str(config.radar.driver_mode)) self._start_hz_input.setText(f"{config.radar.sweep.start_hz:g}") self._stop_hz_input.setText(f"{config.radar.sweep.stop_hz:g}") self._points_input.setText(str(int(config.radar.sweep.points))) self._ifbw_input.setText(f"{config.radar.sweep.if_bandwidth_hz:g}") self._power_input.setText(f"{config.radar.sweep.power_dbm:g}") self._settling_ms.setText(str(int(config.runtime.settling_ms))) self._combos_text.setText(str(gui_state.switches.combos_text)) self._single_combo_output.setText(str(gui_state.switches.single_output)) self._single_combo_input.setText(str(gui_state.switches.single_input)) self._set_combo_selection_mode(gui_state.switches.combo_mode) self._set_combo_current_text(self._processing_mode, gui_state.processing.selected_mode) self._show_magnitude_checkbox.setChecked(bool(gui_state.processing.pass_through.show_magnitude)) self._show_phase_checkbox.setChecked(bool(gui_state.processing.pass_through.show_phase)) self._pass_through_fixed_y_enabled.setChecked(bool(gui_state.processing.pass_through.fixed_y_enabled)) self._pass_through_y_min_db.setValue(float(gui_state.processing.pass_through.y_min_db)) self._pass_through_y_max_db.setValue(float(gui_state.processing.pass_through.y_max_db)) self._set_combo_current_text(self._bscan_axis, gui_state.processing.bscan.axis) self._bscan_cut_m.setValue(float(gui_state.processing.bscan.cut_m)) self._bscan_max_depth_m.setValue(float(gui_state.processing.bscan.max_depth_m)) self._bscan_gain.setValue(float(gui_state.processing.bscan.gain)) self._bscan_start_freq_mhz.setValue(float(gui_state.processing.bscan.start_freq_mhz)) self._bscan_stop_freq_mhz.setValue(float(gui_state.processing.bscan.stop_freq_mhz)) self._set_combo_current_text(self._gpr_config_mode, str(config.gpr.mode)) self._gpr_relative_permittivity.setValue(float(config.gpr.relative_permittivity)) self._gpr_tx_geometry_input.setPlainText( "\n".join( f"{int(entry.output_pos)} {float(entry.x_m):g}" for entry in config.gpr.tx_geometry ) ) self._gpr_rx_geometry_input.setPlainText( "\n".join( f"{int(entry.input_pos)} {float(entry.x_m):g}" for entry in config.gpr.rx_geometry ) ) self._gpr_input_positions_input.setText(str(gui_state.processing.gpr.input_positions)) self._gpr_output_positions_input.setText(str(gui_state.processing.gpr.output_positions)) self._gpr_min_depth_m.setValue(float(gui_state.processing.gpr.min_depth_m)) self._gpr_max_depth_m.setValue(float(gui_state.processing.gpr.max_depth_m)) self._gpr_comp_power.setValue(float(gui_state.processing.gpr.comp_power)) self._gpr_start_freq_mhz.setValue(float(gui_state.processing.gpr.start_freq_mhz)) self._gpr_stop_freq_mhz.setValue(float(gui_state.processing.gpr.stop_freq_mhz)) self._gpr_background_subtract_enabled.setChecked( bool(gui_state.processing.gpr.background_subtract_enabled) ) self._gpr_background_mean_count.setValue(int(gui_state.processing.gpr.background_mean_count)) self._save_count.setValue(int(gui_state.data_actions.save_count)) self._save_path_input.setText(str(gui_state.data_actions.save_path)) self._save_name_input.setText(str(gui_state.data_actions.save_name)) self._defaults_config = config self._gui_defaults = gui_state self._selected_preprocess_sets = selected_preprocess_sets self._preprocess_set_name = str(gui_state.preprocess_dialog.set_name) self._apply_history_limit_from_config(config) self._gpr_geometry_signature = None self._gpr_selected_geometry = None self._sync_pass_through_y_controls() self._refresh_preprocess_summary_labels() if self._preprocess_dialog is not None: with ExitStack() as dialog_blockers: dialog_blockers.enter_context(QSignalBlocker(self._preprocess_dialog._set_name_input)) for combo in self._preprocess_dialog._set_combos.values(): dialog_blockers.enter_context(QSignalBlocker(combo)) self._preprocess_dialog.set_set_name(self._preprocess_set_name) self._preprocess_dialog.set_selected_sets(self._selected_preprocess_sets, emit_signal=False) self._apply_initial_radar_limits() self._on_processing_mode_changed(gui_state.processing.selected_mode) self._update_history_indicator() self._remember_active_profile_path(profile_path) def _build_config(self) -> RunConfigModel: """Build `RunConfigModel` from current GUI widget values.""" config = self._defaults_config.clone() config.radar.sweep.start_hz = float(self._start_hz_input.text().strip()) config.radar.sweep.stop_hz = float(self._stop_hz_input.text().strip()) config.radar.sweep.points = int(self._points_input.text().strip()) config.radar.sweep.if_bandwidth_hz = float(self._ifbw_input.text().strip()) config.radar.sweep.power_dbm = float(self._power_input.text().strip()) config.runtime.settling_ms = int(self._settling_ms.text().strip()) config.runtime.processing_live_config_path = str(self._live_config_writer.path) if self._single_combo_select_button.isChecked(): config.combos = [ ComboModel( input=int(self._single_combo_input.text().strip()), output=int(self._single_combo_output.text().strip()), ) ] else: combo_text = self._combos_text.text() config.combos = parse_combos_from_text(combo_text) config.ensure_combos() if self._switches_are_effectively_static(config): config.combos = [ComboModel(input=0, output=0)] for key in PREPROCESS_ASSET_KEYS: preprocess_asset_model(config, key).bundle_path = "" for key in VISIBLE_PREPROCESS_ASSET_KEYS: preprocess_asset_model(config, key).set_name = self._selected_preprocess_sets.get(key, "") config.gpr.mode = self._gpr_config_mode.currentText() config.gpr.relative_permittivity = float(self._gpr_relative_permittivity.value()) config.gpr.tx_geometry = self._parse_gpr_tx_geometry_text(self._gpr_tx_geometry_input.toPlainText()) config.gpr.rx_geometry = self._parse_gpr_rx_geometry_text(self._gpr_rx_geometry_input.toPlainText()) validate_gpr_model( config.gpr, input_switch_positions=config.input_switch.positions, output_switch_positions=config.output_switch.positions, ) return config def _radar_key(self, config: RunConfigModel) -> str: """Build radar key used by preprocess-set storage lookup.""" return radar_key_from_config( model_name=config.radar.model, serial=config.radar.serial, sweep_start_hz=config.radar.sweep.start_hz, sweep_stop_hz=config.radar.sweep.stop_hz, sweep_points=config.radar.sweep.points, ifbw_hz=config.radar.sweep.if_bandwidth_hz, power_dbm=config.radar.sweep.power_dbm, ) def _live_processing_config(self, *, history_command: str = "none") -> ProcessingLiveConfig: """Build live processing config from current processing widgets.""" self._sync_bscan_frequency_limits_with_radar() self._sync_gpr_frequency_limits_with_radar() y_min_db = float(self._pass_through_y_min_db.value()) y_max_db = float(self._pass_through_y_max_db.value()) return ProcessingLiveConfig( processor_mode=self._processing_mode.currentText(), pass_through_channel="s21", pass_through_fixed_y_enabled=bool(self._pass_through_fixed_y_enabled.isChecked()), pass_through_y_min_db=min(y_min_db, y_max_db), pass_through_y_max_db=max(y_min_db, y_max_db), bscan_axis=self._bscan_axis.currentText(), bscan_channel="s21", bscan_cut_m=float(self._bscan_cut_m.value()), bscan_max_depth_m=float(self._bscan_max_depth_m.value()), bscan_gain=float(self._bscan_gain.value()), bscan_start_freq_mhz=float(self._bscan_start_freq_mhz.value()), bscan_stop_freq_mhz=float(self._bscan_stop_freq_mhz.value()), gpr_input_positions=self._parse_csv_int_list(self._gpr_input_positions_input.text()), gpr_output_positions=self._parse_csv_int_list(self._gpr_output_positions_input.text()), gpr_min_depth_m=float(self._gpr_min_depth_m.value()), gpr_max_depth_m=float(self._gpr_max_depth_m.value()), gpr_comp_power=float(self._gpr_comp_power.value()), gpr_start_freq_mhz=float(self._gpr_start_freq_mhz.value()), gpr_stop_freq_mhz=float(self._gpr_stop_freq_mhz.value()), gpr_background_subtract_enabled=bool(self._gpr_background_subtract_enabled.isChecked()), gpr_background_mean_count=int(self._gpr_background_mean_count.value()), history_command_seq=int(self._history_command_seq), history_command=str(history_command), ) def _write_live_processing_config(self, *, history_command: str = "none", bump_history_seq: bool = False) -> None: """Persist current live processing config to runtime JSON file.""" if bump_history_seq: self._history_command_seq += 1 self._live_config_writer.write(self._live_processing_config(history_command=history_command)) def _on_processing_live_settings_changed(self, *_args) -> None: """Handle live-processing setting changes and trigger redraw when needed.""" try: self._write_live_processing_config() current_mode = self._processing_mode.currentText() if current_mode == "bscan": self._drain_results_until_quiet(timeout_s=0.25, poll_s=0.01) self._sync_bscan_history_from_results() self._draw_bscan_heatmap_from_history() elif current_mode == "gpr": self._drain_results_until_quiet(timeout_s=0.35, poll_s=0.01) if self._result_history: if not self._draw_results(self._result_history[-1]): self._clear_gpr_plot() else: self._clear_gpr_plot() elif self._result_history: self._draw_results(self._result_history[-1]) else: self._clear_trace_plots() except Exception as exc: # noqa: BLE001 self._show_exception("Failed to update live processing settings", exc) def _on_processing_mode_changed(self, mode: str) -> None: """Switch processing parameter page and refresh corresponding visualization.""" mode_to_page = { "pass_through": 0, "bscan": 1, "gpr": 2, } self._set_plot_mode(mode) self._processing_mode_pages.setCurrentIndex(mode_to_page.get(mode, 0)) current_page = self._processing_mode_pages.currentWidget() if current_page is not None: self._processing_mode_pages.setFixedHeight(current_page.sizeHint().height()) self._processing_mode_pages.updateGeometry() self._on_processing_live_settings_changed() if mode == "pass_through": self._log( "Processing mode selected: pass_through " f"(show_magnitude={self._show_magnitude_checkbox.isChecked()}, " f"show_phase={self._show_phase_checkbox.isChecked()}, " f"fixed_y={self._pass_through_fixed_y_enabled.isChecked()}, " f"y_range={self._pass_through_y_min_db.value():g}..{self._pass_through_y_max_db.value():g} dB)" ) elif mode == "bscan": self._log( "Processing mode selected: bscan " f"(axis={self._bscan_axis.currentText()}, " f"cut={self._bscan_cut_m.value():g} m, " f"max_depth={self._bscan_max_depth_m.value():g} m, " f"gain={self._bscan_gain.value():g}, " f"freq={self._bscan_start_freq_mhz.value():g}..{self._bscan_stop_freq_mhz.value():g} MHz)" ) elif mode == "gpr": self._log( "Processing mode selected: gpr " f"(inputs={self._gpr_input_positions_input.text().strip() or ''}, " f"outputs={self._gpr_output_positions_input.text().strip() or ''}, " f"depth={self._gpr_min_depth_m.value():g}..{self._gpr_max_depth_m.value():g} m, " f"freq={self._gpr_start_freq_mhz.value():g}..{self._gpr_stop_freq_mhz.value():g} MHz, " f"background_subtract={self._gpr_background_subtract_enabled.isChecked()}, " f"mean_count={self._gpr_background_mean_count.value()})" ) def _clear_history_mode_caches(self) -> None: """Drop cached render state for pass-through, B-scan, and GPR views.""" self._bscan_history_floor_collection_id = 0 self._clear_bscan_plot_history() if hasattr(self, "_bscan_plot"): self._bscan_plot.clear() self._clear_trace_plots() if hasattr(self, "_gpr_plot"): self._clear_gpr_plot() def _redraw_after_history_deletion(self) -> None: """Refresh plot immediately after destructive history deletion.""" if self._processing_mode.currentText() == "bscan": if self._result_history: self._sync_bscan_history_from_results() if not self._draw_bscan_heatmap_from_history(): self._bscan_plot.clear() return if self._processing_mode.currentText() == "gpr": if self._result_history and self._draw_results(self._result_history[-1]): return self._clear_gpr_plot() return if self._result_history: self._draw_results(self._result_history[-1]) return self._bscan_plot.clear() self._clear_trace_plots() def _on_radar_identity_changed(self, *_args) -> None: """Refresh device limits when radar identity/mode changes.""" if self._radar_mode.currentText() != "native": self._apply_radar_limits_to_ui(None) return changed = self._refresh_radar_limits_from_device() if changed: self._on_processing_live_settings_changed() def _on_radar_sweep_limits_changed(self) -> None: """Clamp processing frequency bounds after sweep start/stop edits.""" if self._sync_processing_frequency_limits_with_radar(): self._on_processing_live_settings_changed() def _refresh_radar_limits_from_device(self) -> bool: """Query native LibreVNA limits and apply them to GUI fields.""" serial = self._serial_input.text().strip() radar_service = LibreVnaService(serial=serial or None) if not radar_service.driver_available: self._fallback_to_mock_mode("LibreVNA Python driver is not available for device limits query") return False try: limits = radar_service.read_device_limits() except Exception as exc: # noqa: BLE001 self._log_exception("Failed to query LibreVNA limits; using UI fallback", exc, level="WARN") self._apply_radar_limits_to_ui(None) return False return self._apply_radar_limits_to_ui(limits) def _fallback_to_mock_mode(self, reason: str) -> None: """Handle unavailable native limits without mutating JSON-backed mode.""" self._log_warning(reason) self._apply_radar_limits_to_ui(None) def _apply_radar_limits_to_ui(self, limits: dict[str, float | int] | None) -> bool: """Apply optional radar limits and clamp dependent GUI fields.""" previous_limits = dict(self._radar_limits) if self._radar_limits is not None else None if limits is None: self._radar_limits = None self._radar_start_label.setText("Start Hz") self._radar_stop_label.setText("Stop Hz") self._radar_points_label.setText("Points") self._radar_ifbw_label.setText("IF BW Hz") self._radar_power_label.setText("Stimulus Power dBm") if self._radar_mode.currentText() == "native": self._radar_limits_hint.setText("Device limits unavailable in native mode (device not connected).") else: self._radar_limits_hint.setText("Mock mode: device limits are not applied.") self._power_input.setToolTip("Device power limits are available only in native mode.") return False min_freq_hz = float(limits["min_frequency_hz"]) max_freq_hz = float(limits["max_frequency_hz"]) min_ifbw_hz = float(limits["min_ifbw_hz"]) max_ifbw_hz = float(limits["max_ifbw_hz"]) max_points = int(limits["max_points"]) min_power_dbm = float(limits["min_power_dbm"]) max_power_dbm = float(limits["max_power_dbm"]) self._radar_limits = limits self._radar_start_label.setText(f"Start Hz ({min_freq_hz:g}..{max_freq_hz:g})") self._radar_stop_label.setText(f"Stop Hz ({min_freq_hz:g}..{max_freq_hz:g})") self._radar_points_label.setText(f"Points (1..{max_points:d})") self._radar_ifbw_label.setText(f"IF BW Hz ({min_ifbw_hz:g}..{max_ifbw_hz:g})") self._radar_power_label.setText(f"Stimulus Power dBm ({min_power_dbm:g}..{max_power_dbm:g})") self._radar_limits_hint.setText( f"Limits: Freq {min_freq_hz:g}..{max_freq_hz:g} Hz, Points 1..{max_points:d}, " f"IF BW {min_ifbw_hz:g}..{max_ifbw_hz:g} Hz, Power {min_power_dbm:g}..{max_power_dbm:g} dBm." ) changed = False prev_start = self._start_hz_input.text().strip() prev_stop = self._stop_hz_input.text().strip() prev_points = self._points_input.text().strip() prev_ifbw = self._ifbw_input.text().strip() prev_power = self._power_input.text().strip() start_hz = self._clamp_line_edit_float(self._start_hz_input, min_freq_hz, max_freq_hz) stop_hz = self._clamp_line_edit_float(self._stop_hz_input, min_freq_hz, max_freq_hz) if start_hz > stop_hz: stop_hz = start_hz self._stop_hz_input.setText(f"{stop_hz:g}") changed = True points = self._clamp_line_edit_int(self._points_input, 1, max_points) ifbw = self._clamp_line_edit_float(self._ifbw_input, min_ifbw_hz, max_ifbw_hz) power = self._clamp_line_edit_float(self._power_input, min_power_dbm, max_power_dbm) self._power_input.setToolTip(f"Device range: {min_power_dbm:g}..{max_power_dbm:g} dBm") changed = ( changed or prev_start != self._start_hz_input.text().strip() or prev_stop != self._stop_hz_input.text().strip() or prev_points != self._points_input.text().strip() or prev_ifbw != self._ifbw_input.text().strip() or prev_power != self._power_input.text().strip() ) applied_limits_changed = previous_limits != limits if applied_limits_changed: self._log( "Applied radar device limits: " f"freq={min_freq_hz:g}..{max_freq_hz:g} Hz, " f"points=1..{max_points}, " f"ifbw={min_ifbw_hz:g}..{max_ifbw_hz:g} Hz, " f"power={min_power_dbm:g}..{max_power_dbm:g} dBm" ) adjustments: list[str] = [] current_start = self._start_hz_input.text().strip() current_stop = self._stop_hz_input.text().strip() current_points = self._points_input.text().strip() current_ifbw = self._ifbw_input.text().strip() current_power = self._power_input.text().strip() if prev_start != current_start: adjustments.append(f"Start Hz: {prev_start or ''} -> {current_start}") if prev_stop != current_stop: adjustments.append(f"Stop Hz: {prev_stop or ''} -> {current_stop}") if prev_points != current_points: adjustments.append(f"Points: {prev_points or ''} -> {current_points}") if prev_ifbw != current_ifbw: adjustments.append(f"IF BW Hz: {prev_ifbw or ''} -> {current_ifbw}") if prev_power != current_power: adjustments.append(f"Stimulus Power dBm: {prev_power or ''} -> {current_power}") if adjustments: self._log_warning( "Radar fields were adjusted to satisfy device limits.", details="\n".join(adjustments), ) self._sync_processing_frequency_limits_with_radar() return changed @staticmethod def _clamp_line_edit_float(widget, min_value: float, max_value: float) -> float: """Clamp float line-edit value to inclusive range and rewrite widget text.""" try: value = float(widget.text().strip()) except ValueError: value = min_value value = min(max(value, min_value), max_value) widget.setText(f"{value:g}") return value @staticmethod def _clamp_line_edit_int(widget, min_value: int, max_value: int) -> int: """Clamp integer line-edit value to inclusive range and rewrite widget text.""" try: value = int(float(widget.text().strip())) except ValueError: value = min_value value = min(max(value, min_value), max_value) widget.setText(str(value)) return value def _sync_processing_frequency_limits_with_radar(self) -> bool: """Synchronize all processing frequency widgets with radar sweep bounds.""" bscan_changed = self._sync_bscan_frequency_limits_with_radar() gpr_changed = self._sync_gpr_frequency_limits_with_radar() return bscan_changed or gpr_changed def _sync_frequency_spinboxes_with_radar(self, widget_names: tuple[str, ...]) -> bool: """Synchronize one or more MHz spin boxes with current radar sweep bounds.""" required_widgets = ("_start_hz_input", "_stop_hz_input", *widget_names) if not all(hasattr(self, widget_name) for widget_name in required_widgets): return False try: radar_start_hz = float(self._start_hz_input.text().strip()) radar_stop_hz = float(self._stop_hz_input.text().strip()) except ValueError: return False radar_min_mhz = min(radar_start_hz, radar_stop_hz) / 1_000_000.0 radar_max_mhz = max(radar_start_hz, radar_stop_hz) / 1_000_000.0 changed = False widget_labels = { "_bscan_start_freq_mhz": "B-scan Start MHz", "_bscan_stop_freq_mhz": "B-scan Stop MHz", "_gpr_start_freq_mhz": "GPR Start MHz", "_gpr_stop_freq_mhz": "GPR Stop MHz", } widgets = [getattr(self, widget_name) for widget_name in widget_names] previous_values = {widget_name: getattr(self, widget_name).value() for widget_name in widget_names} for widget in widgets: if widget.minimum() != radar_min_mhz or widget.maximum() != radar_max_mhz: changed = True widget.blockSignals(True) widget.setRange(radar_min_mhz, radar_max_mhz) widget.blockSignals(False) for widget in widgets: clamped_value = min(max(widget.value(), radar_min_mhz), radar_max_mhz) if clamped_value != widget.value(): changed = True widget.blockSignals(True) widget.setValue(clamped_value) widget.blockSignals(False) clamped_fields = [] for widget_name in widget_names: current_value = getattr(self, widget_name).value() previous_value = previous_values[widget_name] if current_value == previous_value: continue clamped_fields.append( f"{widget_labels.get(widget_name, widget_name)}: {previous_value:g} -> {current_value:g}" ) if clamped_fields: self._log_warning( "Processing frequency limits were clamped to the active radar sweep.", details="\n".join(clamped_fields), ) return changed def _sync_bscan_frequency_limits_with_radar(self) -> bool: """Synchronize B-scan start/stop MHz widget ranges with radar sweep bounds.""" return self._sync_frequency_spinboxes_with_radar( ( "_bscan_start_freq_mhz", "_bscan_stop_freq_mhz", ) ) def _sync_gpr_frequency_limits_with_radar(self) -> bool: """Synchronize GPR start/stop MHz widget ranges with radar sweep bounds.""" return self._sync_frequency_spinboxes_with_radar( ( "_gpr_start_freq_mhz", "_gpr_stop_freq_mhz", ) ) @staticmethod def _switches_are_effectively_static(config: RunConfigModel) -> bool: """Return `True` when switch setup effectively yields one fixed combo.""" has_single_position = config.input_switch.positions <= 1 and config.output_switch.positions <= 1 both_mock = config.input_switch.driver_mode == "mock" and config.output_switch.driver_mode == "mock" return has_single_position or both_mock