Ayzen
d283ecfdf1
The processor appended sweeps the GUI fed back to its own history, so its next replay published them twice and out of order, and the B-scan came out with rotated columns. Fed-back sweeps now stay out of that history, results are kept in timestamp order, picking the displayed combo only redraws and only new sweeps are recorded to disk.
738 lines
35 KiB
Python
738 lines
35 KiB
Python
"""Pipeline runtime lifecycle mixin for the main GUI window."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from PyQt6.QtCore import QCoreApplication, QEventLoop
|
|
|
|
from python_app.gui.runtime.constraints import validate_processing_mode_constraints
|
|
from python_app.gui.runtime.history import (
|
|
build_processor_run_signature,
|
|
build_run_history_signature,
|
|
record_result_history,
|
|
)
|
|
from python_app.hardware_full.kamil_adc import apply_kamil_adc_laser_control
|
|
from python_app.hardware_full.single_radar_service import create_single_radar_service
|
|
from python_app.models.dataset_model import ComboKey, ResultCollection, SweepCollection
|
|
from python_app.models.run_config_model import RunConfigModel
|
|
from python_app.orchestration.preprocess_assets import (
|
|
PREPROCESS_ASSET_SPECS,
|
|
REQUIRED_PREPROCESS_ASSET_KEYS,
|
|
preprocess_asset_model,
|
|
runtime_preprocess_asset_keys,
|
|
)
|
|
from python_app.orchestration.restart_policy import RestartPolicy
|
|
from python_app.orchestration.shm_reader import ShmRingReader
|
|
|
|
|
|
class AppWindowPipelineMixin:
|
|
"""Controls start/stop, readers, and periodic polling of pipeline rings."""
|
|
|
|
# A repeated identical reader error is deduped from the log, but must still be
|
|
# re-logged every this-many polls so a persistent failure stays visible.
|
|
_READER_ERROR_RELOG_EVERY = 200
|
|
# After this many consecutive identical reader errors, attempt one reader
|
|
# reconnect; if it keeps failing past the next threshold, stop the pipeline so
|
|
# a wedged reader cannot stay silently broken forever.
|
|
_READER_ERROR_RECONNECT_AT = 40
|
|
_READER_ERROR_STOP_AT = 400
|
|
# If a pipeline child exits unexpectedly while the run should be live, relaunch
|
|
# the whole pipeline from the last-written runtime config. It retries FOREVER with
|
|
# capped back-off (this is an unattended appliance — it must keep trying to come
|
|
# back, never permanently stop); the failure streak resets once a healthy poll sees
|
|
# data. The "if it breaks it comes back up" contract holds in BOTH GUI and headless.
|
|
_RESTART_POLICY = RestartPolicy(min_interval_s=3.0, max_interval_s=60.0)
|
|
# Safety backstop so a wedged/dropping processor cannot leave a single capture
|
|
# polling forever with no completion and no error. Generous, not a tight deadline.
|
|
_SINGLE_CAPTURE_TIMEOUT_S = 300.0
|
|
|
|
def _processor_requires_restart(self, run_signature: tuple[object, ...]) -> bool:
|
|
"""Return whether alive `data_processor` was started with different stable run settings."""
|
|
return self._supervisor.is_processor_running() and self._processor_run_signature != run_signature
|
|
|
|
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",
|
|
details=self._capture_state_details(),
|
|
)
|
|
return
|
|
if self._supervisor.is_running():
|
|
# A single-shot capture from a running continuous pipeline must
|
|
# restart acquisition with `runtime.continuous=false`; refusing
|
|
# here would leave the previous run streaming and the user could
|
|
# never reach the single-capture termination state.
|
|
if not single_capture:
|
|
self._show_error(
|
|
"Pipeline is already running",
|
|
details=self._process_state_details(),
|
|
)
|
|
return
|
|
self._log("Stopping continuous pipeline before single capture")
|
|
self._stop_run()
|
|
|
|
try:
|
|
processor_was_running = self._supervisor.is_processor_running()
|
|
config = self._build_config()
|
|
run_signature = self._build_run_history_signature(config)
|
|
processor_signature = self._build_processor_run_signature(config)
|
|
if self._processor_requires_restart(processor_signature):
|
|
self._log("Restarting data_processor because stable run settings changed")
|
|
self._stop_all_processes()
|
|
processor_was_running = False
|
|
if not processor_was_running:
|
|
self._reset_runtime_history()
|
|
|
|
self._validate_processing_mode_constraints(config)
|
|
radar_key = self._radar_key(config)
|
|
|
|
missing_assets = [
|
|
PREPROCESS_ASSET_SPECS[key].display_name
|
|
for key in REQUIRED_PREPROCESS_ASSET_KEYS
|
|
if not preprocess_asset_model(config, key).set_name
|
|
]
|
|
if missing_assets:
|
|
raise RuntimeError(
|
|
"Select all required preprocess sets in Preprocessing Panel before Start: "
|
|
+ ", ".join(missing_assets)
|
|
)
|
|
|
|
combo_keys = [ComboKey(input=combo.input, output=combo.output) for combo in config.combos]
|
|
active_preprocess_keys = runtime_preprocess_asset_keys(config)
|
|
preprocess_summary = "; ".join(
|
|
f"{PREPROCESS_ASSET_SPECS[key].display_name}={preprocess_asset_model(config, key).set_name}"
|
|
for key in active_preprocess_keys
|
|
)
|
|
self._log(f"Active preprocess assets for run: {preprocess_summary}")
|
|
|
|
for key in active_preprocess_keys:
|
|
spec = PREPROCESS_ASSET_SPECS[key]
|
|
asset = preprocess_asset_model(config, key)
|
|
if not self._store.has_combo_coverage(spec.set_kind, radar_key, asset.set_name, combo_keys):
|
|
raise RuntimeError(f"Selected {spec.display_name} set does not cover requested run combos")
|
|
|
|
self._config_writer.prepare_preprocess_bundles(self._store, radar_key, config)
|
|
config.runtime.continuous = not single_capture
|
|
|
|
if not single_capture and not config.is_kamil_adc:
|
|
self._prepare_radar_for_native_acquisition(config)
|
|
|
|
config_path = self._config_writer.write(config, self._project_root / "python_app/runtime/run_config.json")
|
|
combo_preview = ", ".join(f"in{combo.input}/out{combo.output}" for combo in config.combos[:6])
|
|
if len(config.combos) > 6:
|
|
combo_preview += ", ..."
|
|
self._log(
|
|
f"Starting pipeline: mode={'single_capture' if single_capture else 'continuous'}, "
|
|
f"config={config_path}, combos={len(config.combos)}"
|
|
f"{', ' + combo_preview if combo_preview else ''}, radar_key={radar_key}"
|
|
)
|
|
|
|
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
|
|
|
|
# Open the single-capture acquisition window BEFORE the producer can emit:
|
|
# any collection the one-shot orchestrator publishes is then strictly newer
|
|
# than this timestamp and is recognized as the target. Reading the clock
|
|
# AFTER start() raced a fast (simulator) orchestrator that had already
|
|
# published — its collection.monotonic_ns < start_ns, so the capture never
|
|
# completed and the GUI hung.
|
|
single_capture_start_ns = time.monotonic_ns() if single_capture else None
|
|
|
|
# `data_preprocessor` is about to become the owner of the preprocessed ring
|
|
# again, so the GUI must stop holding a writer into it. A queued replay would
|
|
# otherwise interleave with the real producer.
|
|
self._bscan_reprocess_timer.stop()
|
|
self._close_replay_ring_writer()
|
|
self._discard_bscan_replay_results()
|
|
|
|
self._supervisor.start(config_path, allow_clean_orchestrator_exit=single_capture)
|
|
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._processor_run_signature = processor_signature
|
|
self._single_capture_active = single_capture
|
|
self._single_capture_start_ns = single_capture_start_ns
|
|
self._single_capture_seen_raw = False
|
|
self._single_capture_target_collection_id = None
|
|
|
|
if not single_capture:
|
|
# Continuous start: discard any stale tail so rendering begins fresh.
|
|
# Single capture must NOT drop here — the drop races with (and can
|
|
# discard) the one-shot result; it instead filters stale data by the
|
|
# start timestamp recorded above.
|
|
self._drop_pending_ring_payloads(include_results=True)
|
|
self._last_reader_error_signature = None
|
|
self._reader_error_repeat_count = 0
|
|
# Record what to relaunch if a child later dies unexpectedly. Only a
|
|
# continuous run auto-restarts; a single capture is bounded by its deadline.
|
|
self._active_run_config = config
|
|
self._active_run_config_path = config_path
|
|
self._pipeline_should_run = not single_capture
|
|
self._pipeline_restart_count = 0
|
|
|
|
pid_map = self._supervisor.pids()
|
|
pid_text = ", ".join(f"{name}={pid}" for name, pid in sorted(pid_map.items())) or "none"
|
|
if single_capture:
|
|
self._status_label.setText("Status: single capture running")
|
|
self._log(f"Single capture started; managed processes: {pid_text}")
|
|
else:
|
|
self._status_label.setText("Status: running")
|
|
self._log(f"Pipeline started; managed processes: {pid_text}")
|
|
except Exception as exc: # noqa: BLE001
|
|
self._single_capture_active = False
|
|
self._single_capture_start_ns = None
|
|
self._stop_all_processes()
|
|
self._show_exception("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",
|
|
details=self._capture_state_details(),
|
|
)
|
|
return
|
|
|
|
was_running = self._supervisor.is_running()
|
|
processor_only_running = self._supervisor.is_processor_running() and not was_running
|
|
if was_running:
|
|
self._stop_run()
|
|
|
|
try:
|
|
config = self._build_config()
|
|
if config.radar.driver_mode == "native":
|
|
self._refresh_radar_limits_from_device()
|
|
processor_signature = self._build_processor_run_signature(config)
|
|
if processor_only_running and self._processor_requires_restart(processor_signature):
|
|
self._stop_all_processes()
|
|
self._reset_runtime_history()
|
|
self._history_run_signature = None
|
|
self._log(
|
|
"Stable run settings changed; stopped data_processor because cached history no longer "
|
|
"matches the current config"
|
|
)
|
|
self._prepare_radar_for_native_acquisition(config)
|
|
points_text = (
|
|
"points=from Kamil ADC stream"
|
|
if config.is_kamil_adc
|
|
else f"points={config.radar.sweep.points}"
|
|
)
|
|
self._log(
|
|
"Radar settings applied: "
|
|
f"start={config.radar.sweep.start_hz:g} Hz, "
|
|
f"stop={config.radar.sweep.stop_hz:g} Hz, "
|
|
f"{points_text}, "
|
|
f"ifbw={config.radar.sweep.if_bandwidth_hz:g} Hz, "
|
|
f"power={config.radar.sweep.power_dbm:g} dBm"
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
self._show_exception("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 single-radar hardware using current sweep settings."""
|
|
if config.radar.model == RunConfigModel.COMPACT_M_K209_MODEL and config.radar.driver_mode != "native":
|
|
raise RuntimeError("Compact-M K209 requires radar.driver_mode='native'")
|
|
if config.radar.driver_mode != "native":
|
|
self._log("Radar pre-configuration skipped (mock mode)")
|
|
return
|
|
if config.is_multi_device:
|
|
self._log("Multi-device raw producer will configure all LibreVNA devices")
|
|
return
|
|
if config.is_matrix_radar:
|
|
self._log("Matrix raw producer will configure the matrix radar")
|
|
return
|
|
if config.is_kamil_adc:
|
|
if apply_kamil_adc_laser_control(config):
|
|
self._log("Kamil ADC laser_control applied via Apply Radar")
|
|
else:
|
|
self._log("Kamil ADC laser_control skipped because it is disabled")
|
|
return
|
|
|
|
radar_service = create_single_radar_service(config)
|
|
if not getattr(radar_service, "driver_available", True):
|
|
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(f"Radar pre-configured via Python driver: model={config.radar.model}")
|
|
|
|
def _stop_run(self) -> None:
|
|
"""Stop acquisition-side processes and close readers as needed."""
|
|
self._pipeline_should_run = False # an explicit stop disables crash auto-restart
|
|
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()
|
|
# Flush any in-progress disk recording so its buffered (not-yet-full) chunk is
|
|
# written rather than lost; the drains above already captured the last data.
|
|
self._finalize_recording_on_stop()
|
|
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."""
|
|
self._pipeline_should_run = False # an explicit stop disables crash auto-restart
|
|
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._processor_run_signature = 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; keep the results reader when ``keep_results``."""
|
|
closed = []
|
|
if self._raw_reader is not None:
|
|
self._raw_reader.close()
|
|
self._raw_reader = None
|
|
closed.append("raw")
|
|
if self._pre_reader is not None:
|
|
self._pre_reader.close()
|
|
self._pre_reader = None
|
|
closed.append("preprocessed")
|
|
if not keep_results and self._result_reader is not None:
|
|
self._result_reader.close()
|
|
self._result_reader = None
|
|
closed.append("results")
|
|
if closed:
|
|
self._log_debug(f"Closed ring readers: {', '.join(closed)}.")
|
|
|
|
def _poll_rings(self) -> None:
|
|
"""Poll readers, ingest history, and trigger rendering."""
|
|
self._web_update_snapshot() # guarded internally; never raises
|
|
self._handle_process_exit_reports()
|
|
|
|
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._poll_recording_writer() # finalize a disk recording once its writer drains
|
|
self._update_history_indicator()
|
|
self._last_reader_error_signature = None
|
|
self._reader_error_repeat_count = 0
|
|
|
|
if self._single_capture_active:
|
|
if self._finish_single_capture_if_ready():
|
|
return
|
|
self._check_single_capture_deadline()
|
|
return
|
|
|
|
if result_latest is not None:
|
|
# Genuine data flowed: the pipeline is healthy, so reset the
|
|
# crash-storm budget (it only caps consecutive crash-restarts).
|
|
self._pipeline_restart_count = 0
|
|
render_started_ns = time.monotonic_ns()
|
|
self._draw_preferred_collection(result_latest=result_latest)
|
|
self._pipeline_metrics.record(
|
|
"rendering", time.monotonic_ns() - render_started_ns
|
|
)
|
|
else:
|
|
self._draw_preferred_collection(result_latest=None)
|
|
except Exception as exc: # noqa: BLE001
|
|
self._handle_reader_poll_error(exc)
|
|
|
|
def _handle_process_exit_reports(self) -> None:
|
|
"""Log child exits and auto-restart the pipeline on an unexpected death.
|
|
|
|
Runs first in the poll tick and is fully guarded: it must never raise, or
|
|
it would abort the Qt slot. An unexpected (non-clean) exit while the run is
|
|
meant to be live triggers a bounded relaunch — in both GUI and headless.
|
|
"""
|
|
unexpected = False
|
|
try:
|
|
for report in self._supervisor.collect_exit_reports():
|
|
if report.level == "INFO":
|
|
self._log(report.format())
|
|
continue
|
|
self._status_label.setText("Status: error")
|
|
self._log_error(report.format())
|
|
unexpected = True
|
|
except Exception as exc: # noqa: BLE001 - the poll tick must survive this
|
|
self._log_exception("Failed to collect process exit reports", exc, level="ERROR")
|
|
return
|
|
|
|
if unexpected and getattr(self, "_pipeline_should_run", False):
|
|
self._recover_pipeline_after_crash()
|
|
|
|
def _recover_pipeline_after_crash(self) -> None:
|
|
"""Relaunch the pipeline after an unexpected child exit (GUI and headless).
|
|
|
|
Re-spawns from the already-written runtime config — no widgets, no dialogs,
|
|
no re-validation — so it is safe to call from the poll tick. Retries forever
|
|
with capped back-off (never gives up); a healthy poll resets the failure streak.
|
|
"""
|
|
now = time.monotonic()
|
|
consecutive_failures = getattr(self, "_pipeline_restart_count", 0)
|
|
if not self._RESTART_POLICY.should_restart_now(
|
|
now_s=now,
|
|
last_restart_s=getattr(self, "_last_pipeline_restart_s", 0.0),
|
|
consecutive_failures=consecutive_failures,
|
|
):
|
|
return # still inside the current back-off window; let it settle
|
|
self._last_pipeline_restart_s = now
|
|
self._pipeline_restart_count = consecutive_failures + 1
|
|
|
|
config = getattr(self, "_active_run_config", None)
|
|
config_path = getattr(self, "_active_run_config_path", None)
|
|
if config is None or config_path is None:
|
|
self._pipeline_should_run = False
|
|
self._log_error("Cannot auto-restart pipeline: no active run configuration recorded.")
|
|
return
|
|
|
|
self._log_error(
|
|
"Pipeline process exited unexpectedly; restarting "
|
|
f"(attempt {self._pipeline_restart_count}, next back-off "
|
|
f"{self._RESTART_POLICY.backoff_for(self._pipeline_restart_count):.0f}s)."
|
|
)
|
|
try:
|
|
# Note: do NOT drain rings here — draining pumps the Qt event loop, which
|
|
# would re-enter _poll_rings mid-restart (and reset the crash-storm count).
|
|
self._supervisor.stop_all()
|
|
self._close_readers(keep_results=False)
|
|
self._supervisor.start(config_path, allow_clean_orchestrator_exit=False)
|
|
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._processor_run_signature = self._build_processor_run_signature(config)
|
|
self._drop_pending_ring_payloads(include_results=True)
|
|
self._status_label.setText("Status: running")
|
|
self._log("Pipeline auto-restarted after crash.")
|
|
except Exception as exc: # noqa: BLE001 - retry on the next crash signal
|
|
self._log_exception("Pipeline auto-restart failed; will retry", exc, level="ERROR")
|
|
|
|
def _check_single_capture_deadline(self) -> None:
|
|
"""Fail a single capture that never completes so it cannot hang forever."""
|
|
start = self._single_capture_start_ns
|
|
if start is None:
|
|
return
|
|
if time.monotonic_ns() - start <= int(self._SINGLE_CAPTURE_TIMEOUT_S * 1e9):
|
|
return
|
|
self._log_error(
|
|
f"Single capture timed out after {self._SINGLE_CAPTURE_TIMEOUT_S:.0f}s "
|
|
"with no result; stopping."
|
|
)
|
|
self._stop_run()
|
|
|
|
def _handle_reader_poll_error(self, exc: Exception) -> None:
|
|
"""Surface a reader-poll failure without spamming the log.
|
|
|
|
Distinct errors log once; an identical recurring error is deduped from the
|
|
log but still drives the status label to error, is periodically re-logged,
|
|
and after escalating thresholds triggers a reader reconnect and finally a
|
|
pipeline stop so a wedged reader cannot fail silently forever.
|
|
"""
|
|
signature = (type(exc).__name__, str(exc))
|
|
# Always reflect a reader failure in the status label, even when deduped.
|
|
self._status_label.setText("Status: error")
|
|
|
|
if self._last_reader_error_signature == signature:
|
|
self._reader_error_repeat_count = getattr(self, "_reader_error_repeat_count", 0) + 1
|
|
else:
|
|
self._last_reader_error_signature = signature
|
|
self._reader_error_repeat_count = 0
|
|
self._log_exception("Reader poll failed", exc, level="ERROR")
|
|
|
|
repeats = self._reader_error_repeat_count
|
|
# Periodically re-log a persistent identical failure so it stays visible.
|
|
if repeats and repeats % self._READER_ERROR_RELOG_EVERY == 0:
|
|
self._log_exception(
|
|
f"Reader poll still failing (repeat #{repeats})", exc, level="ERROR"
|
|
)
|
|
|
|
if repeats >= self._READER_ERROR_STOP_AT:
|
|
# The reader stayed wedged through a reconnect attempt; stop the
|
|
# pipeline so the failure is unmistakable instead of an endless retry.
|
|
self._log_error(
|
|
f"Stopping pipeline after {repeats} consecutive reader poll failures"
|
|
)
|
|
self._reader_error_repeat_count = 0
|
|
self._last_reader_error_signature = None
|
|
self._stop_all_processes()
|
|
elif repeats == self._READER_ERROR_RECONNECT_AT:
|
|
self._reconnect_readers_after_error()
|
|
|
|
def _reconnect_readers_after_error(self) -> None:
|
|
"""Re-open active ring readers in place to recover from a wedged reader."""
|
|
self._log_warning("Attempting ring reader reconnect after repeated poll failures")
|
|
try:
|
|
for attr in ("_raw_reader", "_pre_reader", "_result_reader"):
|
|
reader = getattr(self, attr)
|
|
if reader is None:
|
|
continue
|
|
ring_name = reader._ring_name # noqa: SLF001 - reuse the reader's own ring name
|
|
reader.close()
|
|
setattr(self, attr, ShmRingReader(ring_name))
|
|
# A successful reconnect clears the error state so the next failure
|
|
# logs fresh rather than being swallowed by the stale signature.
|
|
self._last_reader_error_signature = None
|
|
self._reader_error_repeat_count = 0
|
|
self._status_label.setText("Status: running")
|
|
self._log("Ring readers reconnected after repeated poll failures")
|
|
except Exception as exc: # noqa: BLE001
|
|
# Leave the error signature intact so escalation to a stop still fires.
|
|
self._log_exception("Ring reader reconnect failed", exc, level="ERROR")
|
|
|
|
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."""
|
|
if self._raw_reader is None:
|
|
raise RuntimeError("Raw ring reader is not initialised")
|
|
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)
|
|
self._record_collection("raw", collection)
|
|
latest = collection
|
|
# `capture_*_ns` are populated by the C++ sweep_orchestrator with
|
|
# wallclocks captured around the actual device read. Pre-orchestrator
|
|
# producers leave them zero, in which case PipelineMetrics drops it.
|
|
acquisition_ns = int(collection.capture_end_ns) - int(collection.capture_start_ns)
|
|
self._pipeline_metrics.record("acquisition", acquisition_ns)
|
|
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)
|
|
self._record_collection("preprocessed", collection)
|
|
|
|
def _read_all_results(self) -> ResultCollection | None:
|
|
"""Read available result collections from results ring."""
|
|
if self._result_reader is None:
|
|
raise RuntimeError("Result ring reader is not initialised")
|
|
latest: ResultCollection | None = None
|
|
for _ in range(self._max_pop_per_poll):
|
|
collection = self._result_reader.pop_result_collection()
|
|
if collection is None:
|
|
break
|
|
self._pipeline_metrics.record("processing", int(collection.processing_duration_ns))
|
|
# Only a new sweep is a new measurement. Any other result is the processor
|
|
# reprocessing retained sweeps: recording it would duplicate a measurement
|
|
# or put one out of order.
|
|
if record_result_history(self._result_history, collection):
|
|
self._record_collection("results", collection)
|
|
latest = collection
|
|
return latest
|
|
|
|
def _pump_events_during_drain(self, pause_s: float) -> None:
|
|
"""Yield to the Qt event loop for `pause_s` instead of blocking on time.sleep.
|
|
|
|
The bounded drain loops run on the GUI thread; a raw time.sleep here freezes
|
|
the event loop, stalling the keepalive/headless-watchdog timers and starving
|
|
queued signals. Pumping events keeps the daemon responsive while we wait.
|
|
"""
|
|
app = QCoreApplication.instance()
|
|
if app is None:
|
|
# No event loop (e.g. unit context); fall back to a plain short sleep.
|
|
time.sleep(pause_s)
|
|
return
|
|
deadline = time.monotonic() + pause_s
|
|
while True:
|
|
remaining_ms = int((deadline - time.monotonic()) * 1000)
|
|
if remaining_ms <= 0:
|
|
break
|
|
app.processEvents(QEventLoop.ProcessEventsFlag.AllEvents, remaining_ms)
|
|
# processEvents returns immediately when the queue empties; sleep the
|
|
# residual in tiny slices so we neither busy-spin nor block too long.
|
|
time.sleep(min(0.002, max(0.0, deadline - time.monotonic())))
|
|
|
|
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
|
|
# Event-loop-friendly wait so timers/signals keep firing during drain.
|
|
self._pump_events_during_drain(poll_s)
|
|
|
|
def _drain_results_until_quiet(self, *, timeout_s: float, poll_s: float) -> ResultCollection | None:
|
|
"""Drain results until at least one result arrives and the ring becomes quiet."""
|
|
if self._result_reader is None:
|
|
return None
|
|
|
|
deadline = time.monotonic() + timeout_s
|
|
stable_rounds = 0
|
|
latest_seen: ResultCollection | None = None
|
|
|
|
while time.monotonic() < deadline and (latest_seen is None or stable_rounds < 2):
|
|
latest = self._read_all_results()
|
|
if latest is None:
|
|
if latest_seen is not None:
|
|
stable_rounds += 1
|
|
else:
|
|
latest_seen = latest
|
|
stable_rounds = 0
|
|
# Event-loop-friendly wait so timers/signals keep firing during drain.
|
|
self._pump_events_during_drain(poll_s)
|
|
return latest_seen
|
|
|
|
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._discard_bscan_replay_results()
|
|
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 display history should be reset."""
|
|
return build_run_history_signature(config)
|
|
|
|
def _build_processor_run_signature(self, config: RunConfigModel) -> tuple[object, ...]:
|
|
"""Build signature used to decide when the data_processor must be restarted."""
|
|
return build_processor_run_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(),
|
|
)
|
|
|