per-point switching: hardware bring-up, profiles, protocol doc
Verified on the bench that the modified master bitstream excites in every stage and the RF switch routes: 16 traces per sweep, 62 dB between the two switch positions of one master port, 9.9k measurements/s. Adds the built master image, two run profiles, a verification script and librevna_protocol.md, and drops the stale prototype driver copy that the firmware tree still carried.
This commit is contained in:
Ayzen
-10
@@ -1,10 +0,0 @@
|
||||
from .librevna_usb_transport import list_connected_device_serial_numbers
|
||||
from .multi_device_vna_controller import MultiDeviceVnaController
|
||||
from .sweep_measurement_models import SweepConfiguration, SweepMeasurementResult
|
||||
|
||||
__all__ = [
|
||||
"MultiDeviceVnaController",
|
||||
"SweepConfiguration",
|
||||
"SweepMeasurementResult",
|
||||
"list_connected_device_serial_numbers",
|
||||
]
|
||||
-383
@@ -1,383 +0,0 @@
|
||||
LibreVNA вообще, что именно делает наш текущий Python-код, какие пакеты мы используем, как они кодируются и парсятся, и почему `_power_level_centidecibels_milliwatt` сейчас не нужен.
|
||||
|
||||
**1. Где вообще начинается протокол**
|
||||
У LibreVNA поверх USB идёт свой бинарный протокол, а не `tty/serial`. Это видно и в документации, и в firmware:
|
||||
- USB endpoints: [Device_protocol_v13.tex](/home/europa/Documents/LibreVNA/Documentation/DeveloperInfo/Device_protocol_v13.tex:176)
|
||||
- актуальный код firmware: [Protocol.hpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.hpp:573)
|
||||
|
||||
На USB уровне:
|
||||
- `0x01` — host -> device
|
||||
- `0x81` — device -> host
|
||||
- `0x82` — debug ASCII logs
|
||||
|
||||
В нашем коде data transport работает только через:
|
||||
- [librevna_usb_transport.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_usb_transport.py)
|
||||
|
||||
То есть:
|
||||
- мы шлём команды в `0x01`
|
||||
- читаем протокольные пакеты из `0x81`
|
||||
- `0x82` вообще не используем
|
||||
|
||||
**2. Общая структура любого LibreVNA packet**
|
||||
Общий framing один и тот же для почти всех packet types. Это описано в docs и реализовано в firmware:
|
||||
- docs: [Device_protocol_v13.tex](/home/europa/Documents/LibreVNA/Documentation/DeveloperInfo/Device_protocol_v13.tex:208)
|
||||
- firmware decode/encode: [Protocol.cpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.cpp:5)
|
||||
- field offsets/constants: [PacketConstants.h](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/PacketConstants.h:10)
|
||||
|
||||
Структура такая:
|
||||
1. `Header` — 1 байт, всегда `0x5A`
|
||||
2. `Length` — 2 байта, общая длина всего packet целиком
|
||||
3. `Type` — 1 байт, номер packet type
|
||||
4. `Payload` — переменная длина
|
||||
5. `CRC32` — 4 байта
|
||||
|
||||
В Python это делает:
|
||||
- `encode_protocol_frame(...)` в [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:31)
|
||||
- `extract_complete_protocol_frames(...)` в [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:45)
|
||||
|
||||
`encode_protocol_frame(...)`:
|
||||
- считает `total_frame_length = 8 + len(payload)`
|
||||
- пакует header/length/type
|
||||
- считает CRC32
|
||||
- дописывает CRC в конец
|
||||
|
||||
`extract_complete_protocol_frames(...)`:
|
||||
- получает поток байт, а не “готовый packet”
|
||||
- ищет `0x5A`
|
||||
- читает `length`
|
||||
- если полного frame ещё нет, оставляет хвост в буфере
|
||||
- если frame полный, возвращает `(packet_type, payload)`
|
||||
|
||||
Очень важно:
|
||||
- наш Python сейчас **не проверяет CRC**
|
||||
- firmware проверяет CRC для почти всех пакетов, но для `VNADatapoint` у них особый случай: CRC всегда `0`, чтобы не тратить время на вычисление CRC для каждого datapoint, см. [Protocol.cpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.cpp:69) и [Protocol.cpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.cpp:147)
|
||||
|
||||
То есть сейчас Python делает “framing by magic + length”, но не “integrity check”.
|
||||
|
||||
**3. Какие packet types вообще бывают**
|
||||
Полный актуальный enum лежит тут:
|
||||
- [Protocol.hpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.hpp:573)
|
||||
|
||||
Там есть много типов:
|
||||
- `SweepSettings = 2`
|
||||
- `ManualStatus = 3`
|
||||
- `ManualControl = 4`
|
||||
- `DeviceInfo = 5`
|
||||
- `FirmwarePacket = 6`
|
||||
- `Ack = 7`
|
||||
- `ClearFlash = 8`
|
||||
- `PerformFirmwareUpdate = 9`
|
||||
- `Nack = 10`
|
||||
- `Reference = 11`
|
||||
- `Generator = 12`
|
||||
- `SpectrumAnalyzerSettings = 13`
|
||||
- `SpectrumAnalyzerResult = 14`
|
||||
- `RequestDeviceInfo = 15`
|
||||
- ...
|
||||
- `SetIdle = 20`
|
||||
- `DeviceStatus = 25`
|
||||
- `RequestDeviceStatus = 26`
|
||||
- `VNADatapoint = 27`
|
||||
- `SetTrigger = 28`
|
||||
- `ClearTrigger = 29`
|
||||
- `InitiateSweep = 32`
|
||||
- `PerformAction = 33`
|
||||
|
||||
Но наш текущий драйвер использует только маленький поднабор. Это зашито в:
|
||||
- [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:9)
|
||||
|
||||
Мы используем:
|
||||
- `SweepSettings = 2`
|
||||
- `Ack = 7`
|
||||
- `Nack = 10`
|
||||
- `Reference = 11`
|
||||
- `SetIdle = 20`
|
||||
- `VNADatapoint = 27`
|
||||
|
||||
То есть текущий драйвер — это именно VNA sweep driver, а не “полная реализация всех режимов прибора”.
|
||||
|
||||
**4. Какие именно пакеты использует наш драйвер**
|
||||
Разберу их по смыслу.
|
||||
|
||||
**`ReferenceSettings` / packet type `11`**
|
||||
Структура в firmware:
|
||||
- [Protocol.hpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.hpp:186)
|
||||
|
||||
Поля:
|
||||
- `uint32 ExtRefOutputFreq`
|
||||
- битовое поле:
|
||||
- `AutomaticSwitch`
|
||||
- `UseExternalRef`
|
||||
|
||||
В Python это строит:
|
||||
- `build_reference_settings_payload(...)` в [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:74)
|
||||
|
||||
Сейчас мы шлём:
|
||||
- `output_frequency_hz = 0`
|
||||
- `force_external_reference = True`
|
||||
|
||||
То есть:
|
||||
- никакой `Ref Out` не включаем
|
||||
- просто принудительно говорим устройству использовать внешний `10 MHz`
|
||||
|
||||
Это применяется в controller:
|
||||
- [multi_device_vna_controller.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/multi_device_vna_controller.py:183)
|
||||
|
||||
**`SweepSettings` / packet type `2`**
|
||||
Структура в firmware:
|
||||
- [Protocol.hpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.hpp:155)
|
||||
|
||||
Поля:
|
||||
- `f_start`
|
||||
- `f_stop`
|
||||
- `points`
|
||||
- `if_bandwidth`
|
||||
- `cdbm_excitation_start`
|
||||
- bitfield config:
|
||||
- `standby`
|
||||
- `syncMaster`
|
||||
- `suppressPeaks`
|
||||
- `fixedPowerSetting`
|
||||
- `logSweep`
|
||||
- `syncMode`
|
||||
- stage mapping bitfield
|
||||
- `cdbm_excitation_stop`
|
||||
- `dwell_time`
|
||||
|
||||
В Python это строит:
|
||||
- `build_sweep_settings_payload(...)` в [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:92)
|
||||
|
||||
Что мы туда кладём:
|
||||
- start/stop/points/ifbw — из `SweepConfiguration`
|
||||
- power start/stop — одинаковые
|
||||
- `fixedPowerSetting = 1`
|
||||
- `suppressPeaks = 1`
|
||||
- `logSweep = 0`
|
||||
- `syncMode = 3`, если есть slaves
|
||||
- `syncMaster = 1` только у master
|
||||
- `port_1_stimulus_stage = 0` для master, `1` для slave
|
||||
|
||||
Смысл:
|
||||
- master возбуждает `Port 1`
|
||||
- slaves только принимают
|
||||
- sync идёт через hardware trigger path
|
||||
|
||||
Этот packet реально запускает VNA setup в firmware:
|
||||
- [App.cpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/App.cpp:126)
|
||||
|
||||
**`SetIdle` / packet type `20`**
|
||||
Очень простой packet, без payload.
|
||||
В firmware:
|
||||
- [App.cpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/App.cpp:196)
|
||||
|
||||
Он переводит устройство в idle:
|
||||
- выключает активный режим
|
||||
- останавливает sweep path
|
||||
|
||||
У нас он используется только как service-command:
|
||||
- остановить continuous sweep
|
||||
- cleanup при ошибке
|
||||
- `close()`
|
||||
|
||||
Это делает controller:
|
||||
- [multi_device_vna_controller.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/multi_device_vna_controller.py:173)
|
||||
|
||||
**`Ack` / `Nack` / packet types `7` и `10`**
|
||||
Это ответы устройства на команды host.
|
||||
В firmware:
|
||||
- enum: [Protocol.hpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.hpp:581)
|
||||
- отправка `Ack`/`Nack`: [App.cpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/App.cpp:129)
|
||||
|
||||
В transport:
|
||||
- `wait_for_acknowledgement(...)` в [librevna_usb_transport.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_usb_transport.py:62)
|
||||
|
||||
Логика:
|
||||
- читаем входящие пакеты
|
||||
- если пришёл `Ack`, команда считается применённой
|
||||
- если пришёл `Nack`, бросаем exception
|
||||
- прочие asynchronous packets просто пропускаем
|
||||
|
||||
**`VNADatapoint` / packet type `27`**
|
||||
Это главный поток измерения.
|
||||
Структура в firmware:
|
||||
- template `VNADatapoint<s>`: [Protocol.hpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.hpp:25)
|
||||
- декодирование: [Protocol.cpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.cpp:79)
|
||||
|
||||
Payload такой:
|
||||
1. `frequency` — `uint64`, 8 байт
|
||||
2. `cdBm` — `int16`, 2 байта
|
||||
3. `pointNum` — `uint16`, 2 байта
|
||||
4. `real_values` — `n * float32`
|
||||
5. `imag_values` — `n * float32`
|
||||
6. `descr_values` — `n * uint8`
|
||||
|
||||
То есть фиксированный header payload имеет длину 12 байт:
|
||||
- `8 + 2 + 2`
|
||||
|
||||
А дальше идёт `n` значений, каждое занимает:
|
||||
- `4` байта real
|
||||
- `4` байта imag
|
||||
- `1` байт descriptor
|
||||
- итого `9` байт на одно значение
|
||||
|
||||
В Python это парсится тут:
|
||||
- `parse_vna_datapoint_payload(...)` в [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:134)
|
||||
|
||||
Алгоритм:
|
||||
- читаем `frequency_hz`, `power`, `point_index` через `struct.unpack_from("<QhH", ...)`
|
||||
- считаем `value_count = (len(payload) - 12) // 9`
|
||||
- читаем массив `real_components`
|
||||
- читаем массив `imaginary_components`
|
||||
- читаем массив `description_masks`
|
||||
- собираем `dict[mask] = complex(real, imag)`
|
||||
|
||||
**5. Что такое `description_mask` внутри datapoint**
|
||||
Это очень важный байт. Он говорит:
|
||||
- какой это порт
|
||||
- reference это или нет
|
||||
- к какому stage относится значение
|
||||
|
||||
Точные offsets:
|
||||
- [PacketConstants.h](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/PacketConstants.h:41)
|
||||
|
||||
Биты:
|
||||
- bit 0 = `P1`
|
||||
- bit 1 = `P2`
|
||||
- bit 2 = `P3`
|
||||
- bit 3 = `P4`
|
||||
- bit 4 = `Ref`
|
||||
- bits 5..7 = `stage`
|
||||
|
||||
В firmware это же отражено и через enum `Source`:
|
||||
- [Protocol.hpp](/home/europa/Documents/LibreVNA/Software/VNA_embedded/Application/Communication/Protocol.hpp:17)
|
||||
|
||||
В Python поиск по этому байту делает:
|
||||
- `find_receiver_value(...)` в [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:174)
|
||||
|
||||
Она:
|
||||
- фильтрует по `stage`
|
||||
- фильтрует по `reference/non-reference`
|
||||
- если задан `required_port_number`, фильтрует по порту
|
||||
|
||||
Например:
|
||||
- `stage 0 + ref=True` — это master reference receiver
|
||||
- `stage 0 + port=1 + ref=False` — это сигнал на receiver port 1
|
||||
- `stage 0 + port=2 + ref=False` — это сигнал на receiver port 2
|
||||
|
||||
**6. Как `VNADatapoint` превращается в S-параметры**
|
||||
Устройство само `S11/S21/S31` не считает.
|
||||
Оно шлёт только сырые complex receiver values.
|
||||
|
||||
В нашей новой разбивке это делает модуль:
|
||||
- [running_sweep_cycle_collection.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/running_sweep_cycle_collection.py)
|
||||
|
||||
Там:
|
||||
- master datapoint даёт:
|
||||
- master ref
|
||||
- `S11` raw
|
||||
- `S21` raw
|
||||
- каждый slave datapoint даёт:
|
||||
- два своих приёмных канала, которые потом становятся `S31/S41`, `S51/S61`, и так далее
|
||||
|
||||
Нормализация:
|
||||
- `calculate_last_cycle_s_parameters(...)` в [running_sweep_cycle_collection.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/running_sweep_cycle_collection.py:226)
|
||||
|
||||
Формула:
|
||||
- все каналы делятся на `master_reference_measurements[-1]`
|
||||
|
||||
Это принципиально важно:
|
||||
- даже `slave` каналы делятся не на локальный slave `Ref`, а на master reference, потому что именно master возбуждает сигнал
|
||||
|
||||
**7. Почему `_power_level_centidecibels_milliwatt` не используется**
|
||||
Это переменная вот здесь:
|
||||
- [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:143)
|
||||
|
||||
Строка:
|
||||
```python
|
||||
frequency_hz, _power_level_centidecibels_milliwatt, point_index = struct.unpack_from(
|
||||
"<QhH",
|
||||
payload,
|
||||
0,
|
||||
)
|
||||
```
|
||||
|
||||
Почему она есть:
|
||||
- потому что в payload `VNADatapoint` поле мощности реально существует
|
||||
- порядок полей такой: `frequency`, `cdBm`, `pointNum`
|
||||
- чтобы корректно распаковать `point_index`, нам надо прочитать и среднее поле тоже
|
||||
|
||||
Почему она с `_`:
|
||||
- это стандартный Python-сигнал “поле прочитали, но намеренно не используем”
|
||||
|
||||
Почему мы её не используем по смыслу:
|
||||
1. Для сбора sweep-циклов она не нужна.
|
||||
Нам нужны только:
|
||||
- частота
|
||||
- индекс точки
|
||||
- массив complex receiver values
|
||||
2. Для вычисления `S` она тоже не нужна.
|
||||
`S` считаются как отношение signal/ref.
|
||||
3. В нашем текущем sweep builder мы вообще задаём фиксированную мощность:
|
||||
- `cdbm_excitation_start == cdbm_excitation_stop`
|
||||
- `fixedPowerSetting = 1`
|
||||
см. [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py:108)
|
||||
|
||||
То есть в вашем текущем драйвере это поле практически просто дублирует уже известный нам факт о sweep configuration.
|
||||
|
||||
Когда оно могло бы понадобиться:
|
||||
- если бы вы хотели логировать реальную мощность каждой точки
|
||||
- если бы sweep был с изменяющейся мощностью по частоте
|
||||
- если бы нужно было sanity-check’ить, что device реально шлёт ожидаемый power profile
|
||||
- если бы вы строили экспорт “frequency + power + S-parameters”
|
||||
|
||||
Итого:
|
||||
- поле реально есть в протоколе
|
||||
- парсер его читает
|
||||
- но текущему драйверу для работы оно не нужно
|
||||
- поэтому имя с `_` здесь абсолютно осознанное и корректное
|
||||
|
||||
**8. Что именно у нас сейчас реализовано не полностью**
|
||||
Это тоже важно понимать.
|
||||
|
||||
Наш текущий Python:
|
||||
- реализует только subset packet types
|
||||
- не делает `RequestDeviceInfo`
|
||||
- не проверяет `ProtocolVersion`
|
||||
- не проверяет CRC входящих пакетов
|
||||
- не использует debug endpoint `0x82`
|
||||
|
||||
То есть это не “полный client protocol stack”, а достаточно чистый и узкий VNA multi-device driver.
|
||||
|
||||
**9. Краткая карта текущих protocol-функций**
|
||||
В новом пакете роли такие:
|
||||
|
||||
- [librevna_device_protocol.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_device_protocol.py)
|
||||
- описание packet ids
|
||||
- frame encode
|
||||
- frame split
|
||||
- payload builders
|
||||
- datapoint parser
|
||||
- descriptor-mask lookup
|
||||
|
||||
- [librevna_usb_transport.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/librevna_usb_transport.py)
|
||||
- raw USB bulk I/O
|
||||
- receive thread
|
||||
- packet queue
|
||||
- waiting for `Ack`
|
||||
|
||||
- [multi_device_vna_controller.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/multi_device_vna_controller.py)
|
||||
- high-level commands:
|
||||
- configure continuous sweep
|
||||
- collect cycles
|
||||
- stop/close
|
||||
|
||||
- [running_sweep_cycle_collection.py](/home/europa/Documents/LibreVNA/Python/librevna_multi_device_driver/running_sweep_cycle_collection.py)
|
||||
- datapoint collection threads
|
||||
- alignment to full cycle
|
||||
- raw receiver accumulation
|
||||
- normalization to `S`-parameters
|
||||
|
||||
Если хотите, следующим сообщением я могу сделать ещё более низкоуровневый разбор:
|
||||
- прям показать байтовую раскладку одного `SweepSettings` packet
|
||||
или
|
||||
- показать, как из конкретного `VNADatapoint` вручную достать `S21` и `S41`.
|
||||
-191
@@ -1,191 +0,0 @@
|
||||
import binascii
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
from .sweep_measurement_models import SweepConfiguration
|
||||
|
||||
|
||||
class PacketType:
|
||||
SWEEP_SETTINGS = 2
|
||||
ACKNOWLEDGE = 7
|
||||
NEGATIVE_ACKNOWLEDGE = 10
|
||||
REFERENCE_SETTINGS = 11
|
||||
SET_IDLE = 20
|
||||
VNA_DATAPOINT = 27
|
||||
|
||||
|
||||
SYNC_MODE_HARDWARE_TRIGGER = 3
|
||||
FRAME_HEADER_MAGIC = 0x5A
|
||||
|
||||
|
||||
@dataclass
|
||||
class ParsedVnaDatapoint:
|
||||
"""Decoded contents of one VNADatapoint payload."""
|
||||
|
||||
frequency_hz: int
|
||||
point_index: int
|
||||
receiver_values_by_description_mask: dict[int, complex]
|
||||
|
||||
|
||||
def encode_protocol_frame(packet_type: int, payload: bytes = b"") -> bytes:
|
||||
"""Wrap a packet payload in the LibreVNA transport frame."""
|
||||
|
||||
total_frame_length = 8 + len(payload)
|
||||
frame_without_crc = struct.pack(
|
||||
"<BHB",
|
||||
FRAME_HEADER_MAGIC,
|
||||
total_frame_length,
|
||||
packet_type,
|
||||
) + payload
|
||||
crc32 = binascii.crc32(frame_without_crc) & 0xFFFF_FFFF
|
||||
return frame_without_crc + struct.pack("<I", crc32)
|
||||
|
||||
|
||||
def extract_complete_protocol_frames(
|
||||
buffered_bytes: bytes,
|
||||
) -> tuple[list[tuple[int, bytes]], bytes]:
|
||||
"""
|
||||
Extract every complete protocol frame from a byte stream buffer.
|
||||
|
||||
Returns a tuple of:
|
||||
1. a list of `(packet_type, payload)` tuples
|
||||
2. the unconsumed trailing bytes that do not yet form a full frame
|
||||
"""
|
||||
|
||||
extracted_packets: list[tuple[int, bytes]] = []
|
||||
while len(buffered_bytes) >= 4:
|
||||
if buffered_bytes[0] != FRAME_HEADER_MAGIC:
|
||||
buffered_bytes = buffered_bytes[1:]
|
||||
continue
|
||||
|
||||
total_frame_length = struct.unpack_from("<H", buffered_bytes, 1)[0]
|
||||
if len(buffered_bytes) < total_frame_length:
|
||||
break
|
||||
|
||||
packet_type = buffered_bytes[3]
|
||||
payload = buffered_bytes[4 : total_frame_length - 4]
|
||||
extracted_packets.append((packet_type, payload))
|
||||
buffered_bytes = buffered_bytes[total_frame_length:]
|
||||
|
||||
return extracted_packets, buffered_bytes
|
||||
|
||||
|
||||
def build_reference_settings_payload(
|
||||
output_frequency_hz: int,
|
||||
force_external_reference: bool,
|
||||
) -> bytes:
|
||||
"""Build the ReferenceSettings payload."""
|
||||
|
||||
automatic_reference_switch = 0
|
||||
external_input_configuration_bits = (
|
||||
automatic_reference_switch << 0
|
||||
| force_external_reference << 1
|
||||
)
|
||||
return struct.pack(
|
||||
"<IB",
|
||||
output_frequency_hz,
|
||||
external_input_configuration_bits,
|
||||
)
|
||||
|
||||
|
||||
def build_sweep_settings_payload(
|
||||
sweep_configuration: SweepConfiguration,
|
||||
is_synchronization_master: bool,
|
||||
synchronization_enabled: bool,
|
||||
) -> bytes:
|
||||
"""Build the current protocol-v14 SweepSettings payload."""
|
||||
|
||||
excitation_power_centidecibels_milliwatt = int(sweep_configuration.power_dbm * 100)
|
||||
dwell_time_microseconds = max(
|
||||
0,
|
||||
min(int(sweep_configuration.dwell_us), 0xFFFF),
|
||||
)
|
||||
synchronization_mode = (
|
||||
SYNC_MODE_HARDWARE_TRIGGER if synchronization_enabled else 0
|
||||
)
|
||||
|
||||
configuration_bits = (
|
||||
0 << 0
|
||||
| is_synchronization_master << 1
|
||||
| 1 << 2
|
||||
| 1 << 3
|
||||
| 0 << 4
|
||||
| synchronization_mode << 5
|
||||
)
|
||||
|
||||
port_1_stimulus_stage = 0 if is_synchronization_master else 1
|
||||
stage_configuration_bits = 0 | port_1_stimulus_stage << 3 | 1 << 6
|
||||
|
||||
return struct.pack(
|
||||
"<QQHIhBHhH",
|
||||
sweep_configuration.start_hz,
|
||||
sweep_configuration.stop_hz,
|
||||
sweep_configuration.points,
|
||||
sweep_configuration.if_bandwidth,
|
||||
excitation_power_centidecibels_milliwatt,
|
||||
configuration_bits,
|
||||
stage_configuration_bits,
|
||||
excitation_power_centidecibels_milliwatt,
|
||||
dwell_time_microseconds,
|
||||
)
|
||||
|
||||
|
||||
def parse_vna_datapoint_payload(payload: bytes) -> Optional[ParsedVnaDatapoint]:
|
||||
"""Decode one VNADatapoint payload into a structured Python object."""
|
||||
|
||||
fixed_header_length_bytes = 12
|
||||
per_value_storage_length_bytes = 9
|
||||
|
||||
if len(payload) < fixed_header_length_bytes:
|
||||
return None
|
||||
|
||||
frequency_hz, _power_level_centidecibels_milliwatt, point_index = struct.unpack_from(
|
||||
"<QhH",
|
||||
payload,
|
||||
0,
|
||||
)
|
||||
value_count = (len(payload) - fixed_header_length_bytes) // per_value_storage_length_bytes
|
||||
if value_count == 0:
|
||||
return None
|
||||
|
||||
real_components = struct.unpack_from(f"<{value_count}f", payload, 12)
|
||||
imaginary_components = struct.unpack_from(
|
||||
f"<{value_count}f",
|
||||
payload,
|
||||
12 + 4 * value_count,
|
||||
)
|
||||
description_masks = payload[12 + 8 * value_count : 12 + 9 * value_count]
|
||||
|
||||
receiver_values_by_description_mask = {
|
||||
description_masks[value_index]: complex(
|
||||
real_components[value_index],
|
||||
imaginary_components[value_index],
|
||||
)
|
||||
for value_index in range(value_count)
|
||||
}
|
||||
return ParsedVnaDatapoint(
|
||||
frequency_hz=frequency_hz,
|
||||
point_index=point_index,
|
||||
receiver_values_by_description_mask=receiver_values_by_description_mask,
|
||||
)
|
||||
|
||||
|
||||
def find_receiver_value(
|
||||
receiver_values_by_description_mask: dict[int, complex],
|
||||
stage_index: int,
|
||||
is_reference_receiver: bool,
|
||||
required_port_number: int = 0,
|
||||
) -> Optional[complex]:
|
||||
"""Look up one complex receiver value by stage, receiver type, and port number."""
|
||||
|
||||
required_port_bit = (1 << (required_port_number - 1)) if required_port_number else 0
|
||||
for description_mask, complex_value in receiver_values_by_description_mask.items():
|
||||
if (description_mask >> 5) != stage_index:
|
||||
continue
|
||||
if bool(description_mask & 0x10) != is_reference_receiver:
|
||||
continue
|
||||
if required_port_bit and not (description_mask & required_port_bit):
|
||||
continue
|
||||
return complex_value
|
||||
return None
|
||||
-169
@@ -1,169 +0,0 @@
|
||||
import logging
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
|
||||
import usb.core
|
||||
import usb.util
|
||||
|
||||
try:
|
||||
import libusb_package
|
||||
except ImportError:
|
||||
libusb_package = None
|
||||
|
||||
from .librevna_device_protocol import (
|
||||
PacketType,
|
||||
encode_protocol_frame,
|
||||
extract_complete_protocol_frames,
|
||||
)
|
||||
|
||||
|
||||
USB_VENDOR_ID = 0x1209
|
||||
USB_PRODUCT_ID = 0x4121
|
||||
USB_DATA_ENDPOINT_OUT = 0x01
|
||||
USB_DATA_ENDPOINT_IN = 0x81
|
||||
|
||||
transport_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LibreVnaUsbBackendError(RuntimeError):
|
||||
"""Raised when PyUSB cannot access a libusb backend."""
|
||||
|
||||
|
||||
def _build_missing_backend_message() -> str:
|
||||
return (
|
||||
"PyUSB could not find a libusb backend. "
|
||||
"On Windows, install the Python USB backend with "
|
||||
"`python -m pip install -r Python/requirements.txt` from the repository root "
|
||||
"or `python -m pip install libusb-package` inside the active virtual environment. "
|
||||
"If the LibreVNA GUI already sees the device, the USB driver is likely fine and "
|
||||
"only the Python libusb DLL is missing."
|
||||
)
|
||||
|
||||
|
||||
def _usb_find(**kwargs):
|
||||
if libusb_package is not None:
|
||||
return libusb_package.find(**kwargs)
|
||||
|
||||
try:
|
||||
return usb.core.find(**kwargs)
|
||||
except usb.core.NoBackendError as exc:
|
||||
raise LibreVnaUsbBackendError(_build_missing_backend_message()) from exc
|
||||
|
||||
|
||||
class LibreVnaUsbBulkConnection:
|
||||
"""USB bulk transport wrapper for one LibreVNA device."""
|
||||
|
||||
_USB_TRANSFER_TIMEOUT_MS = 1_000
|
||||
|
||||
def __init__(self, serial_number: str) -> None:
|
||||
usb_device = _usb_find(
|
||||
idVendor=USB_VENDOR_ID,
|
||||
idProduct=USB_PRODUCT_ID,
|
||||
serial_number=serial_number,
|
||||
)
|
||||
if usb_device is None:
|
||||
raise RuntimeError(f"LibreVNA with serial {serial_number!r} not found on USB")
|
||||
|
||||
try:
|
||||
if usb_device.is_kernel_driver_active(0):
|
||||
usb_device.detach_kernel_driver(0)
|
||||
except (usb.core.USBError, NotImplementedError):
|
||||
pass
|
||||
|
||||
usb_device.set_configuration()
|
||||
|
||||
self.serial_number = serial_number
|
||||
self._usb_device = usb_device
|
||||
self._received_packets: queue.Queue[tuple[int, bytes]] = queue.Queue()
|
||||
self._stop_requested = threading.Event()
|
||||
self._transmit_lock = threading.Lock()
|
||||
self._receive_thread = threading.Thread(
|
||||
target=self._receive_packets_forever,
|
||||
daemon=True,
|
||||
name=f"vna-rx-{serial_number[:8]}",
|
||||
)
|
||||
self._receive_thread.start()
|
||||
|
||||
def close(self) -> None:
|
||||
self._stop_requested.set()
|
||||
self._receive_thread.join(timeout=0.5)
|
||||
usb.util.dispose_resources(self._usb_device)
|
||||
|
||||
def drain_received_packets(self) -> list[tuple[int, bytes]]:
|
||||
"""Remove all already-buffered packets without blocking."""
|
||||
|
||||
drained_packets: list[tuple[int, bytes]] = []
|
||||
while True:
|
||||
try:
|
||||
drained_packets.append(self._received_packets.get_nowait())
|
||||
except queue.Empty:
|
||||
return drained_packets
|
||||
|
||||
def send_packet(self, packet_type: int, payload: bytes = b"") -> None:
|
||||
with self._transmit_lock:
|
||||
self._usb_device.write(
|
||||
USB_DATA_ENDPOINT_OUT,
|
||||
encode_protocol_frame(packet_type, payload),
|
||||
timeout=self._USB_TRANSFER_TIMEOUT_MS,
|
||||
)
|
||||
|
||||
def receive_packet(self, timeout_seconds: float = 2.0) -> tuple[int, bytes]:
|
||||
"""Wait for the next received packet."""
|
||||
|
||||
return self._received_packets.get(timeout=timeout_seconds)
|
||||
|
||||
def wait_for_acknowledgement(self, timeout_seconds: float = 2.0) -> None:
|
||||
"""Wait for an ACK packet while ignoring unrelated asynchronous packets."""
|
||||
|
||||
end_time_monotonic = time.monotonic() + timeout_seconds
|
||||
while True:
|
||||
remaining_seconds = end_time_monotonic - time.monotonic()
|
||||
if remaining_seconds <= 0:
|
||||
raise TimeoutError(f"ACK not received within {timeout_seconds} s")
|
||||
|
||||
packet_type, _payload = self.receive_packet(timeout_seconds=remaining_seconds)
|
||||
if packet_type == PacketType.ACKNOWLEDGE:
|
||||
return
|
||||
if packet_type == PacketType.NEGATIVE_ACKNOWLEDGE:
|
||||
raise RuntimeError("Device returned NACK")
|
||||
|
||||
def _receive_packets_forever(self) -> None:
|
||||
buffered_bytes = b""
|
||||
while not self._stop_requested.is_set():
|
||||
try:
|
||||
usb_chunk = bytes(
|
||||
self._usb_device.read(
|
||||
USB_DATA_ENDPOINT_IN,
|
||||
4096,
|
||||
timeout=100,
|
||||
)
|
||||
)
|
||||
except usb.core.USBTimeoutError:
|
||||
continue
|
||||
except Exception as exc:
|
||||
if not self._stop_requested.is_set():
|
||||
transport_logger.error(
|
||||
"USB read error on %s: %s",
|
||||
self.serial_number,
|
||||
exc,
|
||||
)
|
||||
break
|
||||
|
||||
buffered_bytes += usb_chunk
|
||||
extracted_packets, buffered_bytes = extract_complete_protocol_frames(
|
||||
buffered_bytes
|
||||
)
|
||||
for packet in extracted_packets:
|
||||
self._received_packets.put(packet)
|
||||
|
||||
|
||||
def list_connected_device_serial_numbers() -> list[str]:
|
||||
"""Return the serial numbers of all connected LibreVNA devices."""
|
||||
|
||||
connected_devices = _usb_find(
|
||||
idVendor=USB_VENDOR_ID,
|
||||
idProduct=USB_PRODUCT_ID,
|
||||
find_all=True,
|
||||
)
|
||||
return [device.serial_number for device in (connected_devices or [])]
|
||||
-223
@@ -1,223 +0,0 @@
|
||||
import time
|
||||
from collections.abc import Iterator, Sequence
|
||||
from dataclasses import replace
|
||||
from typing import Optional
|
||||
|
||||
from .librevna_device_protocol import (
|
||||
PacketType,
|
||||
build_reference_settings_payload,
|
||||
build_sweep_settings_payload,
|
||||
)
|
||||
from .librevna_usb_transport import (
|
||||
LibreVnaUsbBulkConnection,
|
||||
)
|
||||
from .running_sweep_cycle_collection import collect_complete_running_sweep_cycles
|
||||
from .sweep_measurement_models import SweepConfiguration, SweepMeasurementResult
|
||||
|
||||
|
||||
class MultiDeviceVnaController:
|
||||
"""
|
||||
Multi-device LibreVNA controller for synchronized forward S-parameter sweeps.
|
||||
|
||||
Port mapping:
|
||||
master Port 1 / Port 2 → S11 / S21
|
||||
slave 0 Port 1 / Port 2 → S31 / S41
|
||||
slave 1 Port 1 / Port 2 → S51 / S61
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
master_serial_number: str,
|
||||
slave_serial_numbers: Sequence[str] = (),
|
||||
force_external_reference: bool = True,
|
||||
) -> None:
|
||||
self._master_device = LibreVnaUsbBulkConnection(master_serial_number)
|
||||
self._slave_devices = [
|
||||
LibreVnaUsbBulkConnection(slave_serial_number)
|
||||
for slave_serial_number in slave_serial_numbers
|
||||
]
|
||||
self._all_devices = [self._master_device, *self._slave_devices]
|
||||
self._synchronization_enabled = len(self._slave_devices) > 0
|
||||
self._force_external_reference = force_external_reference
|
||||
self._reference_configuration_applied = False
|
||||
self._last_applied_sweep_configuration: Optional[SweepConfiguration] = None
|
||||
self._sweep_is_running = False
|
||||
self._is_closed = False
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *_):
|
||||
self.close()
|
||||
|
||||
def close(self) -> None:
|
||||
if self._is_closed:
|
||||
return
|
||||
|
||||
self._is_closed = True
|
||||
self._send_idle_to_all_devices()
|
||||
for device_connection in self._all_devices:
|
||||
device_connection.close()
|
||||
|
||||
def stop_continuous_sweep(self) -> None:
|
||||
"""Stop the currently running sweep without closing the controller."""
|
||||
|
||||
if self._is_closed:
|
||||
return
|
||||
self._send_idle_to_all_devices()
|
||||
|
||||
def configure_continuous_sweep(
|
||||
self,
|
||||
sweep_configuration: SweepConfiguration,
|
||||
) -> None:
|
||||
"""
|
||||
Apply reference and sweep settings and leave the devices sweeping.
|
||||
|
||||
Reapplying the same sweep configuration is a no-op.
|
||||
"""
|
||||
|
||||
if self._is_closed:
|
||||
raise RuntimeError("Controller is already closed")
|
||||
|
||||
if not self._reference_configuration_applied:
|
||||
self._configure_reference_clocks()
|
||||
|
||||
if (
|
||||
self._sweep_is_running
|
||||
and self._last_applied_sweep_configuration == sweep_configuration
|
||||
):
|
||||
return
|
||||
|
||||
if self._sweep_is_running:
|
||||
self._send_idle_to_all_devices()
|
||||
time.sleep(0.05)
|
||||
|
||||
self._drain_all_received_packets()
|
||||
self._configure_sweep_on_all_devices(sweep_configuration)
|
||||
|
||||
def collect_running_sweep_cycles(
|
||||
self,
|
||||
cycle_count: int = 1,
|
||||
) -> SweepMeasurementResult:
|
||||
"""
|
||||
Collect one or more complete cycles from the already-running sweep.
|
||||
|
||||
Call configure_continuous_sweep() first. The driver aligns itself to the
|
||||
next full cycle in the incoming datapoint stream and then captures
|
||||
exactly `cycle_count` complete cycles.
|
||||
"""
|
||||
|
||||
if self._is_closed:
|
||||
raise RuntimeError("Controller is already closed")
|
||||
if not self._sweep_is_running or self._last_applied_sweep_configuration is None:
|
||||
raise RuntimeError(
|
||||
"No running sweep is configured. Call configure_continuous_sweep() first."
|
||||
)
|
||||
|
||||
try:
|
||||
return collect_complete_running_sweep_cycles(
|
||||
master_device_connection=self._master_device,
|
||||
slave_device_connections=self._slave_devices,
|
||||
active_sweep_configuration=self._last_applied_sweep_configuration,
|
||||
cycle_count=cycle_count,
|
||||
)
|
||||
except Exception:
|
||||
self._send_idle_to_all_devices()
|
||||
raise
|
||||
|
||||
def iter_running_sweep_cycles(self) -> Iterator[SweepMeasurementResult]:
|
||||
"""Yield complete sweep cycles forever from the already-running sweep."""
|
||||
|
||||
while True:
|
||||
yield self.collect_running_sweep_cycles(1)
|
||||
|
||||
def _send_command_and_wait_for_acknowledgement(
|
||||
self,
|
||||
device_connection: LibreVnaUsbBulkConnection,
|
||||
packet_type: int,
|
||||
payload: bytes = b"",
|
||||
timeout_seconds: float = 3.0,
|
||||
retry_count: int = 1,
|
||||
) -> None:
|
||||
last_error: Optional[Exception] = None
|
||||
for _attempt_index in range(retry_count + 1):
|
||||
device_connection.send_packet(packet_type, payload)
|
||||
try:
|
||||
device_connection.wait_for_acknowledgement(
|
||||
timeout_seconds=timeout_seconds
|
||||
)
|
||||
return
|
||||
except Exception as exc:
|
||||
last_error = exc
|
||||
|
||||
assert last_error is not None
|
||||
raise last_error
|
||||
|
||||
def _try_send_command_without_failing(
|
||||
self,
|
||||
device_connection: LibreVnaUsbBulkConnection,
|
||||
packet_type: int,
|
||||
payload: bytes = b"",
|
||||
timeout_seconds: float = 3.0,
|
||||
retry_count: int = 1,
|
||||
) -> None:
|
||||
try:
|
||||
self._send_command_and_wait_for_acknowledgement(
|
||||
device_connection,
|
||||
packet_type,
|
||||
payload,
|
||||
timeout_seconds=timeout_seconds,
|
||||
retry_count=retry_count,
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _send_idle_to_all_devices(self) -> None:
|
||||
for device_connection in self._all_devices:
|
||||
self._try_send_command_without_failing(
|
||||
device_connection,
|
||||
PacketType.SET_IDLE,
|
||||
timeout_seconds=3.0,
|
||||
retry_count=1,
|
||||
)
|
||||
self._sweep_is_running = False
|
||||
|
||||
def _configure_reference_clocks(self) -> None:
|
||||
for device_connection in self._all_devices:
|
||||
self._send_command_and_wait_for_acknowledgement(
|
||||
device_connection,
|
||||
PacketType.REFERENCE_SETTINGS,
|
||||
build_reference_settings_payload(0, self._force_external_reference),
|
||||
timeout_seconds=3.0,
|
||||
retry_count=1,
|
||||
)
|
||||
|
||||
time.sleep(0.05)
|
||||
self._reference_configuration_applied = True
|
||||
|
||||
def _configure_sweep_on_all_devices(
|
||||
self,
|
||||
sweep_configuration: SweepConfiguration,
|
||||
) -> None:
|
||||
sweep_configuration_commands = [
|
||||
*[(slave_device, False) for slave_device in self._slave_devices],
|
||||
(self._master_device, True),
|
||||
]
|
||||
for device_connection, is_synchronization_master in sweep_configuration_commands:
|
||||
self._send_command_and_wait_for_acknowledgement(
|
||||
device_connection,
|
||||
PacketType.SWEEP_SETTINGS,
|
||||
build_sweep_settings_payload(
|
||||
sweep_configuration,
|
||||
is_synchronization_master=is_synchronization_master,
|
||||
synchronization_enabled=self._synchronization_enabled,
|
||||
),
|
||||
timeout_seconds=3.0,
|
||||
retry_count=1,
|
||||
)
|
||||
self._last_applied_sweep_configuration = replace(sweep_configuration)
|
||||
self._sweep_is_running = True
|
||||
|
||||
def _drain_all_received_packets(self) -> None:
|
||||
for device_connection in self._all_devices:
|
||||
device_connection.drain_received_packets()
|
||||
-284
@@ -1,284 +0,0 @@
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Callable, Sequence
|
||||
from typing import Optional
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .librevna_device_protocol import (
|
||||
PacketType,
|
||||
ParsedVnaDatapoint,
|
||||
find_receiver_value,
|
||||
parse_vna_datapoint_payload,
|
||||
)
|
||||
from .librevna_usb_transport import LibreVnaUsbBulkConnection
|
||||
from .sweep_measurement_models import SweepConfiguration, SweepMeasurementResult
|
||||
|
||||
|
||||
def collect_complete_running_sweep_cycles(
|
||||
master_device_connection: LibreVnaUsbBulkConnection,
|
||||
slave_device_connections: Sequence[LibreVnaUsbBulkConnection],
|
||||
active_sweep_configuration: SweepConfiguration,
|
||||
cycle_count: int,
|
||||
) -> SweepMeasurementResult:
|
||||
"""Collect complete cycles from an already-running synchronized sweep."""
|
||||
|
||||
cycle_count = int(cycle_count)
|
||||
if cycle_count < 1:
|
||||
raise ValueError("cycle_count must be >= 1")
|
||||
|
||||
all_device_connections = [master_device_connection, *slave_device_connections]
|
||||
point_count = active_sweep_configuration.points
|
||||
frequencies_hz = np.zeros(point_count, dtype=np.float64)
|
||||
raw_receiver_measurements_by_s_parameter = {
|
||||
s_parameter_name: np.full(
|
||||
(cycle_count, point_count),
|
||||
np.nan + 1j * np.nan,
|
||||
dtype=complex,
|
||||
)
|
||||
for s_parameter_name in build_forward_s_parameter_names(len(slave_device_connections))
|
||||
}
|
||||
master_reference_measurements = np.full(
|
||||
(cycle_count, point_count),
|
||||
np.nan + 1j * np.nan,
|
||||
dtype=complex,
|
||||
)
|
||||
collection_errors: list[Exception] = []
|
||||
datapoint_counts_by_device_serial = {
|
||||
device_connection.serial_number: 0
|
||||
for device_connection in all_device_connections
|
||||
}
|
||||
stop_collection_requested = threading.Event()
|
||||
datapoint_timeout_seconds = max(
|
||||
10.0,
|
||||
active_sweep_configuration.dwell_us / 1_000_000 + 2.0,
|
||||
)
|
||||
|
||||
def collect_datapoints_from_device(
|
||||
device_connection: LibreVnaUsbBulkConnection,
|
||||
handle_datapoint: Callable[[ParsedVnaDatapoint], bool],
|
||||
) -> None:
|
||||
datapoints_received = 0
|
||||
expected_datapoint_count = cycle_count * point_count
|
||||
last_datapoint_timestamp = time.monotonic()
|
||||
|
||||
while datapoints_received < expected_datapoint_count:
|
||||
if stop_collection_requested.is_set():
|
||||
return
|
||||
|
||||
remaining_timeout_seconds = (
|
||||
last_datapoint_timestamp + datapoint_timeout_seconds
|
||||
) - time.monotonic()
|
||||
if remaining_timeout_seconds <= 0:
|
||||
collection_errors.append(
|
||||
TimeoutError(
|
||||
f"No datapoints from {device_connection.serial_number} for "
|
||||
f"{datapoint_timeout_seconds:.1f} s "
|
||||
f"(received {datapoints_received}/{point_count})"
|
||||
)
|
||||
)
|
||||
stop_collection_requested.set()
|
||||
return
|
||||
|
||||
try:
|
||||
packet_type, payload = device_connection.receive_packet(
|
||||
timeout_seconds=min(1.0, remaining_timeout_seconds)
|
||||
)
|
||||
except (queue.Empty, TimeoutError) as exc:
|
||||
if stop_collection_requested.is_set():
|
||||
return
|
||||
if isinstance(exc, queue.Empty):
|
||||
continue
|
||||
collection_errors.append(exc)
|
||||
stop_collection_requested.set()
|
||||
return
|
||||
|
||||
if packet_type != PacketType.VNA_DATAPOINT:
|
||||
continue
|
||||
|
||||
parsed_datapoint = parse_vna_datapoint_payload(payload)
|
||||
if parsed_datapoint and 0 <= parsed_datapoint.point_index < point_count:
|
||||
last_datapoint_timestamp = time.monotonic()
|
||||
datapoint_was_consumed = handle_datapoint(parsed_datapoint)
|
||||
if datapoint_was_consumed:
|
||||
datapoint_counts_by_device_serial[device_connection.serial_number] += 1
|
||||
datapoints_received += 1
|
||||
|
||||
def build_cycle_tracking_handler(
|
||||
cycle_aware_handler: Callable[[ParsedVnaDatapoint, int], None],
|
||||
) -> Callable[[ParsedVnaDatapoint], bool]:
|
||||
cycle_tracking_state = {
|
||||
"current_cycle_index": 0,
|
||||
"previous_point_index": -1,
|
||||
"has_seen_cycle_start": False,
|
||||
}
|
||||
|
||||
def handle_datapoint(parsed_datapoint: ParsedVnaDatapoint) -> bool:
|
||||
current_point_index = parsed_datapoint.point_index
|
||||
if not cycle_tracking_state["has_seen_cycle_start"]:
|
||||
if current_point_index != 0:
|
||||
cycle_tracking_state["previous_point_index"] = current_point_index
|
||||
return False
|
||||
cycle_tracking_state["has_seen_cycle_start"] = True
|
||||
|
||||
if (
|
||||
cycle_tracking_state["previous_point_index"] >= 0
|
||||
and current_point_index < cycle_tracking_state["previous_point_index"]
|
||||
):
|
||||
cycle_tracking_state["current_cycle_index"] += 1
|
||||
|
||||
cycle_tracking_state["previous_point_index"] = current_point_index
|
||||
current_cycle_index = cycle_tracking_state["current_cycle_index"]
|
||||
if current_cycle_index >= cycle_count:
|
||||
return False
|
||||
|
||||
cycle_aware_handler(parsed_datapoint, current_cycle_index)
|
||||
return True
|
||||
|
||||
return handle_datapoint
|
||||
|
||||
def handle_master_datapoint(
|
||||
parsed_datapoint: ParsedVnaDatapoint,
|
||||
cycle_index: int,
|
||||
) -> None:
|
||||
point_index = parsed_datapoint.point_index
|
||||
reference_receiver_value = find_receiver_value(
|
||||
parsed_datapoint.receiver_values_by_description_mask,
|
||||
stage_index=0,
|
||||
is_reference_receiver=True,
|
||||
)
|
||||
if reference_receiver_value is None:
|
||||
return
|
||||
|
||||
frequencies_hz[point_index] = parsed_datapoint.frequency_hz
|
||||
master_reference_measurements[cycle_index, point_index] = reference_receiver_value
|
||||
|
||||
for receiver_port_number, s_parameter_name in ((1, "S11"), (2, "S21")):
|
||||
port_receiver_value = find_receiver_value(
|
||||
parsed_datapoint.receiver_values_by_description_mask,
|
||||
stage_index=0,
|
||||
is_reference_receiver=False,
|
||||
required_port_number=receiver_port_number,
|
||||
)
|
||||
if port_receiver_value is not None:
|
||||
raw_receiver_measurements_by_s_parameter[s_parameter_name][
|
||||
cycle_index,
|
||||
point_index,
|
||||
] = port_receiver_value
|
||||
|
||||
def build_slave_datapoint_handler(
|
||||
slave_index: int,
|
||||
) -> Callable[[ParsedVnaDatapoint], bool]:
|
||||
first_s_parameter_name = f"S{2 * slave_index + 3}1"
|
||||
second_s_parameter_name = f"S{2 * slave_index + 4}1"
|
||||
|
||||
def handle_slave_datapoint(
|
||||
parsed_datapoint: ParsedVnaDatapoint,
|
||||
cycle_index: int,
|
||||
) -> None:
|
||||
point_index = parsed_datapoint.point_index
|
||||
for receiver_port_number, s_parameter_name in (
|
||||
(1, first_s_parameter_name),
|
||||
(2, second_s_parameter_name),
|
||||
):
|
||||
port_receiver_value = find_receiver_value(
|
||||
parsed_datapoint.receiver_values_by_description_mask,
|
||||
stage_index=0,
|
||||
is_reference_receiver=False,
|
||||
required_port_number=receiver_port_number,
|
||||
)
|
||||
if port_receiver_value is not None:
|
||||
raw_receiver_measurements_by_s_parameter[s_parameter_name][
|
||||
cycle_index,
|
||||
point_index,
|
||||
] = port_receiver_value
|
||||
|
||||
return build_cycle_tracking_handler(handle_slave_datapoint)
|
||||
|
||||
collection_threads = [
|
||||
threading.Thread(
|
||||
target=collect_datapoints_from_device,
|
||||
args=(
|
||||
master_device_connection,
|
||||
build_cycle_tracking_handler(handle_master_datapoint),
|
||||
),
|
||||
daemon=True,
|
||||
name="collect-master",
|
||||
),
|
||||
]
|
||||
for slave_index, slave_device_connection in enumerate(slave_device_connections):
|
||||
collection_threads.append(
|
||||
threading.Thread(
|
||||
target=collect_datapoints_from_device,
|
||||
args=(slave_device_connection, build_slave_datapoint_handler(slave_index)),
|
||||
daemon=True,
|
||||
name=f"collect-slave{slave_index}",
|
||||
)
|
||||
)
|
||||
|
||||
for collection_thread in collection_threads:
|
||||
collection_thread.start()
|
||||
for collection_thread in collection_threads:
|
||||
collection_thread.join()
|
||||
|
||||
if collection_errors:
|
||||
raise RuntimeError(
|
||||
f"Sweep collection failed: {collection_errors[0]}"
|
||||
) from collection_errors[0]
|
||||
|
||||
if slave_device_connections and min(datapoint_counts_by_device_serial.values(), default=0) == 0:
|
||||
raise RuntimeError(
|
||||
"No datapoints received from at least one device; "
|
||||
"hardware trigger sync did not start. Check Trigger Out/In loop "
|
||||
"and 10 MHz reference wiring."
|
||||
)
|
||||
|
||||
return SweepMeasurementResult(
|
||||
frequencies_hz=frequencies_hz,
|
||||
s_parameters=calculate_last_cycle_s_parameters(
|
||||
raw_receiver_measurements_by_s_parameter,
|
||||
master_reference_measurements,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def calculate_last_cycle_s_parameters(
|
||||
raw_receiver_measurements_by_s_parameter: dict[str, np.ndarray],
|
||||
master_reference_measurements: np.ndarray,
|
||||
) -> dict[str, np.ndarray]:
|
||||
"""Convert raw receiver captures from the final sweep cycle into S-parameters."""
|
||||
|
||||
if np.isnan(master_reference_measurements).any():
|
||||
missing_reference_count = int(np.isnan(master_reference_measurements).sum())
|
||||
raise RuntimeError(
|
||||
f"Master reference missing for {missing_reference_count} datapoints"
|
||||
)
|
||||
|
||||
s_parameters: dict[str, np.ndarray] = {}
|
||||
for s_parameter_name, raw_receiver_measurements in (
|
||||
raw_receiver_measurements_by_s_parameter.items()
|
||||
):
|
||||
if np.isnan(raw_receiver_measurements).any():
|
||||
missing_measurement_count = int(np.isnan(raw_receiver_measurements).sum())
|
||||
raise RuntimeError(
|
||||
f"Measurement {s_parameter_name} missing for "
|
||||
f"{missing_measurement_count} datapoints"
|
||||
)
|
||||
s_parameters[s_parameter_name] = (
|
||||
raw_receiver_measurements[-1] / master_reference_measurements[-1]
|
||||
)
|
||||
|
||||
return s_parameters
|
||||
|
||||
|
||||
def build_forward_s_parameter_names(slave_device_count: int) -> list[str]:
|
||||
"""Return S-parameter names for the configured forward multi-device topology."""
|
||||
|
||||
s_parameter_names = ["S11", "S21"]
|
||||
for slave_index in range(slave_device_count):
|
||||
s_parameter_names += [
|
||||
f"S{2 * slave_index + 3}1",
|
||||
f"S{2 * slave_index + 4}1",
|
||||
]
|
||||
return s_parameter_names
|
||||
-23
@@ -1,23 +0,0 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
@dataclass
|
||||
class SweepConfiguration:
|
||||
"""User-facing configuration for one VNA sweep."""
|
||||
|
||||
start_hz: int = 100_000
|
||||
stop_hz: int = 3_000_000_000
|
||||
points: int = 201
|
||||
if_bandwidth: int = 1_000
|
||||
power_dbm: float = -10.0
|
||||
dwell_us: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class SweepMeasurementResult:
|
||||
"""Measured sweep data from one or more complete sweep cycles."""
|
||||
|
||||
frequencies_hz: np.ndarray
|
||||
s_parameters: dict[str, np.ndarray]
|
||||
@@ -1,4 +0,0 @@
|
||||
pyusb
|
||||
libusb-package; platform_system == "Windows"
|
||||
numpy
|
||||
matplotlib
|
||||
File diff suppressed because it is too large
Load Diff
Binary file not shown.
|
Before Width: | Height: | Size: 299 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 295 KiB |
BIN
Binary file not shown.
|
Before Width: | Height: | Size: 151 KiB |
BIN
Binary file not shown.
|
Before Width: | Height: | Size: 142 KiB |
@@ -1,336 +0,0 @@
|
||||
"""
|
||||
test_sweep.py - run synchronized sweeps continuously and draw the latest cycle live.
|
||||
|
||||
Usage
|
||||
-----
|
||||
List connected devices:
|
||||
python test_sweep.py
|
||||
|
||||
Single device (S11, S21):
|
||||
python test_sweep.py <master_serial>
|
||||
|
||||
Two devices - master + one slave (S11, S21, S31, S41):
|
||||
python test_sweep.py <master_serial> <slave_serial>
|
||||
|
||||
Multi-device runs always use hardware trigger synchronization.
|
||||
Close the plot window or press Ctrl+C to stop the live sweep.
|
||||
"""
|
||||
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
|
||||
import matplotlib.pyplot as plt
|
||||
import numpy as np
|
||||
from matplotlib.axes import Axes
|
||||
from matplotlib.figure import Figure
|
||||
from matplotlib.lines import Line2D
|
||||
from matplotlib.text import Text
|
||||
|
||||
from librevna_multi_device_driver import (
|
||||
MultiDeviceVnaController,
|
||||
SweepConfiguration,
|
||||
SweepMeasurementResult,
|
||||
list_connected_device_serial_numbers,
|
||||
)
|
||||
|
||||
|
||||
# Set to "internal" to keep each device on its own reference clock.
|
||||
DEFAULT_REFERENCE_SOURCE = "external"
|
||||
DEFAULT_SWEEP_CONFIGURATION = SweepConfiguration(
|
||||
start_hz=100_000_000,
|
||||
stop_hz=6_000_000_000,
|
||||
points=501,
|
||||
if_bandwidth=10000,
|
||||
power_dbm=-10.0,
|
||||
)
|
||||
|
||||
_ACTIVE_MEASUREMENT_CONTROLLER: MultiDeviceVnaController | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class LiveSweepPlot:
|
||||
figure: Figure
|
||||
status_text: Text
|
||||
magnitude_axis: Axes
|
||||
phase_axes: list[Axes]
|
||||
magnitude_lines: dict[str, Line2D]
|
||||
phase_lines: dict[str, Line2D]
|
||||
|
||||
|
||||
def _force_external_reference_from_default_source() -> bool:
|
||||
if DEFAULT_REFERENCE_SOURCE == "external":
|
||||
return True
|
||||
if DEFAULT_REFERENCE_SOURCE == "internal":
|
||||
return False
|
||||
raise ValueError(
|
||||
"DEFAULT_REFERENCE_SOURCE must be either 'external' or 'internal'"
|
||||
)
|
||||
|
||||
|
||||
def _parse_device_serial_numbers_from_command_line(
|
||||
command_line_arguments: list[str],
|
||||
) -> list[str]:
|
||||
for command_line_argument in command_line_arguments:
|
||||
if command_line_argument.startswith("--"):
|
||||
raise SystemExit(f"Unknown option: {command_line_argument}")
|
||||
return command_line_arguments
|
||||
|
||||
|
||||
def _print_usage() -> None:
|
||||
try:
|
||||
connected_device_serial_numbers = list_connected_device_serial_numbers()
|
||||
except RuntimeError as exc:
|
||||
print(exc)
|
||||
return
|
||||
|
||||
if not connected_device_serial_numbers:
|
||||
print("No LibreVNA devices found. Check the USB connection and OS USB access.")
|
||||
return
|
||||
|
||||
print("Connected LibreVNA devices:")
|
||||
for connected_device_serial_number in connected_device_serial_numbers:
|
||||
print(f" {connected_device_serial_number}")
|
||||
print(f"\nUsage: {sys.argv[0]} <master_serial> [slave_serial ...]")
|
||||
|
||||
|
||||
def _install_termination_signal_handlers():
|
||||
def handle_stop_signal(received_signal_number, _current_stack_frame) -> None:
|
||||
global _ACTIVE_MEASUREMENT_CONTROLLER
|
||||
if _ACTIVE_MEASUREMENT_CONTROLLER is not None:
|
||||
try:
|
||||
print(
|
||||
f"\nSignal {received_signal_number} received, stopping devices...",
|
||||
flush=True,
|
||||
)
|
||||
_ACTIVE_MEASUREMENT_CONTROLLER.close()
|
||||
except Exception:
|
||||
pass
|
||||
raise SystemExit(128 + received_signal_number)
|
||||
|
||||
previous_sigint_handler = signal.getsignal(signal.SIGINT)
|
||||
previous_sigterm_handler = signal.getsignal(signal.SIGTERM)
|
||||
signal.signal(signal.SIGINT, handle_stop_signal)
|
||||
signal.signal(signal.SIGTERM, handle_stop_signal)
|
||||
return previous_sigint_handler, previous_sigterm_handler
|
||||
|
||||
|
||||
def _restore_termination_signal_handlers(
|
||||
previous_sigint_handler,
|
||||
previous_sigterm_handler,
|
||||
) -> None:
|
||||
signal.signal(signal.SIGINT, previous_sigint_handler)
|
||||
signal.signal(signal.SIGTERM, previous_sigterm_handler)
|
||||
|
||||
|
||||
def _print_run_header(
|
||||
master_serial_number: str,
|
||||
slave_serial_numbers: list[str],
|
||||
sweep_configuration: SweepConfiguration,
|
||||
) -> None:
|
||||
print(f"Master : {master_serial_number}")
|
||||
for slave_index, slave_serial_number in enumerate(slave_serial_numbers):
|
||||
print(f"Slave {slave_index} : {slave_serial_number}")
|
||||
print(
|
||||
"Sync : hardware trigger"
|
||||
if slave_serial_numbers
|
||||
else "Sync : disabled (single device)"
|
||||
)
|
||||
print(f"Ref : {DEFAULT_REFERENCE_SOURCE} on all devices")
|
||||
print("Sweeps : continuous until Ctrl+C or the plot window is closed")
|
||||
print(
|
||||
f"Sweep : {sweep_configuration.start_hz / 1e6:.3f} MHz - "
|
||||
f"{sweep_configuration.stop_hz / 1e6:.3f} MHz, "
|
||||
f"{sweep_configuration.points} points, "
|
||||
f"IF BW {sweep_configuration.if_bandwidth} Hz"
|
||||
)
|
||||
print("Running live sweep...")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
global _ACTIVE_MEASUREMENT_CONTROLLER
|
||||
|
||||
device_serial_numbers = _parse_device_serial_numbers_from_command_line(
|
||||
sys.argv[1:]
|
||||
)
|
||||
if not device_serial_numbers:
|
||||
_print_usage()
|
||||
return
|
||||
|
||||
master_serial_number = device_serial_numbers[0]
|
||||
slave_serial_numbers = device_serial_numbers[1:]
|
||||
sweep_configuration = DEFAULT_SWEEP_CONFIGURATION
|
||||
force_external_reference = _force_external_reference_from_default_source()
|
||||
sweep_cycle_durations_seconds: list[float] = []
|
||||
|
||||
_print_run_header(
|
||||
master_serial_number,
|
||||
slave_serial_numbers,
|
||||
sweep_configuration,
|
||||
)
|
||||
previous_sigint_handler, previous_sigterm_handler = (
|
||||
_install_termination_signal_handlers()
|
||||
)
|
||||
|
||||
try:
|
||||
with MultiDeviceVnaController(
|
||||
master_serial_number,
|
||||
slave_serial_numbers,
|
||||
force_external_reference=force_external_reference,
|
||||
) as measurement_controller:
|
||||
_ACTIVE_MEASUREMENT_CONTROLLER = measurement_controller
|
||||
measurement_controller.configure_continuous_sweep(sweep_configuration)
|
||||
plt.ion()
|
||||
|
||||
live_plot: LiveSweepPlot | None = None
|
||||
sweep_cycle_count = 0
|
||||
print("Waiting for the first complete sweep cycle...")
|
||||
|
||||
while True:
|
||||
if live_plot is not None and not plt.fignum_exists(live_plot.figure.number):
|
||||
print("\nPlot window closed, stopping sweep.", flush=True)
|
||||
break
|
||||
|
||||
cycle_collection_start_time = time.monotonic()
|
||||
sweep_measurement_result = measurement_controller.collect_running_sweep_cycles(
|
||||
1,
|
||||
)
|
||||
cycle_duration_seconds = time.monotonic() - cycle_collection_start_time
|
||||
sweep_cycle_durations_seconds.append(cycle_duration_seconds)
|
||||
sweep_cycle_count += 1
|
||||
|
||||
if (
|
||||
live_plot is not None
|
||||
and not plt.fignum_exists(live_plot.figure.number)
|
||||
):
|
||||
print("\nPlot window closed, stopping sweep.", flush=True)
|
||||
break
|
||||
|
||||
if live_plot is None:
|
||||
live_plot = _create_live_sweep_plot(sweep_measurement_result)
|
||||
|
||||
_update_live_sweep_plot(
|
||||
live_plot,
|
||||
sweep_measurement_result,
|
||||
sweep_cycle_count=sweep_cycle_count,
|
||||
last_cycle_duration_seconds=cycle_duration_seconds,
|
||||
average_cycle_duration_seconds=float(
|
||||
np.mean(sweep_cycle_durations_seconds)
|
||||
),
|
||||
)
|
||||
finally:
|
||||
_ACTIVE_MEASUREMENT_CONTROLLER = None
|
||||
_restore_termination_signal_handlers(
|
||||
previous_sigint_handler,
|
||||
previous_sigterm_handler,
|
||||
)
|
||||
|
||||
if sweep_cycle_durations_seconds:
|
||||
average_sweep_duration_seconds = float(np.mean(sweep_cycle_durations_seconds))
|
||||
minimum_sweep_duration_seconds = float(np.min(sweep_cycle_durations_seconds))
|
||||
maximum_sweep_duration_seconds = float(np.max(sweep_cycle_durations_seconds))
|
||||
print(
|
||||
"Sweep time (steady-state): "
|
||||
f"{average_sweep_duration_seconds:.6f} s avg, "
|
||||
f"{minimum_sweep_duration_seconds:.6f}-"
|
||||
f"{maximum_sweep_duration_seconds:.6f} s "
|
||||
f"over {len(sweep_cycle_durations_seconds)} cycles"
|
||||
)
|
||||
else:
|
||||
print("Sweep time (steady-state): not enough repeated cycles to estimate")
|
||||
|
||||
print("Sweep stopped.")
|
||||
|
||||
|
||||
def _create_live_sweep_plot(
|
||||
sweep_measurement_result: SweepMeasurementResult,
|
||||
) -> LiveSweepPlot:
|
||||
frequencies_mhz = sweep_measurement_result.frequencies_hz / 1e6
|
||||
s_parameter_names = sorted(sweep_measurement_result.s_parameters)
|
||||
figure_height_inches = 3.5 + 2.5 * len(s_parameter_names)
|
||||
figure, axes = plt.subplots(
|
||||
1 + len(s_parameter_names),
|
||||
1,
|
||||
figsize=(10, figure_height_inches),
|
||||
sharex=True,
|
||||
)
|
||||
axes = np.atleast_1d(axes)
|
||||
magnitude_axis = axes[0]
|
||||
phase_axes = list(axes[1:])
|
||||
magnitude_lines: dict[str, Line2D] = {}
|
||||
phase_lines: dict[str, Line2D] = {}
|
||||
|
||||
for s_parameter_name in s_parameter_names:
|
||||
magnitude_line, = magnitude_axis.plot(
|
||||
frequencies_mhz,
|
||||
np.zeros_like(frequencies_mhz),
|
||||
label=s_parameter_name,
|
||||
linewidth=0.9,
|
||||
)
|
||||
magnitude_lines[s_parameter_name] = magnitude_line
|
||||
|
||||
magnitude_axis.set_title("Multi-port S-parameters (forward, Port 1 stimulus)")
|
||||
magnitude_axis.set_ylabel("Magnitude (dB)")
|
||||
magnitude_axis.set_ylim(-25, 0)
|
||||
magnitude_axis.grid(True, alpha=0.4)
|
||||
magnitude_axis.legend(loc="lower left")
|
||||
|
||||
for phase_axis, s_parameter_name in zip(phase_axes, s_parameter_names):
|
||||
phase_line, = phase_axis.plot(
|
||||
frequencies_mhz,
|
||||
np.zeros_like(frequencies_mhz),
|
||||
label=s_parameter_name,
|
||||
linewidth=0.9,
|
||||
)
|
||||
phase_lines[s_parameter_name] = phase_line
|
||||
phase_axis.set_ylabel("Phase (deg)")
|
||||
phase_axis.set_title(f"{s_parameter_name} phase")
|
||||
phase_axis.set_ylim(-180, 180)
|
||||
phase_axis.grid(True, alpha=0.4)
|
||||
phase_axis.legend(loc="lower left")
|
||||
|
||||
phase_axes[-1].set_xlabel("Frequency (MHz)")
|
||||
|
||||
status_text = figure.suptitle("")
|
||||
figure.tight_layout(rect=(0, 0, 1, 0.97))
|
||||
if figure.canvas.manager is not None:
|
||||
figure.canvas.manager.set_window_title("LibreVNA Live Sweep")
|
||||
plt.show(block=False)
|
||||
|
||||
return LiveSweepPlot(
|
||||
figure=figure,
|
||||
status_text=status_text,
|
||||
magnitude_axis=magnitude_axis,
|
||||
phase_axes=phase_axes,
|
||||
magnitude_lines=magnitude_lines,
|
||||
phase_lines=phase_lines,
|
||||
)
|
||||
|
||||
|
||||
def _update_live_sweep_plot(
|
||||
live_plot: LiveSweepPlot,
|
||||
sweep_measurement_result: SweepMeasurementResult,
|
||||
sweep_cycle_count: int,
|
||||
last_cycle_duration_seconds: float,
|
||||
average_cycle_duration_seconds: float,
|
||||
) -> None:
|
||||
for s_parameter_name, complex_trace in sweep_measurement_result.s_parameters.items():
|
||||
magnitude_db = 20 * np.log10(np.abs(complex_trace) + 1e-15)
|
||||
phase_degrees = np.rad2deg(np.angle(complex_trace))
|
||||
live_plot.magnitude_lines[s_parameter_name].set_ydata(magnitude_db)
|
||||
live_plot.phase_lines[s_parameter_name].set_ydata(phase_degrees)
|
||||
|
||||
live_plot.status_text.set_text(
|
||||
"Cycle "
|
||||
f"{sweep_cycle_count} | "
|
||||
f"Last {last_cycle_duration_seconds:.3f} s | "
|
||||
f"Avg {average_cycle_duration_seconds:.3f} s"
|
||||
)
|
||||
live_plot.figure.canvas.draw_idle()
|
||||
live_plot.figure.canvas.flush_events()
|
||||
plt.pause(0.001)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user