"""Raw acquisition producer for RadioPhotonic radar (RPH).""" from __future__ import annotations from contextlib import ExitStack import logging import threading import time import numpy as np from python_app.hardware_full.rph_radar.service import RphRadarService from python_app.hardware_full.switch_service import SwitchService, select_and_settle from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData from python_app.models.run_config_schema import RunConfigModel from python_app.scripts.producer_common import RawPublisher, close_devices, load_producer_config, retry_open logger = logging.getLogger(__name__) def _open_radar_with_retry( config: RunConfigModel, radar: RphRadarService, input_switch: SwitchService | None, output_switch: SwitchService | None, stop_requested: threading.Event, ) -> bool: """Reopen the ADC and Python-owned switches, cleaning partial opens before retry.""" close_devices(input_switch, output_switch, radar) def attempt() -> RphRadarService: try: radar.open(stop_event=stop_requested) radar.configure(config.radar.sweep) if output_switch is not None: output_switch.open() if input_switch is not None: input_switch.open() except Exception: close_devices(input_switch, output_switch, radar) raise return radar return retry_open(attempt, stop_requested, "RadioPhotonic radar (RPH)") is not None def main() -> int: """Run producer process until config or signal requests exit.""" config_path, config, stop_requested = load_producer_config("Publish RadioPhotonic radar (RPH) raw sweeps to SHM rings") if not config.is_rph_radar: raise RuntimeError("rph_radar_raw_producer requires radar.model='rph_radar'") logger.info( "RadioPhotonic radar (RPH) raw producer starting: config=%s, combos=%d, continuous=%s", config_path, len(config.combos), config.runtime.continuous, ) radar = input_switch = output_switch = None cleanup = ExitStack() try: publisher = RawPublisher(config.rings, cleanup) # Switch-aware mode: with native switches the E-502 collector drives the RF switches # itself and tags each sweep with its combo and source cycle. Otherwise Python # owns the switches and starts a fresh source session after each changed pair. collector_driven = ( config.radar.rph_radar.source == "e502" and config.input_switch.driver_mode == "native" and config.output_switch.driver_mode == "native" ) radar = RphRadarService( config, collector_driven=collector_driven, ) input_switch = None if collector_driven else SwitchService.from_model(config.input_switch) output_switch = None if collector_driven else SwitchService.from_model(config.output_switch) logger.info( "RadioPhotonic radar (RPH) switch control: %s", "collector-driven" if collector_driven else "producer-driven", ) if not _open_radar_with_retry(config, radar, input_switch, output_switch, stop_requested): return 0 # asked to stop before a device became available collection_id = 1 while not stop_requested.is_set(): collection_start = time.monotonic() traces: list[TraceData] = [] cycle_id = None try: for combo in config.combos: if stop_requested.is_set(): break if collector_driven: # The collector already switched and tagged the sweep; just # read the clean capture for this combination. sweep = radar.acquire(combo=(combo.input, combo.output)) if cycle_id is not None and cycle_id != radar.last_cycle_id: break # A dropped or overwritten pair invalidates this cycle. cycle_id = radar.last_cycle_id else: if input_switch is None or output_switch is None: raise RuntimeError("Producer-driven switches are not initialized") if not select_and_settle(output_switch, input_switch, combo.output, combo.input, config.runtime.settling_ms, wait=stop_requested.wait): break radar.prepare_combo(ComboKey(combo.input, combo.output)) sweep = radar.acquire() traces.append( TraceData( combo=ComboKey(input=combo.input, output=combo.output), frequency_hz=np.asarray(sweep.x, dtype=np.float32), s21=np.asarray(sweep.trace("s21"), dtype=np.complex64), capture_start_ns=sweep.capture_start_ns, capture_end_ns=sweep.capture_end_ns, ) ) except Exception as exc: # noqa: BLE001 — reconnect forever, never give up logger.warning( "RadioPhotonic radar (RPH) acquisition failed; reconnecting and waiting for the device: %s", exc, exc_info=True, ) if not _open_radar_with_retry(config, radar, input_switch, output_switch, stop_requested): break # stop requested while waiting to reconnect continue # An incomplete sweep set means the device dropped out (or a stop was # requested mid-collection). Only exit on stop; otherwise reconnect and # wait for the device rather than killing the producer. if len(traces) != len(config.combos): if stop_requested.is_set(): break if collector_driven: continue # Latest-wins transport skipped part of a source cycle. logger.warning( "RadioPhotonic radar (RPH) produced an incomplete collection (%d of %d combos); " "reconnecting and waiting for the device", len(traces), len(config.combos), ) if not _open_radar_with_retry(config, radar, input_switch, output_switch, stop_requested): break # stop requested while waiting to reconnect continue collection = SweepCollection( collection_id=cycle_id if cycle_id is not None else collection_id, monotonic_ns=time.monotonic_ns(), traces=traces, capture_start_ns=min(trace.capture_start_ns for trace in traces), capture_end_ns=max(trace.capture_end_ns for trace in traces), ) publisher.publish(collection) if not config.runtime.continuous: break collection_duration_s = time.monotonic() - collection_start if collection_id == 1 or collection_id % 20 == 0 or collection_duration_s > 2.0: logger.info("RadioPhotonic radar (RPH) collection %d acquired in %.3f s", collection_id, collection_duration_s) collection_id += 1 finally: close_devices(output_switch, input_switch, radar) cleanup.close() logger.info("RadioPhotonic radar (RPH) raw producer stopped") return 0 if __name__ == "__main__": raise SystemExit(main())