"""Backend adapters used by :mod:`python_app.hardware_full.librevna_service`.""" from __future__ import annotations from dataclasses import dataclass, field import math from typing import Any, Protocol import numpy as np from python_app.hardware_full.librevna_driver.models import SweepResult from python_app.models.run_config_model import RadarSweepModel class LibreVnaBackend(Protocol): """Minimal backend contract required by `LibreVnaService`.""" def open(self) -> None: """Open backend resources.""" def close(self) -> None: """Close backend resources.""" @property def is_open(self) -> bool: """Return whether backend currently holds open runtime resources.""" def configure(self, sweep: RadarSweepModel) -> None: """Apply sweep settings.""" def read_device_limits(self) -> dict[str, float | int]: """Query runtime device limits.""" def acquire(self) -> SweepResult: """Acquire one sweep with all available traces.""" @dataclass(slots=True) class NativeLibreVnaBackend: """Native backend using direct USB LibreVNA Python driver.""" serial: str | None strict_protocol_version: int _device: Any | None = field(init=False, default=None, repr=False) _settings: RadarSweepModel | None = field(init=False, default=None, repr=False) _libre_vna_device: type[Any] | None = field(init=False, default=None, repr=False) _vna_sweep_settings: type[Any] | None = field(init=False, default=None, repr=False) def __post_init__(self) -> None: """Import native driver classes lazily and initialize backend state.""" from python_app.hardware_full.librevna_driver import LibreVNADevice, VNASweepSettings self._device: LibreVNADevice | None = None self._settings: RadarSweepModel | None = None self._libre_vna_device = LibreVNADevice self._vna_sweep_settings = VNASweepSettings def open(self) -> None: """Open device connection if it is not already open.""" if self._device is not None: return self._device = self._libre_vna_device() self._device.connect(serial=self.serial, strict_protocol_version=self.strict_protocol_version, timeout_s=2.0) @property def is_open(self) -> bool: """Return `True` when native device connection is open.""" return self._device is not None def close(self) -> None: """Close device connection when open.""" if self._device is None: return self._device.disconnect() self._device = None def configure(self, sweep: RadarSweepModel) -> None: """Store and apply sweep settings to connected hardware.""" self._settings = sweep if self._device is None: return settings = self._vna_sweep_settings( f_start_hz=sweep.start_hz, f_stop_hz=sweep.stop_hz, points=sweep.points, if_bandwidth_hz=sweep.if_bandwidth_hz, power_start_dbm=sweep.power_dbm, power_stop_dbm=sweep.power_dbm, excited_ports=(1, 2), ) self._device.vna.configure(settings) def read_device_limits(self) -> dict[str, float | int]: """Read frequency/IFBW/power/points limits from connected device.""" if self._device is None: raise RuntimeError("Failed to connect to LibreVNA device") device_info = self._device.get_device_info() limits = device_info.limits return { "min_frequency_hz": float(limits.min_frequency_hz), "max_frequency_hz": float(limits.max_frequency_hz), "min_ifbw_hz": float(limits.min_ifbw_hz), "max_ifbw_hz": float(limits.max_ifbw_hz), "max_points": int(limits.max_points), "min_power_dbm": float(limits.min_power_dbm), "max_power_dbm": float(limits.max_power_dbm), } def acquire(self) -> SweepResult: """Acquire one sweep from hardware.""" if self._settings is None: raise RuntimeError("Radar service is not configured") if self._device is None: raise RuntimeError("Device not found") result = self._device.vna.acquire(expected_points=self._settings.points, timeout_s=20.0) return SweepResult( x=np.asarray(result.x, dtype=np.float32), traces={ "s11": np.asarray(result.trace("s11"), dtype=np.complex64), "s21": np.asarray(result.trace("s21"), dtype=np.complex64), }, ) @dataclass(slots=True) class MockLibreVnaBackend: """Synthetic backend used for local development and tests.""" _settings: RadarSweepModel | None = field(init=False, default=None, repr=False) _mock_phase: float = field(init=False, default=0.0, repr=False) def __post_init__(self) -> None: """Initialize mock backend state.""" self._settings = None self._mock_phase = 0.0 def open(self) -> None: """No-op for mock backend.""" def close(self) -> None: """No-op for mock backend.""" @property def is_open(self) -> bool: """Mock backend does not hold external resources.""" return False def configure(self, sweep: RadarSweepModel) -> None: """Store sweep settings used by synthetic acquisition.""" self._settings = sweep def read_device_limits(self) -> dict[str, float | int]: """Mock backend does not support native device limits queries.""" raise RuntimeError("LibreVNA Python driver is not available") def acquire(self) -> SweepResult: """Generate synthetic S11 and S21 values using deterministic envelopes.""" if self._settings is None: raise RuntimeError("Radar service is not configured") points = self._settings.points freq = np.linspace(self._settings.start_hz, self._settings.stop_hz, points, dtype=np.float32) phase = (2.0 * math.pi * np.linspace(0.0, 1.0, points, dtype=np.float32)) + self._mock_phase s21_envelope = 0.6 + 0.4 * np.sin(phase * 0.5) s11_envelope = 0.25 + 0.15 * np.cos(phase * 0.75) s21 = (s21_envelope * np.cos(phase) + 1j * s21_envelope * np.sin(phase)).astype(np.complex64) reflected_phase = phase * 0.6 + 0.8 s11 = ( s11_envelope * np.cos(reflected_phase) + 1j * s11_envelope * np.sin(reflected_phase) ).astype(np.complex64) self._mock_phase += 0.05 return SweepResult( x=freq, traces={ "s11": s11, "s21": s21, }, )