radar_system/python_app/hardware_full/kamil_adc/service.py

"""Service that launches the Kamil ADC collector and serves processed sweeps.

Owns the lifecycle of the external collector process and its TTY reader, and maps
each raw (main, reference) sweep to an :class:`SweepResult` on the fixed
processing grid via :class:`KamilAdcSweepProcessor`.
"""

from __future__ import annotations

from contextlib import suppress
from dataclasses import dataclass, field
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.kamil_adc.processing import (
    KamilAdcProcessingParams,
    KamilAdcSweepProcessor,
)
from python_app.hardware_full.kamil_adc.protocol import RawSweep
from python_app.hardware_full.kamil_adc.tty_reader import (
    KamilAdcTtyReader,
    raise_if_process_exited,
)
from python_app.hardware_full.librevna_driver.models import SweepResult
from python_app.models.run_config_model import RadarSweepModel, RunConfigModel

logger = logging.getLogger(__name__)

# Project root, used to resolve relative collector paths (e.g.
# ``build/bin/kamil_adc_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
# Sweeps to skip after a Python-driven switch change before trusting a capture: one
# for the pre-switch sweep still in the mailbox, one for a possible transition
# straddler in flight. Calibration speed is not critical, so we err on safety.
_SWITCH_DRAIN_SWEEPS = 2


@dataclass(slots=True)
class KamilAdcService:
    """Launch the external Kamil ADC collector and serve its processed sweeps."""

    config: RunConfigModel
    # When set, the collector is launched with ``config:<path>`` so it drives the RF
    # switches itself (switch-aware mode) from this run_config.json. ``None`` keeps
    # the standalone collector that streams a single channel (calibration / mock).
    switch_config_path: str | None = None
    _process: subprocess.Popen[bytes] | None = field(init=False, default=None, repr=False)
    _reader: KamilAdcTtyReader | None = field(init=False, default=None, repr=False)
    _processor: KamilAdcSweepProcessor | None = field(init=False, default=None, repr=False)
    _rejected_count: int = field(init=False, default=0, repr=False)

    def __post_init__(self) -> None:
        self._validate_config()

    @property
    def command(self) -> list[str]:
        """External collector command, including the generated ``tty:`` argument.

        In switch-aware mode (``switch_config_path`` set) the collector also gets
        ``config:<path>`` so it reads the switch/combo configuration and drives the
        switches in lock-step with the sweeps.
        """
        adc = self.config.radar.kamil_adc
        cmd = [str(self._resolve_executable()), *adc.args]
        if self.switch_config_path is not None:
            cmd.append(f"config:{self.switch_config_path}")
        cmd.append(f"tty:{adc.tty_path}")
        return cmd

    def open(self, *, stop_event: threading.Event | None = None) -> None:
        """Launch the collector and start the TTY reader thread.

        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
        previous_tty_identity = _prepare_tty_path_for_collector(self.config.radar.kamil_adc.tty_path)
        try:
            self._start_process()
            self._wait_for_tty(previous_tty_identity, stop_event=stop_event)
            reader = KamilAdcTtyReader(self.config.radar.kamil_adc.tty_path)
            reader.open()
            self._reader = reader
            logger.info("Kamil ADC service opened")
        except Exception:
            logger.exception("Kamil ADC service failed to open; cleaning up")
            self.close()
            raise

    def close(self) -> None:
        """Stop the TTY reader and the external collector process."""
        logger.debug("Closing Kamil ADC service")
        if self._reader is not None:
            with suppress(Exception):
                self._reader.close()
            self._reader = None
        self._stop_process()

    def configure(self, sweep: RadarSweepModel) -> None:
        """Build the sweep processor from the ``radar.kamil_adc`` calibration/band.

        The generic ``sweep`` argument is accepted for interface parity with the
        other radar services but is not used: the Kamil ADC frequency axis comes
        from the reference-phase calibration, and the output grid from
        ``radar.kamil_adc.band`` — never from the nominal sweep bounds.
        """
        self._processor = KamilAdcSweepProcessor(
            KamilAdcProcessingParams.from_kamil_model(self.config.radar.kamil_adc)
        )
        self._rejected_count = 0
        params = self._processor.params
        logger.debug(
            "Kamil ADC configured: band %.6g-%.6g Hz, %d points; calibration "
            "(%.6g rad -> %.6g Hz, %.6g rad -> %.6g Hz)",
            params.band_start_hz, params.band_stop_hz, params.band_points,
            params.phase0_rad, params.freq0_hz, params.phase1_rad, params.freq1_hz,
        )

    def read_device_limits(self) -> dict[str, float | int]:
        """Kamil ADC has no runtime-readable sweep-limit API."""
        raise RuntimeError("Kamil ADC 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("Kamil ADC service is not configured")
        if self._reader is None:
            raise RuntimeError("Kamil ADC service is not open")
        process = self._process
        if process is None or process.poll() is not None:
            return_code = None if process is None else process.poll()
            raise RuntimeError(f"Kamil ADC collector is not running (code={return_code})")

        grid = self._processor.grid_hz
        points = int(grid.size)
        deadline = time.monotonic() + self.config.radar.kamil_adc.sweep_timeout_s
        while True:
            remaining_s = deadline - time.monotonic()
            if remaining_s <= 0.0:
                raise TimeoutError(
                    "Timed out waiting for a Kamil ADC sweep covering the configured band"
                )
            if combo is None:
                raw = self._reader.read_sweep(timeout_s=remaining_s, process=process)
            else:
                raw = self._reader.read_sweep_for(combo, timeout_s=remaining_s, process=process)
            s21 = self._processor.process(raw.main, raw.reference)
            if s21 is not None:
                return SweepResult(
                    x=grid.copy(),
                    traces={
                        "s11": np.zeros(points, dtype=np.complex64),
                        "s21": s21,
                    },
                )
            self._log_rejected_sweep(raw)

    def drain_after_switch(self, sweeps: int = _SWITCH_DRAIN_SWEEPS) -> None:
        """Discard sweeps captured before / across a just-applied switch change.

        The collector free-runs, so right after the RF switches move the reader
        still holds a sweep captured in the *previous* combination, and a sweep that
        straddles the transition may still be in flight. Without this, the next
        :meth:`acquire` would return that stale data and the capture would be
        attributed to the wrong combination (an off-by-one across the sequence).

        Block until ``sweeps`` freshly-published sweeps have gone by, so the next
        :meth:`acquire` returns a sweep captured entirely in the new switch state.
        Used by the Python-driven calibration capture, where the collector does not
        tag sweeps; the switch-aware collector path handles this with combo tags
        instead. No-op when the service is not open.
        """
        if self._reader is None:
            return
        target = self._reader.published_count + max(1, int(sweeps))
        deadline = time.monotonic() + self.config.radar.kamil_adc.sweep_timeout_s
        while self._reader.published_count < target:
            if time.monotonic() > deadline:
                raise TimeoutError("Timed out draining Kamil ADC sweeps after a switch change")
            raise_if_process_exited(self._process)
            time.sleep(0.005)

    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("Kamil ADC service is not open")
        return self._reader.read_sweep(
            timeout_s=self.config.radar.kamil_adc.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 Kamil ADC 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.DEVNULL,
            stderr=subprocess.STDOUT,
            start_new_session=True,
        )

    def _stop_process(self) -> None:
        process = self._process
        self._process = None
        if process is None or process.poll() is not None:
            return
        logger.info("Stopping Kamil ADC 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.kamil_adc.stop_timeout_s, _STOP_KILL_GRACE_S)
        with suppress(ProcessLookupError):
            os.killpg(process.pid, signal.SIGTERM)
        try:
            process.wait(timeout=grace_s)
            return
        except subprocess.TimeoutExpired:
            pass
        logger.warning("Kamil ADC 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)

    def _wait_for_tty(
        self,
        previous_identity: tuple[object, ...] | None,
        *,
        stop_event: threading.Event | None = None,
    ) -> None:
        adc = self.config.radar.kamil_adc
        deadline = time.monotonic() + adc.startup_timeout_s
        while time.monotonic() < deadline:
            if stop_event is not None and stop_event.is_set():
                raise RuntimeError("Kamil ADC 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("Kamil ADC startup aborted before TTY became available")
            else:
                time.sleep(0.05)
        raise TimeoutError(
            f"Timed out waiting for Kamil ADC 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.kamil_adc.executable_path)

    def _resolve_project_dir(self) -> Path:
        project_dir = self.config.radar.kamil_adc.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.kamil_adc
        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_kamil_adc:
            raise RuntimeError("KamilAdcService requires radar.model='kamil_adc'")
        if self.config.radar.driver_mode != "native":
            raise RuntimeError("Kamil ADC requires radar.driver_mode='native'")

        adc = self.config.radar.kamil_adc
        if not adc.executable_path:
            raise ValueError("radar.kamil_adc.executable_path is required")
        if not adc.tty_path:
            raise ValueError("radar.kamil_adc.tty_path is required")
        if any(arg.startswith("tty:") for arg in adc.args):
            raise ValueError("radar.kamil_adc.args must not contain tty:<path>; use tty_path instead")
        for name in ("startup_timeout_s", "sweep_timeout_s", "stop_timeout_s"):
            if getattr(adc, name) <= 0.0:
                raise ValueError(f"radar.kamil_adc.{name} must be > 0")

        project_dir = self._resolve_project_dir()
        if not project_dir.is_dir():
            raise RuntimeError(f"radar.kamil_adc.project_dir is not a directory: {project_dir}")
        executable_path = self._resolve_executable()
        if not executable_path.is_file():
            raise RuntimeError(f"radar.kamil_adc.executable_path is not a file: {executable_path}")
        if not os.access(executable_path, os.X_OK):
            raise RuntimeError(f"radar.kamil_adc.executable_path is not executable: {executable_path}")

        # Surface a malformed calibration/band at open time, not mid-acquisition.
        KamilAdcProcessingParams.from_kamil_model(adc)

    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
        params = self._processor.params  # type: ignore[union-attr]
        try:
            freqs = self._processor.reference_frequency_axis(raw.reference)  # type: ignore[union-attr]
            covered = f"[{float(np.min(freqs)):.6g}, {float(np.max(freqs)):.6g}]"
        except Exception:  # noqa: BLE001 — diagnostics must never raise
            covered = "<unavailable>"
        logger.warning(
            "Kamil ADC 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, params.band_start_hz, params.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