933 lines
44 KiB
Python
933 lines
44 KiB
Python
"""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,
|
|
speed_m_s=0.0,
|
|
look_angle_deg=0.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()),
|
|
speed_m_s=float(self._gpr_speed_m_s.value()),
|
|
look_angle_deg=float(self._gpr_look_angle_deg.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():
|
|
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:
|
|
normalized_path = self._normalize_profile_path(Path(selected_path))
|
|
profile = GuiProfileModel.load_from_path(normalized_path)
|
|
processor_running = self._supervisor.is_processor_running()
|
|
self._apply_loaded_profile(profile, normalized_path)
|
|
profile_kind = "legacy run config" if profile.gui is None else "full GUI profile"
|
|
message = (
|
|
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()}"
|
|
)
|
|
if processor_running:
|
|
message += (
|
|
"; data_processor is still running, so live processing settings were applied immediately "
|
|
"and stable settings are now staged in the UI for the next Start"
|
|
)
|
|
self._log(message)
|
|
except Exception as exc: # noqa: BLE001
|
|
self._show_exception("Failed to load config profile", exc)
|
|
|
|
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_speed_m_s,
|
|
self._gpr_look_angle_deg,
|
|
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_speed_m_s.setValue(float(gui_state.processing.gpr.speed_m_s))
|
|
self._gpr_look_angle_deg.setValue(float(gui_state.processing.gpr.look_angle_deg))
|
|
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_speed_m_s=float(self._gpr_speed_m_s.value()),
|
|
gpr_look_angle_deg=float(self._gpr_look_angle_deg.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 '<all>'}, "
|
|
f"outputs={self._gpr_output_positions_input.text().strip() or '<all>'}, "
|
|
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"speed={self._gpr_speed_m_s.value():g} m/s, "
|
|
f"look_angle={self._gpr_look_angle_deg.value():g} deg, "
|
|
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 '<empty>'} -> {current_start}")
|
|
if prev_stop != current_stop:
|
|
adjustments.append(f"Stop Hz: {prev_stop or '<empty>'} -> {current_stop}")
|
|
if prev_points != current_points:
|
|
adjustments.append(f"Points: {prev_points or '<empty>'} -> {current_points}")
|
|
if prev_ifbw != current_ifbw:
|
|
adjustments.append(f"IF BW Hz: {prev_ifbw or '<empty>'} -> {current_ifbw}")
|
|
if prev_power != current_power:
|
|
adjustments.append(f"Stimulus Power dBm: {prev_power or '<empty>'} -> {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
|