106 lines
4.0 KiB
Python
106 lines
4.0 KiB
Python
"""Shared CLI, interruptible device retry, and raw-ring publication."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections.abc import Callable
|
|
from contextlib import ExitStack, suppress
|
|
import logging
|
|
from pathlib import Path
|
|
import signal
|
|
import threading
|
|
from typing import Protocol, TypeVar
|
|
|
|
from python_app.logging_setup import coerce_level
|
|
from python_app.models.dataset_model import SweepCollection
|
|
from python_app.models.run_config_schema import RingEndpointModel, RingsModel, RunConfigModel
|
|
from python_app.models.run_config_validation import validate_acquisition_config
|
|
from python_app.orchestration.shm.decoder import RAW_MAGIC
|
|
from python_app.orchestration.shm.ring_writer import ShmRingWriter
|
|
from python_app.storage.npz.serialize import serialize_trace_collection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
T = TypeVar("T")
|
|
|
|
|
|
def load_producer_config(description: str) -> tuple[Path, RunConfigModel, threading.Event]:
|
|
parser = argparse.ArgumentParser(description=description)
|
|
parser.add_argument("--config", required=True, type=Path, help="Path to run_config.json")
|
|
path = parser.parse_args().config
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
|
|
stop_requested = threading.Event()
|
|
|
|
def request_stop(_signum: int, _frame: object) -> None:
|
|
stop_requested.set()
|
|
|
|
signal.signal(signal.SIGINT, request_stop)
|
|
signal.signal(signal.SIGTERM, request_stop)
|
|
config = RunConfigModel.load_from_path(path)
|
|
config.ensure_combos()
|
|
validate_acquisition_config(config)
|
|
logging.getLogger("python_app").setLevel(coerce_level(config.logging.level))
|
|
return path, config, stop_requested
|
|
|
|
|
|
class _Closable(Protocol):
|
|
def close(self) -> None: ...
|
|
|
|
|
|
def close_devices(*devices: _Closable | None) -> None:
|
|
"""Attempt every hardware cleanup even if one disconnected device raises."""
|
|
for device in devices:
|
|
if device is not None:
|
|
with suppress(Exception):
|
|
device.close()
|
|
|
|
|
|
def retry_open(
|
|
attempt: Callable[[], T], stop_requested: threading.Event, device_name: str,
|
|
) -> T | None:
|
|
"""Retry an attempt that cleans up on failure, with capped, interruptible backoff."""
|
|
failures = 0
|
|
delay = 1.0
|
|
while not stop_requested.is_set():
|
|
try:
|
|
device = attempt()
|
|
except Exception as exc:
|
|
failures += 1
|
|
if failures == 1 or failures % 30 == 0:
|
|
logger.warning("%s unavailable (attempt %d); retrying until present: %s", device_name, failures, exc)
|
|
if stop_requested.wait(delay):
|
|
break
|
|
delay = min(delay * 2, 10.0)
|
|
else:
|
|
if failures:
|
|
logger.info("%s opened after %d attempt(s)", device_name, failures + 1)
|
|
return device
|
|
return None
|
|
|
|
|
|
class RawPublisher:
|
|
"""Publish one validated frame to raw and, when enabled, its independent GUI tap."""
|
|
|
|
def __init__(self, rings: RingsModel, cleanup: ExitStack) -> None:
|
|
def open_writer(endpoint: RingEndpointModel) -> ShmRingWriter:
|
|
writer = ShmRingWriter(endpoint.name, endpoint.capacity, endpoint.slot_size_bytes)
|
|
cleanup.callback(writer.close)
|
|
return writer
|
|
|
|
self._raw = open_writer(rings.raw)
|
|
self._tap = open_writer(rings.raw_tap) if rings.raw_tap.name else None
|
|
self._dropped = 0
|
|
|
|
def publish(self, collection: SweepCollection) -> bool:
|
|
payload = serialize_trace_collection(collection, RAW_MAGIC, require_capture_window=True)
|
|
if not self._raw.push(payload):
|
|
self._dropped += 1
|
|
if self._dropped == 1 or self._dropped % 100 == 0:
|
|
logger.error(
|
|
"Raw payload %d B exceeds ring slot %d B; dropping collection %d (drops=%d)",
|
|
len(payload), self._raw.slot_size_bytes, collection.collection_id, self._dropped,
|
|
)
|
|
return False
|
|
if self._tap is not None:
|
|
self._tap.push(payload) # independent latest-wins tap; never backpressure raw
|
|
return True
|