From 6693ece0aa74b84b0b26de6847acb4ec88b227eb Mon Sep 17 00:00:00 2001 From: Ayzen Date: Wed, 9 Sep 2026 21:26:56 +0300 Subject: [PATCH] per-point stimulus switching: host side MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Матрица 4x4 за одну развёртку: каждый порт мастера измеряется на каждой частоте дважды, с двумя положениями выходного ключа. Ключ дёргает контроллер в триггерной петле, поэтому расбери переключателями больше не управляет — SwitchedMatrixRadarService уходит из тракта, все 16 комбинаций снимаются в пределах одной точки вместо разброса между тремя развёртками. Выходы нумеруются позиция-мажорно, стадии порт-мажорно (старшие биты счётчика стадий выбирают порт, младший — положение ключа). Из этого следует, что положение ключа = младший бит номера стадии, а период 2 нацело делит 4 стадии — поэтому контроллеру достаточно инвертировать ногу на каждый цикл триггера, без абсолютного счётчика. Выходы 0 и 1 при этом сохраняют прежний смысл. Приёмники слейвов пронумерованы от числа выходов + 1, а не от фиксированного 3: иначе при 4 выходах прямой член «приёмник 3 <- выход 3» назывался бы так же, как отражение выхода 3. При двух выходах даёт прежние порты 3..6. Слейвам порт-стадии выставляются в 7: стоковый приём «одна за последней стадией» перестаёт работать, когда порт выбирается старшими битами. Всё за флагом multi_device.stimulus_switch_positions (1 = прежнее поведение, байт в байт). Спецификация правок FPGA/STM в device_firmware/per_point_switching.md. Co-Authored-By: Claude Opus 5 (1M context) --- device_firmware/per_point_switching.md | 178 ++++++++++++++++++ .../controller.py | 13 +- .../cycle_collection.py | 107 +++++++---- .../librevna_multi_device_driver/models.py | 78 ++++++++ .../librevna_multi_device_driver/protocol.py | 47 ++++- .../hardware_full/matrix_radar_service.py | 3 +- .../hardware_full/multi_device_service.py | 59 +++++- python_app/models/run_config_codec.py | 10 +- python_app/models/run_config_schema.py | 33 +++- python_app/tests/test_librevna_multidevice.py | 145 +++++++++++++- .../tests/test_switched_matrix_capture.py | 36 ++++ 11 files changed, 645 insertions(+), 64 deletions(-) create mode 100644 device_firmware/per_point_switching.md diff --git a/device_firmware/per_point_switching.md b/device_firmware/per_point_switching.md new file mode 100644 index 0000000..6aad6e1 --- /dev/null +++ b/device_firmware/per_point_switching.md @@ -0,0 +1,178 @@ +# Переключение стимула внутри частотной точки + +Режим, в котором одна развёртка даёт матрицу 4×4 вместо 2×4: каждый порт мастера +измеряется на **каждой** частоте дважды — с двумя положениями выходного ключа. + +``` +4 выхода = 2 порта мастера × 2 положения ключа +4 входа = 2 слейва × 2 порта (входной ключ не участвует) +``` + +Выигрыш не в скорости (стадий столько же, сколько было развёрток), а в +когерентности: все 16 комбинаций снимаются в пределах ~0.4 мс на одной частоте +вместо разброса в сотни миллисекунд между развёртками. + +## Раскладка стадий + +| Стадия | Порт мастера | Ключ | Выход | +|---|---|---|---| +| 0 | 1 | A | 0 | +| 1 | 1 | B | 2 | +| 2 | 2 | A | 1 | +| 3 | 2 | B | 3 | + +Стадии нумеруются **порт-мажорно**: старшие биты счётчика стадий выбирают порт, +младший — положение ключа. Выходы нумеруются **позиция-мажорно**, поэтому выходы +0 и 1 сохраняют тот же смысл, что и без переключения. + +Из этого следует главное свойство: **положение ключа — это младший бит номера +стадии**, а период 2 нацело делит 4 стадии. Значит через границу точки чередование +продолжается без разрыва, и контроллеру достаточно инвертировать ногу на каждый +цикл триггера — абсолютный счётчик стадий не нужен, нужна только правильная +чётность, и та только один раз на старте развёртки. + +## Контракт по проводу + +Хост уже отправляет это (см. `build_sweep_settings_payload`), прошивка должна +считать. Слово стадий в `SweepSettings` (16 бит, смещение 25 в payload): + +| Биты | Поле | Значение в этом режиме | +|---|---|---| +| 0..2 | `stages` | 3 (то есть 4 стадии) | +| 3..5 | `port1Stage` | 0 — **индекс порта**, а не номер стадии | +| 6..8 | `port2Stage` | 1 | +| 9..11 | `port3Stage` → ширина под-стадии | `log2(положений ключа)` = 1 | +| 12..14 | `port4Stage` | не используется | +| 15 | `unused2` | не используется | + +Поле `port3Stage` переиспользовано под ширину блока под-стадий: двухпортовая +прошивка его игнорирует, поэтому формат пакета не меняется. **Ноль там — +стоковая раскладка**, так что хост с `stimulus_switch_positions = 1` остаётся +полностью совместим с немодифицированным прибором. + +Слейвам обе порт-стадии выставляются в **7**. Стоковый приём «одна за последней +стадией» здесь не работает: при 4 стадиях старшие биты счётчика принимают только +значения 0 и 1, свободного значения не остаётся, и слейв начал бы излучать. + +## FPGA мастера + +### `Sweep.vhd` + +Сейчас порт включается по точному совпадению, поэтому излучает ровно в одной +стадии, и во второй под-стадии зондирующего сигнала просто не будет: + +```vhdl +if std_logic_vector(stage_cnt) = PORT1_STAGE then +``` + +Меняется на сравнение по старшим битам плюс явное «порт выключен». Сдвиг на +переменное число бит синтезируется в трёхбитный barrel shifter: + +```vhdl +-- в entity +SUBSTAGE_BITS : in STD_LOGIC_VECTOR (2 downto 0); + +-- в architecture +signal substage_shift : natural range 0 to 7; +substage_shift <= to_integer(unsigned(SUBSTAGE_BITS)); + +-- в состоянии Settling, вместо каждой из двух проверок: +if unsigned(PORT1_STAGE) <= unsigned(STAGES) + and shift_right(stage_cnt, substage_shift) = unsigned(PORT1_STAGE) then + PORT1_ACTIVE <= '1'; + source_active <= '1'; +else + PORT1_ACTIVE <= '0'; +end if; +``` + +Проверка совместимости: + +* `SUBSTAGE_BITS = 0`, `STAGES = 1`, порты 0 и 1 — сдвиг нулевой, сравнение + вырождается в исходное. Стоковое поведение сохраняется бит в бит. +* Слейв с `port_stage = 7`: `7 <= 1` ложно → порт выключен. Старый хост слал 2 — + `2 <= 1` тоже ложно, так что новый битстрим работает и со старым хостом. +* `SUBSTAGE_BITS = 1`, `STAGES = 3`: порт 0 активен в стадиях 0 и 1, порт 1 — в + 2 и 3. + +### `SPIConfig.vhd` + +В регистре `SweepSetup` (адрес 6) заняты биты 15:13, 12, 5:3 и 2:0 — **биты 11:6 +свободны**. Берём три из них: + +```vhdl +when 6 => STAGES <= spi_buf_out(15 downto 13); + SYNC_ENABLED <= spi_buf_out(12); + SUBSTAGE_BITS <= spi_buf_out(8 downto 6); -- новое + PORT1_STAGE <= spi_buf_out(5 downto 3); + PORT2_STAGE <= spi_buf_out(2 downto 0); +``` + +### `top.vhd` + +Пробросить `SUBSTAGE_BITS` от `SPIConfig` к `Sweep` — новый сигнал рядом с +`sweep_stages` / `sweep_port1_stage`. + +## Прошивка STM32 LibreVNA + +Для самих стадий менять нечего: `stageCnt` уже считается до `settings.stages`, +значения тегируются номером стадии, `VNADatapoint<32>` вмещает 4 стадии × 3 +значения. Нужно только протащить новое поле: + +```c +// FPGA.cpp +void FPGA::SetupSweep(uint8_t stages, uint8_t port1_stage, uint8_t port2_stage, + uint8_t substage_bits, bool synchronize, bool syncMaster) { + ... + value |= (uint16_t) (substage_bits & 0x07) << 6; + +// VNA.cpp +FPGA::SetupSweep(s.stages, s.port1Stage, s.port2Stage, s.port3Stage, + s.syncMode != 0, s.syncMaster); +``` + +Собирать так же, как описано в [`reproduce/README.md`](multi_libreVNA_firmware/reproduce/README.md). +Отдельно: битстрим мастера единственный в своём роде, `top.bin` перезаписывается +каждой сборкой ISE — переименовывать сразу. + +## Контроллер ключа (отдельный STM32) + +Врезается в триггерную петлю между `TRIGGER_OUT` мастера и `TRIGGER_IN` слейва. +Электрически это чисто: в режиме `Mode::Trigger` прошивка LibreVNA уводит STM32 +прибора в Hi-Z на обеих линиях (`GPIO_MODE_INPUT`, `NOPULL`, EXTI снят), так что +обе линии свободны. + +Три требования: + +1. **Прозрачный повторитель обоих фронтов.** Задержка суб-микросекундная + (bare-metal EXTI), иначе она добавляется к каждой из 4 стадий каждой точки. +2. **Ключ переключается по СПАДУ**, а не по фронту. Спад означает «оцифровка + стадии закончилась»; дальше идёт `SamplingDone` → `Settling` следующей стадии, + то есть окно успокоения. При `dwell_us = 2` и штатной задержке PLL это окно + ~62 мкс — запас огромный. Переключение по фронту заставило бы удерживать + кольцо на время установления ключа. +3. **Одна нога на два ключа.** Ключи на порту 1 и на порту 2 ходят синхронно; + нужное положение — просто инверсия на каждом цикле триггера. + +Чётность привязывается к началу развёртки. Между развёртками FPGA держится в +сбросе (`StartSweep` = снять и поднять AUX3), а хост успевает получить пакет +статуса, поэтому пауза на линии триггера на порядок длиннее межстадийной +(~100 мкс против ≥1 мс — померить осциллографом и взять порог посередине). Если +есть физический доступ к AUX3 мастера (нога FPGA P74) — лучше завести её на +контроллер как точный маркер вместо порога по паузе. + +Сбой чётности меняет местами антенны 0↔2 и 1↔3 **молча**, поэтому стоит вывести +текущее положение ключа на отладочную ногу для сверки осциллографом. + +## Хост + +Уже сделано. Включается одним полем в `run_config.json`: + +```json +"multi_device": { "stimulus_switch_positions": 2 }, +"sweep": { "dwell_us": 2 } +``` + +При этом расбери **перестаёт трогать ключ**: `SwitchedMatrixRadarService` уходит +из тракта, матрица 4×4 собирается за одну развёртку. `stimulus_switch_positions: +1` возвращает всё в прежний режим. diff --git a/python_app/hardware_full/librevna_multi_device_driver/controller.py b/python_app/hardware_full/librevna_multi_device_driver/controller.py index fb40ada..879a96b 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/controller.py +++ b/python_app/hardware_full/librevna_multi_device_driver/controller.py @@ -17,6 +17,7 @@ from python_app.hardware_full.librevna_multi_device_driver.models import ( SweepConfiguration, SweepMeasurementResult, normalize_master_stimulus_ports, + normalize_switch_substage_count, ) from python_app.hardware_full.librevna_multi_device_driver.protocol import ( PacketType, @@ -36,13 +37,21 @@ class MultiDeviceVnaController: master_serial_number: str, slave_serial_numbers: Sequence[str] = (), force_external_reference: bool = True, + switch_substage_count: int = 1, ) -> None: - """Open all configured devices and initialize controller state.""" + """Open all configured devices and initialize controller state. + + ``switch_substage_count`` is a property of the wiring, not of a sweep: it is + how many positions the stimulus switch is stepped through WITHIN one frequency + point. ``1`` is the stock behaviour; anything larger needs the modified master + bitstream and the switch controller sitting in the trigger loop. + """ self._master_device: LibreVnaUsbBulkConnection | None = None self._slave_devices: list[LibreVnaUsbBulkConnection] = [] self._all_devices: list[LibreVnaUsbBulkConnection] = [] self._synchronization_enabled = bool(slave_serial_numbers) self._force_external_reference = bool(force_external_reference) + self._switch_substage_count = normalize_switch_substage_count(switch_substage_count) self._reference_configuration_applied = False self._last_applied_sweep_configuration: Optional[SweepConfiguration] = None self._last_master_stimulus_ports: tuple[int, ...] | None = None @@ -194,6 +203,7 @@ class MultiDeviceVnaController: active_sweep_configuration=self._last_applied_sweep_configuration, cycle_count=cycle_count, master_stimulus_ports=self._last_master_stimulus_ports, + switch_substage_count=self._switch_substage_count, datapoint_timeout_seconds=datapoint_timeout_seconds, ) except Exception: @@ -332,6 +342,7 @@ class MultiDeviceVnaController: is_synchronization_master=is_synchronization_master, synchronization_enabled=self._synchronization_enabled, master_stimulus_ports=master_stimulus_ports, + switch_substage_count=self._switch_substage_count, ), timeout_seconds=1.0, retry_count=1, diff --git a/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py b/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py index e32cc5d..58d14bd 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py +++ b/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py @@ -14,8 +14,10 @@ from python_app.hardware_full.librevna_multi_device_driver.exceptions import ( TransientCollectionError, ) from python_app.hardware_full.librevna_multi_device_driver.models import ( + OutputChannel, SweepConfiguration, SweepMeasurementResult, + build_output_channels, normalize_master_stimulus_ports, ) from python_app.hardware_full.librevna_multi_device_driver.protocol import ( @@ -173,6 +175,7 @@ def collect_complete_running_sweep_cycles( active_sweep_configuration: SweepConfiguration, cycle_count: int, master_stimulus_ports: Sequence[int], + switch_substage_count: int = 1, datapoint_timeout_seconds: float | None = None, ) -> SweepMeasurementResult: """Collect complete cycles from an already-running synchronized sweep.""" @@ -180,7 +183,9 @@ def collect_complete_running_sweep_cycles( if cycle_count < 1: raise ValueError("cycle_count must be >= 1") stimulus_ports = normalize_master_stimulus_ports(master_stimulus_ports) - stage_by_master_port = {port: stage for stage, port in enumerate(stimulus_ports)} + # One output per (master port, stimulus switch position). With no per-point + # switching this is one output per port, exactly as before. + output_channels = build_output_channels(stimulus_ports, switch_substage_count) all_device_connections = [master_device_connection, *slave_device_connections] point_count = active_sweep_configuration.points @@ -197,13 +202,18 @@ def collect_complete_running_sweep_cycles( ) else: frequencies_hz = np.array([float(active_sweep_configuration.start_hz)], dtype=np.float64) - master_reference_measurements_by_port = { - port: np.full((cycle_count, point_count), np.nan + 1j * np.nan, dtype=complex) - for port in stimulus_ports + # Keyed by 1-based OUTPUT number, not by master port: with per-point switching two + # outputs share a stimulus port but are measured in different stages, so each needs + # its own incident reference. + master_reference_measurements_by_output = { + channel.number: np.full((cycle_count, point_count), np.nan + 1j * np.nan, dtype=complex) + for channel in output_channels } raw_receiver_measurements_by_s_parameter = { name: np.full((cycle_count, point_count), np.nan + 1j * np.nan, dtype=complex) - for name in build_forward_s_parameter_names(len(slave_device_connections), stimulus_ports) + for name in build_forward_s_parameter_names( + len(slave_device_connections), output_channels + ) } collection_errors: list[Exception] = [] @@ -250,7 +260,7 @@ def collect_complete_running_sweep_cycles( # ~150 ms sweep; this estimates 30 ms, tolerance 15 ms, so an off-by-one of # ~150 ms is flagged with ~10x margin and ~1 ms jitter clears it with ~7x.) # A spread above tolerance is rejected and retried, never silently emitted. - stage_count = max(1, len(stimulus_ports)) + stage_count = max(1, len(output_channels)) expected_cycle_seconds = _expected_sweep_cycle_seconds( point_count, stage_count, active_sweep_configuration.if_bandwidth ) @@ -541,41 +551,41 @@ def collect_complete_running_sweep_cycles( def handle_master_datapoint(parsed_datapoint: ParsedVnaDatapoint, cycle_index: int) -> None: """Store the master device's frequency, reference, and reflection values. - Records the sweep-point frequency and, per active master stimulus port, the - reference receiver value and the matching reflection (S11/S22) into the - cycle/point measurement buffers. + Records the sweep-point frequency and, per output channel, the reference + receiver value and the matching reflection into the cycle/point measurement + buffers. Two outputs may share a stimulus port (different switch positions); + they are told apart by the stage tag. """ point_index = parsed_datapoint.point_index frequencies_hz[point_index] = parsed_datapoint.frequency_hz - for master_stimulus_port, stage_index in stage_by_master_port.items(): + for channel in output_channels: reference_receiver_value = find_receiver_value( parsed_datapoint.receiver_values_by_description_mask, - stage_index=stage_index, + stage_index=channel.stage_index, is_reference_receiver=True, - required_port_number=master_stimulus_port, + required_port_number=channel.stimulus_port, ) if reference_receiver_value is None: reference_receiver_value = find_receiver_value( parsed_datapoint.receiver_values_by_description_mask, - stage_index=stage_index, + stage_index=channel.stage_index, is_reference_receiver=True, ) if reference_receiver_value is None: continue - master_reference_measurements_by_port[master_stimulus_port][ + master_reference_measurements_by_output[channel.number][ cycle_index, point_index, ] = reference_receiver_value - receiver_port_number = master_stimulus_port - s_parameter_name = master_reflection_s_parameter_name(master_stimulus_port) + s_parameter_name = master_reflection_s_parameter_name(channel) port_receiver_value = find_receiver_value( parsed_datapoint.receiver_values_by_description_mask, - stage_index=stage_index, + stage_index=channel.stage_index, is_reference_receiver=False, - required_port_number=receiver_port_number, + required_port_number=channel.stimulus_port, ) if port_receiver_value is not None: raw_receiver_measurements_by_s_parameter[s_parameter_name][ @@ -588,25 +598,25 @@ def collect_complete_running_sweep_cycles( ) -> Callable[[ParsedVnaDatapoint], bool]: """Build a cycle-tracking datapoint handler for the given slave device. - The slave's two receivers map to ports ``2*slave_index + 3`` and ``+ 4``, - producing forward S-parameters (e.g. S3x/S4x) for each active master - stimulus port. + The slave's two receivers map to the virtual receiver ports numbered just + above the outputs (see :func:`slave_receiver_base_port`), producing one + forward S-parameter per output channel. """ - receiver_base_port = 2 * slave_index + 3 + receiver_base_port = slave_receiver_base_port(slave_index, output_channels) def handle_slave_datapoint(parsed_datapoint: ParsedVnaDatapoint, cycle_index: int) -> None: """Store this slave's forward receiver values into the measurement buffers.""" point_index = parsed_datapoint.point_index - for master_stimulus_port, stage_index in stage_by_master_port.items(): - first_s_parameter_name = f"S{receiver_base_port}{master_stimulus_port}" - second_s_parameter_name = f"S{receiver_base_port + 1}{master_stimulus_port}" + for channel in output_channels: + first_s_parameter_name = f"S{receiver_base_port}{channel.number}" + second_s_parameter_name = f"S{receiver_base_port + 1}{channel.number}" 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=stage_index, + stage_index=channel.stage_index, is_reference_receiver=False, required_port_number=receiver_port_number, ) @@ -729,29 +739,42 @@ def collect_complete_running_sweep_cycles( frequencies_hz=frequencies_hz, s_parameters=calculate_last_cycle_s_parameters( raw_receiver_measurements_by_s_parameter, - master_reference_measurements_by_port, + master_reference_measurements_by_output, ), ) -def master_reflection_s_parameter_name(master_stimulus_port: int) -> str: - """Return master reflection trace name for active output port.""" - if master_stimulus_port == 1: - return "S11" - return "S22" +def slave_receiver_base_port(slave_index: int, output_channels: Sequence[OutputChannel]) -> int: + """Return the first virtual receiver port number of one slave device. + + Receivers are numbered directly above the outputs, so the whole topology reads + as one virtual VNA whose first ``N`` ports transmit and whose remaining ports + receive. Numbering them from a fixed 3 instead would make a 4-output setup + collide with itself: the reflection of output 3 and the forward term + ``receiver 3 <- output 3`` would both be called ``S33``. With two outputs this + still yields the historical ports 3..6. + """ + return len(output_channels) + 2 * slave_index + 1 -def build_forward_s_parameter_names(slave_device_count: int, master_stimulus_ports: Sequence[int]) -> list[str]: +def master_reflection_s_parameter_name(channel: OutputChannel) -> str: + """Return the reflection trace name for one output channel (``S11``, ``S22``, ...).""" + return f"S{channel.number}{channel.number}" + + +def build_forward_s_parameter_names( + slave_device_count: int, output_channels: Sequence[OutputChannel] +) -> list[str]: """Return S-parameter names for the configured forward multi-device topology.""" names: list[str] = [] - for master_stimulus_port in master_stimulus_ports: - names.append(master_reflection_s_parameter_name(master_stimulus_port)) + for channel in output_channels: + names.append(master_reflection_s_parameter_name(channel)) for slave_index in range(slave_device_count): - receiver_base_port = 2 * slave_index + 3 + receiver_base_port = slave_receiver_base_port(slave_index, output_channels) names.extend( [ - f"S{receiver_base_port}{master_stimulus_port}", - f"S{receiver_base_port + 1}{master_stimulus_port}", + f"S{receiver_base_port}{channel.number}", + f"S{receiver_base_port + 1}{channel.number}", ] ) return names @@ -759,7 +782,7 @@ def build_forward_s_parameter_names(slave_device_count: int, master_stimulus_por def calculate_last_cycle_s_parameters( raw_receiver_measurements_by_s_parameter: dict[str, np.ndarray], - master_reference_measurements_by_port: dict[int, np.ndarray], + master_reference_measurements_by_output: dict[int, np.ndarray], ) -> dict[str, np.ndarray]: """Convert raw receiver captures from the final sweep cycle into S-parameters. @@ -775,8 +798,10 @@ def calculate_last_cycle_s_parameters( zero_filled_points_by_trace: dict[str, int] = {} point_count = 0 for s_parameter_name, raw_receiver_measurements in raw_receiver_measurements_by_s_parameter.items(): - master_stimulus_port = int(s_parameter_name[-1]) - master_reference_measurements = master_reference_measurements_by_port[master_stimulus_port] + # The trailing digit of every name is the 1-based OUTPUT number, which is what + # selects the incident reference (see build_forward_s_parameter_names). + output_number = int(s_parameter_name[-1]) + master_reference_measurements = master_reference_measurements_by_output[output_number] last_cycle_reference = master_reference_measurements[-1] last_cycle_measurement = raw_receiver_measurements[-1] diff --git a/python_app/hardware_full/librevna_multi_device_driver/models.py b/python_app/hardware_full/librevna_multi_device_driver/models.py index 9341bb1..00261e5 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/models.py +++ b/python_app/hardware_full/librevna_multi_device_driver/models.py @@ -7,6 +7,11 @@ from dataclasses import dataclass import numpy as np +# Hard ceiling from the wire format: SweepSettings.stages is 3 bits, so a point can +# carry at most 8 stages (and one VNADatapoint holds 3 values per stage, well within +# its 32-value capacity). +MAX_SWEEP_STAGES = 8 + def normalize_master_stimulus_ports(master_stimulus_ports: Sequence[int]) -> tuple[int, ...]: """Validate master stimulus ports and return them as a tuple of ints. @@ -25,6 +30,79 @@ def normalize_master_stimulus_ports(master_stimulus_ports: Sequence[int]) -> tup return stimulus_ports +def normalize_switch_substage_count(switch_substage_count: int) -> int: + """Validate the per-point stimulus switch position count. + + ``1`` means no per-point switching (one stage per master port, the stock + LibreVNA behaviour). Larger values require the modified master bitstream, + which keeps a port excited across a GROUP of stages instead of a single one. + The ceiling is the 3-bit ``stages`` field: ``ports * substages <= 8``. + """ + substage_count = int(switch_substage_count) + if substage_count < 1: + raise ValueError("switch_substage_count must be >= 1") + if substage_count & (substage_count - 1): + # The master bitstream selects the port from the UPPER bits of the stage + # counter, so the substage block must be a power of two. + raise ValueError("switch_substage_count must be a power of two") + return substage_count + + +@dataclass(frozen=True, slots=True) +class OutputChannel: + """One stimulus output of the widened matrix: a master port at a switch position. + + ``index`` is the canonical output position used by the radar combo matrix, + ``stage_index`` is the tag the device stamps on every value of this channel + (``descr >> 5`` in the datapoint), and ``stimulus_port`` is the master port + that physically excites it. + """ + + index: int + stimulus_port: int + stage_index: int + + @property + def number(self) -> int: + """Return the 1-based output number used in S-parameter names.""" + return self.index + 1 + + +def build_output_channels( + master_stimulus_ports: Sequence[int], switch_substage_count: int = 1 +) -> tuple[OutputChannel, ...]: + """Map every (master port, switch position) pair to an output and a device stage. + + Outputs are numbered switch-position-major, so positions of the FIRST switch + setting keep the output indices they have without per-point switching (with + ``switch_substage_count=1`` this is the identity). Stages are numbered + port-major, because the master bitstream takes the excited port from the upper + bits of the stage counter and the switch position from the lower bits — which + is also what makes the switch line a plain alternation the controller can drive + without tracking an absolute stage number. + + This is the single source of truth shared by the protocol builder, the cycle + collector and the acquisition service, so the mapping cannot drift between them. + """ + stimulus_ports = normalize_master_stimulus_ports(master_stimulus_ports) + substage_count = normalize_switch_substage_count(switch_substage_count) + stage_count = len(stimulus_ports) * substage_count + if stage_count > MAX_SWEEP_STAGES: + raise ValueError( + f"{len(stimulus_ports)} master port(s) x {substage_count} switch position(s) " + f"needs {stage_count} stages, but the device supports at most {MAX_SWEEP_STAGES}" + ) + return tuple( + OutputChannel( + index=substage * len(stimulus_ports) + port_index, + stimulus_port=port, + stage_index=port_index * substage_count + substage, + ) + for substage in range(substage_count) + for port_index, port in enumerate(stimulus_ports) + ) + + @dataclass(frozen=True, slots=True) class SweepConfiguration: """User-facing configuration for one VNA sweep.""" diff --git a/python_app/hardware_full/librevna_multi_device_driver/protocol.py b/python_app/hardware_full/librevna_multi_device_driver/protocol.py index c78cdb8..a9012ee 100644 --- a/python_app/hardware_full/librevna_multi_device_driver/protocol.py +++ b/python_app/hardware_full/librevna_multi_device_driver/protocol.py @@ -8,7 +8,9 @@ from dataclasses import dataclass from python_app.hardware_full.librevna_multi_device_driver.models import ( SweepConfiguration, + build_output_channels, normalize_master_stimulus_ports, + normalize_switch_substage_count, ) @@ -26,6 +28,13 @@ class PacketType: SYNC_MODE_HARDWARE_TRIGGER = 3 FRAME_HEADER_MAGIC = 0x5A +# Stage index handed to a device whose ports must never excite. The modified master +# bitstream matches a port against the UPPER bits of the stage counter, so the stock +# "one past the last stage" sentinel stops working once substages exist (with 4 stages +# the upper bits only ever hold 0 or 1, leaving no free value). The bitstream instead +# treats any port stage >= the stage count as "port disabled", and 7 is always >= it. +DISABLED_PORT_STAGE = 7 + @dataclass(frozen=True, slots=True) class ParsedVnaDatapoint: @@ -52,15 +61,31 @@ def build_sweep_settings_payload( is_synchronization_master: bool, synchronization_enabled: bool, master_stimulus_ports: Sequence[int], + switch_substage_count: int = 1, ) -> bytes: - """Build protocol-v14 SweepSettings for staged master outputs or synchronized receivers.""" + """Build protocol-v14 SweepSettings for staged master outputs or synchronized receivers. + + With ``switch_substage_count > 1`` every master port is excited across a BLOCK of + consecutive stages instead of a single one, so the same frequency point is measured + once per stimulus switch position. That requires the modified master bitstream: the + stock one matches ``stage_cnt`` against the port stage exactly, so a port excites in + exactly one stage and the extra substages would carry no stimulus. + + The substage block width is sent as ``log2(switch_substage_count)`` in the otherwise + unused ``port3Stage`` field (bits 9..11 of the stage word) — it tells the bitstream + how many LOW bits of the stage counter are the switch position rather than the port + selector. Zero there is the stock layout, so this payload stays compatible with an + unmodified device as long as ``switch_substage_count`` is 1. + """ stimulus_ports = normalize_master_stimulus_ports(master_stimulus_ports) + substage_count = normalize_switch_substage_count(switch_substage_count) + output_channels = build_output_channels(stimulus_ports, substage_count) excitation_power_centidecibels_milliwatt = int(round(sweep_configuration.power_dbm * 100.0)) dwell_time_microseconds = max(0, min(int(sweep_configuration.dwell_us), 0xFFFF)) synchronization_mode = SYNC_MODE_HARDWARE_TRIGGER if synchronization_enabled else 0 - last_stage_index = len(stimulus_ports) - 1 - inactive_stage_index = min(last_stage_index + 1, 7) + last_stage_index = len(output_channels) - 1 + substage_bits = substage_count.bit_length() - 1 configuration_bits = ( 0 << 0 @@ -72,17 +97,23 @@ def build_sweep_settings_payload( ) if is_synchronization_master: - stage_by_port = {port: stage for stage, port in enumerate(stimulus_ports)} - port_1_stimulus_stage = stage_by_port.get(1, inactive_stage_index) - port_2_stimulus_stage = stage_by_port.get(2, inactive_stage_index) + # The port selector lives in the stage bits ABOVE the substage block, so the + # value sent here is the port's index, not its first stage number. + port_stage_by_port = { + port: index for index, port in enumerate(stimulus_ports) + } + port_1_stimulus_stage = port_stage_by_port.get(1, DISABLED_PORT_STAGE) + port_2_stimulus_stage = port_stage_by_port.get(2, DISABLED_PORT_STAGE) else: - port_1_stimulus_stage = inactive_stage_index - port_2_stimulus_stage = inactive_stage_index + # Receiver-only devices must never excite. + port_1_stimulus_stage = DISABLED_PORT_STAGE + port_2_stimulus_stage = DISABLED_PORT_STAGE stage_configuration_bits = ( (last_stage_index & 0x07) | (port_1_stimulus_stage & 0x07) << 3 | (port_2_stimulus_stage & 0x07) << 6 + | (substage_bits & 0x07) << 9 ) return struct.pack( diff --git a/python_app/hardware_full/matrix_radar_service.py b/python_app/hardware_full/matrix_radar_service.py index 6651b20..4ca527c 100644 --- a/python_app/hardware_full/matrix_radar_service.py +++ b/python_app/hardware_full/matrix_radar_service.py @@ -52,6 +52,7 @@ def create_matrix_radar_service(config: RunConfigModel) -> MatrixRadarService: force_external_reference=config.radar.multi_device.force_external_reference, recovery_attempts=config.radar.multi_device.recovery_attempts, backend_mode=config.radar.driver_mode, + switch_substage_count=config.matrix_stimulus_switch_positions, ) out_physical = config.matrix_output_switch_positions in_physical = config.matrix_input_switch_positions @@ -67,7 +68,7 @@ def create_matrix_radar_service(config: RunConfigModel) -> MatrixRadarService: inner=inner, output_switch=build_physical_switch(config.output_switch, out_physical, config.radar.driver_mode), input_switch=build_physical_switch(config.input_switch, in_physical, config.radar.driver_mode), - inner_output_positions=RunConfigModel.MULTI_DEVICE_OUTPUT_POSITIONS, + inner_output_positions=config.matrix_virtual_output_positions, inner_input_positions=RunConfigModel.MULTI_DEVICE_INPUT_POSITIONS, settling_ms=config.runtime.settling_ms, ) diff --git a/python_app/hardware_full/multi_device_service.py b/python_app/hardware_full/multi_device_service.py index 8d80dca..aa1a729 100644 --- a/python_app/hardware_full/multi_device_service.py +++ b/python_app/hardware_full/multi_device_service.py @@ -13,7 +13,10 @@ import numpy as np from python_app.hardware_full.librevna_multi_device_driver.cycle_collection import LIBREVNA_NATIVE_SWEEP_TIMEOUT_SECONDS from python_app.hardware_full.librevna_multi_device_driver.exceptions import TransientCollectionError -from python_app.hardware_full.librevna_multi_device_driver.models import SweepConfiguration +from python_app.hardware_full.librevna_multi_device_driver.models import ( + SweepConfiguration, + normalize_switch_substage_count, +) from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData from python_app.models.run_config_model import RadarSweepModel, RunConfigModel @@ -43,10 +46,33 @@ _MAX_RECOVERY_WALL_SECONDS: float = 10.0 # previously caused the periodic freeze. _MAX_TRANSIENT_COLLECTION_RETRIES: int = 2 -_INPUT_S_PARAMETERS_BY_OUTPUT: dict[int, tuple[str, ...]] = { - 0: ("s31", "s41", "s51", "s61"), - 1: ("s32", "s42", "s52", "s62"), -} +def _build_s_parameter_layout( + output_count: int, slave_count: int +) -> tuple[dict[int, tuple[str, ...]], dict[int, str]]: + """Return the (forward, reflection) trace names for each output position. + + Mirrors the naming in ``cycle_collection``: outputs are virtual ports 1..N and + the slaves' receivers are numbered directly above them, so a 2-output setup + keeps the historical ``s31``/``s41``/``s51``/``s61`` layout while a 4-output one + (per-point stimulus switching) shifts the receivers to ports 5..8 to stay + collision-free. + """ + receiver_ports = [ + output_count + 2 * slave_index + port_offset + for slave_index in range(slave_count) + for port_offset in (1, 2) + ] + forward_by_output = { + output_index: tuple( + f"s{receiver_port}{output_index + 1}" for receiver_port in receiver_ports + ) + for output_index in range(output_count) + } + reflection_by_output = { + output_index: f"s{output_index + 1}{output_index + 1}" + for output_index in range(output_count) + } + return forward_by_output, reflection_by_output @dataclass(slots=True) class MultiDeviceLibreVnaService: @@ -57,6 +83,10 @@ class MultiDeviceLibreVnaService: force_external_reference: bool = True recovery_attempts: int = 3 backend_mode: str = "auto" + # Stimulus switch positions stepped through WITHIN one frequency point. 1 keeps the + # stock 2-output matrix; 2 doubles the outputs (4) without a second sweep, at the + # cost of requiring the modified master bitstream and the trigger-loop controller. + switch_substage_count: int = 1 _controller: "MultiDeviceVnaController | None" = field(init=False, default=None, repr=False) _sweep_configuration: SweepConfiguration | None = field(init=False, default=None, repr=False) _using_mock_backend: bool = field(init=False, default=False, repr=False) @@ -75,6 +105,12 @@ class MultiDeviceLibreVnaService: raise ValueError(f"Unsupported multi-device backend mode: {self.backend_mode}") self.backend_mode = mode self._using_mock_backend = mode == "mock" + self.switch_substage_count = normalize_switch_substage_count(self.switch_substage_count) + + @property + def output_position_count(self) -> int: + """Return how many stimulus outputs one acquisition produces.""" + return RunConfigModel.MULTI_DEVICE_MASTER_PORTS * self.switch_substage_count @property def using_mock_backend(self) -> bool: @@ -92,6 +128,7 @@ class MultiDeviceLibreVnaService: master_serial_number=self.master_serial, slave_serial_numbers=self.slave_serials, force_external_reference=self.force_external_reference, + switch_substage_count=self.switch_substage_count, ) logger.info( "Multi-device controller opened (master=%s, slaves=%s)", @@ -198,6 +235,7 @@ class MultiDeviceLibreVnaService: points=int(sweep.points), if_bandwidth=int(round(float(sweep.if_bandwidth_hz))), power_dbm=float(sweep.power_dbm), + dwell_us=int(sweep.dwell_us), ) logger.debug( "Multi-device configured: %s-%s Hz, %s points, IFBW=%s Hz, %s dBm", @@ -347,14 +385,17 @@ class MultiDeviceLibreVnaService: } frequencies = np.asarray(result.frequencies_hz, dtype=np.float32) + forward_by_output, reflection_by_output = _build_s_parameter_layout( + self.output_position_count, len(self.slave_serials) + ) traces: list[TraceData] = [] - for output_pos in range(RunConfigModel.MULTI_DEVICE_OUTPUT_POSITIONS): + for output_pos in range(self.output_position_count): reflection = self._required_s_parameter( normalized_s_parameters, - "s11" if output_pos == 0 else "s22", + reflection_by_output[output_pos], ) - for input_pos, s_parameter_name in enumerate(_INPUT_S_PARAMETERS_BY_OUTPUT[output_pos]): + for input_pos, s_parameter_name in enumerate(forward_by_output[output_pos]): traces.append( TraceData( combo=ComboKey(input=input_pos, output=output_pos), @@ -388,7 +429,7 @@ class MultiDeviceLibreVnaService: ) base_phase = 2.0 * math.pi * np.linspace(0.0, 1.0, points, dtype=np.float32) + self._mock_phase traces: list[TraceData] = [] - for output_pos in range(RunConfigModel.MULTI_DEVICE_OUTPUT_POSITIONS): + for output_pos in range(self.output_position_count): reflected_phase = base_phase * (0.55 + 0.05 * output_pos) + 0.7 * (output_pos + 1) s11 = ( (0.22 + 0.04 * output_pos) * np.cos(reflected_phase) diff --git a/python_app/models/run_config_codec.py b/python_app/models/run_config_codec.py index 0654bb5..d604749 100644 --- a/python_app/models/run_config_codec.py +++ b/python_app/models/run_config_codec.py @@ -193,6 +193,7 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: sweep_payload, "if_bandwidth_hz", model.radar.sweep.if_bandwidth_hz ) model.radar.sweep.power_dbm = _read_float(sweep_payload, "stimulus_power_dbm", model.radar.sweep.power_dbm) + model.radar.sweep.dwell_us = _read_int(sweep_payload, "dwell_us", model.radar.sweep.dwell_us) slave_serials_payload = multi_device_payload.get( "slave_serials", multi_device_payload.get("slave_serial_numbers", model.radar.multi_device.slave_serials), @@ -229,6 +230,11 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: "input_switch_positions", model.radar.multi_device.input_switch_positions, ) + model.radar.multi_device.stimulus_switch_positions = _read_int( + multi_device_payload, + "stimulus_switch_positions", + model.radar.multi_device.stimulus_switch_positions, + ) model.radar.kamil_adc.project_dir = _read_str( kamil_adc_payload, "project_dir", model.radar.kamil_adc.project_dir ) @@ -491,6 +497,7 @@ def run_config_to_dict(model: RunConfigModel) -> dict[str, Any]: "stop_hz": model.radar.sweep.stop_hz, "if_bandwidth_hz": model.radar.sweep.if_bandwidth_hz, "stimulus_power_dbm": model.radar.sweep.power_dbm, + "dwell_us": model.radar.sweep.dwell_us, } if not model.is_kamil_adc: sweep_payload["points"] = model.radar.sweep.points @@ -509,7 +516,8 @@ def run_config_to_dict(model: RunConfigModel) -> dict[str, Any]: "force_external_reference": model.radar.multi_device.force_external_reference, "recovery_attempts": model.radar.multi_device.recovery_attempts, "output_switch_positions": model.radar.multi_device.output_switch_positions, - "input_switch_positions": model.radar.multi_device.input_switch_positions + "input_switch_positions": model.radar.multi_device.input_switch_positions, + "stimulus_switch_positions": model.radar.multi_device.stimulus_switch_positions }, "kamil_adc": { "project_dir": model.radar.kamil_adc.project_dir, diff --git a/python_app/models/run_config_schema.py b/python_app/models/run_config_schema.py index b3c6dfc..70b668f 100644 --- a/python_app/models/run_config_schema.py +++ b/python_app/models/run_config_schema.py @@ -31,6 +31,10 @@ class RadarSweepModel: points: int = 1 if_bandwidth_hz: float = 1.0 power_dbm: float = -30.0 + # Extra per-stage settling requested from the device, on top of its own PLL + # settling delay (60 us by default). With per-point stimulus switching this is the + # window the switch controller has to move the switch between stages. + dwell_us: int = 0 @dataclass(slots=True) @@ -42,6 +46,10 @@ class RadarMultiDeviceModel: recovery_attempts: int = 3 output_switch_positions: int = 1 # 1 = свитча нет input_switch_positions: int = 1 # 1 = свитча нет + # Positions of the stimulus switch stepped through WITHIN one frequency point, + # driven by the controller in the hardware trigger loop rather than by this host. + # 1 = выключено (стоковое поведение). Требует доработанного битстрима мастера. + stimulus_switch_positions: int = 1 @dataclass(slots=True) @@ -354,6 +362,10 @@ class RunConfigModel: SN9000_MODEL = "sn9000" MULTI_DEVICE_INPUT_POSITIONS = 4 MULTI_DEVICE_OUTPUT_POSITIONS = 2 + # Stimulus ports the master itself owns. The virtual output axis is this times the + # per-point stimulus switch positions, so it equals MULTI_DEVICE_OUTPUT_POSITIONS + # unless per-point switching is enabled. + MULTI_DEVICE_MASTER_PORTS = 2 @staticmethod def build_full_combos(input_positions: int, output_positions: int) -> list[ComboModel]: @@ -406,6 +418,25 @@ class RunConfigModel: return 1 return max(1, int(self.radar.multi_device.input_switch_positions)) + @property + def matrix_stimulus_switch_positions(self) -> int: + """Positions the stimulus switch steps through inside one frequency point.""" + if not self.is_multi_device: + return 1 + return max(1, int(self.radar.multi_device.stimulus_switch_positions)) + + @property + def matrix_virtual_output_positions(self) -> int: + """Virtual output axis size produced by one acquisition. + + Per-point stimulus switching multiplies the master's own ports, because each + port is measured once per switch position at every frequency — so those extra + outputs cost no extra sweep and are NOT driven by this host. + """ + if not self.is_multi_device: + return self.MULTI_DEVICE_OUTPUT_POSITIONS + return self.MULTI_DEVICE_MASTER_PORTS * self.matrix_stimulus_switch_positions + def build_runtime_combos(self) -> list[ComboModel]: """Build the combo matrix from the effective switch axis sizes.""" return self.build_full_combos(self.input_switch.positions, self.output_switch.positions) @@ -477,7 +508,7 @@ class RunConfigModel: self.output_switch.driver_mode = "mock" self.output_switch.driver = self.output_switch.driver or "h7992" self.output_switch.radar_port = 1 - self.output_switch.positions = out_physical * self.MULTI_DEVICE_OUTPUT_POSITIONS + self.output_switch.positions = out_physical * self.matrix_virtual_output_positions self.output_switch.default_position = 0 self.input_switch.name = self.input_switch.name or "virtual_input" diff --git a/python_app/tests/test_librevna_multidevice.py b/python_app/tests/test_librevna_multidevice.py index 6cd4645..96e7961 100644 --- a/python_app/tests/test_librevna_multidevice.py +++ b/python_app/tests/test_librevna_multidevice.py @@ -21,15 +21,26 @@ from python_app.hardware_full.librevna_multi_device_driver.cycle_collection impo _CrossDeviceCycleSynchronizer, _expected_sweep_cycle_seconds, _is_transient_collection_error, + build_forward_s_parameter_names, calculate_last_cycle_s_parameters, ) from python_app.hardware_full.librevna_multi_device_driver.exceptions import TransientCollectionError -from python_app.hardware_full.librevna_multi_device_driver.models import normalize_master_stimulus_ports -from python_app.hardware_full.librevna_multi_device_driver.protocol import parse_vna_datapoint_payload +from python_app.hardware_full.librevna_multi_device_driver.models import ( + SweepConfiguration, + build_output_channels, + normalize_master_stimulus_ports, +) +from python_app.hardware_full.librevna_multi_device_driver.protocol import ( + DISABLED_PORT_STAGE, + build_sweep_settings_payload, + parse_vna_datapoint_payload, +) from python_app.hardware_full.multi_device_service import ( _MAX_TRANSIENT_COLLECTION_RETRIES, MultiDeviceLibreVnaService, + _build_s_parameter_layout, ) +from python_app.models.run_config_model import RadarSweepModel def _control_frame(packet_type: NativePacketType, payload: bytes) -> bytes: @@ -372,5 +383,135 @@ class TransientRetryTierTest(unittest.TestCase): self.assertEqual(service.collection_calls, 1) # fatal escalates immediately, no in-place retry +class PerPointStimulusSwitchingTest(unittest.TestCase): + """Per-point stimulus switching: one output per (master port, switch position). + + Without hardware these pin the contract the modified master bitstream and the + trigger-loop switch controller have to agree with. + """ + + def test_disabled_switching_keeps_the_stock_layout(self) -> None: + channels = build_output_channels((1, 2), 1) + self.assertEqual( + [(c.index, c.stimulus_port, c.stage_index) for c in channels], + [(0, 1, 0), (1, 2, 1)], + ) + + def test_two_positions_give_four_outputs(self) -> None: + channels = build_output_channels((1, 2), 2) + # Outputs are switch-position-major, so outputs 0/1 stay exactly what they are + # without switching; stages are port-major, matching the bitstream taking the + # port from the stage counter's UPPER bits. + self.assertEqual( + [(c.index, c.stimulus_port, c.stage_index) for c in channels], + [(0, 1, 0), (1, 2, 2), (2, 1, 1), (3, 2, 3)], + ) + + def test_switch_position_is_the_low_stage_bit(self) -> None: + # This is what lets the controller just invert its pin on every trigger cycle + # instead of tracking an absolute stage number. + channels = build_output_channels((1, 2), 2) + for channel in channels: + expected_position = channel.index // 2 # switch-position-major output order + self.assertEqual(channel.stage_index & 1, expected_position) + + def test_stage_budget_is_enforced(self) -> None: + build_output_channels((1, 2), 4) # 8 stages exactly fits + with self.assertRaises(ValueError): + build_output_channels((1, 2), 8) # 16 stages does not + for bad in (0, 3, -1): + with self.assertRaises(ValueError): + build_output_channels((1, 2), bad) + + def test_trace_names_do_not_collide_and_stay_stable(self) -> None: + two_output_names = build_forward_s_parameter_names(2, build_output_channels((1, 2), 1)) + self.assertEqual( + two_output_names, + ["S11", "S31", "S41", "S51", "S61", "S22", "S32", "S42", "S52", "S62"], + ) + four_output_names = build_forward_s_parameter_names(2, build_output_channels((1, 2), 2)) + # Receivers move to ports 5..8 so no forward term can be named like a + # reflection (a fixed base of 3 would make receiver 3 <- output 3 collide + # with the reflection of output 3). + self.assertEqual(len(four_output_names), len(set(four_output_names))) + self.assertEqual(len(four_output_names), 4 * 5) + for reflection in ("S11", "S22", "S33", "S44"): + self.assertIn(reflection, four_output_names) + + def test_service_and_collector_agree_on_trace_names(self) -> None: + # The service looks these up by name in whatever the collector produced, so a + # silent divergence would surface only as a missing-trace failure on hardware. + for substage_count in (1, 2): + channels = build_output_channels((1, 2), substage_count) + produced = {name.lower() for name in build_forward_s_parameter_names(2, channels)} + forward_by_output, reflection_by_output = _build_s_parameter_layout( + len(channels), slave_count=2 + ) + requested = set(reflection_by_output.values()) + for names in forward_by_output.values(): + requested.update(names) + self.assertEqual(requested, produced, msg=f"substages={substage_count}") + + def test_sweep_settings_carry_stages_and_substage_width(self) -> None: + configuration = SweepConfiguration( + start_hz=1_000_000, stop_hz=2_000_000, points=101, if_bandwidth=1000, power_dbm=-10.0 + ) + master = build_sweep_settings_payload( + configuration, + is_synchronization_master=True, + synchronization_enabled=True, + master_stimulus_ports=(1, 2), + switch_substage_count=2, + ) + stage_word = struct.unpack_from(" last index 3 + self.assertEqual((stage_word >> 3) & 0x07, 0) # port 1 selected by upper stage bits 0 + self.assertEqual((stage_word >> 6) & 0x07, 1) # port 2 -> 1 + self.assertEqual((stage_word >> 9) & 0x07, 1) # log2(2 positions) = 1 substage bit + + slave = build_sweep_settings_payload( + configuration, + is_synchronization_master=False, + synchronization_enabled=True, + master_stimulus_ports=(1, 2), + switch_substage_count=2, + ) + slave_stage_word = struct.unpack_from("> 3) & 0x07, DISABLED_PORT_STAGE) + self.assertEqual((slave_stage_word >> 6) & 0x07, DISABLED_PORT_STAGE) + + def test_disabled_switching_sends_the_stock_stage_word(self) -> None: + configuration = SweepConfiguration( + start_hz=1_000_000, stop_hz=2_000_000, points=101, if_bandwidth=1000, power_dbm=-10.0 + ) + payload = build_sweep_settings_payload( + configuration, + is_synchronization_master=True, + synchronization_enabled=True, + master_stimulus_ports=(1, 2), + ) + stage_word = struct.unpack_from("> 3) & 0x07, 0) + self.assertEqual((stage_word >> 6) & 0x07, 1) + self.assertEqual((stage_word >> 9) & 0x07, 0) # no substage bits -> stock bitstream + + def test_service_reports_four_outputs_when_switching(self) -> None: + service = MultiDeviceLibreVnaService( + master_serial="M", slave_serials=["A", "B"], backend_mode="mock", switch_substage_count=2 + ) + self.assertEqual(service.output_position_count, 4) + service.configure(RadarSweepModel(start_hz=1e6, stop_hz=2e6, points=4, if_bandwidth_hz=1e3)) + collection = service.acquire_collection(collection_id=1) + self.assertEqual(len(collection.traces), 4 * 4) + self.assertEqual( + sorted({trace.combo.output for trace in collection.traces}), [0, 1, 2, 3] + ) + + if __name__ == "__main__": unittest.main() diff --git a/python_app/tests/test_switched_matrix_capture.py b/python_app/tests/test_switched_matrix_capture.py index 71fbf38..98cb499 100644 --- a/python_app/tests/test_switched_matrix_capture.py +++ b/python_app/tests/test_switched_matrix_capture.py @@ -13,6 +13,7 @@ from unittest import mock import numpy as np +from python_app.hardware_full.matrix_radar_service import create_matrix_radar_service from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService from python_app.hardware_full.switched_matrix_radar_service import SwitchedMatrixRadarService from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData @@ -213,5 +214,40 @@ class ManualComboCaptureUsesTargetedAcquisitionTest(unittest.TestCase): self.assertEqual(inner_acquire.call_count, 2) +class PerPointStimulusSwitchingConfigTest(unittest.TestCase): + """Per-point stimulus switching widens the matrix without a host-driven switch.""" + + @staticmethod + def _config(stimulus_switch_positions: int) -> RunConfigModel: + config = RunConfigModel() + config.radar.model = RunConfigModel.LIBREVNA_MULTI_MODEL + config.radar.driver_mode = "mock" + config.radar.multi_device.slave_serials = ["SLAVE_A", "SLAVE_B"] + config.radar.multi_device.stimulus_switch_positions = stimulus_switch_positions + config.apply_device_model_constraints() + return config + + def test_two_positions_double_the_output_axis(self) -> None: + config = self._config(2) + self.assertEqual(config.matrix_virtual_output_positions, 4) + self.assertEqual(config.output_switch.positions, 4) + self.assertEqual(config.input_switch.positions, 4) + self.assertEqual(len(config.combos), 16) + + def test_host_does_not_drive_any_switch(self) -> None: + # The switch is stepped by the controller in the hardware trigger loop, so the + # host must acquire the whole 4x4 matrix in ONE sweep and never wrap the + # service in the sweep-per-step widener. + service = create_matrix_radar_service(self._config(2)) + self.assertIsInstance(service, MultiDeviceLibreVnaService) + self.assertNotIsInstance(service, SwitchedMatrixRadarService) + self.assertEqual(service.output_position_count, 4) + + def test_disabled_switching_keeps_the_two_output_matrix(self) -> None: + config = self._config(1) + self.assertEqual(config.matrix_virtual_output_positions, 2) + self.assertEqual(len(config.combos), 8) + + if __name__ == "__main__": unittest.main()