118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""Main user-facing LibreVNA direct-USB device class."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from types import TracebackType
|
|
|
|
from .api.config import ConfigController
|
|
from .api.generator import GeneratorController
|
|
from .api.vna import VNAController
|
|
from .enums import PacketType
|
|
from .models import DeviceInfo, DeviceStatus, Packet, USBDeviceDescriptor
|
|
from .session import LibreVNASession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LibreVNADevice:
|
|
"""Main entry point for direct USB control of LibreVNA."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Create device facade with session-backed VNA/config controllers."""
|
|
self._session = LibreVNASession()
|
|
|
|
self.vna = VNAController(self._session)
|
|
self.generator = GeneratorController(self._session)
|
|
self.config = ConfigController(self._session)
|
|
|
|
def __enter__(self) -> LibreVNADevice:
|
|
"""Return self to support context-managed lifetime in host applications."""
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc: BaseException | None,
|
|
tb: TracebackType | None,
|
|
) -> None:
|
|
"""Always close USB session when leaving context manager block."""
|
|
self.disconnect()
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
"""Return connection state of underlying USB session."""
|
|
return self._session.is_connected
|
|
|
|
@property
|
|
def connected_serial(self) -> str | None:
|
|
"""Return currently connected serial number when available."""
|
|
return self._session.connected_serial
|
|
|
|
@staticmethod
|
|
def list_devices() -> list[USBDeviceDescriptor]:
|
|
"""Return all currently discoverable LibreVNA USB devices."""
|
|
devices = LibreVNASession.list_devices()
|
|
logger.debug("Discovered %d LibreVNA USB device(s)", len(devices))
|
|
return devices
|
|
|
|
def connect(
|
|
self,
|
|
serial: str | None = None,
|
|
strict_protocol_version: int = 14,
|
|
timeout_s: float = 1.0,
|
|
) -> None:
|
|
"""Connect to a LibreVNA device by optional serial number."""
|
|
logger.info(
|
|
"Connecting to LibreVNA (serial=%s, strict_protocol=%d, timeout=%.2fs)",
|
|
serial,
|
|
strict_protocol_version,
|
|
timeout_s,
|
|
)
|
|
self._session.connect(
|
|
serial=serial,
|
|
strict_protocol_version=strict_protocol_version,
|
|
timeout_s=timeout_s,
|
|
)
|
|
logger.info("Connected to LibreVNA (serial=%s)", self.connected_serial)
|
|
|
|
def disconnect(self) -> None:
|
|
"""Disconnect USB transport and clear runtime state."""
|
|
logger.info("Disconnecting LibreVNA session")
|
|
self._session.disconnect()
|
|
logger.info("LibreVNA session disconnected")
|
|
|
|
def send(self, packet: Packet, *, require_ack: bool = True, timeout_s: float = 0.5) -> None:
|
|
"""Send low-level protocol packet."""
|
|
logger.debug(
|
|
"Sending packet %s (require_ack=%s, timeout=%.2fs)",
|
|
packet.type.name,
|
|
require_ack,
|
|
timeout_s,
|
|
)
|
|
self._session.send(packet, require_ack=require_ack, timeout_s=timeout_s)
|
|
|
|
def request(
|
|
self,
|
|
packet_type: PacketType,
|
|
response_type: PacketType,
|
|
*,
|
|
timeout_s: float = 1.0,
|
|
) -> Packet:
|
|
"""Send no-payload request packet and wait for a response packet."""
|
|
logger.debug(
|
|
"Request packet=%s response=%s timeout=%.2fs",
|
|
packet_type.name,
|
|
response_type.name,
|
|
timeout_s,
|
|
)
|
|
return self._session.request(packet_type, response_type, timeout_s=timeout_s)
|
|
|
|
def get_device_info(self) -> DeviceInfo:
|
|
"""Return cached device info from handshake."""
|
|
return self._session.get_device_info()
|
|
|
|
def get_device_status(self, *, timeout_s: float = 1.0) -> DeviceStatus:
|
|
"""Query and return current device status."""
|
|
return self._session.get_device_status(timeout_s=timeout_s)
|