164 lines
6.0 KiB
Python
164 lines
6.0 KiB
Python
"""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.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_s21(self) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Acquire one S21 trace."""
|
|
|
|
|
|
@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_s21(self) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Acquire one S21 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 np.asarray(result.x, dtype=np.float32), 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_s21(self) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Generate synthetic S21 values using deterministic phase envelope."""
|
|
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
|
|
envelope = 0.6 + 0.4 * np.sin(phase * 0.5)
|
|
s21 = (envelope * np.cos(phase) + 1j * envelope * np.sin(phase)).astype(np.complex64)
|
|
self._mock_phase += 0.05
|
|
return freq, s21
|