"""Main GUI composition root. This module wires UI/controller mixins together and owns application-level state shared across them (runtime services, readers, history buffers, timer). """ from __future__ import annotations from collections import deque import json from pathlib import Path from PyQt6.QtCore import QTimer from PyQt6.QtWidgets import QMainWindow, QMessageBox from python_app.gui.controllers.app_window_config_mixin import AppWindowConfigMixin from python_app.gui.controllers.app_window_pipeline_mixin import AppWindowPipelineMixin from python_app.gui.controllers.app_window_plot_mixin import AppWindowPlotMixin from python_app.gui.controllers.app_window_preprocess_mixin import AppWindowPreprocessMixin from python_app.gui.controllers.app_window_snapshot_mixin import AppWindowSnapshotMixin from python_app.gui.controllers.app_window_ui_mixin import AppWindowUiMixin from python_app.gui.preprocess_dialog import PreprocessDialog from python_app.models.dataset_model import ResultCollection, SweepCollection from python_app.models.run_config_model import RunConfigModel from python_app.orchestration.config_writer import ConfigWriter from python_app.orchestration.live_processing_config import ProcessingLiveConfigWriter from python_app.orchestration.preprocess_assets import PREPROCESS_ASSET_KEYS, preprocess_asset_model from python_app.orchestration.process_supervisor import ProcessSupervisor from python_app.orchestration.shm_reader import ShmRingReader from python_app.storage.npz_store import NpzStore from python_app.workflows.sequential_capture_workflow import SequentialCaptureSession class AppWindow( AppWindowUiMixin, AppWindowConfigMixin, AppWindowPreprocessMixin, AppWindowPlotMixin, AppWindowPipelineMixin, AppWindowSnapshotMixin, QMainWindow, ): """Top-level window coordinating GUI state and acquisition runtime.""" def __init__(self, project_root: Path) -> None: """Initialize all app subsystems in deterministic order.""" super().__init__() self._init_paths_and_defaults(project_root) self._init_runtime_services() self._init_reader_handles() self._init_preprocess_state() self._init_capture_state() self._init_history_state() self._init_runtime_limits() self._init_polling_timer() self._bootstrap_ui_runtime() def _init_paths_and_defaults(self, project_root: Path) -> None: """Initialize project paths and baseline run configuration.""" self._project_root = project_root self._defaults_config_path = project_root / "run_config.json" self._defaults_config = RunConfigModel.load_from_path(self._defaults_config_path) def _init_runtime_services(self) -> None: """Initialize long-lived service objects used by mixins.""" runtime_dir = self._project_root / "python_app/runtime" self._store = NpzStore(self._project_root / "python_app/data") self._config_writer = ConfigWriter(runtime_dir) self._supervisor = ProcessSupervisor(self._project_root) self._live_config_writer = ProcessingLiveConfigWriter(runtime_dir / "processing_live.json") def _init_reader_handles(self) -> None: """Initialize SHM readers as detached (not connected) handles.""" self._raw_reader: ShmRingReader | None = None self._pre_reader: ShmRingReader | None = None self._result_reader: ShmRingReader | None = None def _init_preprocess_state(self) -> None: """Initialize preprocessing dialog and selected set names.""" self._preprocess_dialog: PreprocessDialog | None = None self._selected_preprocess_sets = { key: str(preprocess_asset_model(self._defaults_config, key).set_name) for key in PREPROCESS_ASSET_KEYS } def _init_capture_state(self) -> None: """Initialize one-shot capture and sequence-control flags.""" self._capture_session: SequentialCaptureSession | None = None self._resume_pipeline_after_capture = False self._single_capture_active = False self._single_capture_start_ns: int | None = None self._single_capture_seen_raw = False self._single_capture_target_collection_id: int | None = None def _init_history_state(self) -> None: """Initialize runtime history buffers and render-cache state.""" history_limit = self._history_limit_from_config() self._raw_history: deque[SweepCollection] = deque(maxlen=history_limit) self._pre_history: deque[SweepCollection] = deque(maxlen=history_limit) self._result_history: deque[ResultCollection] = deque(maxlen=history_limit) # Sequence id must survive GUI restarts so history commands stay monotonic. self._history_command_seq = self._load_history_command_seq(self._live_config_writer.path) self._bscan_history_limit = history_limit self._bscan_history_by_combo = {} self._bscan_depth_axis_by_combo = {} self._bscan_history_floor_collection_id = 0 self._bscan_render_signature = None self._gpr_lookup_table = None self._gpr_image_item = None self._gpr_tx_item = None self._gpr_rx_item = None self._gpr_points_item = None self._gpr_region_centers_item = None self._gpr_point_labels = [] self._gpr_region_center_labels = [] self._gpr_region_mask_items = [] self._gpr_region_contours = [] self._gpr_geometry_signature = None self._gpr_selected_geometry = None self._phase_viewbox = None self._history_run_signature = None self._radar_limits: dict[str, float | int] | None = None def _history_limit_from_config(self) -> int: """Return unified GUI history limit derived from configured ring capacities.""" return max( 1, min( int(self._defaults_config.rings.raw_tap.capacity), int(self._defaults_config.rings.preprocessed_tap.capacity), int(self._defaults_config.rings.results.capacity), ), ) def _init_runtime_limits(self) -> None: """Initialize read/drain loop limits used by polling and snapshot code.""" self._max_pop_per_poll = 256 self._max_pop_per_snapshot_drain = 4096 def _init_polling_timer(self) -> None: """Create periodic timer that polls SHM rings for new data.""" self._timer = QTimer(self) self._timer.setInterval(50) self._timer.timeout.connect(self._poll_rings) def _bootstrap_ui_runtime(self) -> None: """Build UI and apply initial runtime-bound state after widgets exist.""" self._build_ui() self._refresh_preprocess_summary_labels() self._apply_initial_radar_limits() self._write_live_processing_config() self._timer.start() def _apply_initial_radar_limits(self) -> None: """Apply startup radar-limits strategy according to selected radar mode.""" if self._radar_mode.currentText() == "native": self._refresh_radar_limits_from_device() return self._apply_radar_limits_to_ui(None) def _log(self, text: str) -> None: """Append a line to the runtime log panel.""" self._log_box.appendPlainText(text) @staticmethod def _load_history_command_seq(config_path: Path) -> int: """Load previously used live-command sequence from runtime config file.""" try: payload = json.loads(config_path.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 # Missing or malformed file should not block startup. return 0 raw_value = payload.get("history_command_seq", 0) if isinstance(raw_value, bool): return 0 if isinstance(raw_value, (int, float)): return max(0, int(raw_value)) return 0 def _show_error(self, message: str) -> None: """Log and present an error in a modal dialog.""" self._log(f"ERROR: {message}") QMessageBox.critical(self, "Error", message) def closeEvent(self, event) -> None: # noqa: N802 """Ensure workers and dialogs are closed before window destruction.""" try: self._resume_pipeline_after_capture = False # 1) Abort active capture first (releases exclusive hardware resources). self._abort_capture_sequence(resume_pipeline=False) # 2) Stop all managed processes/readers. self._stop_all_processes() # 3) Close auxiliary dialog windows. if self._preprocess_dialog is not None: self._preprocess_dialog.close() finally: super().closeEvent(event)