per-point stimulus switching: host side

Матрица 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) <noreply@anthropic.com>

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("<H", master, 25)[0]
+        self.assertEqual(stage_word & 0x07, 3)            # 4 stages -> 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("<H", slave, 25)[0]
+        self.assertEqual(slave_stage_word & 0x07, 3)      # same stage count as the master
+        # Receivers must never excite. The stock "one past the last stage" sentinel is
+        # unusable once the port comes from the upper stage bits, so both ports are
+        # parked at the explicit disabled value instead.
+        self.assertEqual((slave_stage_word >> 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("<H", payload, 25)[0]
+        self.assertEqual(stage_word & 0x07, 1)            # 2 stages
+        self.assertEqual((stage_word >> 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()