163 lines
8.1 KiB
Python
163 lines
8.1 KiB
Python
"""Radar-specific editor constraints and device limits."""
|
|
|
|
from collections.abc import Callable
|
|
|
|
from PyQt6.QtCore import QSignalBlocker
|
|
from PyQt6.QtWidgets import QCheckBox, QComboBox, QLineEdit, QPlainTextEdit
|
|
|
|
from python_app.gui.controllers.log_controller import LogController
|
|
from python_app.gui.controllers.sections.views import RadarView, SwitchView, ProcessingView, connect_setting
|
|
from python_app.hardware_full.librevna_service import LibreVnaService
|
|
from python_app.hardware_full.single_radar_service import create_single_radar_service
|
|
from python_app.models.run_config_schema import RunConfigModel
|
|
|
|
|
|
class RadarSettingsController:
|
|
def __init__(
|
|
self, radar: RadarView, switches: SwitchView, processing: ProcessingView,
|
|
defaults: Callable[[], RunConfigModel], log: LogController, *,
|
|
reset_preprocess: Callable[[], None], live_changed: Callable[[], None],
|
|
) -> None:
|
|
self._radar, self._switches, self._processing = radar, switches, processing
|
|
self._defaults, self._log = defaults, log
|
|
self._reset_preprocess, self._live_changed = reset_preprocess, live_changed
|
|
self.limits: dict[str, float | int] | None = None
|
|
switches.select_text.clicked.connect(lambda: self.set_combo_mode("text"))
|
|
switches.select_single.clicked.connect(lambda: self.set_combo_mode("single"))
|
|
for widget in radar.sweep_edits:
|
|
widget.editingFinished.connect(self.sweep_changed)
|
|
for name, widget in radar.fields.items():
|
|
if name.startswith(("_adc_", "_optical_")) and isinstance(widget, (QLineEdit, QPlainTextEdit, QComboBox, QCheckBox)):
|
|
connect_setting(widget, self._optical_changed if widget in (radar.optical_enabled, radar.optical_mode)
|
|
else self._reset_preprocess)
|
|
|
|
def set_combo_mode(self, mode: str) -> None:
|
|
config = self._defaults()
|
|
matrix = config.is_multi_device
|
|
text_selected = matrix or mode != "single"
|
|
for button, selected in ((self._switches.select_text, text_selected),
|
|
(self._switches.select_single, not text_selected)):
|
|
button.setChecked(selected)
|
|
button.setEnabled(not matrix)
|
|
combos = self._switches.combos
|
|
if matrix:
|
|
combos.setText(",".join(f"{combo.input}:{combo.output}" for combo in config.build_runtime_combos()))
|
|
combos.setEnabled(text_selected and not matrix)
|
|
for widget in (self._switches.single_input, self._switches.single_output):
|
|
widget.setEnabled(not text_selected and not matrix)
|
|
|
|
def sync_profile(self, combo_mode: str) -> None:
|
|
self.set_combo_mode(combo_mode)
|
|
adc = self._defaults().is_rph_radar
|
|
self._radar.vna_panel.setVisible(not adc)
|
|
self._radar.adc_panel.setVisible(adc)
|
|
self.sync_adc_controls()
|
|
|
|
def sync_adc_controls(self) -> None:
|
|
enabled = self._radar.optical_enabled.isChecked()
|
|
mode = self._radar.optical_mode.currentText()
|
|
for active, widgets in ((enabled, self._radar.optical_common),
|
|
(enabled and mode == "manual", self._radar.optical_manual),
|
|
(enabled and mode == "variation", self._radar.optical_variation)):
|
|
for widget in widgets:
|
|
widget.setEnabled(active)
|
|
self._radar.manual_panel.setVisible(mode == "manual")
|
|
self._radar.variation_panel.setVisible(mode == "variation")
|
|
|
|
def _optical_changed(self) -> None:
|
|
self.sync_adc_controls()
|
|
self._reset_preprocess()
|
|
|
|
def initial_limits(self) -> None:
|
|
mode = self._defaults().radar.driver_mode
|
|
if mode == "native":
|
|
self._log.debug("Querying radar limits from device (native driver mode).")
|
|
self.refresh_limits()
|
|
else:
|
|
self._log.debug(f"Skipping device radar-limit query (driver mode={mode}).")
|
|
self.apply_limits(None)
|
|
|
|
def refresh_limits(self) -> bool:
|
|
config = self._defaults()
|
|
if config.is_rph_radar:
|
|
return self.apply_limits(None)
|
|
service = (LibreVnaService(serial=config.radar.serial or None) if config.is_multi_device
|
|
else create_single_radar_service(config))
|
|
return self.apply_limits(service.read_device_limits())
|
|
|
|
def apply_limits(self, limits: dict[str, float | int] | None) -> bool:
|
|
previous, self.limits = self.limits, limits
|
|
labels = ("Start (Hz)", "Stop (Hz)", "Points", "IF bandwidth (Hz)", "Stimulus power (dBm)")
|
|
edits = self._radar.sweep_edits
|
|
hint = self._radar.limits_hint
|
|
if limits is None:
|
|
for widget, label in zip(self._radar.sweep_labels, labels, strict=True):
|
|
widget.setText(label)
|
|
hint.setText("Device limits are not available.")
|
|
edits[-1].setToolTip("Stimulus power configured in the active profile.")
|
|
return False
|
|
|
|
frequency = (float(limits["min_frequency_hz"]), float(limits["max_frequency_hz"]))
|
|
ifbw = (float(limits["min_ifbw_hz"]), float(limits["max_ifbw_hz"]))
|
|
power = (float(limits["min_power_dbm"]), float(limits["max_power_dbm"]))
|
|
max_points = int(limits["max_points"])
|
|
ranges = (frequency, frequency, (1, max_points), ifbw, power)
|
|
before = [edit.text().strip() for edit in edits]
|
|
for index, (widget, label, edit, bounds) in enumerate(zip(self._radar.sweep_labels, labels, edits, ranges, strict=True)):
|
|
minimum, maximum = bounds
|
|
widget.setText(f"{label} ({minimum:g}..{maximum:g})")
|
|
self._clamp(edit, minimum, maximum, integer=index == 2)
|
|
if float(edits[0].text()) > float(edits[1].text()):
|
|
edits[1].setText(edits[0].text())
|
|
edits[-1].setToolTip(f"Device range: {power[0]:g}..{power[1]:g} dBm")
|
|
summary = (f"freq={frequency[0]:g}..{frequency[1]:g} Hz, points=1..{max_points}, "
|
|
f"ifbw={ifbw[0]:g}..{ifbw[1]:g} Hz, power={power[0]:g}..{power[1]:g} dBm")
|
|
hint.setText(f"Limits: {summary}")
|
|
if previous != limits:
|
|
self._log.info(f"Applied radar device limits: {summary}")
|
|
after = [edit.text().strip() for edit in edits]
|
|
adjustments = [f"{label}: {old or '<empty>'} -> {new}"
|
|
for label, old, new in zip(labels, before, after, strict=True) if old != new]
|
|
if adjustments:
|
|
self._log.warning("Radar fields were adjusted to satisfy device limits.", details="\n".join(adjustments))
|
|
self.sync_frequency_limits()
|
|
return before != after
|
|
|
|
@staticmethod
|
|
def _clamp(widget: QLineEdit, minimum: float, maximum: float, *, integer: bool = False) -> None:
|
|
try:
|
|
value = float(widget.text().strip())
|
|
if integer:
|
|
value = int(value)
|
|
except ValueError:
|
|
value = minimum
|
|
value = min(max(value, minimum), maximum)
|
|
widget.setText(str(int(value)) if integer else f"{value:g}")
|
|
|
|
def sync_frequency_limits(self) -> bool:
|
|
try:
|
|
config = self._defaults()
|
|
band = config.radar.rph_radar.band
|
|
bounds = ([band.start_hz / 1e6, band.stop_hz / 1e6] if config.is_rph_radar else
|
|
sorted(float(widget.text()) / 1e6 for widget in self._radar.sweep_edits[:2]))
|
|
except ValueError:
|
|
return False
|
|
changed = False
|
|
adjustments: list[str] = []
|
|
for index, widget in enumerate(self._processing.frequency_controls):
|
|
before = widget.value()
|
|
changed |= (widget.minimum(), widget.maximum()) != tuple(bounds)
|
|
with QSignalBlocker(widget):
|
|
widget.setRange(*bounds)
|
|
if before != widget.value():
|
|
adjustments.append(f"Frequency control {index + 1}: {before:g} -> {widget.value():g}")
|
|
if adjustments:
|
|
self._log.warning("Processing frequency limits were clamped to the active radar sweep.",
|
|
details="\n".join(adjustments))
|
|
return changed or bool(adjustments)
|
|
|
|
def sweep_changed(self) -> None:
|
|
self._reset_preprocess()
|
|
if self.sync_frequency_limits():
|
|
self._live_changed()
|