65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
"""Factory for single-radar Python acquisition services."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Protocol
|
|
|
|
from python_app.hardware_full.kamil_adc import KamilAdcService
|
|
from python_app.hardware_full.librevna_driver.models import SweepResult
|
|
from python_app.hardware_full.librevna_service import LibreVnaService
|
|
from python_app.hardware_full.remote_compact_m_k209_service import RemoteCompactMK209Service
|
|
from python_app.models.run_config_model import RadarSweepModel, RunConfigModel
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SingleRadarService(Protocol):
|
|
"""Common API used by single-radar workflows."""
|
|
|
|
def open(self) -> None:
|
|
"""Open radar connection."""
|
|
|
|
def close(self) -> None:
|
|
"""Close radar connection."""
|
|
|
|
def configure(self, sweep: RadarSweepModel) -> None:
|
|
"""Apply sweep settings."""
|
|
|
|
def read_device_limits(self) -> dict[str, float | int]:
|
|
"""Read device capability limits."""
|
|
|
|
def acquire(self) -> SweepResult:
|
|
"""Acquire one sweep."""
|
|
|
|
|
|
def create_single_radar_service(config: RunConfigModel) -> SingleRadarService:
|
|
"""Create the Python service for a non-matrix-radar config."""
|
|
if config.is_matrix_radar:
|
|
raise RuntimeError(
|
|
"single-radar service factory does not support matrix radars; "
|
|
"use create_matrix_radar_service instead"
|
|
)
|
|
|
|
model = config.radar.model or RunConfigModel.LIBREVNA_MODEL
|
|
logger.debug("Creating single-radar service for model=%s (driver_mode=%s)", model, config.radar.driver_mode)
|
|
if model == RunConfigModel.LIBREVNA_MODEL:
|
|
# Forward driver_mode (mirrors the matrix path): 'native' must require real
|
|
# hardware and 'mock' must use the synthetic backend — never silently the wrong one.
|
|
return LibreVnaService(serial=config.radar.serial or None, backend_mode=config.radar.driver_mode)
|
|
|
|
if model == RunConfigModel.COMPACT_M_K209_MODEL:
|
|
if config.radar.driver_mode != "native":
|
|
raise RuntimeError("Compact-M K209 requires radar.driver_mode='native'")
|
|
return RemoteCompactMK209Service(
|
|
host=config.radar.remote_host,
|
|
port=config.radar.remote_port,
|
|
)
|
|
|
|
if model == RunConfigModel.KAMIL_ADC_MODEL:
|
|
if config.radar.driver_mode != "native":
|
|
raise RuntimeError("Kamil ADC requires radar.driver_mode='native'")
|
|
return KamilAdcService(config)
|
|
|
|
raise RuntimeError(f"Unsupported single-radar model: {model}")
|