408 lines
14 KiB
Python
408 lines
14 KiB
Python
"""Payload encoders/decoders for LibreVNA protocol packets."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import logging
|
|
import struct
|
|
|
|
import numpy as np
|
|
|
|
from ..enums import HardwareFamily, PacketType
|
|
from ..exceptions import ParseError, UnsupportedHardwareError
|
|
from ..models import (
|
|
DeviceConfigVariant,
|
|
DeviceInfo,
|
|
DeviceLimits,
|
|
DeviceStatus,
|
|
Packet,
|
|
VNADatapointPacket,
|
|
VNASweepSettings,
|
|
)
|
|
from .structs import DeviceStatusUnion
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
_DEVICE_INFO_STRUCT = struct.Struct("<HBBBBcQQIIHhhIIBQBH")
|
|
_SWEEP_SETTINGS_STRUCT = struct.Struct("<QQHIhBHhH")
|
|
_DEVICE_CONFIG_V1_STRUCT = struct.Struct("<IBHB")
|
|
_DEVICE_CONFIG_VFF_STRUCT = struct.Struct("<IIIBH")
|
|
_DEVICE_CONFIG_VFE_STRUCT = struct.Struct("<H")
|
|
_DEVICE_CONFIG_VD0_STRUCT = struct.Struct("<HIB")
|
|
|
|
_DEVICE_STATUS_VARIANT_BY_FAMILY: dict[HardwareFamily, str] = {
|
|
HardwareFamily.V1: "V1",
|
|
HardwareFamily.VFF: "VFF",
|
|
HardwareFamily.VFE: "VFE",
|
|
HardwareFamily.VD0: "VD0",
|
|
HardwareFamily.VE0: "VD0",
|
|
}
|
|
|
|
|
|
def _family_from_hardware_version(hardware_version: int) -> HardwareFamily:
|
|
"""Map hardware version byte to typed hardware family enum."""
|
|
try:
|
|
return HardwareFamily(hardware_version)
|
|
except ValueError as exc:
|
|
raise UnsupportedHardwareError(
|
|
f"Unsupported hardware_version in DeviceInfo: 0x{hardware_version:02X}"
|
|
) from exc
|
|
|
|
|
|
def _ensure_payload_length(payload: bytes, expected: int, *, packet_name: str) -> None:
|
|
"""Validate exact payload length for fixed-size packet structures."""
|
|
if len(payload) != expected:
|
|
raise ParseError(
|
|
f"{packet_name} payload length mismatch: expected {expected}, got {len(payload)}"
|
|
)
|
|
|
|
|
|
def parse_device_info(payload: bytes) -> DeviceInfo:
|
|
"""Parse `DeviceInfo` packet payload into typed model."""
|
|
_ensure_payload_length(payload, _DEVICE_INFO_STRUCT.size, packet_name="DeviceInfo")
|
|
|
|
(
|
|
protocol_version,
|
|
fw_major,
|
|
fw_minor,
|
|
fw_patch,
|
|
hardware_version,
|
|
hw_revision_raw,
|
|
min_freq,
|
|
max_freq,
|
|
min_ifbw,
|
|
max_ifbw,
|
|
max_points,
|
|
min_cdbm,
|
|
max_cdbm,
|
|
min_rbw,
|
|
max_rbw,
|
|
max_amplitude_points,
|
|
max_harmonic,
|
|
num_ports,
|
|
max_dwell_time_us,
|
|
) = _DEVICE_INFO_STRUCT.unpack(payload)
|
|
|
|
family = _family_from_hardware_version(hardware_version)
|
|
hw_revision = hw_revision_raw.decode("ascii", errors="replace")
|
|
|
|
limits = DeviceLimits(
|
|
min_frequency_hz=float(min_freq),
|
|
max_frequency_hz=float(max_freq),
|
|
max_frequency_harmonic_hz=float(max_harmonic),
|
|
min_ifbw_hz=float(min_ifbw),
|
|
max_ifbw_hz=float(max_ifbw),
|
|
max_points=int(max_points),
|
|
min_power_dbm=float(min_cdbm) / 100.0,
|
|
max_power_dbm=float(max_cdbm) / 100.0,
|
|
min_rbw_hz=float(min_rbw),
|
|
max_rbw_hz=float(max_rbw),
|
|
max_amplitude_points=int(max_amplitude_points),
|
|
max_dwell_time_s=float(max_dwell_time_us) * 1e-6,
|
|
)
|
|
|
|
device_info = DeviceInfo(
|
|
protocol_version=int(protocol_version),
|
|
firmware_major=int(fw_major),
|
|
firmware_minor=int(fw_minor),
|
|
firmware_patch=int(fw_patch),
|
|
firmware_version=f"{fw_major}.{fw_minor}.{fw_patch}",
|
|
hardware_version=int(hardware_version),
|
|
hardware_revision=hw_revision,
|
|
hardware_family=family,
|
|
limits=limits,
|
|
num_ports=int(num_ports),
|
|
)
|
|
logger.debug(
|
|
"Decoded DeviceInfo: protocol=%d fw=%s family=%s ports=%d",
|
|
device_info.protocol_version,
|
|
device_info.firmware_version,
|
|
device_info.hardware_family.name,
|
|
device_info.num_ports,
|
|
)
|
|
return device_info
|
|
|
|
|
|
def _device_status_variant_name_for_family(family: HardwareFamily) -> str:
|
|
"""Resolve `DeviceStatusUnion` variant name for specific hardware family."""
|
|
try:
|
|
return _DEVICE_STATUS_VARIANT_BY_FAMILY[family]
|
|
except KeyError as exc:
|
|
raise UnsupportedHardwareError(
|
|
f"Unsupported hardware family for DeviceStatus: {family!r}"
|
|
) from exc
|
|
|
|
|
|
def _structure_fields_dict(struct_obj: ctypes.Structure) -> dict[str, int | float | bool]:
|
|
"""Convert ctypes structure fields into plain Python dictionary."""
|
|
values: dict[str, int | float | bool] = {}
|
|
for entry in struct_obj._fields_:
|
|
field_name = entry[0]
|
|
if field_name.startswith("_"):
|
|
continue
|
|
raw_value = getattr(struct_obj, field_name)
|
|
|
|
if len(entry) == 3:
|
|
values[field_name] = bool(raw_value)
|
|
continue
|
|
|
|
field_type = entry[1]
|
|
if field_type in {ctypes.c_float, ctypes.c_double}:
|
|
values[field_name] = float(raw_value)
|
|
else:
|
|
values[field_name] = int(raw_value)
|
|
|
|
return values
|
|
|
|
|
|
def parse_device_status(payload: bytes, family: HardwareFamily) -> DeviceStatus:
|
|
"""Parse `DeviceStatus` payload according to active hardware family."""
|
|
_ensure_payload_length(payload, ctypes.sizeof(DeviceStatusUnion), packet_name="DeviceStatus")
|
|
|
|
union = DeviceStatusUnion()
|
|
ctypes.memmove(ctypes.addressof(union), payload, len(payload))
|
|
variant_name = _device_status_variant_name_for_family(family)
|
|
variant = getattr(union, variant_name)
|
|
raw = _structure_fields_dict(variant)
|
|
|
|
source_locked = bool(raw.get("source_locked")) if "source_locked" in raw else None
|
|
lo_locked = None
|
|
if "LO_locked" in raw:
|
|
lo_locked = bool(raw["LO_locked"])
|
|
elif "LO1_locked" in raw:
|
|
lo_locked = bool(raw["LO1_locked"])
|
|
|
|
adc_overload = bool(raw.get("ADC_overload")) if "ADC_overload" in raw else None
|
|
unlevel = bool(raw.get("unlevel")) if "unlevel" in raw else None
|
|
|
|
temperatures_c: list[float] = []
|
|
for key in ("temp_source", "temp_LO1", "temp_MCU"):
|
|
if key in raw:
|
|
temperatures_c.append(float(raw[key]))
|
|
if "temp_eCal" in raw:
|
|
temperatures_c.append(float(raw["temp_eCal"]) / 100.0)
|
|
|
|
return DeviceStatus(
|
|
family=family,
|
|
source_locked=source_locked,
|
|
lo_locked=lo_locked,
|
|
adc_overload=adc_overload,
|
|
unlevel=unlevel,
|
|
temperatures_c=temperatures_c,
|
|
raw=raw,
|
|
)
|
|
|
|
|
|
def decode_vna_datapoint_payload(payload: bytes) -> VNADatapointPacket:
|
|
"""Decode variable-size VNADatapoint payload."""
|
|
if len(payload) < 12:
|
|
raise ParseError("VNADatapoint payload is too short")
|
|
|
|
values_block = len(payload) - 12
|
|
if values_block % 9 != 0:
|
|
raise ParseError("VNADatapoint payload length is not aligned to value tuple size")
|
|
|
|
num_values = values_block // 9
|
|
|
|
(frequency_or_time,) = struct.unpack_from("<Q", payload, 0)
|
|
(cdbm,) = struct.unpack_from("<h", payload, 8)
|
|
(point_number,) = struct.unpack_from("<H", payload, 10)
|
|
|
|
real = np.frombuffer(payload, dtype="<f4", count=num_values, offset=12).astype(np.float64, copy=False)
|
|
imag = np.frombuffer(payload, dtype="<f4", count=num_values, offset=12 + 4 * num_values).astype(
|
|
np.float64,
|
|
copy=False,
|
|
)
|
|
flags = np.frombuffer(payload, dtype=np.uint8, count=num_values, offset=12 + 8 * num_values)
|
|
|
|
return VNADatapointPacket(
|
|
frequency_or_time=int(frequency_or_time),
|
|
cdbm=int(cdbm),
|
|
point_number=int(point_number),
|
|
real=real,
|
|
imag=imag,
|
|
flags=flags,
|
|
)
|
|
|
|
|
|
def decode_packet_payload(packet: Packet) -> object:
|
|
"""Decode packet payload for known response packet types."""
|
|
if packet.type == PacketType.DEVICE_INFO:
|
|
return parse_device_info(packet.payload)
|
|
if packet.type == PacketType.VNA_DATAPOINT:
|
|
return decode_vna_datapoint_payload(packet.payload)
|
|
return packet.payload
|
|
|
|
|
|
def encode_sweep_settings_payload(settings: VNASweepSettings) -> bytes:
|
|
"""Encode `SweepSettings` payload."""
|
|
if len(settings.excited_ports) > 4:
|
|
raise ValueError("Protocol supports at most four excited ports")
|
|
sync_mode = settings.sync_mode if settings.sync_mode is not None else 0
|
|
|
|
stage_by_port = {port: stage for stage, port in enumerate(settings.excited_ports)}
|
|
|
|
flags1 = 0
|
|
flags1 |= int(bool(settings.standby)) << 0
|
|
flags1 |= int(bool(settings.sync_master)) << 1
|
|
flags1 |= int(bool(settings.suppress_invalid_peaks)) << 2
|
|
flags1 |= int(bool(settings.fixed_power_setting)) << 3
|
|
flags1 |= int(settings.sweep_scale.value == "log") << 4
|
|
flags1 |= (int(sync_mode) & 0x03) << 5
|
|
|
|
flags2 = 0
|
|
stages = len(settings.excited_ports) - 1
|
|
flags2 |= stages & 0x07
|
|
flags2 |= (stage_by_port.get(1, 0) & 0x07) << 3
|
|
flags2 |= (stage_by_port.get(2, 0) & 0x07) << 6
|
|
flags2 |= (stage_by_port.get(3, 0) & 0x07) << 9
|
|
flags2 |= (stage_by_port.get(4, 0) & 0x07) << 12
|
|
|
|
dwell_us = int(round(settings.dwell_s * 1_000_000.0))
|
|
dwell_us = max(0, min(0xFFFF, dwell_us))
|
|
|
|
return _SWEEP_SETTINGS_STRUCT.pack(
|
|
int(round(settings.f_start_hz)),
|
|
int(round(settings.f_stop_hz)),
|
|
int(settings.points),
|
|
int(round(settings.if_bandwidth_hz)),
|
|
int(round(settings.power_start_dbm * 100.0)),
|
|
flags1,
|
|
flags2,
|
|
int(round(settings.power_stop_dbm * 100.0)),
|
|
dwell_us,
|
|
)
|
|
|
|
|
|
def parse_device_config(payload: bytes, family: HardwareFamily) -> DeviceConfigVariant:
|
|
"""Decode family-specific `DeviceConfiguration` payload."""
|
|
if len(payload) != 15:
|
|
raise ParseError(f"DeviceConfiguration payload length mismatch: expected 15, got {len(payload)}")
|
|
|
|
values: dict[str, int | float | bool]
|
|
if family == HardwareFamily.V1:
|
|
if1, adc_prescaler, dft_phase_inc, pll_delay = _DEVICE_CONFIG_V1_STRUCT.unpack(payload[:8])
|
|
values = {
|
|
"IF1": int(if1),
|
|
"ADCprescaler": int(adc_prescaler),
|
|
"DFTphaseInc": int(dft_phase_inc),
|
|
"PLLSettlingDelay": int(pll_delay),
|
|
}
|
|
elif family == HardwareFamily.VFF:
|
|
ip, mask, gw, flags1, flags2 = _DEVICE_CONFIG_VFF_STRUCT.unpack(payload)
|
|
values = {
|
|
"ip": int(ip),
|
|
"mask": int(mask),
|
|
"gw": int(gw),
|
|
"dhcp": bool(flags1 & 0x01),
|
|
"autogain": bool(flags2 & 0x01),
|
|
"portGain": int((flags2 >> 1) & 0x0F),
|
|
"refGain": int((flags2 >> 5) & 0x0F),
|
|
}
|
|
elif family == HardwareFamily.VFE:
|
|
(flags,) = _DEVICE_CONFIG_VFE_STRUCT.unpack(payload[:2])
|
|
values = {
|
|
"autogain": bool(flags & 0x01),
|
|
"portGain": int((flags >> 1) & 0x0F),
|
|
"refGain": int((flags >> 5) & 0x0F),
|
|
}
|
|
elif family in {HardwareFamily.VD0, HardwareFamily.VE0}:
|
|
dft_phase_inc, adc_rate, pll_delay = _DEVICE_CONFIG_VD0_STRUCT.unpack(payload[:7])
|
|
values = {
|
|
"DFTphaseInc": int(dft_phase_inc),
|
|
"ADCrate": int(adc_rate),
|
|
"PLLSettlingDelay": int(pll_delay),
|
|
}
|
|
else:
|
|
raise UnsupportedHardwareError(f"Unsupported hardware family for DeviceConfiguration: {family!r}")
|
|
|
|
return DeviceConfigVariant(family=family, values=values)
|
|
|
|
|
|
def encode_device_config_payload(config: DeviceConfigVariant) -> bytes:
|
|
"""Encode family-specific `DeviceConfiguration` payload."""
|
|
family = config.family
|
|
values = config.values
|
|
|
|
def _require_value(key: str) -> int | float | bool:
|
|
"""Fetch required device-config value by key or raise parse error."""
|
|
if key not in values:
|
|
raise ParseError(f"Missing required device configuration field: '{key}'")
|
|
return values[key]
|
|
|
|
if family == HardwareFamily.V1:
|
|
payload = _DEVICE_CONFIG_V1_STRUCT.pack(
|
|
int(_require_value("IF1")),
|
|
int(_require_value("ADCprescaler")),
|
|
int(_require_value("DFTphaseInc")),
|
|
int(_require_value("PLLSettlingDelay")),
|
|
)
|
|
elif family == HardwareFamily.VFF:
|
|
flags1 = int(bool(_require_value("dhcp")))
|
|
flags2 = 0
|
|
flags2 |= int(bool(_require_value("autogain"))) << 0
|
|
flags2 |= (int(_require_value("portGain")) & 0x0F) << 1
|
|
flags2 |= (int(_require_value("refGain")) & 0x0F) << 5
|
|
payload = _DEVICE_CONFIG_VFF_STRUCT.pack(
|
|
int(_require_value("ip")),
|
|
int(_require_value("mask")),
|
|
int(_require_value("gw")),
|
|
flags1,
|
|
flags2,
|
|
)
|
|
elif family == HardwareFamily.VFE:
|
|
flags = 0
|
|
flags |= int(bool(_require_value("autogain"))) << 0
|
|
flags |= (int(_require_value("portGain")) & 0x0F) << 1
|
|
flags |= (int(_require_value("refGain")) & 0x0F) << 5
|
|
payload = _DEVICE_CONFIG_VFE_STRUCT.pack(flags)
|
|
elif family in {HardwareFamily.VD0, HardwareFamily.VE0}:
|
|
payload = _DEVICE_CONFIG_VD0_STRUCT.pack(
|
|
int(_require_value("DFTphaseInc")),
|
|
int(_require_value("ADCrate")),
|
|
int(_require_value("PLLSettlingDelay")),
|
|
)
|
|
else:
|
|
raise UnsupportedHardwareError(f"Unsupported hardware family for DeviceConfiguration: {family!r}")
|
|
|
|
return payload.ljust(15, b"\x00")
|
|
|
|
|
|
def encode_packet_payload(packet_type: PacketType, payload: object) -> bytes:
|
|
"""Generic payload encoder for `LibreVNASession.send()` raw API."""
|
|
if payload is None:
|
|
return b""
|
|
if isinstance(payload, (bytes, bytearray, memoryview)):
|
|
return bytes(payload)
|
|
|
|
if packet_type == PacketType.SWEEP_SETTINGS and isinstance(payload, VNASweepSettings):
|
|
return encode_sweep_settings_payload(payload)
|
|
if packet_type == PacketType.DEVICE_CONFIGURATION and isinstance(payload, DeviceConfigVariant):
|
|
return encode_device_config_payload(payload)
|
|
|
|
logger.error(
|
|
"Unsupported payload object for packet %s: %s",
|
|
packet_type.name,
|
|
type(payload).__name__,
|
|
)
|
|
raise TypeError(f"Unsupported payload object for packet {packet_type.name}: {type(payload).__name__}")
|
|
|
|
|
|
NO_PAYLOAD_PACKET_TYPES = {
|
|
PacketType.ACK,
|
|
PacketType.NACK,
|
|
PacketType.REQUEST_DEVICE_INFO,
|
|
PacketType.REQUEST_DEVICE_CONFIGURATION,
|
|
PacketType.REQUEST_DEVICE_STATUS,
|
|
PacketType.INITIATE_SWEEP,
|
|
PacketType.RESET_DEVICE_CONFIGURATION,
|
|
}
|
|
|
|
|
|
def ensure_no_payload_types(packet_type: PacketType, payload: bytes) -> None:
|
|
"""Validate empty payload requirement for no-payload packet types."""
|
|
if packet_type in NO_PAYLOAD_PACKET_TYPES and payload:
|
|
logger.error("No-payload packet %s received payload length %d", packet_type.name, len(payload))
|
|
raise ParseError(f"Packet {packet_type.name} does not support payload")
|