418 lines
19 KiB
Python
418 lines
19 KiB
Python
"""Service that serves processed RadioPhotonic radar (RPH) sweeps.
|
|
|
|
With ``source == "e502"`` it owns the external collector process and its TTY; with
|
|
``"rfg_udp"`` it reads the RFG board's UDP stream directly. Each raw (main,
|
|
reference) sweep maps to a :class:`SweepResult` on the fixed processing grid via
|
|
:class:`RphRadarSweepProcessor`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import suppress
|
|
from dataclasses import asdict, dataclass, field
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import stat
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
from python_app.hardware_full.rph_radar.processing import RphRadarSweepProcessor
|
|
from python_app.hardware_full.laser_control.conversions import variation_step_count
|
|
from python_app.hardware_full.rph_radar.protocol import RawSweep, RfgSweepAssembler
|
|
from python_app.hardware_full.rph_radar.stream_reader import RphRadarStreamReader, raise_if_process_exited
|
|
from python_app.hardware_full.librevna_driver.models import SweepResult
|
|
from python_app.models.dataset_model import ComboKey
|
|
from python_app.models.run_config_schema import RadarSweepModel, RunConfigModel
|
|
from python_app.models.run_config_validation import validate_rfg_source, validate_rph_radar_model
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Project root, used to resolve relative collector paths (e.g.
|
|
# ``build/bin/rph_radar_collector``) independent of the launching CWD.
|
|
_REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
# Default home of the proprietary L-Card runtime libraries the collector loads
|
|
# via dlopen. Prepended to LD_LIBRARY_PATH unless the config pins it explicitly.
|
|
_DEFAULT_LCARD_LIB_DIR = "~/.local/lib"
|
|
# Rejected-sweep logging is throttled so a persistently mis-set band/calibration
|
|
# does not flood the log: log the first rejection, then every Nth.
|
|
_REJECT_LOG_EVERY = 50
|
|
# The collector's graceful X502 teardown can block, so a stop gives it only this
|
|
# brief window to release the device cleanly before escalating to SIGKILL. Caps
|
|
# the configured stop_timeout_s so a stop can never hang.
|
|
_STOP_KILL_GRACE_S = 0.5
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RphRadarService:
|
|
"""Serve processed RadioPhotonic radar (RPH) sweeps from the E-502 collector or the RFG stream."""
|
|
|
|
config: RunConfigModel
|
|
collector_driven: bool = False
|
|
_process: subprocess.Popen[bytes] | None = field(init=False, default=None, repr=False)
|
|
_reader: RphRadarStreamReader | None = field(init=False, default=None, repr=False)
|
|
_processor: RphRadarSweepProcessor | None = field(init=False, default=None, repr=False)
|
|
_rejected_count: int = field(init=False, default=0, repr=False)
|
|
_prepared_combo: ComboKey | None = field(init=False, default=None, repr=False)
|
|
last_cycle_id: int = field(init=False, default=0)
|
|
_output_thread: threading.Thread | None = field(init=False, default=None, repr=False)
|
|
_stop_event: threading.Event | None = field(init=False, default=None, repr=False)
|
|
|
|
def __post_init__(self) -> None:
|
|
self._validate_config()
|
|
|
|
@property
|
|
def command(self) -> list[str]:
|
|
"""Pass the current typed settings once; the collector never rereads the profile."""
|
|
adc = self.config.radar.rph_radar
|
|
cmd = [str(self._resolve_executable()), *adc.args]
|
|
if self.collector_driven:
|
|
switching = {"input": asdict(self.config.input_switch), "output": asdict(self.config.output_switch),
|
|
"combos": [[combo.input, combo.output] for combo in self.config.combos],
|
|
"settling_ms": self.config.runtime.settling_ms}
|
|
cmd.append("switching:" + json.dumps(switching, separators=(",", ":")))
|
|
cmd.append(f"tty:{adc.tty_path}")
|
|
return cmd
|
|
|
|
def open(self, *, stop_event: threading.Event | None = None) -> None:
|
|
"""Bind the RFG UDP stream, or launch the E-502 collector and start its TTY reader.
|
|
|
|
An optional ``stop_event`` lets a caller abort the TTY-wait loop promptly
|
|
(e.g. on shutdown) instead of blocking for the full startup timeout.
|
|
"""
|
|
if self._reader is not None:
|
|
return
|
|
adc = self.config.radar.rph_radar
|
|
if adc.source == "rfg_udp":
|
|
variation = self.config.radar.laser_control.variation
|
|
steps = variation_step_count(variation.min_value, variation.max_value, variation.step)
|
|
step_ns = variation.time_step * 1000
|
|
# Lower bound of the sweep period: the firmware idles at least delay_time - 1 ms.
|
|
period_ns = steps * step_ns + (variation.delay_time - 1) * 1_000_000
|
|
reader = RphRadarStreamReader(adc.udp_bind, RfgSweepAssembler(steps, step_ns, period_ns))
|
|
reader.open()
|
|
self._reader = reader
|
|
logger.info("RadioPhotonic radar (RPH) RFG stream opened on %s (%d steps per sweep)", adc.udp_bind, steps)
|
|
return
|
|
if self._process is not None and self._process.poll() is None:
|
|
raise RuntimeError("Previous RPH collector did not stop; cannot start a fresh capture")
|
|
self._stop_event = stop_event
|
|
previous_tty_identity = _prepare_tty_path_for_collector(self.config.radar.rph_radar.tty_path)
|
|
try:
|
|
self._start_process()
|
|
self._wait_for_tty(previous_tty_identity, stop_event=stop_event)
|
|
reader = RphRadarStreamReader(self.config.radar.rph_radar.tty_path)
|
|
reader.open()
|
|
self._reader = reader
|
|
logger.info("RadioPhotonic radar (RPH) service opened")
|
|
except Exception:
|
|
logger.exception("RadioPhotonic radar (RPH) service failed to open; cleaning up")
|
|
self.close()
|
|
raise
|
|
|
|
def close(self) -> None:
|
|
"""Stop the reader and, for E-502, the external collector process."""
|
|
logger.debug("Closing RadioPhotonic radar (RPH) service")
|
|
self._prepared_combo = None
|
|
if self._reader is not None:
|
|
with suppress(Exception):
|
|
self._reader.close()
|
|
self._reader = None
|
|
self._stop_process()
|
|
if self._output_thread is not None:
|
|
self._output_thread.join(timeout=1.0)
|
|
self._output_thread = None
|
|
|
|
def configure(self, sweep: RadarSweepModel) -> None:
|
|
"""Build the sweep processor from the ``radar.rph_radar`` calibration/band.
|
|
|
|
The generic ``sweep`` argument is accepted for interface parity with the
|
|
other radar services but is not used: the RadioPhotonic radar (RPH) frequency axis comes
|
|
from the reference-phase calibration, and the output grid from
|
|
``radar.rph_radar.band`` — never from the nominal sweep bounds.
|
|
"""
|
|
adc = self.config.radar.rph_radar
|
|
self._processor = RphRadarSweepProcessor(adc)
|
|
self._rejected_count = 0
|
|
phase, band = adc.phase_calibration, adc.band
|
|
logger.debug(
|
|
"RadioPhotonic radar (RPH) configured: band %.6g-%.6g Hz, %d points; calibration "
|
|
"(%.6g rad -> %.6g Hz, %.6g rad -> %.6g Hz)",
|
|
band.start_hz, band.stop_hz, band.points,
|
|
phase.phase0_rad, phase.freq0_hz, phase.phase1_rad, phase.freq1_hz,
|
|
)
|
|
|
|
def read_device_limits(self) -> dict[str, float | int]:
|
|
"""RadioPhotonic radar (RPH) has no runtime-readable sweep-limit API."""
|
|
raise RuntimeError("RadioPhotonic radar (RPH) device limits are not available")
|
|
|
|
def acquire(self, combo: tuple[int, int] | None = None) -> SweepResult:
|
|
"""Return the next sweep that covers the band, as S21 on the fixed grid.
|
|
|
|
Sweeps whose floated frequency range does not span the configured band are
|
|
rejected and the next sweep is read, until one passes or the sweep timeout
|
|
elapses (which then surfaces as a :class:`TimeoutError`).
|
|
|
|
When ``combo`` is given (switch-aware mode), only the clean sweep captured
|
|
under that switch combination is returned; the collector drives the switches
|
|
and tags each sweep. When ``None`` (calibration / non-switch mode), the
|
|
single newest sweep is returned regardless of combination.
|
|
"""
|
|
if self._processor is None:
|
|
raise RuntimeError("RadioPhotonic radar (RPH) service is not configured")
|
|
if self._reader is None:
|
|
raise RuntimeError("RadioPhotonic radar (RPH) service is not open")
|
|
process = self._process
|
|
if self.config.radar.rph_radar.source == "e502" and (process is None or process.poll() is not None):
|
|
return_code = None if process is None else process.poll()
|
|
raise RuntimeError(f"RadioPhotonic radar (RPH) collector is not running (code={return_code})")
|
|
|
|
grid = self._processor.grid_hz
|
|
deadline = time.monotonic() + self.config.radar.rph_radar.sweep_timeout_s
|
|
while True:
|
|
remaining_s = deadline - time.monotonic()
|
|
if remaining_s <= 0.0:
|
|
raise TimeoutError(
|
|
"Timed out waiting for a RadioPhotonic radar (RPH) sweep covering the configured band"
|
|
)
|
|
raw = self._reader.read_sweep(timeout_s=remaining_s, combo=combo, process=process)
|
|
s21 = self._processor.process(raw.main, raw.reference)
|
|
if s21 is not None:
|
|
self.last_cycle_id = raw.cycle_id
|
|
return SweepResult(
|
|
x=grid,
|
|
traces={
|
|
"s21": s21,
|
|
},
|
|
capture_start_ns=raw.capture_start_ns,
|
|
capture_end_ns=raw.capture_end_ns,
|
|
)
|
|
self._log_rejected_sweep(raw)
|
|
|
|
def prepare_combo(self, combo: ComboKey) -> None:
|
|
"""Start a fresh ADC session after switching; retain phase branch tracking."""
|
|
if self._prepared_combo == combo:
|
|
return
|
|
self.close()
|
|
if self.config.runtime.settling_ms < 1:
|
|
time.sleep(.001) # The caller already waited any configured settling.
|
|
self.open(stop_event=self._stop_event)
|
|
self._prepared_combo = combo
|
|
|
|
def read_raw_sweep(self) -> RawSweep:
|
|
"""Return the next raw (main, reference) sweep without any processing.
|
|
|
|
Bypasses the frequency mapping, normalization, crop and resample of
|
|
:meth:`acquire` — intended for calibration tooling that needs the
|
|
unprocessed reference samples. Raises if the service is not open.
|
|
"""
|
|
if self._reader is None:
|
|
raise RuntimeError("RadioPhotonic radar (RPH) service is not open")
|
|
return self._reader.read_sweep(
|
|
timeout_s=self.config.radar.rph_radar.sweep_timeout_s,
|
|
process=self._process,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Process / TTY lifecycle
|
|
# ------------------------------------------------------------------
|
|
|
|
def _start_process(self) -> None:
|
|
if self._process is not None and self._process.poll() is None:
|
|
return
|
|
logger.info("Starting RadioPhotonic radar (RPH) collector: %s", " ".join(self.command))
|
|
self._process = subprocess.Popen(
|
|
self.command,
|
|
cwd=str(self._resolve_project_dir()),
|
|
env=self._build_env(),
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
start_new_session=True,
|
|
)
|
|
self._output_thread = threading.Thread(target=self._log_output, args=(self._process,),
|
|
name="rph-radar-log", daemon=True)
|
|
self._output_thread.start()
|
|
|
|
@staticmethod
|
|
def _log_output(process: subprocess.Popen[bytes]) -> None:
|
|
if process.stdout is not None:
|
|
with process.stdout:
|
|
for line in process.stdout:
|
|
logger.info("Collector: %s", line.decode("utf-8", errors="replace").rstrip())
|
|
|
|
def _stop_process(self) -> None:
|
|
process = self._process
|
|
if process is None or process.poll() is not None:
|
|
self._process = None
|
|
return
|
|
logger.info("Stopping RadioPhotonic radar (RPH) collector (pid=%d)", process.pid)
|
|
# The collector's graceful X502 teardown can block, so give it only a brief
|
|
# window to release the device cleanly, then SIGKILL the whole group hard.
|
|
grace_s = min(self.config.radar.rph_radar.stop_timeout_s, _STOP_KILL_GRACE_S)
|
|
with suppress(ProcessLookupError):
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
try:
|
|
process.wait(timeout=grace_s)
|
|
self._process = None
|
|
return
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
logger.warning("RadioPhotonic radar (RPH) collector (pid=%d) did not stop in %.1fs; sending SIGKILL", process.pid, grace_s)
|
|
with suppress(ProcessLookupError):
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
# close() must never raise: a collector wedged in uninterruptible I/O
|
|
# (USB D-state in the L-Card driver) may not be reaped within the grace
|
|
# window even after SIGKILL. Best-effort wait; the OS reaps it eventually.
|
|
with suppress(subprocess.TimeoutExpired):
|
|
process.wait(timeout=1.0)
|
|
# Retain ownership in D-state so retries cannot launch a second collector.
|
|
if process.poll() is not None:
|
|
self._process = None
|
|
|
|
def _wait_for_tty(
|
|
self,
|
|
previous_identity: tuple[object, ...] | None,
|
|
*,
|
|
stop_event: threading.Event | None = None,
|
|
) -> None:
|
|
adc = self.config.radar.rph_radar
|
|
deadline = time.monotonic() + adc.startup_timeout_s
|
|
while time.monotonic() < deadline:
|
|
if stop_event is not None and stop_event.is_set():
|
|
raise RuntimeError("RadioPhotonic radar (RPH) startup aborted before TTY became available")
|
|
raise_if_process_exited(self._process)
|
|
identity = _tty_identity(adc.tty_path)
|
|
if identity is not None and identity != previous_identity:
|
|
return
|
|
if stop_event is not None:
|
|
if stop_event.wait(0.05):
|
|
raise RuntimeError("RadioPhotonic radar (RPH) startup aborted before TTY became available")
|
|
else:
|
|
time.sleep(0.05)
|
|
raise TimeoutError(
|
|
f"Timed out waiting for RadioPhotonic radar (RPH) TTY `{adc.tty_path}` to be created by the collector"
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Path / environment resolution
|
|
# ------------------------------------------------------------------
|
|
|
|
def _resolve_executable(self) -> Path:
|
|
return self._resolve_path(self.config.radar.rph_radar.executable_path)
|
|
|
|
def _resolve_project_dir(self) -> Path:
|
|
project_dir = self.config.radar.rph_radar.project_dir
|
|
return self._resolve_path(project_dir) if project_dir else _REPO_ROOT
|
|
|
|
@staticmethod
|
|
def _resolve_path(path_str: str) -> Path:
|
|
"""Expand ``~`` and resolve a relative path against the project root."""
|
|
path = Path(path_str).expanduser()
|
|
return path if path.is_absolute() else (_REPO_ROOT / path)
|
|
|
|
def _build_env(self) -> dict[str, str]:
|
|
"""Child environment: inherited env + config env, with the L-Card lib path.
|
|
|
|
Config ``env`` values are ``~``/``$VAR`` expanded. Unless the config pins
|
|
``LD_LIBRARY_PATH`` itself, the default L-Card library directory is
|
|
prepended so the collector's dlopen of libx502api/libe502api succeeds.
|
|
"""
|
|
adc = self.config.radar.rph_radar
|
|
env = os.environ.copy()
|
|
for key, value in adc.env.items():
|
|
env[key] = os.path.expandvars(os.path.expanduser(value))
|
|
if "LD_LIBRARY_PATH" not in adc.env:
|
|
lcard_dir = os.path.expanduser(_DEFAULT_LCARD_LIB_DIR)
|
|
existing = env.get("LD_LIBRARY_PATH", "")
|
|
env["LD_LIBRARY_PATH"] = f"{lcard_dir}{os.pathsep}{existing}" if existing else lcard_dir
|
|
return env
|
|
|
|
# ------------------------------------------------------------------
|
|
# Validation / diagnostics
|
|
# ------------------------------------------------------------------
|
|
|
|
def _validate_config(self) -> None:
|
|
if not self.config.is_rph_radar:
|
|
raise RuntimeError("RphRadarService requires radar.model='rph_radar'")
|
|
if self.config.radar.driver_mode != "native":
|
|
raise RuntimeError("RadioPhotonic radar (RPH) requires radar.driver_mode='native'")
|
|
|
|
adc = self.config.radar.rph_radar
|
|
validate_rph_radar_model(adc)
|
|
if adc.source == "rfg_udp":
|
|
validate_rfg_source(self.config)
|
|
return
|
|
if not adc.executable_path:
|
|
raise ValueError("radar.rph_radar.executable_path is required")
|
|
if not adc.tty_path:
|
|
raise ValueError("radar.rph_radar.tty_path is required")
|
|
if any(arg.startswith("tty:") for arg in adc.args):
|
|
raise ValueError("radar.rph_radar.args must not contain tty:<path>; use tty_path instead")
|
|
|
|
project_dir = self._resolve_project_dir()
|
|
if not project_dir.is_dir():
|
|
raise RuntimeError(f"radar.rph_radar.project_dir is not a directory: {project_dir}")
|
|
executable_path = self._resolve_executable()
|
|
if not executable_path.is_file():
|
|
raise RuntimeError(f"radar.rph_radar.executable_path is not a file: {executable_path}")
|
|
if not os.access(executable_path, os.X_OK):
|
|
raise RuntimeError(f"radar.rph_radar.executable_path is not executable: {executable_path}")
|
|
|
|
|
|
def _log_rejected_sweep(self, raw) -> None:
|
|
"""Log a band-coverage rejection (throttled) with the measured span."""
|
|
self._rejected_count += 1
|
|
if self._rejected_count != 1 and self._rejected_count % _REJECT_LOG_EVERY != 0:
|
|
return
|
|
assert self._processor is not None
|
|
band = self.config.radar.rph_radar.band
|
|
try:
|
|
freqs = self._processor.reference_frequency_axis(raw.reference)
|
|
covered = f"[{float(np.min(freqs)):.6g}, {float(np.max(freqs)):.6g}]"
|
|
except Exception: # noqa: BLE001 — diagnostics must never raise
|
|
covered = "<unavailable>"
|
|
logger.warning(
|
|
"RadioPhotonic radar (RPH) sweep rejected (count=%d): covered %s Hz does not span band "
|
|
"[%.6g, %.6g] Hz (usable points=%d). Check phase_calibration/band.",
|
|
self._rejected_count, covered, band.start_hz, band.stop_hz, raw.size,
|
|
)
|
|
|
|
|
|
def _tty_identity(path: str) -> tuple[object, ...] | None:
|
|
try:
|
|
if os.path.islink(path):
|
|
stat_result = os.lstat(path)
|
|
return (
|
|
"link",
|
|
os.readlink(path),
|
|
int(stat_result.st_dev),
|
|
int(stat_result.st_ino),
|
|
int(stat_result.st_mtime_ns),
|
|
)
|
|
stat_result = os.stat(path)
|
|
except FileNotFoundError:
|
|
return None
|
|
return (
|
|
"node",
|
|
int(stat_result.st_dev),
|
|
int(stat_result.st_ino),
|
|
int(stat_result.st_mtime_ns),
|
|
)
|
|
|
|
|
|
def _prepare_tty_path_for_collector(path: str) -> tuple[object, ...] | None:
|
|
"""Remove a stale generated TTY symlink/file before starting the collector."""
|
|
try:
|
|
stat_result = os.lstat(path)
|
|
except FileNotFoundError:
|
|
return None
|
|
if stat.S_ISLNK(stat_result.st_mode) or stat.S_ISREG(stat_result.st_mode):
|
|
os.unlink(path)
|
|
return None
|