108 lines
4.2 KiB
Python
108 lines
4.2 KiB
Python
"""High-level orchestration service for configuring and querying LibreVNA."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
import logging
|
|
|
|
from python_app.hardware_full.librevna_backends import LibreVnaBackend, MockLibreVnaBackend, NativeLibreVnaBackend
|
|
from python_app.hardware_full.librevna_driver.models import SweepResult
|
|
from python_app.models.run_config_model import RadarSweepModel
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LibreVnaService:
|
|
"""Provide stable API for GUI/workflows while hiding backend details."""
|
|
|
|
serial: str | None = None
|
|
strict_protocol_version: int = 14
|
|
backend_mode: str = "auto"
|
|
_driver_available: bool = field(init=False, default=False, repr=False)
|
|
_backend: LibreVnaBackend | None = field(init=False, default=None, repr=False)
|
|
_using_mock_backend: bool = field(init=False, default=False, repr=False)
|
|
|
|
def __post_init__(self) -> None:
|
|
"""Initialize selected backend and detect driver availability."""
|
|
self._driver_available = False
|
|
self._backend = None
|
|
self._using_mock_backend = False
|
|
|
|
mode = self.backend_mode.strip().lower()
|
|
if mode not in {"auto", "native", "mock"}:
|
|
raise ValueError(f"Unsupported LibreVnaService backend mode: {self.backend_mode}")
|
|
|
|
if mode == "mock":
|
|
self._backend = MockLibreVnaBackend()
|
|
self._using_mock_backend = True
|
|
return
|
|
|
|
try:
|
|
self._backend = NativeLibreVnaBackend(
|
|
serial=self.serial,
|
|
strict_protocol_version=self.strict_protocol_version,
|
|
)
|
|
self._driver_available = True
|
|
except Exception as exc:
|
|
# 'native' demands real hardware — never substitute synthetic data.
|
|
if mode == "native":
|
|
raise
|
|
# 'auto' falls back to the mock backend ONLY when the native driver
|
|
# library itself is unavailable (a dev host without the USB stack) —
|
|
# this is not device-absence (that surfaces later from open()). Log it
|
|
# loudly so it is never a silent surprise; a deployed appliance should
|
|
# set driver_mode='native' to forbid the fallback entirely.
|
|
logger.warning("LibreVNA native backend unavailable; falling back to mock (mode=auto): %s", exc)
|
|
self._backend = MockLibreVnaBackend()
|
|
self._using_mock_backend = True
|
|
|
|
@property
|
|
def driver_available(self) -> bool:
|
|
"""Return `True` when native Python LibreVNA driver is available."""
|
|
return self._driver_available
|
|
|
|
def open(self) -> None:
|
|
"""Open backend resources."""
|
|
if not self._driver_available and not self._using_mock_backend:
|
|
return
|
|
if self._backend is None:
|
|
return
|
|
self._backend.open()
|
|
|
|
def close(self) -> None:
|
|
"""Close backend resources."""
|
|
if self._backend is None:
|
|
return
|
|
self._backend.close()
|
|
|
|
def configure(self, sweep: RadarSweepModel) -> None:
|
|
"""Apply sweep settings to active backend."""
|
|
if self._backend is None:
|
|
raise RuntimeError("LibreVNA backend is not initialized")
|
|
self._backend.configure(sweep)
|
|
|
|
def read_device_limits(self) -> dict[str, float | int]:
|
|
"""Read native device limits from connected LibreVNA."""
|
|
if not self._driver_available:
|
|
raise RuntimeError("LibreVNA Python driver is not available")
|
|
if self._backend is None:
|
|
raise RuntimeError("LibreVNA backend is not initialized")
|
|
|
|
opened_here = not self._backend.is_open
|
|
try:
|
|
if opened_here:
|
|
self.open()
|
|
return self._backend.read_device_limits()
|
|
finally:
|
|
if opened_here:
|
|
self.close()
|
|
|
|
def acquire(self) -> SweepResult:
|
|
"""Acquire one sweep with all available traces from active backend."""
|
|
if self._backend is None:
|
|
raise RuntimeError("LibreVNA backend is not initialized")
|
|
if self._using_mock_backend and not self._driver_available and self.backend_mode != "mock":
|
|
raise RuntimeError("Device not found")
|
|
return self._backend.acquire()
|