199 lines
9.8 KiB
Python
199 lines
9.8 KiB
Python
"""GUI profile persistence, startup selection and application to concrete editors."""
|
|
|
|
from copy import deepcopy
|
|
from dataclasses import dataclass
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtWidgets import QFileDialog, QWidget
|
|
|
|
from python_app.gui.controllers.config_controller import ConfigController
|
|
from python_app.gui.controllers.sections.views import PrimaryView
|
|
from python_app.file_io import atomic_write_json
|
|
from python_app.gui.controllers.live_settings_controller import LiveSettingsController
|
|
from python_app.gui.controllers.log_controller import LogController
|
|
from python_app.gui.controllers.pipeline_controller import PipelineController
|
|
from python_app.gui.controllers.preprocess_controller import PreprocessController
|
|
from python_app.gui.controllers.radar_settings_controller import RadarSettingsController
|
|
from python_app.gui.settings_defaults import cpp_bscan_replay_window_for_config, default_gui_state_for_config
|
|
from python_app.models.gui_profile_schema import GuiProfileModel, GuiStateModel
|
|
from python_app.models.run_config_schema import RunConfigModel
|
|
from python_app.orchestration.gui_session_state import GuiSessionState, GuiSessionStateStore
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProfileState:
|
|
config: RunConfigModel
|
|
gui: GuiStateModel
|
|
path: Path
|
|
session: GuiSessionStateStore
|
|
|
|
|
|
def _remember(state: ProfileState, path: Path, store: GuiSessionStateStore, log: LogController) -> None:
|
|
state.path = path.expanduser().resolve(strict=False)
|
|
try:
|
|
store.write(GuiSessionState(last_profile_path=str(state.path)))
|
|
except Exception as exc:
|
|
log.exception("Failed to update GUI session-state with the active config profile path", exc, level="WARN")
|
|
|
|
|
|
def load_startup(project_root: Path, log: LogController) -> ProfileState:
|
|
store = GuiSessionStateStore(project_root / "python_app/runtime/gui_session_state.json")
|
|
root_path = (project_root / "run_config.json").resolve(strict=False)
|
|
requested = os.environ.get("RADAR_SYSTEM_PROFILE", "").strip()
|
|
if not requested:
|
|
try:
|
|
requested = store.load().last_profile_path.strip()
|
|
except Exception as exc:
|
|
log.warning("Failed to read GUI session-state; using root run_config.json.", details=log.exception_details(exc))
|
|
path = Path(requested).expanduser() if requested else root_path
|
|
if not path.is_absolute():
|
|
path = project_root / path
|
|
path = path.resolve(strict=False)
|
|
try:
|
|
profile = GuiProfileModel.load_from_path(path)
|
|
except Exception as exc:
|
|
if path == root_path:
|
|
raise
|
|
log.warning("Failed to load the last selected config profile; falling back to root run_config.json.",
|
|
details=log.exception_details(exc))
|
|
profile, path = GuiProfileModel.load_from_path(root_path), root_path
|
|
config = profile.run_config.clone()
|
|
log.set_level(config.logging.level)
|
|
gui = profile.gui or default_gui_state_for_config(config, project_root)
|
|
if profile.gui is None and path != root_path:
|
|
log.info(f"Loaded legacy run config without GUI defaults: {path}")
|
|
state = ProfileState(config, gui, path, store)
|
|
_remember(state, path, store, log)
|
|
return state
|
|
|
|
|
|
class ProfileController:
|
|
def __init__(
|
|
self, parent: QWidget, project_root: Path, state: ProfileState, config: ConfigController,
|
|
preprocess: PreprocessController, pipeline: PipelineController, radar: RadarSettingsController,
|
|
live: LiveSettingsController, log: LogController,
|
|
) -> None:
|
|
self.state = state
|
|
self._parent, self._project_root, self._config = parent, project_root, config
|
|
self._preprocess, self._pipeline, self._radar, self._live, self._log = preprocess, pipeline, radar, live, log
|
|
self._session = state.session
|
|
|
|
@staticmethod
|
|
def write_profile(
|
|
path: Path, profile: GuiProfileModel, active: RunConfigModel | None = None,
|
|
*, allow_overwrite: bool = True, dataset: bool = False,
|
|
) -> GuiProfileModel:
|
|
"""Atomically save settings or a dataset profile with its acquisition provenance."""
|
|
if not allow_overwrite and path.exists():
|
|
raise FileExistsError(f"Config profile output already exists: {path}")
|
|
if dataset and active is not None:
|
|
profile = profile.clone()
|
|
profile.run_config.radar = deepcopy(active.radar)
|
|
profile.run_config.input_switch = deepcopy(active.input_switch)
|
|
profile.run_config.output_switch = deepcopy(active.output_switch)
|
|
profile.run_config.combos = deepcopy(active.combos)
|
|
profile.run_config.preprocess = deepcopy(active.preprocess)
|
|
profile.run_config.gpr.tx_geometry = deepcopy(active.gpr.tx_geometry)
|
|
profile.run_config.gpr.rx_geometry = deepcopy(active.gpr.rx_geometry)
|
|
atomic_write_json(path, profile.to_dict(), durable=True)
|
|
return profile
|
|
|
|
def bind_actions(self, actions: PrimaryView) -> None:
|
|
actions.load.clicked.connect(self.load_dialog)
|
|
actions.save.clicked.connect(self.save_dialog)
|
|
|
|
def build_config(self) -> RunConfigModel:
|
|
return self._config.build_config(self.state.config, self._preprocess.selected_sets,
|
|
self._live.writer.path, self._log.level.currentText().lower())
|
|
|
|
def build_gui_state(self) -> GuiStateModel:
|
|
return self._config.build_gui_state(self.state.config, self._preprocess.settings())
|
|
|
|
def build_profile(self) -> GuiProfileModel:
|
|
return GuiProfileModel(run_config=self.build_config(), gui=self.build_gui_state())
|
|
|
|
def radar_key(self) -> str:
|
|
return self._config.radar_key(self.state.config)
|
|
|
|
def write(self, path: Path, *, allow_overwrite: bool = True, dataset: bool = False) -> GuiProfileModel:
|
|
return self.write_profile(path, self.build_profile(), self._pipeline.active_config,
|
|
allow_overwrite=allow_overwrite, dataset=dataset)
|
|
|
|
def save_dialog(self) -> None:
|
|
try:
|
|
selected, _ = QFileDialog.getSaveFileName(self._parent, "Save Config Profile", str(self.state.path),
|
|
"JSON Files (*.json);;All Files (*)")
|
|
if not selected:
|
|
return
|
|
path = Path(selected).expanduser().resolve(strict=False)
|
|
if not path.suffix:
|
|
path = path.with_suffix(".json")
|
|
profile = self.write(path)
|
|
self.state.config = profile.run_config.clone()
|
|
self.state.gui = profile.gui or default_gui_state_for_config(profile.run_config, self._project_root)
|
|
_remember(self.state, path, self._session, self._log)
|
|
config, sweep = profile.run_config, profile.run_config.radar.sweep
|
|
points = "from RadioPhotonic radar (RPH) stream" if config.is_rph_radar else str(sweep.points)
|
|
self._log.info(
|
|
f"Config profile saved: path={path}, combos={len(config.combos)}, "
|
|
f"sweep={sweep.start_hz:g}..{sweep.stop_hz:g} Hz, points={points}, "
|
|
f"ifbw={sweep.if_bandwidth_hz:g} Hz, power={sweep.power_dbm:g} dBm, "
|
|
f"processing_mode={self.state.gui.processing.selected_mode}"
|
|
)
|
|
except Exception as exc:
|
|
self._log.show_exception("Failed to save config profile", exc)
|
|
|
|
def _ready_to_load(self) -> bool:
|
|
if self._pipeline.closing:
|
|
return False
|
|
reason = self._preprocess.busy_reason()
|
|
if reason:
|
|
self._log.show_error(reason, details=self._preprocess.capture_state_details())
|
|
return False
|
|
if self._pipeline.running or self._pipeline.should_run or self._pipeline.paused:
|
|
self._log.show_error("Stop all pipeline processes before loading a config profile")
|
|
return False
|
|
return True
|
|
|
|
def load(self, path: Path) -> None:
|
|
if not self._ready_to_load():
|
|
return
|
|
try:
|
|
path = path.expanduser().resolve(strict=False)
|
|
profile = GuiProfileModel.load_from_path(path)
|
|
self.apply(profile, path)
|
|
kind = "legacy run config" if profile.gui is None else "full GUI profile"
|
|
self._log.info(f"Config profile loaded: path={path}, kind={kind}, combos={len(self.state.config.combos)}, "
|
|
f"processing_mode={self._live.active_mode}")
|
|
except Exception as exc:
|
|
self._log.show_exception("Failed to load config profile", exc)
|
|
|
|
def load_dialog(self) -> None:
|
|
if not self._ready_to_load():
|
|
return
|
|
selected, _ = QFileDialog.getOpenFileName(self._parent, "Load Config Profile", str(self.state.path),
|
|
"JSON Files (*.json);;All Files (*)")
|
|
if selected:
|
|
self.load(Path(selected))
|
|
|
|
def apply(self, profile: GuiProfileModel, path: Path) -> None:
|
|
self._log.debug(f"Applying loaded profile: path={path}, has_gui_state={profile.gui is not None}.")
|
|
config = profile.run_config.clone()
|
|
gui = profile.gui or default_gui_state_for_config(config, self._project_root)
|
|
self.state.config, self.state.gui = config, gui
|
|
self._config.apply_profile_widgets(config, gui)
|
|
self._radar.sync_profile(gui.switches.combo_mode)
|
|
self._live.sync_controls()
|
|
self._log.set_level(self._log.level.currentText())
|
|
self._preprocess.apply_profile(config, gui.preprocess_dialog)
|
|
self._pipeline.plots.live_window = cpp_bscan_replay_window_for_config(config)
|
|
self._pipeline.plots.clear_bscan()
|
|
self._live.set_mode_page(gui.processing.selected_mode)
|
|
self._radar.initial_limits()
|
|
if self._preprocess.dialog is not None:
|
|
self._preprocess.refresh_sets()
|
|
self._live.mode_changed(gui.processing.selected_mode)
|
|
self._pipeline.history_changed()
|
|
_remember(self.state, path, self._session, self._log)
|