393 lines
16 KiB
Python
393 lines
16 KiB
Python
"""Pipeline runtime lifecycle mixin for the main GUI window."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from python_app.gui.runtime.constraints import validate_processing_mode_constraints
|
|
from python_app.gui.runtime.history import build_run_history_signature, record_result_history
|
|
from python_app.hardware_full.librevna_service import LibreVnaService
|
|
from python_app.models.dataset_model import ComboKey, ResultCollection, SweepCollection
|
|
from python_app.models.run_config_model import RunConfigModel
|
|
from python_app.orchestration.shm_reader import ShmRingReader
|
|
|
|
|
|
class AppWindowPipelineMixin:
|
|
"""Controls start/stop, readers, and periodic polling of pipeline rings."""
|
|
|
|
def _start_single_capture(self) -> None:
|
|
"""Start acquisition in single-capture mode."""
|
|
self._start_run(single_capture=True)
|
|
|
|
def _start_run(self, *, single_capture: bool = False) -> None:
|
|
"""Start pipeline processes and ring readers."""
|
|
if self._capture_session is not None:
|
|
self._show_error("Cannot start pipeline during active capture sequence")
|
|
return
|
|
if self._supervisor.is_running():
|
|
self._show_error("Pipeline is already running")
|
|
return
|
|
|
|
try:
|
|
processor_was_running = self._supervisor.is_processor_running()
|
|
if not processor_was_running:
|
|
self._reset_runtime_history()
|
|
|
|
config = self._build_config()
|
|
self._validate_processing_mode_constraints(config)
|
|
run_signature = self._build_run_history_signature(config)
|
|
radar_key = self._radar_key(config)
|
|
|
|
if not config.preprocess.calibration_set or not config.preprocess.reference_set:
|
|
raise RuntimeError("Select calibration and reference sets in Preprocessing Panel before Start")
|
|
|
|
combo_keys = [ComboKey(input_pos=combo.input, output_pos=combo.output) for combo in config.combos]
|
|
|
|
if not self._store.has_combo_coverage(
|
|
"calibration", radar_key, config.preprocess.calibration_set, combo_keys
|
|
):
|
|
raise RuntimeError("Selected calibration set does not cover requested run combos")
|
|
if not self._store.has_combo_coverage(
|
|
"reference", radar_key, config.preprocess.reference_set, combo_keys
|
|
):
|
|
raise RuntimeError("Selected reference set does not cover requested run combos")
|
|
|
|
calibration_bundle, reference_bundle = self._config_writer.prepare_bundles(
|
|
self._store,
|
|
radar_key,
|
|
config.preprocess.calibration_set,
|
|
config.preprocess.reference_set,
|
|
)
|
|
config.preprocess.calibration_bundle_path = str(calibration_bundle)
|
|
config.preprocess.reference_bundle_path = str(reference_bundle)
|
|
config.runtime.continuous = not single_capture
|
|
|
|
if not single_capture:
|
|
self._prepare_radar_for_native_acquisition(config)
|
|
|
|
config_path = self._config_writer.write(config, self._project_root / "python_app/runtime/run_config.json")
|
|
|
|
if not single_capture:
|
|
should_reset_history = (
|
|
self._history_run_signature is not None and self._history_run_signature != run_signature
|
|
)
|
|
if should_reset_history:
|
|
self._reset_runtime_history()
|
|
self._log("History reset because run settings changed")
|
|
self._history_run_signature = run_signature
|
|
|
|
self._supervisor.start(config_path)
|
|
self._close_readers()
|
|
self._raw_reader = ShmRingReader(config.rings.raw_tap.name)
|
|
self._pre_reader = ShmRingReader(config.rings.preprocessed_tap.name)
|
|
self._result_reader = ShmRingReader(config.rings.results.name)
|
|
self._single_capture_active = single_capture
|
|
self._single_capture_start_ns = None
|
|
self._single_capture_seen_raw = False
|
|
self._single_capture_target_collection_id = None
|
|
|
|
# Always drop unread payloads for all stages so single-capture starts
|
|
# from a clean boundary and does not retain stale results-only tail.
|
|
self._drop_pending_ring_payloads(include_results=True)
|
|
if single_capture:
|
|
self._single_capture_start_ns = time.monotonic_ns()
|
|
|
|
if single_capture:
|
|
self._status_label.setText("Status: single capture running")
|
|
self._log("Single capture started")
|
|
else:
|
|
self._status_label.setText("Status: running")
|
|
self._log("Pipeline started")
|
|
except Exception as exc: # noqa: BLE001
|
|
self._single_capture_active = False
|
|
self._single_capture_start_ns = None
|
|
self._stop_all_processes()
|
|
self._show_error(f"Failed to start pipeline: {exc}")
|
|
|
|
def _apply_radar_settings(self) -> None:
|
|
"""Apply current radar settings by preconfiguring native device."""
|
|
if self._capture_session is not None:
|
|
self._show_error("Finish or abort capture sequence before applying radar settings")
|
|
return
|
|
|
|
was_running = self._supervisor.is_running()
|
|
if was_running:
|
|
self._stop_run()
|
|
|
|
try:
|
|
if self._radar_mode.currentText() == "native":
|
|
self._refresh_radar_limits_from_device()
|
|
config = self._build_config()
|
|
self._prepare_radar_for_native_acquisition(config)
|
|
self._log("Radar settings applied")
|
|
except Exception as exc: # noqa: BLE001
|
|
self._show_error(f"Failed to apply radar settings: {exc}")
|
|
finally:
|
|
if was_running:
|
|
self._start_run()
|
|
|
|
def _prepare_radar_for_native_acquisition(self, config: RunConfigModel) -> None:
|
|
"""Preconfigure native LibreVNA using current sweep settings."""
|
|
if config.radar.driver_mode != "native":
|
|
self._log("Radar pre-configuration skipped (mock mode)")
|
|
return
|
|
|
|
radar_service = LibreVnaService(serial=config.radar.serial or None)
|
|
if not radar_service.driver_available:
|
|
raise RuntimeError("LibreVNA Python driver is not available for native pre-configuration")
|
|
|
|
try:
|
|
radar_service.open()
|
|
radar_service.configure(config.radar.sweep)
|
|
finally:
|
|
radar_service.close()
|
|
|
|
self._log("Radar pre-configured via Python driver")
|
|
|
|
def _stop_run(self) -> None:
|
|
"""Stop acquisition-side processes and close readers as needed."""
|
|
was_running = self._supervisor.is_running()
|
|
if was_running:
|
|
self._supervisor.stop_orchestrator()
|
|
self._drain_rings_until_quiet(timeout_s=0.35, poll_s=0.02)
|
|
self._supervisor.stop_preprocessor()
|
|
self._drain_rings_until_quiet(timeout_s=0.25, poll_s=0.02)
|
|
else:
|
|
self._supervisor.stop()
|
|
self._drain_rings_once_for_history()
|
|
keep_results_reader = self._supervisor.is_processor_running()
|
|
self._close_readers(keep_results=keep_results_reader)
|
|
self._single_capture_active = False
|
|
self._single_capture_start_ns = None
|
|
self._single_capture_seen_raw = False
|
|
self._single_capture_target_collection_id = None
|
|
self._update_history_indicator()
|
|
self._status_label.setText("Status: idle")
|
|
if was_running:
|
|
if keep_results_reader:
|
|
self._log("Acquisition stopped (data_processor kept running)")
|
|
else:
|
|
self._log("Pipeline stopped")
|
|
|
|
def _stop_all_processes(self) -> None:
|
|
"""Stop all managed pipeline processes and close all readers."""
|
|
was_running = self._supervisor.is_running() or self._supervisor.is_processor_running()
|
|
self._supervisor.stop_all()
|
|
self._drain_rings_until_quiet(timeout_s=0.25, poll_s=0.02)
|
|
self._close_readers(keep_results=False)
|
|
self._single_capture_active = False
|
|
self._single_capture_start_ns = None
|
|
self._single_capture_seen_raw = False
|
|
self._single_capture_target_collection_id = None
|
|
self._update_history_indicator()
|
|
self._status_label.setText("Status: idle")
|
|
if was_running:
|
|
self._log("All pipeline processes stopped")
|
|
|
|
def _close_readers(self, *, keep_results: bool = False) -> None:
|
|
"""Close active ring readers."""
|
|
if self._raw_reader is not None:
|
|
self._raw_reader.close()
|
|
self._raw_reader = None
|
|
if self._pre_reader is not None:
|
|
self._pre_reader.close()
|
|
self._pre_reader = None
|
|
if not keep_results and self._result_reader is not None:
|
|
self._result_reader.close()
|
|
self._result_reader = None
|
|
|
|
def _poll_rings(self) -> None:
|
|
"""Poll readers, ingest history, and trigger rendering."""
|
|
for report in self._supervisor.collect_crash_reports():
|
|
self._status_label.setText("Status: error")
|
|
self._log(report)
|
|
|
|
try:
|
|
if self._raw_reader is not None:
|
|
self._read_all_raw()
|
|
self._read_all_preprocessed()
|
|
result_latest = self._read_all_results() if self._result_reader is not None else None
|
|
self._update_history_indicator()
|
|
|
|
if self._single_capture_active:
|
|
if self._finish_single_capture_if_ready():
|
|
return
|
|
return
|
|
|
|
self._draw_preferred_collection(result_latest=result_latest)
|
|
except Exception as exc: # noqa: BLE001
|
|
self._log(f"Reader error: {exc}")
|
|
|
|
def _finish_single_capture_if_ready(self) -> bool:
|
|
"""Finalize single capture when the exact target result becomes available."""
|
|
if not self._single_capture_active:
|
|
return False
|
|
if self._single_capture_start_ns is None:
|
|
return False
|
|
if not self._single_capture_seen_raw:
|
|
return False
|
|
target_result = self._find_single_capture_target_result()
|
|
if target_result is None:
|
|
return False
|
|
|
|
self._draw_results(target_result)
|
|
self._log("Single capture completed")
|
|
self._stop_run()
|
|
return True
|
|
|
|
def _find_single_capture_target_result(self) -> ResultCollection | None:
|
|
"""Return the exact result collection corresponding to the captured target raw sweep."""
|
|
if self._single_capture_target_collection_id is None:
|
|
return None
|
|
if self._single_capture_start_ns is None:
|
|
return None
|
|
|
|
for collection in reversed(self._result_history):
|
|
if collection.collection_id != self._single_capture_target_collection_id:
|
|
continue
|
|
if collection.monotonic_ns < self._single_capture_start_ns:
|
|
continue
|
|
if not self._result_collection_has_trace(collection):
|
|
continue
|
|
return collection
|
|
|
|
return None
|
|
|
|
def _read_all_raw(self) -> SweepCollection | None:
|
|
"""Read available raw collections from raw ring."""
|
|
assert self._raw_reader is not None
|
|
latest: SweepCollection | None = None
|
|
for _ in range(self._max_pop_per_poll):
|
|
collection = self._raw_reader.pop_raw_collection()
|
|
if collection is None:
|
|
break
|
|
self._raw_history.append(collection)
|
|
latest = collection
|
|
if self._single_capture_active and self._single_capture_start_ns is not None:
|
|
if collection.monotonic_ns >= self._single_capture_start_ns:
|
|
self._single_capture_seen_raw = True
|
|
if self._single_capture_target_collection_id is None:
|
|
self._single_capture_target_collection_id = collection.collection_id
|
|
return latest
|
|
|
|
def _read_all_preprocessed(self) -> None:
|
|
"""Read available preprocessed collections from preprocessed ring."""
|
|
if self._pre_reader is None:
|
|
return
|
|
|
|
for _ in range(self._max_pop_per_poll):
|
|
collection = self._pre_reader.pop_preprocessed_collection()
|
|
if collection is None:
|
|
break
|
|
self._pre_history.append(collection)
|
|
|
|
def _read_all_results(self) -> ResultCollection | None:
|
|
"""Read available result collections from results ring."""
|
|
assert self._result_reader is not None
|
|
latest: ResultCollection | None = None
|
|
for _ in range(self._max_pop_per_poll):
|
|
collection = self._result_reader.pop_result_collection()
|
|
if collection is None:
|
|
break
|
|
if record_result_history(self._result_history, collection):
|
|
latest = collection
|
|
return latest
|
|
|
|
def _drain_rings_once_for_history(self) -> None:
|
|
"""Perform one non-blocking read pass to extend histories."""
|
|
if self._raw_reader is not None:
|
|
self._read_all_raw()
|
|
self._read_all_preprocessed()
|
|
if self._result_reader is not None:
|
|
self._read_all_results()
|
|
|
|
def _drain_rings_until_quiet(self, *, timeout_s: float, poll_s: float) -> None:
|
|
"""Drain rings until history sizes stabilize or timeout expires."""
|
|
deadline = time.monotonic() + timeout_s
|
|
stable_rounds = 0
|
|
previous = (
|
|
len(self._raw_history),
|
|
len(self._pre_history),
|
|
len(self._result_history),
|
|
)
|
|
|
|
while time.monotonic() < deadline and stable_rounds < 2:
|
|
self._drain_rings_once_for_history()
|
|
current = (
|
|
len(self._raw_history),
|
|
len(self._pre_history),
|
|
len(self._result_history),
|
|
)
|
|
if current == previous:
|
|
stable_rounds += 1
|
|
else:
|
|
stable_rounds = 0
|
|
previous = current
|
|
time.sleep(poll_s)
|
|
|
|
def _drain_results_until_quiet(self, *, timeout_s: float, poll_s: float) -> None:
|
|
"""Drain only results ring until size stabilizes or timeout expires."""
|
|
if self._result_reader is None:
|
|
return
|
|
|
|
deadline = time.monotonic() + timeout_s
|
|
stable_rounds = 0
|
|
|
|
while time.monotonic() < deadline and stable_rounds < 2:
|
|
latest = self._read_all_results()
|
|
if latest is None:
|
|
stable_rounds += 1
|
|
else:
|
|
stable_rounds = 0
|
|
time.sleep(poll_s)
|
|
|
|
def _update_history_indicator(self) -> None:
|
|
"""Update UI label with current history buffer sizes."""
|
|
self._history_label.setText(
|
|
f"History: raw={len(self._raw_history)}, "
|
|
f"preprocessed={len(self._pre_history)}, "
|
|
f"results={len(self._result_history)}"
|
|
)
|
|
|
|
def _reset_runtime_history(self) -> None:
|
|
"""Reset runtime history and B-scan caches."""
|
|
self._replace_runtime_history(retained_raw=[], retained_pre=[], retained_result=[])
|
|
self._bscan_history_floor_collection_id = 0
|
|
self._clear_history_mode_caches()
|
|
self._update_history_indicator()
|
|
|
|
def _replace_runtime_history(
|
|
self,
|
|
*,
|
|
retained_raw: list[SweepCollection],
|
|
retained_pre: list[SweepCollection],
|
|
retained_result: list[ResultCollection],
|
|
) -> None:
|
|
"""Replace history deques with provided retained tails."""
|
|
raw_tail = retained_raw[-self._raw_history.maxlen :] if self._raw_history.maxlen is not None else retained_raw
|
|
pre_tail = retained_pre[-self._pre_history.maxlen :] if self._pre_history.maxlen is not None else retained_pre
|
|
result_tail = (
|
|
retained_result[-self._result_history.maxlen :]
|
|
if self._result_history.maxlen is not None
|
|
else retained_result
|
|
)
|
|
|
|
self._raw_history.clear()
|
|
self._pre_history.clear()
|
|
self._result_history.clear()
|
|
self._raw_history.extend(raw_tail)
|
|
self._pre_history.extend(pre_tail)
|
|
self._result_history.extend(result_tail)
|
|
|
|
def _build_run_history_signature(self, config: RunConfigModel) -> tuple[object, ...]:
|
|
"""Build signature used to decide when history should be reset."""
|
|
return build_run_history_signature(config)
|
|
|
|
def _validate_processing_mode_constraints(self, config: RunConfigModel) -> None:
|
|
"""Validate processing-mode constraints for run start."""
|
|
validate_processing_mode_constraints(
|
|
self._processing_mode.currentText(),
|
|
config,
|
|
self._live_processing_config(),
|
|
)
|