774 lines
38 KiB
Python
774 lines
38 KiB
Python
"""Own pipeline processes, shared ingestion, recovery, and exclusive operations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from contextlib import AbstractContextManager, nullcontext
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import time
|
|
|
|
from PyQt6.QtCore import QCoreApplication, QEventLoop, QTimer
|
|
|
|
from python_app.gui.controllers.bscan_replay import BscanReplay
|
|
from python_app.gui.controllers.history_controller import HistoryController
|
|
from python_app.gui.controllers.sections.views import PrimaryView
|
|
from python_app.gui.controllers.plot_controller import PlotController
|
|
from python_app.gui.controllers.recording_controller import RecordingController
|
|
from python_app.gui.controllers.renderers.sar3d import SAR3D_STATES, sar3d_status
|
|
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
|
|
from python_app.gui.runtime.pipeline_pause import PipelinePause
|
|
from python_app.gui.settings_defaults import env_enabled
|
|
from python_app.hardware_full.rph_radar.laser import apply_rph_radar_laser_control
|
|
from python_app.hardware_full.single_radar_service import create_single_radar_service
|
|
from python_app.models.collection_identity import collection_key, index_collections
|
|
from python_app.models.dataset_model import ComboKey, ResultCollection, SweepCollection
|
|
from python_app.models.run_config_schema import RunConfigModel
|
|
from python_app.models.run_config_validation import validate_acquisition_config
|
|
from python_app.orchestration.config_writer import ConfigWriter
|
|
from python_app.orchestration.live_processing_config import ProcessingLiveConfig
|
|
from python_app.orchestration.pipeline_metrics import PipelineMetrics
|
|
from python_app.orchestration.preprocess_assets import (
|
|
PREPROCESS_ASSET_NAMES, REQUIRED_PREPROCESS_ASSET_KEYS,
|
|
preprocess_asset_model, runtime_preprocess_asset_keys,
|
|
)
|
|
from python_app.orchestration.process_supervisor import ProcessSupervisor
|
|
from python_app.orchestration.restart_policy import RestartPolicy
|
|
from python_app.orchestration.shm.ring_reader import ShmRingReader
|
|
from python_app.storage.npz.store import NpzStore
|
|
from python_app.workflows.object_calibration import validate_object_set
|
|
|
|
logger = logging.getLogger(__name__)
|
|
Readers = tuple[ShmRingReader | None, ShmRingReader | None, ShmRingReader | None]
|
|
|
|
|
|
class PipelineController:
|
|
_READER_ERROR_RELOG_EVERY = 200
|
|
_READER_ERROR_RECONNECT_AT = 40
|
|
_READER_ERROR_STOP_AT = 400
|
|
_RESTART_POLICY = RestartPolicy(min_interval_s=3.0, max_interval_s=60.0)
|
|
SINGLE_CAPTURE_TIMEOUT_S = 300.0
|
|
|
|
def __init__(
|
|
self, project_root: Path, store: NpzStore, config_writer: ConfigWriter,
|
|
history: HistoryController, plots: PlotController, recording: RecordingController,
|
|
*, build_config: Callable[[], RunConfigModel], live_config: Callable[[], ProcessingLiveConfig],
|
|
publish_live: Callable[[], None], capture_guard: Callable[[], str | None],
|
|
refresh_radar_limits: Callable[[], bool], status: Callable[[str], None],
|
|
history_changed: Callable[[], None], poll_ui: Callable[[], None],
|
|
show_exception: Callable[[str, Exception], None], deferred_close: Callable[[], object],
|
|
metrics_every: int | None = None,
|
|
event_scope: Callable[[], AbstractContextManager[object]] = nullcontext,
|
|
) -> None:
|
|
self._project_root, self._store, self._config_writer = project_root, store, config_writer
|
|
self.history, self.plots, self.recording = history, plots, recording
|
|
self._build_config, self._live_config, self.publish_live = build_config, live_config, publish_live
|
|
self._capture_guard, self._refresh_radar_limits = capture_guard, refresh_radar_limits
|
|
self._status, self.history_changed, self._poll_ui = status, history_changed, poll_ui
|
|
self._show_exception, self._deferred_close = show_exception, deferred_close
|
|
self._event_scope = event_scope
|
|
self.supervisor = ProcessSupervisor(project_root)
|
|
self.metrics = PipelineMetrics(report_every=self._metrics_every() if metrics_every is None else metrics_every,
|
|
log_sink=logger.info)
|
|
self.timer = QTimer()
|
|
self.timer.setInterval(50)
|
|
self.timer.timeout.connect(self.poll)
|
|
self.replay = BscanReplay(history, self.reprocess_history, self._warning, logger.debug)
|
|
self._warnings: set[str] = set()
|
|
self._raw_reader: ShmRingReader | None = None
|
|
self._pre_reader: ShmRingReader | None = None
|
|
self._result_reader: ShmRingReader | None = None
|
|
# Results read anywhere (polling, Stop/pause drains, snapshots) await display;
|
|
# the next poll shows the newest acquisition, not merely the last one read.
|
|
self._display_pending = False
|
|
self.active_config: RunConfigModel | None = None
|
|
self.active_config_path: Path | None = None
|
|
self.stop_generation = 0
|
|
self.should_run = False
|
|
self._pause: PipelinePause | None = None
|
|
self._stopping = self._closing = self._close_deferred = self._stop_all_requested = False
|
|
self._processor_signature: tuple[object, ...] | None = None
|
|
self.single_capture_active = False
|
|
self._single_capture_start_ns: int | None = None
|
|
self._single_capture_target_key: tuple[int, int] | None = None
|
|
self.max_pop_per_poll, self.max_pop_per_snapshot_drain = 256, 4096
|
|
self._last_reader_error_signature: tuple[str, str] | None = None
|
|
self._reader_error_repeat_count = self._restart_count = 0
|
|
self._last_restart_s = 0.0
|
|
self._ui_errors: set[tuple[str, str, str]] = set()
|
|
|
|
def bind_actions(self, view: PrimaryView) -> None:
|
|
view.start.clicked.connect(lambda: self.start())
|
|
view.single_capture.clicked.connect(lambda: self.start(single_capture=True))
|
|
view.stop.clicked.connect(self.stop)
|
|
view.apply_radar.clicked.connect(self.apply_radar)
|
|
|
|
@staticmethod
|
|
def _metrics_every() -> int:
|
|
raw = os.environ.get("RADAR_SYSTEM_METRICS_REPORT_EVERY", "").strip()
|
|
try:
|
|
value = int(raw) if raw else 50
|
|
if value >= 1:
|
|
return value
|
|
except ValueError:
|
|
pass
|
|
logger.warning("Ignoring invalid RADAR_SYSTEM_METRICS_REPORT_EVERY=%r; using default 50.", raw)
|
|
return 50
|
|
|
|
@property
|
|
def running(self) -> bool:
|
|
return self.supervisor.is_running()
|
|
|
|
@property
|
|
def processor_running(self) -> bool:
|
|
return self.supervisor.is_processor_running()
|
|
|
|
@property
|
|
def acquisition_healthy(self) -> bool:
|
|
return {"acquisition", "data_preprocessor"}.issubset(self.supervisor.pids())
|
|
|
|
@property
|
|
def paused(self) -> bool:
|
|
return self._pause is not None
|
|
|
|
@property
|
|
def stopping(self) -> bool:
|
|
return self._stopping
|
|
|
|
@property
|
|
def closing(self) -> bool:
|
|
return self._closing
|
|
|
|
@property
|
|
def readers(self) -> Readers:
|
|
return self._raw_reader, self._pre_reader, self._result_reader
|
|
|
|
def _warning(self, text: str, *, details: str | None = None, once_key: str | None = None) -> None:
|
|
if once_key is not None:
|
|
if once_key in self._warnings:
|
|
return
|
|
self._warnings.add(once_key)
|
|
logger.warning("%s%s", text, f"\n{details}" if details else "")
|
|
|
|
def can_resume(self, generation: int) -> bool:
|
|
return not self._closing and self.stop_generation == generation
|
|
|
|
def operation_cancelled(self) -> None:
|
|
if not self._closing:
|
|
logger.info("Pipeline operation cancelled by Stop")
|
|
|
|
def schedule_startup(self) -> None:
|
|
if not env_enabled("RADAR_SYSTEM_AUTO_START"):
|
|
return
|
|
generation = self.stop_generation
|
|
|
|
def start() -> None:
|
|
if not self.can_resume(generation):
|
|
return
|
|
started = self.start()
|
|
if not started and env_enabled("RADAR_SYSTEM_HEADLESS") and self.can_resume(generation):
|
|
logger.error("Headless auto-start did not bring the pipeline up")
|
|
app = QCoreApplication.instance()
|
|
if app is not None:
|
|
app.exit(1)
|
|
|
|
def apply_then_start() -> None:
|
|
if self.can_resume(generation):
|
|
self.apply_radar()
|
|
QTimer.singleShot(1000, start)
|
|
|
|
logger.info("Auto-start requested by launcher")
|
|
QTimer.singleShot(500, apply_then_start if env_enabled("RADAR_SYSTEM_AUTO_APPLY_RADAR") else start)
|
|
|
|
def pause(self) -> PipelinePause:
|
|
if self.paused or self._stopping or self._closing:
|
|
raise RuntimeError("Wait for the current pipeline operation to finish")
|
|
generation, resume = self.stop_generation, self.should_run
|
|
|
|
def release(requested: bool) -> None:
|
|
if self._pause is operation:
|
|
self._pause = None
|
|
if requested and resume and self.can_resume(generation):
|
|
if not self.start() and self.can_resume(generation):
|
|
self.should_run = True
|
|
self._last_restart_s = time.monotonic()
|
|
self._restart_count += 1
|
|
logger.error("Pipeline resume failed; automatic restart will retry")
|
|
|
|
operation = PipelinePause(lambda: self._pause is operation and self.can_resume(generation), release)
|
|
# Ownership precedes shutdown: drain pumps Qt and may receive Stop or Close.
|
|
self._pause = operation
|
|
try:
|
|
error = self.stop_pipeline(keep_processor=True)
|
|
if error is not None:
|
|
raise error
|
|
if not operation.valid:
|
|
raise InterruptedError("Pipeline operation cancelled")
|
|
self._display_pending = False # the pausing operation owns the display
|
|
return operation
|
|
except BaseException:
|
|
operation.finish(resume=False)
|
|
raise
|
|
|
|
def _validate_assets(self, config: RunConfigModel) -> None:
|
|
missing = [PREPROCESS_ASSET_NAMES[key] for key in REQUIRED_PREPROCESS_ASSET_KEYS
|
|
if not preprocess_asset_model(config, key).set_name]
|
|
if missing:
|
|
raise RuntimeError("Select all required preprocess sets in Preprocessing Panel before Start: " + ", ".join(missing))
|
|
radar_key = config.radar_key()
|
|
combos = [ComboKey(input=combo.input, output=combo.output) for combo in config.combos]
|
|
keys = runtime_preprocess_asset_keys(config)
|
|
logger.info("Active preprocess assets for run: %s", "; ".join(
|
|
f"{PREPROCESS_ASSET_NAMES[key]}={preprocess_asset_model(config, key).set_name}" for key in keys))
|
|
for key in keys:
|
|
asset = preprocess_asset_model(config, key)
|
|
if not self._store.has_combo_coverage(key, radar_key, asset.set_name, combos):
|
|
raise RuntimeError(f"Selected {PREPROCESS_ASSET_NAMES[key]} set does not cover requested run combos")
|
|
name = config.preprocess.s21.object_calibration.set_name
|
|
if name:
|
|
collection, metadata = self._store.load_set_with_metadata("s21_object_calibration", radar_key, name)
|
|
through = self._store.load_set("s21_calibration", radar_key, config.preprocess.s21.calibration.set_name)
|
|
validate_object_set(collection, metadata, config.to_dict(), radar_key, through)
|
|
|
|
def start(self, *, single_capture: bool = False, config_override: RunConfigModel | None = None,
|
|
pause: PipelinePause | None = None) -> bool:
|
|
if self._closing:
|
|
return False
|
|
generation = self.stop_generation
|
|
starting = False
|
|
try:
|
|
if self._stopping:
|
|
raise RuntimeError("Wait for pipeline shutdown before starting again")
|
|
if config_override is None or pause is None:
|
|
reason = self._capture_guard()
|
|
if reason:
|
|
raise RuntimeError(reason)
|
|
if self._pause is not pause or (pause is not None and not pause.valid):
|
|
raise RuntimeError("Wait for the current pipeline operation to finish")
|
|
config = config_override.clone() if config_override is not None else self._build_config()
|
|
validate_acquisition_config(config)
|
|
if config_override is None:
|
|
live = self._live_config()
|
|
validate_processing_mode_constraints(live.processor_mode, config, live)
|
|
self._validate_assets(config)
|
|
run_signature, processor_signature = build_run_history_signature(config), build_processor_run_signature(config)
|
|
if self.running:
|
|
if not single_capture:
|
|
raise RuntimeError("Pipeline is already running")
|
|
starting = True
|
|
logger.info("Stopping continuous pipeline before single capture")
|
|
self.stop_pipeline(keep_processor=True)
|
|
if not self.can_resume(generation):
|
|
return False
|
|
starting = True
|
|
processor_was_running = self.processor_running
|
|
if self._processor_requires_restart(processor_signature):
|
|
logger.info("Restarting data_processor because stable run settings changed")
|
|
self.stop_all()
|
|
if not self.can_resume(generation):
|
|
return False
|
|
processor_was_running = False
|
|
self._config_writer.prepare_preprocess_bundles(self._store, config.radar_key(), config)
|
|
if not processor_was_running:
|
|
self.reset_history()
|
|
config.runtime.continuous = not single_capture
|
|
config_path = self._config_writer.write(config, self._project_root / "python_app/runtime/run_config.json")
|
|
logger.info("Starting pipeline: mode=%s, config=%s, combos=%d, radar_key=%s",
|
|
"single_capture" if single_capture else "continuous", config_path, len(config.combos), config.radar_key())
|
|
if self.history.run_signature is not None and self.history.run_signature != run_signature:
|
|
if processor_was_running:
|
|
self.reset_history()
|
|
self.history.apply_deletion("clear_all")
|
|
self.publish_live()
|
|
logger.info("History reset because run settings changed")
|
|
self.history.run_signature = run_signature
|
|
# Timestamp before spawn; a fast one-shot producer can publish inside start().
|
|
started_ns = time.monotonic_ns() if single_capture else None
|
|
self.replay.close()
|
|
self.discard_replay()
|
|
if not self.can_resume(generation) or (pause is not None and not pause.valid):
|
|
return False
|
|
self.supervisor.start(config_path, allow_clean_acquisition_exit=single_capture)
|
|
self._open_readers(config)
|
|
self._processor_signature = processor_signature
|
|
self.single_capture_active, self._single_capture_start_ns = single_capture, started_ns
|
|
self._single_capture_target_key = None
|
|
if not single_capture:
|
|
self.drop_pending()
|
|
self._last_reader_error_signature, self._reader_error_repeat_count = None, 0
|
|
self.active_config, self.active_config_path = config, config_path
|
|
self.should_run, self._restart_count = not single_capture and pause is None, 0
|
|
self._status("Status: single capture running" if single_capture else "Status: running")
|
|
logger.info("Pipeline started; managed processes: %s", self.supervisor.pids())
|
|
return self.can_resume(generation) and self.running
|
|
except Exception as exc:
|
|
# Refused actions must not stop another operation or an existing run.
|
|
if starting:
|
|
self.stop_all()
|
|
self._show_exception("Failed to start pipeline", exc)
|
|
return False
|
|
|
|
def _processor_requires_restart(self, signature: tuple[object, ...]) -> bool:
|
|
return self.processor_running and self._processor_signature != signature
|
|
|
|
def apply_radar(self) -> None:
|
|
try:
|
|
reason = self._capture_guard()
|
|
if reason:
|
|
raise RuntimeError(reason)
|
|
config = self._build_config()
|
|
validate_acquisition_config(config)
|
|
processor_only = self.processor_running and not self.running
|
|
with self.pause() as operation:
|
|
if config.radar.driver_mode == "native":
|
|
self._refresh_radar_limits()
|
|
if not operation.valid:
|
|
return
|
|
if processor_only and self._processor_requires_restart(build_processor_run_signature(config)):
|
|
self.stop_all()
|
|
if not operation.valid:
|
|
return
|
|
self.reset_history()
|
|
self.history.run_signature = None
|
|
self._prepare_radar(config)
|
|
logger.info("Radar settings applied: start=%g Hz, stop=%g Hz, points=%s, ifbw=%g Hz, power=%g dBm",
|
|
config.radar.sweep.start_hz, config.radar.sweep.stop_hz,
|
|
"from RadioPhotonic radar (RPH) stream" if config.is_rph_radar else config.radar.sweep.points,
|
|
config.radar.sweep.if_bandwidth_hz, config.radar.sweep.power_dbm)
|
|
except InterruptedError:
|
|
self.operation_cancelled()
|
|
except Exception as exc:
|
|
self._show_exception("Failed to apply radar settings", exc)
|
|
|
|
@staticmethod
|
|
def _prepare_radar(config: RunConfigModel) -> None:
|
|
if config.radar.driver_mode != "native":
|
|
logger.info("Radar pre-configuration skipped (mock mode)")
|
|
elif config.is_multi_device:
|
|
logger.info("Multi-device raw producer will configure all LibreVNA devices")
|
|
elif config.is_rph_radar:
|
|
applied = apply_rph_radar_laser_control(config)
|
|
logger.info("RadioPhotonic radar (RPH) laser_control %s", "applied via Apply Radar" if applied else "disabled")
|
|
else:
|
|
service = create_single_radar_service(config)
|
|
try:
|
|
service.open()
|
|
service.configure(config.radar.sweep)
|
|
finally:
|
|
service.close()
|
|
logger.info("Radar pre-configured via Python driver: model=%s", config.radar.model)
|
|
|
|
def stop(self) -> None:
|
|
self.stop_generation += 1
|
|
self.stop_pipeline(keep_processor=True)
|
|
|
|
def stop_all(self) -> None:
|
|
self.stop_pipeline(keep_processor=False)
|
|
|
|
def stop_pipeline(self, *, keep_processor: bool) -> Exception | None:
|
|
self.should_run = False
|
|
self.replay.close()
|
|
if self._stopping:
|
|
self._stop_all_requested |= not keep_processor
|
|
return
|
|
self._stopping, self._stop_all_requested = True, not keep_processor
|
|
failure: tuple[str, Exception] | None = None
|
|
try:
|
|
drain_error: Exception | None = None
|
|
try:
|
|
if keep_processor and self.running:
|
|
self.supervisor.stop_acquisition()
|
|
self.drain_rings(timeout_s=0.35, poll_s=0.02)
|
|
self.supervisor.stop_preprocessor()
|
|
self.drain_rings(timeout_s=0.25, poll_s=0.02)
|
|
elif keep_processor:
|
|
self.supervisor.stop()
|
|
self.drain_once()
|
|
if self._stop_all_requested:
|
|
self.supervisor.stop_all()
|
|
self.drain_rings(timeout_s=0.25, poll_s=0.02)
|
|
except Exception as exc:
|
|
drain_error = exc
|
|
self.stop_generation += 1
|
|
(self.supervisor.stop_all if self._stop_all_requested else self.supervisor.stop)()
|
|
if self._stop_all_requested:
|
|
self._processor_signature = None
|
|
self.recording.finish(reason="pipeline pause" if self.paused else "pipeline closed" if self._closing else "pipeline stopped")
|
|
keep_results = not self._stop_all_requested and self.processor_running
|
|
ring_name = self._result_reader.name if self._result_reader is not None else None
|
|
self._close_readers(keep_results=keep_results and drain_error is None)
|
|
self.single_capture_active, self._single_capture_start_ns, self._single_capture_target_key = False, None, None
|
|
self.history_changed()
|
|
self._status("Status: idle")
|
|
logger.info("Acquisition stopped (data_processor kept running)" if keep_results else "Pipeline stopped")
|
|
if drain_error is not None:
|
|
self._status("Status: error")
|
|
failure = "Pipeline stopped, but final ring data could not be read", drain_error
|
|
if keep_results and ring_name and not self._closing:
|
|
self._result_reader = ShmRingReader(ring_name)
|
|
except Exception as exc:
|
|
self.stop_generation += 1
|
|
self._status("Status: error")
|
|
failure = "Failed to stop pipeline", exc
|
|
finally:
|
|
self._stopping = False
|
|
if self._close_deferred:
|
|
self._close_deferred = False
|
|
self._deferred_close()
|
|
if failure is not None:
|
|
if self.paused or self._closing:
|
|
logger.error(failure[0], exc_info=failure[1])
|
|
else:
|
|
self._show_exception(*failure)
|
|
return failure[1]
|
|
return None
|
|
|
|
def begin_close(self) -> bool:
|
|
if not self._closing:
|
|
self.stop_generation += 1
|
|
self._closing = True
|
|
self.timer.stop()
|
|
self.replay.close()
|
|
if self._stopping:
|
|
self._stop_all_requested = self._close_deferred = True
|
|
return False
|
|
return True
|
|
|
|
def close(self) -> None:
|
|
if not self.begin_close():
|
|
return
|
|
if self.recording.active:
|
|
self.stop_pipeline(keep_processor=True)
|
|
self.stop_all()
|
|
|
|
def _close_readers(self, *, keep_results: bool = False) -> None:
|
|
raw, pre, result = self.readers
|
|
self._raw_reader = self._pre_reader = None
|
|
if not keep_results:
|
|
self._result_reader = None
|
|
for reader in (raw, pre, None if keep_results else result):
|
|
if reader is not None:
|
|
reader.close()
|
|
|
|
def _open_readers(self, config: RunConfigModel) -> None:
|
|
self._close_readers()
|
|
self._raw_reader = ShmRingReader(config.rings.raw_tap.name) if config.rings.raw_tap.name else None
|
|
self._pre_reader = ShmRingReader(config.rings.preprocessed_tap.name) if config.rings.preprocessed_tap.name else None
|
|
self._result_reader = ShmRingReader(config.rings.results.name)
|
|
|
|
def poll(self) -> None:
|
|
if self._closing or self._stopping:
|
|
return
|
|
self._poll_ui()
|
|
self._handle_process_exit_reports()
|
|
try:
|
|
latest = self.read_raw(), self.read_preprocessed(), self.read_results()
|
|
except Exception as exc:
|
|
self._handle_reader_poll_error(exc)
|
|
return
|
|
if any(collection is not None for collection in latest):
|
|
self._last_reader_error_signature, self._reader_error_repeat_count = None, 0
|
|
for action in (self.recording.poll, self.history_changed):
|
|
try:
|
|
action()
|
|
except Exception as exc:
|
|
self._report_ui_error("Runtime display/recording", exc)
|
|
try:
|
|
if self.single_capture_active:
|
|
try:
|
|
self._finish_single_capture_if_ready()
|
|
finally:
|
|
if self.single_capture_active:
|
|
self._check_single_capture_deadline()
|
|
elif self._display_pending:
|
|
if latest[2] is not None:
|
|
self._restart_count = 0 # results flow again
|
|
self._display_pending = False
|
|
if self.history.results:
|
|
started = time.monotonic_ns()
|
|
self.plots.draw(self.history.results[-1])
|
|
self.metrics.record("rendering", time.monotonic_ns() - started)
|
|
except Exception as exc:
|
|
self._report_ui_error("Result display", exc)
|
|
|
|
def _report_ui_error(self, context: str, exc: Exception) -> None:
|
|
self._status(f"Status: {context.lower()} error (acquisition continues)")
|
|
signature = context, type(exc).__name__, str(exc)
|
|
if signature not in self._ui_errors:
|
|
self._ui_errors.add(signature)
|
|
logger.error("%s failed; acquisition continues", context, exc_info=exc)
|
|
|
|
def _handle_process_exit_reports(self) -> None:
|
|
unexpected = False
|
|
try:
|
|
for report in self.supervisor.collect_exit_reports():
|
|
if report.level == "INFO":
|
|
logger.info(report.format())
|
|
else:
|
|
self._status("Status: error")
|
|
logger.error(report.format())
|
|
unexpected = True
|
|
if not self.paused and self.should_run and (unexpected or not
|
|
{"data_processor", "data_preprocessor", "acquisition"}.issubset(self.supervisor.pids())):
|
|
self._recover_pipeline_after_crash()
|
|
except Exception:
|
|
logger.exception("Failed to collect process exit reports")
|
|
|
|
def _recover_pipeline_after_crash(self) -> None:
|
|
if self.paused or self._closing:
|
|
return
|
|
now = time.monotonic()
|
|
if not self._RESTART_POLICY.should_restart_now(now_s=now, last_restart_s=self._last_restart_s,
|
|
consecutive_failures=self._restart_count):
|
|
return
|
|
self._last_restart_s, self._restart_count = now, self._restart_count + 1
|
|
config, path = self.active_config, self.active_config_path
|
|
if config is None or path is None:
|
|
self.should_run = False
|
|
logger.error("Cannot auto-restart pipeline: no active run configuration recorded.")
|
|
return
|
|
logger.error("Pipeline process exited unexpectedly; restarting (attempt %d, next back-off %.0fs).",
|
|
self._restart_count, self._RESTART_POLICY.backoff_for(self._restart_count))
|
|
try:
|
|
# No event-pumping drain in recovery: polling must not re-enter this restart.
|
|
self.supervisor.stop_all()
|
|
self._close_readers()
|
|
self.supervisor.start(path, allow_clean_acquisition_exit=False)
|
|
self._open_readers(config)
|
|
self._processor_signature = build_processor_run_signature(config)
|
|
self.drop_pending()
|
|
self._status("Status: running")
|
|
logger.info("Pipeline auto-restarted after crash.")
|
|
except Exception:
|
|
logger.exception("Pipeline auto-restart failed; will retry")
|
|
|
|
def _handle_reader_poll_error(self, exc: Exception) -> None:
|
|
signature = type(exc).__name__, str(exc)
|
|
self._status("Status: error")
|
|
self._reader_error_repeat_count += 1
|
|
if self._last_reader_error_signature != signature:
|
|
self._last_reader_error_signature = signature
|
|
logger.error("Reader poll failed", exc_info=exc)
|
|
repeats = self._reader_error_repeat_count
|
|
if repeats % self._READER_ERROR_RELOG_EVERY == 0:
|
|
logger.error("Reader poll still failing (repeat #%d)", repeats, exc_info=exc)
|
|
if repeats >= self._READER_ERROR_STOP_AT:
|
|
logger.error("Stopping pipeline after %d consecutive reader poll failures", repeats)
|
|
self._last_reader_error_signature, self._reader_error_repeat_count = None, 0
|
|
self.stop_all()
|
|
elif repeats == self._READER_ERROR_RECONNECT_AT:
|
|
try:
|
|
self._raw_reader = self._reconnect(self._raw_reader)
|
|
self._pre_reader = self._reconnect(self._pre_reader)
|
|
self._result_reader = self._reconnect(self._result_reader)
|
|
logger.info("Ring readers reconnected after repeated poll failures")
|
|
except Exception:
|
|
logger.exception("Ring reader reconnect failed")
|
|
|
|
@staticmethod
|
|
def _reconnect(reader: ShmRingReader | None) -> ShmRingReader | None:
|
|
if reader is None:
|
|
return None
|
|
name = reader.name
|
|
reader.close()
|
|
return ShmRingReader(name)
|
|
|
|
def _finish_single_capture_if_ready(self) -> bool:
|
|
if not self.single_capture_active or self._single_capture_target_key is None:
|
|
return False
|
|
result = index_collections(self.history.results).get(self._single_capture_target_key)
|
|
if result is None or not self.plots.has_output(result):
|
|
return False
|
|
try:
|
|
self.plots.draw(result)
|
|
except Exception as exc:
|
|
self._report_ui_error("Result display", exc)
|
|
status = sar3d_status(result) if self.plots.stack.currentWidget() is self.plots.sar3d.widget else None
|
|
self.stop()
|
|
if status is not None and status[0] != 0:
|
|
state = SAR3D_STATES[status[0]]
|
|
self._status(f"Status: {state}")
|
|
(logger.error if status[0] == 2 else logger.warning)("Single capture: SAR3D %s: %s", state, status[1])
|
|
else:
|
|
logger.info("Single capture completed")
|
|
return True
|
|
|
|
def _check_single_capture_deadline(self) -> None:
|
|
start = self._single_capture_start_ns
|
|
if start is not None and time.monotonic_ns() - start > self.SINGLE_CAPTURE_TIMEOUT_S * 1e9:
|
|
logger.error("Single capture timed out after %.0fs with no result; stopping.", self.SINGLE_CAPTURE_TIMEOUT_S)
|
|
self.stop()
|
|
|
|
def read_raw(self, *, max_collections: int | None = None) -> SweepCollection | None:
|
|
latest = None
|
|
if self._raw_reader is not None:
|
|
for _ in range(self.max_pop_per_poll if max_collections is None else max_collections):
|
|
collection = self._raw_reader.pop_raw_collection()
|
|
if collection is None:
|
|
break
|
|
self.history.raw.append(collection)
|
|
self.recording.record("raw", collection)
|
|
self.metrics.record("acquisition", collection.capture_end_ns - collection.capture_start_ns)
|
|
self._select_single_target(collection)
|
|
latest = collection
|
|
return latest
|
|
|
|
def read_preprocessed(self, *, max_collections: int | None = None) -> SweepCollection | None:
|
|
latest = None
|
|
if self._pre_reader is not None:
|
|
for _ in range(self.max_pop_per_poll if max_collections is None else max_collections):
|
|
collection = self._pre_reader.pop_preprocessed_collection()
|
|
if collection is None:
|
|
break
|
|
self.history.preprocessed.append(collection)
|
|
self.recording.record("preprocessed", collection)
|
|
latest = collection
|
|
return latest
|
|
|
|
def read_results(self, *, max_collections: int | None = None) -> ResultCollection | None:
|
|
latest = None
|
|
if self._result_reader is not None:
|
|
for _ in range(self.max_pop_per_poll if max_collections is None else max_collections):
|
|
collection = self._result_reader.pop_result_collection()
|
|
if collection is None:
|
|
break
|
|
if self.history.is_removed(collection):
|
|
continue
|
|
if self._raw_reader is None:
|
|
self._select_single_target(collection)
|
|
self.metrics.record("processing", collection.processing_duration_ns)
|
|
if self.history.record_result(collection):
|
|
self.recording.record("results", collection)
|
|
self._display_pending = True
|
|
latest = collection
|
|
return latest
|
|
|
|
def _select_single_target(self, collection: SweepCollection | ResultCollection) -> None:
|
|
if (self.single_capture_active and self._single_capture_target_key is None
|
|
and self._single_capture_start_ns is not None and collection.capture_start_ns >= self._single_capture_start_ns):
|
|
self._single_capture_target_key = collection_key(collection)
|
|
|
|
def pump_events(self, pause_s: float) -> None:
|
|
app = QCoreApplication.instance()
|
|
if app is None:
|
|
time.sleep(pause_s)
|
|
return
|
|
deadline = time.monotonic() + pause_s
|
|
while (remaining_ms := int((deadline - time.monotonic()) * 1000)) > 0:
|
|
with self._event_scope():
|
|
app.processEvents(QEventLoop.ProcessEventsFlag.AllEvents, remaining_ms)
|
|
time.sleep(min(0.002, max(0.0, deadline - time.monotonic())))
|
|
|
|
def drain_once(self) -> bool:
|
|
return any(collection is not None for collection in (self.read_raw(), self.read_preprocessed(), self.read_results()))
|
|
|
|
def drain_rings(self, *, timeout_s: float, poll_s: float) -> None:
|
|
deadline, stable = time.monotonic() + timeout_s, 0
|
|
while time.monotonic() < deadline and stable < 2:
|
|
stable = 0 if self.drain_once() else stable + 1
|
|
self.pump_events(poll_s)
|
|
|
|
def drain_results(self, *, timeout_s: float, poll_s: float, revision: int | None = None) -> None:
|
|
"""Read results until the timeout; stop two quiet polls after results arrive or,
|
|
given a revision, as soon as the newest result carries it."""
|
|
deadline, seen, stable = time.monotonic() + timeout_s, False, 0
|
|
while self._result_reader is not None and time.monotonic() < deadline:
|
|
if self.read_results() is not None:
|
|
seen, stable = True, 0
|
|
elif seen:
|
|
stable += 1
|
|
if revision is None and stable >= 2:
|
|
return
|
|
if revision is not None and self.history.results and self.history.results[-1].settings_revision == revision:
|
|
return
|
|
self.pump_events(poll_s)
|
|
|
|
def drain_snapshot(self) -> None:
|
|
initial_readers = self.readers
|
|
if all(reader is None for reader in initial_readers):
|
|
return
|
|
try:
|
|
observed = [False, False, False]
|
|
deadline = time.monotonic() + 1.2
|
|
while time.monotonic() < deadline and not self._closing:
|
|
latest = (self.read_raw(max_collections=self.max_pop_per_snapshot_drain),
|
|
self.read_preprocessed(max_collections=self.max_pop_per_snapshot_drain),
|
|
self.read_results(max_collections=self.max_pop_per_snapshot_drain))
|
|
observed = [seen or value is not None for seen, value in zip(observed, latest)]
|
|
if any(value is not None for value in latest):
|
|
continue
|
|
missing = any(reader is not None and not seen for reader, seen in zip(initial_readers, observed))
|
|
awaiting = self._result_reader is not None and max(len(self.history.raw), len(self.history.preprocessed)) > len(self.history.results)
|
|
if (any(observed) and missing) or awaiting:
|
|
self.pump_events(0.01)
|
|
else:
|
|
break
|
|
except Exception:
|
|
logger.warning("Snapshot drain warning", exc_info=True)
|
|
|
|
def drop_pending(self, *, include_results: bool = True) -> tuple[int, int, int]:
|
|
raw, pre, result = self.readers
|
|
dropped = (raw.drop_all() if raw is not None else 0, pre.drop_all() if pre is not None else 0,
|
|
result.drop_all() if include_results and result is not None else 0)
|
|
if any(dropped):
|
|
logger.info("Ring reset: raw=%d, preprocessed=%d, results=%d", *dropped)
|
|
return dropped
|
|
|
|
def settle_deleted_results(self) -> None:
|
|
if self.processor_running:
|
|
self.drain_results(timeout_s=0.35, poll_s=0.01)
|
|
if self._result_reader is not None:
|
|
self._result_reader.drop_all()
|
|
|
|
def reset_history(self) -> None:
|
|
self.history.raw.clear()
|
|
self.history.preprocessed.clear()
|
|
self.history.results.clear()
|
|
self.discard_replay()
|
|
self.plots.clear()
|
|
self.history_changed()
|
|
|
|
def preprocessing_changed(self, *, restart_processor: bool = False) -> None:
|
|
self.history.run_signature = None
|
|
if restart_processor:
|
|
self._processor_signature = None
|
|
try:
|
|
if self.processor_running:
|
|
self.history.apply_deletion("clear_all")
|
|
self.publish_live()
|
|
finally:
|
|
self.reset_history()
|
|
|
|
def can_reprocess_history(self, *, mode: str | None = None) -> bool:
|
|
bscan = mode == "bscan" if mode is not None else self.plots.stack.currentWidget() is self.plots.bscan.widget
|
|
return (not self._closing and bscan
|
|
and bool(self.history.preprocessed) and not self.running and self.processor_running
|
|
and self._result_reader is not None)
|
|
|
|
def schedule_replay(self) -> None:
|
|
self.replay.schedule(self.can_reprocess_history())
|
|
|
|
def discard_replay(self) -> None:
|
|
self.replay.discard()
|
|
self.plots.bscan.signature = None
|
|
|
|
def reprocess_history(self) -> None:
|
|
if self.replay.active or not self.can_reprocess_history():
|
|
return
|
|
config, reader, generation = self.active_config, self._result_reader, self.stop_generation
|
|
polling = self.timer.isActive()
|
|
self.timer.stop()
|
|
try:
|
|
if self.replay.run(window=self.plots.window_scans(), revision=self.plots.required_revision or 0,
|
|
ring=config.rings.preprocessed if config is not None else None, reader=reader,
|
|
pump_events=self.pump_events, continue_replay=lambda: self.can_resume(generation)
|
|
and not self._stopping and self.can_reprocess_history() and self._result_reader is reader):
|
|
self.plots.redraw()
|
|
except Exception as exc:
|
|
self._show_exception("Failed to recompute the B-scan", exc)
|
|
finally:
|
|
if polling and not self._closing:
|
|
self.timer.start()
|