From 1ea2aabf878595434da2eb65820df9d465b0266e Mon Sep 17 00:00:00 2001 From: Ayzen Date: Tue, 28 Apr 2026 17:29:21 +0300 Subject: [PATCH] added multidevice support --- Makefile | 5 +- python_app/gui/app_window.py | 10 +- .../app_window_config/profile_io_mixin.py | 31 +- .../app_window_config/radar_limits_mixin.py | 19 +- .../app_window_config/state_builders.py | 65 +++- .../controllers/app_window_pipeline_mixin.py | 7 +- .../app_window_plot/bscan_plot_mixin.py | 47 ++- .../controllers/app_window_snapshot_mixin.py | 4 +- .../gui/controllers/sections/radar_section.py | 17 +- .../controllers/sections/switch_section.py | 2 + python_app/gui/main.py | 3 +- python_app/gui/runtime/history.py | 3 + python_app/gui/theme.py | 3 + .../hardware_full/librevna_driver/__init__.py | 11 +- .../librevna_multi_device_driver/__init__.py | 23 ++ .../controller.py | 253 ++++++++++++++ .../cycle_collection.py | 309 ++++++++++++++++++ .../librevna_multi_device_driver/models.py | 27 ++ .../librevna_multi_device_driver/protocol.py | 152 +++++++++ .../librevna_multi_device_driver/transport.py | 112 +++++++ .../hardware_full/multi_device_service.py | 222 +++++++++++++ python_app/models/run_config_codec.py | 32 ++ python_app/models/run_config_model.py | 2 + python_app/models/run_config_schema.py | 54 ++- .../orchestration/process_supervisor.py | 37 ++- python_app/orchestration/shm/__init__.py | 2 + python_app/orchestration/shm/ring_writer.py | 119 +++++++ python_app/orchestration/shm_reader.py | 2 + python_app/scripts/manual_smoke_run.py | 5 + .../scripts/multi_device_raw_producer.py | 95 ++++++ python_app/storage/npz/paths.py | 7 +- python_app/workflows/calibration_workflow.py | 7 + .../workflows/multi_radar_capture_workflow.py | 134 ++++++-- python_app/workflows/radar_config_variants.py | 5 + python_app/workflows/reference_workflow.py | 29 ++ .../workflows/sequential_capture_workflow.py | 116 +++++-- run_config.json | 27 +- 37 files changed, 1860 insertions(+), 138 deletions(-) create mode 100644 python_app/hardware_full/librevna_multi_device_driver/__init__.py create mode 100644 python_app/hardware_full/librevna_multi_device_driver/controller.py create mode 100644 python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py create mode 100644 python_app/hardware_full/librevna_multi_device_driver/models.py create mode 100644 python_app/hardware_full/librevna_multi_device_driver/protocol.py create mode 100644 python_app/hardware_full/librevna_multi_device_driver/transport.py create mode 100644 python_app/hardware_full/multi_device_service.py create mode 100644 python_app/orchestration/shm/ring_writer.py create mode 100644 python_app/scripts/multi_device_raw_producer.py diff --git a/Makefile b/Makefile index a208742..2bb5860 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ CXX := g++ -CXXFLAGS := -std=c++20 -O2 -Wall -Wextra -Wpedantic -pthread +CXXFLAGS := -std=c++20 -O2 -Wall -Wextra -Wpedantic -pthread -MMD -MP LDFLAGS := -pthread -lrt ORCH_LDFLAGS := $(LDFLAGS) -lusb-1.0 @@ -54,6 +54,7 @@ PROCESSOR_SOURCES := \ SWEEP_ORCH_OBJS := $(addprefix $(BUILD_DIR)/,$(COMMON_SOURCES:.cpp=.o) $(ORCH_SOURCES:.cpp=.o)) PREPROCESSOR_OBJS := $(addprefix $(BUILD_DIR)/,$(COMMON_SOURCES:.cpp=.o) $(PREPROC_SOURCES:.cpp=.o)) DATA_PROCESSOR_OBJS := $(addprefix $(BUILD_DIR)/,$(COMMON_SOURCES:.cpp=.o) $(PROCESSOR_SOURCES:.cpp=.o)) +DEPFILES := $(sort $(SWEEP_ORCH_OBJS:.o=.d) $(PREPROCESSOR_OBJS:.o=.d) $(DATA_PROCESSOR_OBJS:.o=.d)) TARGETS := \ $(BIN_DIR)/sweep_orchestrator \ @@ -80,5 +81,7 @@ $(BUILD_DIR)/%.o: %.cpp @mkdir -p $(dir $@) $(CXX) $(CXXFLAGS) $(INCLUDES) -c $< -o $@ +-include $(DEPFILES) + clean: rm -rf $(BUILD_DIR) diff --git a/python_app/gui/app_window.py b/python_app/gui/app_window.py index f875833..4b4092b 100644 --- a/python_app/gui/app_window.py +++ b/python_app/gui/app_window.py @@ -325,7 +325,7 @@ class AppWindow( def _apply_initial_radar_limits(self) -> None: """Apply startup radar-limits strategy according to selected radar mode.""" - if self._radar_mode.currentText() == "native": + if self._defaults_config.radar.driver_mode == "native": self._refresh_radar_limits_from_device() return self._apply_radar_limits_to_ui(None) @@ -364,15 +364,15 @@ class AppWindow( level_upper = level.upper() palette = { - "INFO": ("#7fb1ff", "#dce8f8", "#8ba2be"), - "WARN": ("#f4bf4f", "#f4dca0", "#9f8b55"), - "ERROR": ("#ff5f6d", "#ffd1d5", "#b5878c"), + "INFO": ("#1d5fbf", "#1f2937", "#526277"), + "WARN": ("#9a5b00", "#5c4300", "#7a6640"), + "ERROR": ("#c43d4d", "#6b1f2a", "#8b5d66"), } accent_color, message_color, detail_color = palette.get(level_upper, palette["INFO"]) timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] header_html = ( f"{html.escape(level_upper)}" - f" {html.escape(timestamp)}" + f" {html.escape(timestamp)}" f" {self._escape_log_text(text)}" ) diff --git a/python_app/gui/controllers/app_window_config/profile_io_mixin.py b/python_app/gui/controllers/app_window_config/profile_io_mixin.py index 2431a52..aa9ff94 100644 --- a/python_app/gui/controllers/app_window_config/profile_io_mixin.py +++ b/python_app/gui/controllers/app_window_config/profile_io_mixin.py @@ -11,6 +11,7 @@ from PyQt6.QtCore import QSignalBlocker from PyQt6.QtWidgets import QFileDialog from python_app.models.gui_profile_model import GuiProfileModel +from python_app.models.run_config_model import RunConfigModel from python_app.orchestration.preprocess_assets import ( VISIBLE_PREPROCESS_ASSET_KEYS, preprocess_asset_model, @@ -32,12 +33,36 @@ class AppWindowConfigProfileIOMixin: def _set_combo_selection_mode(self, mode: str) -> None: """Highlight current combo mode and enable only the relevant editors.""" + if self._is_multi_device_model_selected(): + self._run_combos_select_button.setChecked(True) + self._single_combo_select_button.setChecked(False) + self._combos_text.setText(self._fixed_multi_combo_text()) + self._combos_text.setEnabled(False) + self._single_combo_output.setEnabled(True) + self._single_combo_input.setEnabled(True) + self._run_combos_select_button.setEnabled(False) + self._single_combo_select_button.setEnabled(False) + return + text_selected = mode != "single" self._run_combos_select_button.setChecked(text_selected) self._single_combo_select_button.setChecked(not text_selected) self._combos_text.setEnabled(text_selected) self._single_combo_output.setEnabled(not text_selected) self._single_combo_input.setEnabled(not text_selected) + self._run_combos_select_button.setEnabled(True) + self._single_combo_select_button.setEnabled(True) + + def _is_multi_device_model_selected(self) -> bool: + """Return whether the loaded config targets LibreVNA multi-device acquisition.""" + return bool(self._defaults_config.is_multi_device) + + def _fixed_multi_combo_text(self) -> str: + """Return the canonical virtual combo matrix shown for multi-device mode.""" + return ",".join( + f"{int(combo.input)}:{int(combo.output)}" + for combo in RunConfigModel.build_multi_device_virtual_combos() + ) def _sync_pass_through_y_controls(self) -> None: """Enable Y-range editors only when fixed Y mode is active.""" @@ -138,14 +163,13 @@ class AppWindowConfigProfileIOMixin: """Apply already parsed profile to GUI state without restarting the pipeline.""" config = profile.run_config.clone() gui_state = profile.gui if profile.gui is not None else self._default_gui_state_for_config(config) + self._defaults_config = config selected_preprocess_sets = { key: str(preprocess_asset_model(config, key).set_name) for key in VISIBLE_PREPROCESS_ASSET_KEYS } radio_widgets = ( - self._serial_input, - self._radar_mode, self._start_hz_input, self._stop_hz_input, self._points_input, @@ -202,8 +226,6 @@ class AppWindowConfigProfileIOMixin: for widget in radio_widgets: blockers.enter_context(QSignalBlocker(widget)) - self._serial_input.setText(str(config.radar.serial)) - self._set_combo_current_text(self._radar_mode, str(config.radar.driver_mode)) self._start_hz_input.setText(f"{config.radar.sweep.start_hz:g}") self._stop_hz_input.setText(f"{config.radar.sweep.stop_hz:g}") self._points_input.setText(str(int(config.radar.sweep.points))) @@ -271,7 +293,6 @@ class AppWindowConfigProfileIOMixin: self._save_path_input.setText(str(gui_state.data_actions.save_path)) self._save_name_input.setText(str(gui_state.data_actions.save_name)) - self._defaults_config = config self._gui_defaults = gui_state self._selected_preprocess_sets = selected_preprocess_sets self._selected_preprocess_radar_key = self._radar_key(config) diff --git a/python_app/gui/controllers/app_window_config/radar_limits_mixin.py b/python_app/gui/controllers/app_window_config/radar_limits_mixin.py index f42c57d..c470e76 100644 --- a/python_app/gui/controllers/app_window_config/radar_limits_mixin.py +++ b/python_app/gui/controllers/app_window_config/radar_limits_mixin.py @@ -8,16 +8,6 @@ from python_app.hardware_full.librevna_service import LibreVnaService class AppWindowRadarLimitsMixin: """Handle LibreVNA capability probing and dependent UI clamping.""" - def _on_radar_identity_changed(self, *_args) -> None: - """Refresh device limits when radar identity/mode changes.""" - self._reset_preprocess_selection_after_radar_key_change() - if self._radar_mode.currentText() != "native": - self._apply_radar_limits_to_ui(None) - return - changed = self._refresh_radar_limits_from_device() - if changed: - self._on_processing_live_settings_changed() - def _on_radar_sweep_limits_changed(self) -> None: """Clamp processing frequency bounds after sweep start/stop edits.""" self._reset_preprocess_selection_after_radar_key_change() @@ -26,7 +16,7 @@ class AppWindowRadarLimitsMixin: def _refresh_radar_limits_from_device(self) -> bool: """Query native LibreVNA limits and apply them to GUI fields.""" - serial = self._serial_input.text().strip() + serial = self._defaults_config.radar.serial radar_service = LibreVnaService(serial=serial or None) if not radar_service.driver_available: self._fallback_to_mock_mode("LibreVNA Python driver is not available for device limits query") @@ -56,11 +46,8 @@ class AppWindowRadarLimitsMixin: self._radar_points_label.setText("Points") self._radar_ifbw_label.setText("IF BW Hz") self._radar_power_label.setText("Stimulus Power dBm") - if self._radar_mode.currentText() == "native": - self._radar_limits_hint.setText("Device limits unavailable in native mode (device not connected).") - else: - self._radar_limits_hint.setText("Mock mode: device limits are not applied.") - self._power_input.setToolTip("Device power limits are available only in native mode.") + self._radar_limits_hint.setText("Device limits are not available.") + self._power_input.setToolTip("Stimulus power configured in the active profile.") return False min_freq_hz = float(limits["min_frequency_hz"]) diff --git a/python_app/gui/controllers/app_window_config/state_builders.py b/python_app/gui/controllers/app_window_config/state_builders.py index d4f50e4..1bc6437 100644 --- a/python_app/gui/controllers/app_window_config/state_builders.py +++ b/python_app/gui/controllers/app_window_config/state_builders.py @@ -82,6 +82,11 @@ class AppWindowConfigStateBuildersMixin: @staticmethod def _format_combos_text_from_config(config: RunConfigModel) -> str: """Render configured combos for UI text editor, keeping full matrix as empty.""" + if config.is_multi_device: + return ",".join( + f"{int(combo.input)}:{int(combo.output)}" + for combo in RunConfigModel.build_multi_device_virtual_combos() + ) combos = list(config.combos) full_combos = config.build_full_combos(config.input_switch.positions, config.output_switch.positions) if len(combos) == len(full_combos) and all( @@ -215,8 +220,18 @@ class AppWindowConfigStateBuildersMixin: """Build GUI-only persistent state from current widget values.""" return GuiStateModel( switches=GuiSwitchStateModel( - combo_mode="single" if self._single_combo_select_button.isChecked() else "text", - combos_text=self._combos_text.text().strip(), + combo_mode=( + "text" + if self._is_multi_device_model_selected() + else "single" + if self._single_combo_select_button.isChecked() + else "text" + ), + combos_text=( + self._fixed_multi_combo_text() + if self._is_multi_device_model_selected() + else self._combos_text.text().strip() + ), single_input=self._single_combo_input.text().strip(), single_output=self._single_combo_output.text().strip(), ), @@ -292,19 +307,24 @@ class AppWindowConfigStateBuildersMixin: config.runtime.settling_ms = int(self._settling_ms.text().strip()) config.runtime.processing_live_config_path = str(self._live_config_writer.path) - if self._single_combo_select_button.isChecked(): - config.combos = [ - ComboModel( - input=int(self._single_combo_input.text().strip()), - output=int(self._single_combo_output.text().strip()), - ) - ] + if config.is_multi_device: + if len(config.radar.multi_device.slave_serials) != 2: + raise ValueError("LibreVNA multi-device mode requires exactly two slave serials") + config.apply_device_model_constraints() else: - combo_text = self._combos_text.text() - config.combos = parse_combos_from_text(combo_text) - config.ensure_combos() - if self._switches_are_effectively_static(config): - config.combos = [ComboModel(input=0, output=0)] + if self._single_combo_select_button.isChecked(): + config.combos = [ + ComboModel( + input=int(self._single_combo_input.text().strip()), + output=int(self._single_combo_output.text().strip()), + ) + ] + else: + combo_text = self._combos_text.text() + config.combos = parse_combos_from_text(combo_text) + config.ensure_combos() + if self._switches_are_effectively_static(config): + config.combos = [ComboModel(input=0, output=0)] for key in PREPROCESS_ASSET_KEYS: preprocess_asset_model(config, key).bundle_path = "" @@ -331,23 +351,36 @@ class AppWindowConfigStateBuildersMixin: sweep_points=config.radar.sweep.points, ifbw_hz=config.radar.sweep.if_bandwidth_hz, power_dbm=config.radar.sweep.power_dbm, + extra_serials=( + config.radar.multi_device.slave_serials + if config.is_multi_device + else None + ), ) def _radar_key_from_ui(self) -> str: """Build current radar key directly from radar widgets only.""" + model_name = self._defaults_config.radar.model or RunConfigModel.LIBREVNA_MODEL return radar_key_from_config( - model_name=self._defaults_config.radar.model, - serial=self._serial_input.text().strip(), + model_name=model_name, + serial=self._defaults_config.radar.serial, sweep_start_hz=float(self._start_hz_input.text().strip()), sweep_stop_hz=float(self._stop_hz_input.text().strip()), sweep_points=int(self._points_input.text().strip()), ifbw_hz=float(self._ifbw_input.text().strip()), power_dbm=float(self._power_input.text().strip()), + extra_serials=( + self._defaults_config.radar.multi_device.slave_serials + if model_name == RunConfigModel.LIBREVNA_MULTI_MODEL + else None + ), ) @staticmethod def _switches_are_effectively_static(config: RunConfigModel) -> bool: """Return `True` when switch setup effectively yields one fixed combo.""" + if config.is_multi_device: + return False has_single_position = config.input_switch.positions <= 1 and config.output_switch.positions <= 1 both_mock = config.input_switch.driver_mode == "mock" and config.output_switch.driver_mode == "mock" return has_single_position or both_mock diff --git a/python_app/gui/controllers/app_window_pipeline_mixin.py b/python_app/gui/controllers/app_window_pipeline_mixin.py index c373d70..a835fcb 100644 --- a/python_app/gui/controllers/app_window_pipeline_mixin.py +++ b/python_app/gui/controllers/app_window_pipeline_mixin.py @@ -153,9 +153,9 @@ class AppWindowPipelineMixin: self._stop_run() try: - if self._radar_mode.currentText() == "native": - self._refresh_radar_limits_from_device() config = self._build_config() + if config.radar.driver_mode == "native": + self._refresh_radar_limits_from_device() run_signature = self._build_run_history_signature(config) if processor_only_running and self._processor_requires_restart(run_signature): self._stop_all_processes() @@ -185,6 +185,9 @@ class AppWindowPipelineMixin: if config.radar.driver_mode != "native": self._log("Radar pre-configuration skipped (mock mode)") return + if config.is_multi_device: + self._log("Multi-device raw producer will configure all LibreVNA devices") + return radar_service = LibreVnaService(serial=config.radar.serial or None) if not radar_service.driver_available: diff --git a/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py b/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py index 77698cb..a23b273 100644 --- a/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py +++ b/python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py @@ -128,10 +128,13 @@ def rebuild_bscan_history_from_results( def pick_bscan_display_key( history_by_combo: dict[tuple[int, int], deque[np.ndarray]], + preferred_key: tuple[int, int] | None = None, ) -> tuple[int, int] | None: """Choose combo key to display when multiple histories are present.""" if not history_by_combo: return None + if preferred_key is not None and preferred_key in history_by_combo: + return preferred_key return next(iter(history_by_combo.keys())) @@ -267,9 +270,31 @@ class AppWindowBscanPlotMixin: def _pick_bscan_display_key(self) -> tuple[int, int] | None: """Choose combo history key to render.""" - display_key = pick_bscan_display_key(self._bscan_history_by_combo) + requested_key = self._requested_bscan_display_key() + display_key = pick_bscan_display_key( + self._bscan_history_by_combo, + preferred_key=requested_key, + ) available_keys = sorted(self._bscan_history_by_combo.keys()) + if requested_key is not None and requested_key != display_key and available_keys: + combo_signature = ",".join(f"{input_pos}:{output_pos}" for input_pos, output_pos in available_keys) + details = "\n".join( + f"- in{input_pos}/out{output_pos}" + for input_pos, output_pos in available_keys + ) + self._log_warning( + f"B-scan display combo in{requested_key[0]}/out{requested_key[1]} is not available; " + f"rendering in{display_key[0]}/out{display_key[1]} instead.", + details=details, + once_key=( + f"bscan_requested_combo_missing_{requested_key[0]}_{requested_key[1]}_" + f"{combo_signature}" + ), + ) + return display_key if display_key is not None and len(available_keys) > 1: + if requested_key == display_key: + return display_key combo_signature = ",".join(f"{input_pos}:{output_pos}" for input_pos, output_pos in available_keys) details = "\n".join( f"- in{input_pos}/out{output_pos}" @@ -286,6 +311,26 @@ class AppWindowBscanPlotMixin: ) return display_key + def _requested_bscan_display_key(self) -> tuple[int, int] | None: + """Return the multi-device B-scan display combo requested by the GUI.""" + if not ( + hasattr(self, "_is_multi_device_model_selected") + and self._is_multi_device_model_selected() + and hasattr(self, "_single_combo_input") + and hasattr(self, "_single_combo_output") + ): + return None + + input_text = self._single_combo_input.text().strip() + output_text = self._single_combo_output.text().strip() + if not input_text or not output_text: + return None + + try: + return int(input_text), int(output_text) + except ValueError: + return None + def _bscan_lookup_table(self, axis_mode: str) -> np.ndarray: """Return lookup table for current B-scan axis mode.""" return bscan_lookup_table(axis_mode) diff --git a/python_app/gui/controllers/app_window_snapshot_mixin.py b/python_app/gui/controllers/app_window_snapshot_mixin.py index 1eeb882..697016c 100644 --- a/python_app/gui/controllers/app_window_snapshot_mixin.py +++ b/python_app/gui/controllers/app_window_snapshot_mixin.py @@ -102,7 +102,9 @@ class AppWindowSnapshotMixin: config_profile_path = self._vna_json_config_profile_path(output_dir) try: - self._write_gui_profile_to_path(config_profile_path, allow_overwrite=False) + # Snapshot and VNA JSON exports intentionally share the same output directory stem. + # Refresh the companion GUI profile in place so a prior dataset save does not block JSON export. + self._write_gui_profile_to_path(config_profile_path, allow_overwrite=True) except Exception as exc: # noqa: BLE001 exported_preview = "\n".join(str(path) for path in output_paths[:8]) if len(output_paths) > 8: diff --git a/python_app/gui/controllers/sections/radar_section.py b/python_app/gui/controllers/sections/radar_section.py index 415c0ea..65e824e 100644 --- a/python_app/gui/controllers/sections/radar_section.py +++ b/python_app/gui/controllers/sections/radar_section.py @@ -2,7 +2,7 @@ from __future__ import annotations -from PyQt6.QtWidgets import QComboBox, QGroupBox, QLabel, QLineEdit, QVBoxLayout +from PyQt6.QtWidgets import QGroupBox, QLabel, QLineEdit, QVBoxLayout from python_app.gui.controllers.sections.layout_helpers import build_two_column_form_widget @@ -15,22 +15,12 @@ def build_radar_group(owner) -> QGroupBox: layout.setSpacing(8) defaults = owner._defaults_config.radar - owner._serial_input = QLineEdit(defaults.serial) - owner._serial_input.setPlaceholderText("Optional: empty = auto-detect first LibreVNA") - - owner._radar_mode = QComboBox() - owner._radar_mode.addItems(["mock", "native"]) - owner._radar_mode.setToolTip("mock: synthetic signal, native: real LibreVNA hardware") - owner._set_combo_current_text(owner._radar_mode, defaults.driver_mode) - owner._start_hz_input = QLineEdit(f"{defaults.sweep.start_hz:g}") owner._stop_hz_input = QLineEdit(f"{defaults.sweep.stop_hz:g}") owner._points_input = QLineEdit(str(defaults.sweep.points)) owner._ifbw_input = QLineEdit(f"{defaults.sweep.if_bandwidth_hz:g}") owner._power_input = QLineEdit(f"{defaults.sweep.power_dbm:g}") - owner._power_input.setToolTip("Device power limits are available only in native mode.") - owner._serial_input.editingFinished.connect(owner._on_radar_identity_changed) - owner._radar_mode.currentTextChanged.connect(owner._on_radar_identity_changed) + owner._power_input.setToolTip("Stimulus power configured in the active profile.") owner._start_hz_input.editingFinished.connect(owner._on_radar_sweep_limits_changed) owner._stop_hz_input.editingFinished.connect(owner._on_radar_sweep_limits_changed) owner._points_input.editingFinished.connect(owner._on_radar_sweep_limits_changed) @@ -43,7 +33,7 @@ def build_radar_group(owner) -> QGroupBox: owner._radar_ifbw_label = QLabel("IF BW Hz") owner._radar_power_label = QLabel("Stimulus Power dBm") - owner._radar_limits_hint = QLabel("Mock mode: device limits are not applied.") + owner._radar_limits_hint = QLabel("Device limits are not available.") owner._radar_limits_hint.setObjectName("hintLabel") layout.addWidget( @@ -56,7 +46,6 @@ def build_radar_group(owner) -> QGroupBox: (owner._radar_ifbw_label, owner._ifbw_input), (owner._radar_power_label, owner._power_input), ], - split_index=3, ) ) layout.addWidget(owner._radar_limits_hint) diff --git a/python_app/gui/controllers/sections/switch_section.py b/python_app/gui/controllers/sections/switch_section.py index d0ce5ce..e2349f8 100644 --- a/python_app/gui/controllers/sections/switch_section.py +++ b/python_app/gui/controllers/sections/switch_section.py @@ -63,6 +63,8 @@ def build_switch_group(owner) -> QGroupBox: owner._run_combos_select_button.clicked.connect(lambda: owner._set_combo_selection_mode("text")) owner._single_combo_select_button.clicked.connect(lambda: owner._set_combo_selection_mode("single")) + owner._single_combo_output.editingFinished.connect(owner._on_processing_live_settings_changed) + owner._single_combo_input.editingFinished.connect(owner._on_processing_live_settings_changed) owner._set_combo_selection_mode(str(switch_defaults.combo_mode)) return group diff --git a/python_app/gui/main.py b/python_app/gui/main.py index 1fae873..8a20e78 100644 --- a/python_app/gui/main.py +++ b/python_app/gui/main.py @@ -21,7 +21,8 @@ def main() -> int: """Run Qt event loop and show main radar control window.""" app = QApplication(sys.argv) apply_light_theme(app) - pg.setConfigOptions(antialias=True, background="#ffffff", foreground="#334155") + # PyQtGraph foreground controls axis lines, tick text, labels, and titles. + pg.setConfigOptions(antialias=True, background="#ffffff", foreground="#ffffff") window = AppWindow(PROJECT_ROOT) window.showMaximized() return app.exec() diff --git a/python_app/gui/runtime/history.py b/python_app/gui/runtime/history.py index 9441fe1..c01b68f 100644 --- a/python_app/gui/runtime/history.py +++ b/python_app/gui/runtime/history.py @@ -66,8 +66,11 @@ def build_run_history_signature( combos_signature = tuple((int(combo.input), int(combo.output)) for combo in config.combos) preprocess_signature = tuple(preprocess_asset_model(config, key).set_name for key in PREPROCESS_ASSET_KEYS) return ( + str(config.radar.model), str(config.radar.driver_mode), str(config.radar.serial), + tuple(str(value) for value in config.radar.multi_device.slave_serials), + bool(config.radar.multi_device.force_external_reference), float(config.radar.sweep.start_hz), float(config.radar.sweep.stop_hz), int(config.radar.sweep.points), diff --git a/python_app/gui/theme.py b/python_app/gui/theme.py index 72db41e..bb7fc7f 100644 --- a/python_app/gui/theme.py +++ b/python_app/gui/theme.py @@ -103,6 +103,9 @@ QTextEdit#runtimeLogBox { font-family: "DejaVu Sans Mono"; font-size: 12px; padding: 6px 8px; + color: #1f2937; + background-color: #fbfdff; + border-color: #c9d4e1; } QComboBox::drop-down { diff --git a/python_app/hardware_full/librevna_driver/__init__.py b/python_app/hardware_full/librevna_driver/__init__.py index 6263112..ecb75cd 100644 --- a/python_app/hardware_full/librevna_driver/__init__.py +++ b/python_app/hardware_full/librevna_driver/__init__.py @@ -2,7 +2,6 @@ import logging -from .device import LibreVNADevice from .enums import ( HardwareFamily, PacketType, @@ -71,4 +70,14 @@ __all__ = [ "VNASweepSettings", ] + +def __getattr__(name: str): + """Load USB-backed device class only when native hardware access is requested.""" + if name == "LibreVNADevice": + from .device import LibreVNADevice + + return LibreVNADevice + raise AttributeError(name) + + logging.getLogger(__name__).addHandler(logging.NullHandler()) diff --git a/python_app/hardware_full/librevna_multi_device_driver/__init__.py b/python_app/hardware_full/librevna_multi_device_driver/__init__.py new file mode 100644 index 0000000..a9ffd9d --- /dev/null +++ b/python_app/hardware_full/librevna_multi_device_driver/__init__.py @@ -0,0 +1,23 @@ +"""Synchronized multi-device LibreVNA driver.""" + +from python_app.hardware_full.librevna_multi_device_driver.models import SweepConfiguration, SweepMeasurementResult + +__all__ = [ + "MultiDeviceVnaController", + "SweepConfiguration", + "SweepMeasurementResult", + "list_connected_device_serial_numbers", +] + + +def __getattr__(name: str): + """Lazily import native USB pieces only when they are actually needed.""" + if name == "MultiDeviceVnaController": + from python_app.hardware_full.librevna_multi_device_driver.controller import MultiDeviceVnaController + + return MultiDeviceVnaController + if name == "list_connected_device_serial_numbers": + from python_app.hardware_full.librevna_multi_device_driver.transport import list_connected_device_serial_numbers + + return list_connected_device_serial_numbers + raise AttributeError(name) diff --git a/python_app/hardware_full/librevna_multi_device_driver/controller.py b/python_app/hardware_full/librevna_multi_device_driver/controller.py new file mode 100644 index 0000000..be6a7cb --- /dev/null +++ b/python_app/hardware_full/librevna_multi_device_driver/controller.py @@ -0,0 +1,253 @@ +"""Controller for synchronized multi-device LibreVNA sweeps.""" + +from __future__ import annotations + +from collections.abc import Iterator, Sequence +from dataclasses import replace +from typing import Optional +import time + +from python_app.hardware_full.librevna_multi_device_driver.cycle_collection import ( + collect_complete_running_sweep_cycles, +) +from python_app.hardware_full.librevna_multi_device_driver.models import ( + SweepConfiguration, + SweepMeasurementResult, +) +from python_app.hardware_full.librevna_multi_device_driver.protocol import ( + PacketType, + build_reference_settings_payload, + build_sweep_settings_payload, +) +from python_app.hardware_full.librevna_multi_device_driver.transport import LibreVnaUsbBulkConnection + + +class MultiDeviceVnaController: + """Coordinate one master LibreVNA and receiver slave LibreVNAs.""" + + def __init__( + self, + master_serial_number: str, + slave_serial_numbers: Sequence[str] = (), + force_external_reference: bool = True, + ) -> None: + """Open all configured devices and initialize controller state.""" + 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._reference_configuration_applied = False + self._last_applied_sweep_configuration: Optional[SweepConfiguration] = None + self._last_master_stimulus_ports: tuple[int, ...] | None = None + self._reconfigure_delay_s = 0.005 + self._sweep_is_running = False + self._is_closed = False + + try: + 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] + except Exception: + self.close() + raise + + def __enter__(self) -> MultiDeviceVnaController: + """Return this controller as a context manager resource.""" + return self + + def __exit__(self, *_args: object) -> None: + """Close all devices when leaving a context manager scope.""" + self.close() + + def close(self) -> None: + """Stop sweeping and close every opened device transport.""" + 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 device transports.""" + if self._is_closed: + return + self._send_idle_to_all_devices() + + def configure_continuous_sweep( + self, + sweep_configuration: SweepConfiguration, + *, + master_stimulus_ports: Sequence[int] = (1, 2), + ) -> None: + """Apply reference/sweep settings and leave devices sweeping.""" + if self._is_closed: + raise RuntimeError("Controller is already closed") + stimulus_ports = self._normalize_master_stimulus_ports(master_stimulus_ports) + + if not self._reference_configuration_applied: + self._configure_reference_clocks() + + if ( + self._sweep_is_running + and self._last_applied_sweep_configuration == sweep_configuration + and self._last_master_stimulus_ports == stimulus_ports + ): + return + + if self._sweep_is_running: + self._send_idle_to_all_devices() + time.sleep(self._reconfigure_delay_s) + + self._drain_all_received_packets() + self._configure_sweep_on_all_devices( + sweep_configuration, + master_stimulus_ports=stimulus_ports, + ) + + def collect_running_sweep_cycles( + self, + cycle_count: int = 1, + *, + datapoint_timeout_seconds: float | None = None, + ) -> SweepMeasurementResult: + """Collect complete cycles from the already-running synchronized sweep.""" + if self._is_closed: + raise RuntimeError("Controller is already closed") + if ( + not self._sweep_is_running + or self._last_applied_sweep_configuration is None + or self._last_master_stimulus_ports is None + or self._master_device 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, + master_stimulus_ports=self._last_master_stimulus_ports, + datapoint_timeout_seconds=datapoint_timeout_seconds, + ) + except Exception: + self._send_idle_to_all_devices() + raise + + def iter_running_sweep_cycles(self) -> Iterator[SweepMeasurementResult]: + """Yield complete sweep cycles forever from the configured 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: Exception | None = 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: # noqa: BLE001 + 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, + *, + master_stimulus_ports: tuple[int, ...], + ) -> None: + if self._master_device is None: + raise RuntimeError("Master device is not open") + + 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, + master_stimulus_ports=master_stimulus_ports, + ), + timeout_seconds=3.0, + retry_count=1, + ) + self._last_applied_sweep_configuration = replace(sweep_configuration) + self._last_master_stimulus_ports = master_stimulus_ports + self._sweep_is_running = True + + def _drain_all_received_packets(self) -> None: + for device_connection in self._all_devices: + device_connection.drain_received_packets() + + @staticmethod + def _normalize_master_stimulus_ports(master_stimulus_ports: Sequence[int]) -> tuple[int, ...]: + stimulus_ports = tuple(int(port) for port in master_stimulus_ports) + if not stimulus_ports: + raise ValueError("master_stimulus_ports must not be empty") + if len(stimulus_ports) > 2 or set(stimulus_ports) - {1, 2}: + raise ValueError("master_stimulus_ports may contain only ports 1 and 2") + if len(set(stimulus_ports)) != len(stimulus_ports): + raise ValueError("master_stimulus_ports must not contain duplicates") + return stimulus_ports 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 new file mode 100644 index 0000000..0030cfd --- /dev/null +++ b/python_app/hardware_full/librevna_multi_device_driver/cycle_collection.py @@ -0,0 +1,309 @@ +"""Collect complete synchronized sweep cycles from already-running devices.""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +import queue +import threading +import time + +import numpy as np + +from python_app.hardware_full.librevna_multi_device_driver.models import ( + SweepConfiguration, + SweepMeasurementResult, +) +from python_app.hardware_full.librevna_multi_device_driver.protocol import ( + PacketType, + ParsedVnaDatapoint, + find_receiver_value, + parse_vna_datapoint_payload, +) +from python_app.hardware_full.librevna_multi_device_driver.transport import LibreVnaUsbBulkConnection + +LIBREVNA_NATIVE_SWEEP_TIMEOUT_SECONDS = 1.5 + + +def collect_complete_running_sweep_cycles( + *, + master_device_connection: LibreVnaUsbBulkConnection, + slave_device_connections: Sequence[LibreVnaUsbBulkConnection], + active_sweep_configuration: SweepConfiguration, + cycle_count: int, + master_stimulus_ports: Sequence[int], + datapoint_timeout_seconds: float | None = None, +) -> 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") + stimulus_ports = tuple(int(port) for port in master_stimulus_ports) + if not stimulus_ports: + raise ValueError("master_stimulus_ports must not be empty") + if len(stimulus_ports) > 2 or set(stimulus_ports) - {1, 2}: + raise ValueError("master_stimulus_ports may contain only ports 1 and 2") + if len(set(stimulus_ports)) != len(stimulus_ports): + raise ValueError("master_stimulus_ports must not contain duplicates") + stage_by_master_port = {port: stage for stage, port in enumerate(stimulus_ports)} + + all_device_connections = [master_device_connection, *slave_device_connections] + point_count = active_sweep_configuration.points + frequencies_hz = np.zeros(point_count, 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 + } + 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) + } + + 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() + if datapoint_timeout_seconds is None: + datapoint_timeout_seconds = LIBREVNA_NATIVE_SWEEP_TIMEOUT_SECONDS + else: + datapoint_timeout_seconds = max(0.5, float(datapoint_timeout_seconds)) + + 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 + except Exception as exc: # noqa: BLE001 + 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 + cycle_tracking_state["previous_point_index"] = current_point_index + cycle_aware_handler(parsed_datapoint, 0) + return 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 + frequencies_hz[point_index] = parsed_datapoint.frequency_hz + + for master_stimulus_port, stage_index in stage_by_master_port.items(): + reference_receiver_value = find_receiver_value( + parsed_datapoint.receiver_values_by_description_mask, + stage_index=stage_index, + is_reference_receiver=True, + required_port_number=master_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, + is_reference_receiver=True, + ) + if reference_receiver_value is None: + continue + + master_reference_measurements_by_port[master_stimulus_port][ + cycle_index, + point_index, + ] = reference_receiver_value + + receiver_port_number = master_stimulus_port + s_parameter_name = master_reflection_s_parameter_name(master_stimulus_port) + port_receiver_value = find_receiver_value( + parsed_datapoint.receiver_values_by_description_mask, + stage_index=stage_index, + 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]: + receiver_base_port = 2 * slave_index + 3 + + def handle_slave_datapoint(parsed_datapoint: ParsedVnaDatapoint, cycle_index: int) -> None: + 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 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, + 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_by_port, + ), + ) + + +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 build_forward_s_parameter_names(slave_device_count: int, master_stimulus_ports: Sequence[int]) -> 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 slave_index in range(slave_device_count): + receiver_base_port = 2 * slave_index + 3 + names.extend( + [ + f"S{receiver_base_port}{master_stimulus_port}", + f"S{receiver_base_port + 1}{master_stimulus_port}", + ] + ) + return names + + +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], +) -> dict[str, np.ndarray]: + """Convert raw receiver captures from the final sweep cycle into S-parameters.""" + s_parameters: dict[str, np.ndarray] = {} + 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] + if np.isnan(master_reference_measurements).any(): + missing_reference_count = int(np.isnan(master_reference_measurements).sum()) + raise RuntimeError( + f"Master port {master_stimulus_port} reference missing for {missing_reference_count} datapoints" + ) + 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 {missing_measurement_count} datapoints" + ) + s_parameters[s_parameter_name.lower()] = raw_receiver_measurements[-1] / master_reference_measurements[-1] + return s_parameters diff --git a/python_app/hardware_full/librevna_multi_device_driver/models.py b/python_app/hardware_full/librevna_multi_device_driver/models.py new file mode 100644 index 0000000..256bc2e --- /dev/null +++ b/python_app/hardware_full/librevna_multi_device_driver/models.py @@ -0,0 +1,27 @@ +"""Models used by synchronized multi-device LibreVNA acquisition.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np + + +@dataclass(frozen=True, slots=True) +class SweepConfiguration: + """User-facing configuration for one VNA sweep.""" + + start_hz: int + stop_hz: int + points: int + if_bandwidth: int + power_dbm: float + dwell_us: int = 0 + + +@dataclass(slots=True) +class SweepMeasurementResult: + """Measured synchronized sweep for one active master output port.""" + + frequencies_hz: np.ndarray + s_parameters: dict[str, np.ndarray] diff --git a/python_app/hardware_full/librevna_multi_device_driver/protocol.py b/python_app/hardware_full/librevna_multi_device_driver/protocol.py new file mode 100644 index 0000000..30464ec --- /dev/null +++ b/python_app/hardware_full/librevna_multi_device_driver/protocol.py @@ -0,0 +1,152 @@ +"""Small LibreVNA protocol subset used by synchronized multi-device sweeps.""" + +from __future__ import annotations + +import struct +from collections.abc import Sequence +from dataclasses import dataclass + +from python_app.hardware_full.librevna_multi_device_driver.models import SweepConfiguration + + +class PacketType: + """Packet identifiers used by the multi-device sweep controller.""" + + 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(frozen=True, slots=True) +class ParsedVnaDatapoint: + """Decoded contents of one VNADatapoint payload.""" + + frequency_hz: int + point_index: int + receiver_values_by_description_mask: dict[int, complex] + + +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 + | int(bool(force_external_reference)) << 1 + ) + return struct.pack(" bytes: + """Build protocol-v14 SweepSettings for staged master outputs or synchronized receivers.""" + stimulus_ports = tuple(int(port) for port in master_stimulus_ports) + if not stimulus_ports: + raise ValueError("master_stimulus_ports must not be empty") + if len(stimulus_ports) > 2 or set(stimulus_ports) - {1, 2}: + raise ValueError("master_stimulus_ports may contain only ports 1 and 2") + if len(set(stimulus_ports)) != len(stimulus_ports): + raise ValueError("master_stimulus_ports must not contain duplicates") + + 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) + + configuration_bits = ( + 0 << 0 + | int(bool(is_synchronization_master)) << 1 + | 1 << 2 + | 1 << 3 + | 0 << 4 + | synchronization_mode << 5 + ) + + 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) + else: + port_1_stimulus_stage = inactive_stage_index + port_2_stimulus_stage = inactive_stage_index + + stage_configuration_bits = ( + (last_stage_index & 0x07) + | (port_1_stimulus_stage & 0x07) << 3 + | (port_2_stimulus_stage & 0x07) << 6 + ) + + return struct.pack( + " ParsedVnaDatapoint | None: + """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(" complex | None: + """Look up one complex receiver value by stage, receiver type, and optional port.""" + 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 diff --git a/python_app/hardware_full/librevna_multi_device_driver/transport.py b/python_app/hardware_full/librevna_multi_device_driver/transport.py new file mode 100644 index 0000000..4a175b5 --- /dev/null +++ b/python_app/hardware_full/librevna_multi_device_driver/transport.py @@ -0,0 +1,112 @@ +"""USB transport wrapper for one LibreVNA in a synchronized device group.""" + +from __future__ import annotations + +import logging +import queue +import threading +import time + +from python_app.hardware_full.librevna_driver.enums import PacketType as NativePacketType +from python_app.hardware_full.librevna_driver.models import Packet +from python_app.hardware_full.librevna_driver.protocol import FrameScanner, encode_frame +from python_app.hardware_full.librevna_driver.transport.usb import USBTransport +from python_app.hardware_full.librevna_multi_device_driver.protocol import PacketType + +logger = logging.getLogger(__name__) + + +class LibreVnaUsbBulkConnection: + """Minimal packet transport for one LibreVNA device.""" + + def __init__(self, serial_number: str) -> None: + if not serial_number: + raise ValueError("serial_number is required for multi-device acquisition") + self.serial_number = serial_number + self._scanner = FrameScanner() + self._received_packets: queue.Queue[tuple[int, bytes]] = queue.Queue() + self._fatal_error: Exception | None = None + self._fatal_lock = threading.Lock() + self._transport = USBTransport( + on_data=self._on_data, + on_disconnect=self._on_disconnect, + read_chunk_size=4096, + ) + self._transport.connect(serial=serial_number, timeout_s=2.0) + + def close(self) -> None: + """Close USB resources.""" + self._transport.disconnect() + + 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: + """Send one framed packet.""" + self._raise_if_failed() + native_packet_type = NativePacketType(int(packet_type)) + self._transport.write( + encode_frame(Packet(native_packet_type, payload)), + timeout_s=1.0, + ) + + def receive_packet(self, timeout_seconds: float = 2.0) -> tuple[int, bytes]: + """Wait for the next received packet.""" + deadline = time.monotonic() + timeout_seconds + while True: + self._raise_if_failed() + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError(f"Packet not received from {self.serial_number} within {timeout_seconds} s") + try: + return self._received_packets.get(timeout=min(0.2, remaining)) + except queue.Empty: + continue + + def wait_for_acknowledgement(self, timeout_seconds: float = 2.0) -> None: + """Wait for an ACK packet while ignoring unrelated asynchronous packets.""" + deadline = time.monotonic() + timeout_seconds + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError(f"ACK not received from {self.serial_number} within {timeout_seconds} s") + packet_type, _payload = self.receive_packet(timeout_seconds=remaining) + if packet_type == PacketType.ACKNOWLEDGE: + return + if packet_type == PacketType.NEGATIVE_ACKNOWLEDGE: + raise RuntimeError(f"Device {self.serial_number} returned NACK") + + def _on_data(self, chunk: bytes) -> None: + try: + packets = self._scanner.feed(chunk) + except Exception as exc: # noqa: BLE001 + self._set_fatal_error(exc) + return + for packet in packets: + self._received_packets.put((int(packet.type), bytes(packet.payload))) + + def _on_disconnect(self, exc: Exception) -> None: + self._set_fatal_error(exc) + + def _set_fatal_error(self, exc: Exception) -> None: + with self._fatal_lock: + if self._fatal_error is None: + logger.error("LibreVNA USB transport failed for %s: %s", self.serial_number, exc) + self._fatal_error = exc + + def _raise_if_failed(self) -> None: + with self._fatal_lock: + if self._fatal_error is None: + return + raise RuntimeError(f"USB transport failed for {self.serial_number}: {self._fatal_error}") from self._fatal_error + + +def list_connected_device_serial_numbers() -> list[str]: + """Return serial numbers of all connected LibreVNA devices.""" + return [device.serial for device in USBTransport.list_devices()] diff --git a/python_app/hardware_full/multi_device_service.py b/python_app/hardware_full/multi_device_service.py new file mode 100644 index 0000000..b0dd3d2 --- /dev/null +++ b/python_app/hardware_full/multi_device_service.py @@ -0,0 +1,222 @@ +"""High-level service for LibreVNA multi-device acquisition.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import logging +import math +import time +from typing import TYPE_CHECKING + +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.models import SweepConfiguration +from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData +from python_app.models.run_config_model import RadarSweepModel, RunConfigModel + +if TYPE_CHECKING: + from python_app.hardware_full.librevna_multi_device_driver.controller import MultiDeviceVnaController + +logger = logging.getLogger(__name__) + +_INPUT_S_PARAMETERS_BY_OUTPUT: dict[int, tuple[str, ...]] = { + 0: ("s31", "s41", "s51", "s61"), + 1: ("s32", "s42", "s52", "s62"), +} + +@dataclass(slots=True) +class MultiDeviceLibreVnaService: + """Acquire a virtual 2x4 switch matrix from synchronized LibreVNA devices.""" + + master_serial: str + slave_serials: list[str] + force_external_reference: bool = True + recovery_attempts: int = 3 + backend_mode: str = "auto" + _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) + _mock_phase: float = field(init=False, default=0.0, repr=False) + + def __post_init__(self) -> None: + """Validate static topology and backend selection.""" + self.master_serial = str(self.master_serial).strip() + self.slave_serials = [str(value).strip() for value in self.slave_serials if str(value).strip()] + if len(self.slave_serials) != 2: + raise ValueError("LibreVNA multi-device mode requires exactly two slave serial numbers") + self.recovery_attempts = max(0, int(self.recovery_attempts)) + + mode = self.backend_mode.strip().lower() + if mode not in {"auto", "native", "mock"}: + raise ValueError(f"Unsupported multi-device backend mode: {self.backend_mode}") + self.backend_mode = mode + self._using_mock_backend = mode == "mock" + + @property + def using_mock_backend(self) -> bool: + """Return whether this service is generating synthetic data.""" + return self._using_mock_backend + + def open(self) -> None: + """Open native device transports when not in mock mode.""" + if self._using_mock_backend or self._controller is not None: + return + try: + from python_app.hardware_full.librevna_multi_device_driver.controller import MultiDeviceVnaController + + self._controller = MultiDeviceVnaController( + master_serial_number=self.master_serial, + slave_serial_numbers=self.slave_serials, + force_external_reference=self.force_external_reference, + ) + except Exception: + if self.backend_mode == "native": + raise + self._using_mock_backend = True + self._controller = None + + def close(self) -> None: + """Close native device transports.""" + if self._controller is not None: + self._controller.close() + self._controller = None + + def recover(self) -> None: + """Reopen native device transports after a failed acquisition.""" + if self._using_mock_backend: + return + self.close() + time.sleep(0.25) + self.open() + + def configure(self, sweep: RadarSweepModel) -> None: + """Store sweep settings for subsequent full-matrix acquisitions.""" + self._sweep_configuration = SweepConfiguration( + start_hz=int(round(float(sweep.start_hz))), + stop_hz=int(round(float(sweep.stop_hz))), + points=int(sweep.points), + if_bandwidth=int(round(float(sweep.if_bandwidth_hz))), + power_dbm=float(sweep.power_dbm), + ) + + def acquire_collection(self, collection_id: int = 1) -> SweepCollection: + """Acquire one complete virtual 2x4 matrix in canonical combo order.""" + if self._sweep_configuration is None: + raise RuntimeError("Multi-device service is not configured") + capture_start_ns = time.monotonic_ns() + if self._using_mock_backend: + collection = self._acquire_mock_collection(collection_id, capture_start_ns) + else: + collection = self._acquire_native_collection_with_recovery(collection_id, capture_start_ns) + collection.capture_end_ns = time.monotonic_ns() + return collection + + def _acquire_native_collection_with_recovery( + self, + collection_id: int, + capture_start_ns: int, + ) -> SweepCollection: + last_error: Exception | None = None + for attempt_index in range(self.recovery_attempts + 1): + try: + return self._acquire_native_collection(collection_id, capture_start_ns) + except Exception as exc: # noqa: BLE001 + last_error = exc + if attempt_index >= self.recovery_attempts: + break + logger.warning( + "multi-device acquisition failed, reconnecting devices (%d/%d): %s", + attempt_index + 1, + self.recovery_attempts, + exc, + exc_info=True, + ) + self.recover() + + assert last_error is not None + raise last_error + + def _acquire_native_collection(self, collection_id: int, capture_start_ns: int) -> SweepCollection: + if self._controller is None: + raise RuntimeError("Multi-device controller is not open") + assert self._sweep_configuration is not None + + self._controller.configure_continuous_sweep(self._sweep_configuration) + result = self._controller.collect_running_sweep_cycles( + 1, + datapoint_timeout_seconds=LIBREVNA_NATIVE_SWEEP_TIMEOUT_SECONDS, + ) + normalized_s_parameters = { + str(name).lower(): np.asarray(values, dtype=np.complex64) + for name, values in result.s_parameters.items() + } + frequencies = np.asarray(result.frequencies_hz, dtype=np.float32) + + traces: list[TraceData] = [] + for output_pos in range(RunConfigModel.MULTI_DEVICE_OUTPUT_POSITIONS): + reflection = self._required_s_parameter( + normalized_s_parameters, + "s11" if output_pos == 0 else "s22", + ) + + for input_pos, s_parameter_name in enumerate(_INPUT_S_PARAMETERS_BY_OUTPUT[output_pos]): + traces.append( + TraceData( + combo=ComboKey(input_pos=input_pos, output_pos=output_pos), + frequency_hz=frequencies, + s11=reflection, + s21=self._required_s_parameter(normalized_s_parameters, s_parameter_name), + ) + ) + + return SweepCollection( + collection_id=int(collection_id), + monotonic_ns=time.monotonic_ns(), + traces=traces, + capture_start_ns=capture_start_ns, + ) + + def _acquire_mock_collection(self, collection_id: int, capture_start_ns: int) -> SweepCollection: + assert self._sweep_configuration is not None + points = int(self._sweep_configuration.points) + frequencies = np.linspace( + self._sweep_configuration.start_hz, + self._sweep_configuration.stop_hz, + points, + dtype=np.float32, + ) + 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): + 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) + + 1j * (0.22 + 0.04 * output_pos) * np.sin(reflected_phase) + ).astype(np.complex64) + for input_pos in range(RunConfigModel.MULTI_DEVICE_INPUT_POSITIONS): + gain = 0.45 + 0.08 * input_pos + 0.03 * output_pos + phase = base_phase * (1.0 + 0.03 * input_pos) + (0.4 * input_pos + 0.9 * output_pos) + s21 = (gain * np.cos(phase) + 1j * gain * np.sin(phase)).astype(np.complex64) + traces.append( + TraceData( + combo=ComboKey(input_pos=input_pos, output_pos=output_pos), + frequency_hz=frequencies, + s11=s11, + s21=s21, + ) + ) + self._mock_phase += 0.05 + return SweepCollection( + collection_id=int(collection_id), + monotonic_ns=time.monotonic_ns(), + traces=traces, + capture_start_ns=capture_start_ns, + ) + + @staticmethod + def _required_s_parameter(s_parameters: dict[str, np.ndarray], name: str) -> np.ndarray: + values = s_parameters.get(name) + if values is None: + raise RuntimeError(f"Multi-device sweep is missing required {name.upper()} trace") + return values diff --git a/python_app/models/run_config_codec.py b/python_app/models/run_config_codec.py index f0c0c78..94e427b 100644 --- a/python_app/models/run_config_codec.py +++ b/python_app/models/run_config_codec.py @@ -53,6 +53,7 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: run_payload.get("locator_server", run_payload.get("locator")), "run.locator_server", ) + multi_device_payload = _as_dict(radar_payload.get("multi_device"), "radar.multi_device") model.radar.model = str(radar_payload.get("model", model.radar.model)) model.radar.serial = str(radar_payload.get("serial", model.radar.serial)) @@ -66,9 +67,34 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: sweep_payload.get("if_bandwidth_hz", model.radar.sweep.if_bandwidth_hz) ) model.radar.sweep.power_dbm = float(sweep_payload.get("stimulus_power_dbm", model.radar.sweep.power_dbm)) + slave_serials_payload = multi_device_payload.get( + "slave_serials", + multi_device_payload.get("slave_serial_numbers", model.radar.multi_device.slave_serials), + ) + if isinstance(slave_serials_payload, list): + model.radar.multi_device.slave_serials = [str(value).strip() for value in slave_serials_payload if str(value).strip()] + elif isinstance(slave_serials_payload, str): + model.radar.multi_device.slave_serials = [ + value.strip() + for value in slave_serials_payload.split(",") + if value.strip() + ] + model.radar.multi_device.force_external_reference = bool( + multi_device_payload.get( + "force_external_reference", + model.radar.multi_device.force_external_reference, + ) + ) + model.radar.multi_device.recovery_attempts = int( + multi_device_payload.get( + "recovery_attempts", + model.radar.multi_device.recovery_attempts, + ) + ) load_switch_payload(port1_payload, model.output_switch) load_switch_payload(port2_payload, model.input_switch) + model.apply_device_model_constraints() model.runtime.settling_ms = int(run_payload.get("settling_ms", model.runtime.settling_ms)) model.runtime.idle_sleep_ms = int(run_payload.get("idle_sleep_ms", model.runtime.idle_sleep_ms)) @@ -177,6 +203,7 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: x_m=float(entry_payload.get("x_m", 0.0)), ) ) + model.apply_device_model_constraints() validate_gpr_model( model.gpr, input_switch_positions=model.input_switch.positions, @@ -214,6 +241,11 @@ def run_config_to_dict(model: RunConfigModel) -> dict[str, Any]: "serial": model.radar.serial, "driver_mode": model.radar.driver_mode, "mock_signal_hz": model.radar.mock_signal_hz, + "multi_device": { + "slave_serials": list(model.radar.multi_device.slave_serials), + "force_external_reference": model.radar.multi_device.force_external_reference, + "recovery_attempts": model.radar.multi_device.recovery_attempts, + }, "sweep": { "start_hz": model.radar.sweep.start_hz, "stop_hz": model.radar.sweep.stop_hz, diff --git a/python_app/models/run_config_model.py b/python_app/models/run_config_model.py index 08b59c4..21c62c6 100644 --- a/python_app/models/run_config_model.py +++ b/python_app/models/run_config_model.py @@ -10,6 +10,7 @@ from python_app.models.run_config_schema import ( PreprocessAssetModel, PreprocessNotchModel, PreprocessModel, + RadarMultiDeviceModel, RadarModel, RadarSweepModel, RingEndpointModel, @@ -36,6 +37,7 @@ __all__ = [ "PreprocessAssetModel", "PreprocessNotchModel", "PreprocessModel", + "RadarMultiDeviceModel", "RadarModel", "RadarSweepModel", "RingEndpointModel", diff --git a/python_app/models/run_config_schema.py b/python_app/models/run_config_schema.py index 5377e97..40b1b83 100644 --- a/python_app/models/run_config_schema.py +++ b/python_app/models/run_config_schema.py @@ -28,15 +28,25 @@ class RadarSweepModel: power_dbm: float = -30.0 +@dataclass(slots=True) +class RadarMultiDeviceModel: + """Multi-device LibreVNA topology settings.""" + + slave_serials: list[str] = field(default_factory=list) + force_external_reference: bool = True + recovery_attempts: int = 3 + + @dataclass(slots=True) class RadarModel: """Radar section of run configuration.""" - model: str = "" + model: str = "librevna" serial: str = "" driver_mode: str = "mock" mock_signal_hz: float = 1_000_000.0 sweep: RadarSweepModel = field(default_factory=RadarSweepModel) + multi_device: RadarMultiDeviceModel = field(default_factory=RadarMultiDeviceModel) @dataclass(slots=True) @@ -190,6 +200,11 @@ class RunConfigModel: gpr: GprModel = field(default_factory=GprModel) combos: list[ComboModel] = field(default_factory=list) + LIBREVNA_MODEL = "librevna" + LIBREVNA_MULTI_MODEL = "librevna_multi" + MULTI_DEVICE_INPUT_POSITIONS = 4 + MULTI_DEVICE_OUTPUT_POSITIONS = 2 + @staticmethod def build_full_combos(input_positions: int, output_positions: int) -> list[ComboModel]: """Build full Cartesian product of input/output switch positions.""" @@ -199,8 +214,45 @@ class RunConfigModel: for input_pos in range(input_positions) ] + @classmethod + def build_multi_device_virtual_combos(cls) -> list[ComboModel]: + """Build fixed virtual combo matrix for one master and two slave devices.""" + return cls.build_full_combos( + cls.MULTI_DEVICE_INPUT_POSITIONS, + cls.MULTI_DEVICE_OUTPUT_POSITIONS, + ) + + @property + def is_multi_device(self) -> bool: + """Return whether this config targets synchronized multi-device acquisition.""" + return self.radar.model == self.LIBREVNA_MULTI_MODEL + + def apply_device_model_constraints(self) -> None: + """Apply only required wire-format constraints for the selected device model.""" + if not self.is_multi_device: + return + + self.radar.model = self.LIBREVNA_MULTI_MODEL + self.output_switch.name = self.output_switch.name or "virtual_output" + 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 = self.MULTI_DEVICE_OUTPUT_POSITIONS + self.output_switch.default_position = 0 + + self.input_switch.name = self.input_switch.name or "virtual_input" + self.input_switch.driver_mode = "mock" + self.input_switch.driver = self.input_switch.driver or "h7992" + self.input_switch.radar_port = 2 + self.input_switch.positions = self.MULTI_DEVICE_INPUT_POSITIONS + self.input_switch.default_position = 0 + self.combos = self.build_multi_device_virtual_combos() + def ensure_combos(self) -> None: """Populate combos with full matrix when no explicit run combos are set.""" + if self.is_multi_device: + self.apply_device_model_constraints() + return if self.combos: return self.combos = self.build_full_combos(self.input_switch.positions, self.output_switch.positions) diff --git a/python_app/orchestration/process_supervisor.py b/python_app/orchestration/process_supervisor.py index fe262b0..2901f76 100644 --- a/python_app/orchestration/process_supervisor.py +++ b/python_app/orchestration/process_supervisor.py @@ -3,9 +3,11 @@ from __future__ import annotations from dataclasses import dataclass +import json from pathlib import Path import shlex import subprocess +import sys import time from typing import Iterable from typing import Sequence @@ -83,6 +85,7 @@ class ProcessSupervisor: if self.is_running(): raise RuntimeError("Acquisition processes are already running") + acquisition_command = self._acquisition_command(config_path) command_specs = { "data_processor": [ str(self._project_root / "build/bin/data_processor"), @@ -94,11 +97,7 @@ class ProcessSupervisor: "--config", str(config_path), ], - "sweep_orchestrator": [ - str(self._project_root / "build/bin/sweep_orchestrator"), - "--config", - str(config_path), - ], + "sweep_orchestrator": acquisition_command, } processor_was_running = self.is_processor_running() required_processes: list[str] = ["data_preprocessor"] @@ -168,6 +167,34 @@ class ProcessSupervisor: handle=handle, ) + def _acquisition_command(self, config_path: Path) -> list[str]: + """Return acquisition producer command selected by radar.model.""" + radar_model = self._read_radar_model(config_path) + if radar_model == "librevna_multi": + return [ + sys.executable, + "-m", + "python_app.scripts.multi_device_raw_producer", + "--config", + str(config_path), + ] + return [ + str(self._project_root / "build/bin/sweep_orchestrator"), + "--config", + str(config_path), + ] + + @staticmethod + def _read_radar_model(config_path: Path) -> str: + """Read `radar.model` cheaply without constructing the full config model.""" + payload = json.loads(config_path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + return "librevna" + radar_payload = payload.get("radar") + if not isinstance(radar_payload, dict): + return "librevna" + return str(radar_payload.get("model", "librevna")) + def _stop_processes(self, names: Iterable[str]) -> None: """Gracefully terminate processes, then force-kill on timeout.""" ordered_names = list(names) diff --git a/python_app/orchestration/shm/__init__.py b/python_app/orchestration/shm/__init__.py index 52bb9ac..bb34e31 100644 --- a/python_app/orchestration/shm/__init__.py +++ b/python_app/orchestration/shm/__init__.py @@ -9,6 +9,7 @@ from python_app.orchestration.shm.decoder import ( decode_trace_collection, ) from python_app.orchestration.shm.ring_reader import ShmRingReader +from python_app.orchestration.shm.ring_writer import ShmRingWriter __all__ = [ "ByteCursor", @@ -16,6 +17,7 @@ __all__ = [ "RAW_MAGIC", "RESULT_MAGIC", "ShmRingReader", + "ShmRingWriter", "decode_result_collection", "decode_trace_collection", ] diff --git a/python_app/orchestration/shm/ring_writer.py b/python_app/orchestration/shm/ring_writer.py new file mode 100644 index 0000000..807732f --- /dev/null +++ b/python_app/orchestration/shm/ring_writer.py @@ -0,0 +1,119 @@ +"""POSIX shared-memory ring writer compatible with the C++ IPC ring.""" + +from __future__ import annotations + +import mmap +import os +from pathlib import Path +import struct +from typing import Final + +_HEADER_SIZE: Final[int] = 64 +_SLOT_HEADER_SIZE: Final[int] = 16 +_MAGIC: Final[bytes] = b"RDRRING2" +_VERSION: Final[int] = 1 + + +class ShmRingWriter: + """Write binary payloads into the shared-memory ring used by C++ workers.""" + + def __init__(self, ring_name: str, capacity: int, slot_size_bytes: int) -> None: + """Open or create a POSIX SHM ring by name.""" + if not ring_name.startswith("/"): + raise ValueError("ring_name must start with '/'") + if capacity <= 0: + raise ValueError("capacity must be > 0") + if slot_size_bytes <= 0: + raise ValueError("slot_size_bytes must be > 0") + + self._ring_name = ring_name + self._capacity = int(capacity) + self._slot_size_bytes = int(slot_size_bytes) + self._mapped_size = _HEADER_SIZE + self._capacity * (_SLOT_HEADER_SIZE + self._slot_size_bytes) + self._path = Path("/dev/shm") / ring_name[1:] + + created = not self._path.exists() + fd = os.open(self._path, os.O_RDWR | os.O_CREAT, 0o660) + self._file = os.fdopen(fd, "r+b", buffering=0) + if created or self._path.stat().st_size != self._mapped_size: + self._file.truncate(self._mapped_size) + created = True + + self._mmap = mmap.mmap(self._file.fileno(), self._mapped_size) + if created: + self._initialize_header() + else: + self._validate_header() + + def close(self) -> None: + """Close mmap and file handle.""" + self._mmap.close() + self._file.close() + + def push(self, payload: bytes) -> bool: + """Push one payload with overwrite-oldest semantics on overflow.""" + if len(payload) > self._slot_size_bytes: + return False + + write_seq = self._read_u64(24) + read_seq = self._read_u64(32) + if max(0, write_seq - read_seq) >= self._capacity: + self._write_u64(32, read_seq + 1) + dropped = self._read_u64(40) + self._write_u64(40, dropped + 1) + + index = write_seq % self._capacity + slot_offset = _HEADER_SIZE + index * (_SLOT_HEADER_SIZE + self._slot_size_bytes) + payload_offset = slot_offset + _SLOT_HEADER_SIZE + + self._write_u32(slot_offset, len(payload)) + self._write_u32(slot_offset + 4, 0) + self._mmap[payload_offset : payload_offset + len(payload)] = payload + self._write_u64(slot_offset + 8, write_seq + 1) + self._write_u64(24, write_seq + 1) + return True + + @property + def ring_name(self) -> str: + """Return the POSIX SHM ring name.""" + return self._ring_name + + @property + def slot_size_bytes(self) -> int: + """Return maximum payload size per slot.""" + return self._slot_size_bytes + + def _initialize_header(self) -> None: + self._mmap[:] = b"\x00" * self._mapped_size + self._mmap[:8] = _MAGIC + self._write_u32(8, _VERSION) + self._write_u32(12, self._capacity) + self._write_u32(16, self._slot_size_bytes) + self._write_u32(20, 0) + self._write_u64(24, 0) + self._write_u64(32, 0) + self._write_u64(40, 0) + + def _validate_header(self) -> None: + magic = self._mmap[:8] + version = self._read_u32(8) + capacity = self._read_u32(12) + slot_size_bytes = self._read_u32(16) + if magic != _MAGIC: + raise RuntimeError(f"Shared memory ring magic mismatch for {self._ring_name}") + if version != _VERSION: + raise RuntimeError(f"Shared memory ring version mismatch for {self._ring_name}") + if capacity != self._capacity or slot_size_bytes != self._slot_size_bytes: + raise RuntimeError(f"Shared memory ring geometry mismatch for {self._ring_name}") + + def _read_u32(self, offset: int) -> int: + return struct.unpack_from(" int: + return struct.unpack_from(" None: + struct.pack_into(" None: + struct.pack_into(" int: sweep_points=config.radar.sweep.points, ifbw_hz=config.radar.sweep.if_bandwidth_hz, power_dbm=config.radar.sweep.power_dbm, + extra_serials=( + config.radar.multi_device.slave_serials + if config.is_multi_device + else None + ), ) s21_calibration_set = build_synthetic_collection(config, value_scale=1.0, s11_scale=0.15, s11_phase_offset=0.4) diff --git a/python_app/scripts/multi_device_raw_producer.py b/python_app/scripts/multi_device_raw_producer.py new file mode 100644 index 0000000..a18fdf6 --- /dev/null +++ b/python_app/scripts/multi_device_raw_producer.py @@ -0,0 +1,95 @@ +"""Raw acquisition producer for LibreVNA multi-device mode.""" + +from __future__ import annotations + +import argparse +import logging +from pathlib import Path +import signal +import threading +import time + +from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService +from python_app.models.run_config_model import RunConfigModel +from python_app.orchestration.shm import ShmRingWriter +from python_app.storage.npz.serialize import RAW_MAGIC, serialize_trace_collection + +logger = logging.getLogger(__name__) + + +def main() -> int: + """Run producer process until config or signal requests exit.""" + parser = argparse.ArgumentParser(description="Publish LibreVNA multi-device raw sweeps to SHM rings") + parser.add_argument("--config", required=True, type=Path, help="Path to run_config.json") + args = parser.parse_args() + + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") + stop_requested = threading.Event() + + def request_stop(_signum: int, _frame: object) -> None: + stop_requested.set() + + signal.signal(signal.SIGINT, request_stop) + signal.signal(signal.SIGTERM, request_stop) + + config = RunConfigModel.load_from_path(args.config) + config.apply_device_model_constraints() + if not config.is_multi_device: + raise RuntimeError("multi_device_raw_producer requires radar.model='librevna_multi'") + + raw_writer = ShmRingWriter( + config.rings.raw.name, + config.rings.raw.capacity, + config.rings.raw.slot_size_bytes, + ) + raw_tap_writer = ShmRingWriter( + config.rings.raw_tap.name, + config.rings.raw_tap.capacity, + config.rings.raw_tap.slot_size_bytes, + ) + radar = MultiDeviceLibreVnaService( + master_serial=config.radar.serial, + slave_serials=list(config.radar.multi_device.slave_serials), + force_external_reference=config.radar.multi_device.force_external_reference, + recovery_attempts=config.radar.multi_device.recovery_attempts, + backend_mode=config.radar.driver_mode, + ) + + try: + radar.open() + radar.configure(config.radar.sweep) + collection_id = 1 + while not stop_requested.is_set(): + collection_start = time.monotonic() + collection = radar.acquire_collection(collection_id=collection_id) + + payload = serialize_trace_collection(collection, RAW_MAGIC) + if not raw_writer.push(payload): + raise RuntimeError( + f"Raw payload size {len(payload)} exceeds ring slot size {raw_writer.slot_size_bytes}" + ) + if not raw_tap_writer.push(payload): + raise RuntimeError( + f"Raw tap payload size {len(payload)} exceeds ring slot size {raw_tap_writer.slot_size_bytes}" + ) + if not config.runtime.continuous: + break + collection_duration_s = time.monotonic() - collection_start + if collection_id == 1 or collection_id % 20 == 0 or collection_duration_s > 2.0: + logger.info( + "multi-device collection %d acquired in %.3f s", + collection_id, + collection_duration_s, + ) + collection_id += 1 + finally: + radar.close() + raw_tap_writer.close() + raw_writer.close() + + logger.info("multi-device raw producer stopped") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python_app/storage/npz/paths.py b/python_app/storage/npz/paths.py index 6279ac4..22a41e1 100644 --- a/python_app/storage/npz/paths.py +++ b/python_app/storage/npz/paths.py @@ -3,6 +3,7 @@ from __future__ import annotations import re +from collections.abc import Sequence def radar_key_from_config( @@ -13,9 +14,13 @@ def radar_key_from_config( sweep_points: int, ifbw_hz: float, power_dbm: float, + extra_serials: Sequence[str] | None = None, ) -> str: """Build deterministic key for calibration/reference set lookup.""" - serial_part = serial or "no_serial" + serial_parts = [serial or "no_serial"] + if extra_serials: + serial_parts.extend(str(value).strip() or "no_serial" for value in extra_serials) + serial_part = "_".join(sanitize_path_component(value) for value in serial_parts) start_token = _format_float_for_key(sweep_start_hz) stop_token = _format_float_for_key(sweep_stop_hz) ifbw_token = _format_float_for_key(ifbw_hz) diff --git a/python_app/workflows/calibration_workflow.py b/python_app/workflows/calibration_workflow.py index 3d00981..0c0f707 100644 --- a/python_app/workflows/calibration_workflow.py +++ b/python_app/workflows/calibration_workflow.py @@ -17,6 +17,13 @@ def capture_calibration_set( store: NpzStore, ) -> tuple[str, SweepCollection]: """Capture all switch combinations and persist them as calibration set.""" + if config.is_multi_device: + raise RuntimeError( + "LibreVNA multi-device S21 through calibration is not supported by this one-shot full-set helper. " + "Use the sequential preprocess capture flow so each virtual combo can be connected through " + "and captured explicitly." + ) + combos = RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions) radar = LibreVnaService(serial=config.radar.serial or None) diff --git a/python_app/workflows/multi_radar_capture_workflow.py b/python_app/workflows/multi_radar_capture_workflow.py index 3606b10..2a4d9d8 100644 --- a/python_app/workflows/multi_radar_capture_workflow.py +++ b/python_app/workflows/multi_radar_capture_workflow.py @@ -9,12 +9,17 @@ import time import numpy as np from python_app.hardware_full.librevna_service import LibreVnaService +from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService from python_app.hardware_full.switch_service import SwitchService from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData from python_app.models.run_config_model import ComboModel, RunConfigModel from python_app.storage.npz_store import NpzStore from python_app.workflows.radar_config_variants import RadarConfigVariant -from python_app.workflows.sequential_capture_workflow import SequentialCaptureState +from python_app.workflows.sequential_capture_workflow import ( + MULTI_DEVICE_MANUAL_CAPTURE_KINDS, + SequentialCaptureState, + select_trace_for_combo, +) @dataclass(frozen=True, slots=True) @@ -63,9 +68,15 @@ class MultiRadarSequentialCaptureSession: self._kind = kind self._set_name = set_name self._radar_variants = list(radar_variants) - self._combos = RunConfigModel.build_full_combos( - base_config.input_switch.positions, - base_config.output_switch.positions, + self._is_multi_device = base_config.is_multi_device + self._manual_multi_device_capture = self._is_multi_device and kind in MULTI_DEVICE_MANUAL_CAPTURE_KINDS + self._combos = ( + RunConfigModel.build_multi_device_virtual_combos() + if self._is_multi_device + else RunConfigModel.build_full_combos( + base_config.input_switch.positions, + base_config.output_switch.positions, + ) ) if not self._combos: raise RuntimeError("No switch combinations available for capture") @@ -78,27 +89,38 @@ class MultiRadarSequentialCaptureSession: self._next_index = 0 self._opened = False - self._radar = LibreVnaService(serial=base_config.radar.serial or None) - self._input_switch = SwitchService( - name=base_config.input_switch.name, - positions=base_config.input_switch.positions, - mode=base_config.input_switch.driver_mode, - driver=base_config.input_switch.driver, - gpio_chip=base_config.input_switch.gpio_chip, - pin_a=base_config.input_switch.pin_a, - pin_b=base_config.input_switch.pin_b, - invert_logic=base_config.input_switch.invert_logic, - ) - self._output_switch = SwitchService( - name=base_config.output_switch.name, - positions=base_config.output_switch.positions, - mode=base_config.output_switch.driver_mode, - driver=base_config.output_switch.driver, - gpio_chip=base_config.output_switch.gpio_chip, - pin_a=base_config.output_switch.pin_a, - pin_b=base_config.output_switch.pin_b, - invert_logic=base_config.output_switch.invert_logic, - ) + if self._is_multi_device: + self._radar = MultiDeviceLibreVnaService( + master_serial=base_config.radar.serial, + slave_serials=list(base_config.radar.multi_device.slave_serials), + force_external_reference=base_config.radar.multi_device.force_external_reference, + recovery_attempts=base_config.radar.multi_device.recovery_attempts, + backend_mode=base_config.radar.driver_mode, + ) + self._input_switch = None + self._output_switch = None + else: + self._radar = LibreVnaService(serial=base_config.radar.serial or None) + self._input_switch = SwitchService( + name=base_config.input_switch.name, + positions=base_config.input_switch.positions, + mode=base_config.input_switch.driver_mode, + driver=base_config.input_switch.driver, + gpio_chip=base_config.input_switch.gpio_chip, + pin_a=base_config.input_switch.pin_a, + pin_b=base_config.input_switch.pin_b, + invert_logic=base_config.input_switch.invert_logic, + ) + self._output_switch = SwitchService( + name=base_config.output_switch.name, + positions=base_config.output_switch.positions, + mode=base_config.output_switch.driver_mode, + driver=base_config.output_switch.driver, + gpio_chip=base_config.output_switch.gpio_chip, + pin_a=base_config.output_switch.pin_a, + pin_b=base_config.output_switch.pin_b, + invert_logic=base_config.output_switch.invert_logic, + ) @property def kind(self) -> str: @@ -118,8 +140,10 @@ class MultiRadarSequentialCaptureSession: try: self._radar.open() self._radar.configure(self._base_config.radar.sweep) - self._input_switch.open() - self._output_switch.open() + if self._input_switch is not None: + self._input_switch.open() + if self._output_switch is not None: + self._output_switch.open() except Exception: self.close() raise @@ -127,9 +151,11 @@ class MultiRadarSequentialCaptureSession: def close(self) -> None: """Close all opened hardware resources.""" with suppress(Exception): - self._output_switch.close() + if self._output_switch is not None: + self._output_switch.close() with suppress(Exception): - self._input_switch.close() + if self._input_switch is not None: + self._input_switch.close() with suppress(Exception): self._radar.close() self._opened = False @@ -140,7 +166,11 @@ class MultiRadarSequentialCaptureSession: return SequentialCaptureState( kind=self._kind, set_name=self._set_name, - captured_count=len(self._captured_batches), + captured_count=( + self._next_index + if self._is_multi_device and not self._manual_multi_device_capture + else len(self._captured_batches) + ), total_count=len(self._combos), current_combo=current_combo, can_undo=bool(self._captured_batches), @@ -156,6 +186,39 @@ class MultiRadarSequentialCaptureSession: if combo is None: raise RuntimeError("Capture session is already complete") + if self._is_multi_device: + traces: list[TraceData] = [] + variant_labels: list[str] = [] + for variant in self._radar_variants: + self._radar.configure(variant.config.radar.sweep) + if self._base_config.runtime.settling_ms > 0: + time.sleep(self._base_config.runtime.settling_ms / 1000.0) + collection = self._radar.acquire_collection(collection_id=1) + if not collection.traces: + raise RuntimeError(f"Multi-device variant {variant.display_name} returned no traces") + if self._manual_multi_device_capture: + trace = select_trace_for_combo(collection, combo) + self._traces_by_radar_key[variant.radar_key].append(trace) + traces.append(trace) + else: + self._traces_by_radar_key[variant.radar_key].extend(collection.traces) + traces.append(collection.traces[-1]) + variant_labels.append(variant.display_name) + + batch = MultiRadarCaptureBatch( + combo=combo, + traces=tuple(traces), + variant_labels=tuple(variant_labels), + ) + self._captured_batches.append(batch) + if self._manual_multi_device_capture: + self._next_index += 1 + else: + self._next_index = len(self._combos) + return batch + + assert self._input_switch is not None + assert self._output_switch is not None self._output_switch.switch_to(combo.output) self._input_switch.switch_to(combo.input) if self._base_config.runtime.settling_ms > 0: @@ -194,6 +257,17 @@ class MultiRadarSequentialCaptureSession: if not self._captured_batches or self._next_index <= 0: raise RuntimeError("No captured combo is available to undo") + if self._is_multi_device and not self._manual_multi_device_capture: + removed_batch = self._captured_batches[-1] + for variant in self._radar_variants: + traces = self._traces_by_radar_key[variant.radar_key] + if len(traces) < len(self._combos): + raise RuntimeError("Capture session state is inconsistent; missing multi-device traces") + del traces[-len(self._combos) :] + self._next_index = 0 + self._captured_batches.pop() + return removed_batch + expected_combo = self._combos[self._next_index - 1] removed_batch = self._captured_batches[-1] if ( diff --git a/python_app/workflows/radar_config_variants.py b/python_app/workflows/radar_config_variants.py index 2c02521..0020510 100644 --- a/python_app/workflows/radar_config_variants.py +++ b/python_app/workflows/radar_config_variants.py @@ -145,6 +145,11 @@ def _load_radar_config_variant(path: Path, *, base_config: RunConfigModel) -> Ra sweep_points=config.radar.sweep.points, ifbw_hz=config.radar.sweep.if_bandwidth_hz, power_dbm=config.radar.sweep.power_dbm, + extra_serials=( + config.radar.multi_device.slave_serials + if config.is_multi_device + else None + ), ) return RadarConfigVariant( source_path=path, diff --git a/python_app/workflows/reference_workflow.py b/python_app/workflows/reference_workflow.py index 5002573..dcd7e42 100644 --- a/python_app/workflows/reference_workflow.py +++ b/python_app/workflows/reference_workflow.py @@ -5,6 +5,7 @@ from __future__ import annotations import time from python_app.hardware_full.librevna_service import LibreVnaService +from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService from python_app.hardware_full.switch_service import SwitchService from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData from python_app.models.run_config_model import RunConfigModel @@ -17,6 +18,34 @@ def capture_reference_set( store: NpzStore, ) -> tuple[str, SweepCollection]: """Capture all switch combinations and persist them as reference set.""" + if config.is_multi_device: + radar = MultiDeviceLibreVnaService( + master_serial=config.radar.serial, + slave_serials=list(config.radar.multi_device.slave_serials), + force_external_reference=config.radar.multi_device.force_external_reference, + recovery_attempts=config.radar.multi_device.recovery_attempts, + backend_mode=config.radar.driver_mode, + ) + try: + radar.open() + radar.configure(config.radar.sweep) + collection = radar.acquire_collection(collection_id=1) + finally: + radar.close() + + radar_key = radar_key_from_config( + model_name=config.radar.model, + serial=config.radar.serial, + sweep_start_hz=config.radar.sweep.start_hz, + sweep_stop_hz=config.radar.sweep.stop_hz, + sweep_points=config.radar.sweep.points, + ifbw_hz=config.radar.sweep.if_bandwidth_hz, + power_dbm=config.radar.sweep.power_dbm, + extra_serials=config.radar.multi_device.slave_serials, + ) + store.save_set("s21_reference", radar_key, set_name, collection) + return radar_key, collection + combos = RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions) radar = LibreVnaService(serial=config.radar.serial or None) diff --git a/python_app/workflows/sequential_capture_workflow.py b/python_app/workflows/sequential_capture_workflow.py index 529c666..64d1792 100644 --- a/python_app/workflows/sequential_capture_workflow.py +++ b/python_app/workflows/sequential_capture_workflow.py @@ -9,11 +9,14 @@ import time import numpy as np from python_app.hardware_full.librevna_service import LibreVnaService +from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService from python_app.hardware_full.switch_service import SwitchService from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData from python_app.models.run_config_model import ComboModel, RunConfigModel from python_app.storage.npz_store import NpzStore, radar_key_from_config +MULTI_DEVICE_MANUAL_CAPTURE_KINDS = frozenset({"s21_calibration", "s11_open", "s11_short", "s11_load"}) + @dataclass(slots=True) class SequentialCaptureState: @@ -42,7 +45,13 @@ class SequentialCaptureSession: self._config = config self._kind = kind self._set_name = set_name - self._combos = RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions) + self._is_multi_device = config.is_multi_device + self._manual_multi_device_capture = self._is_multi_device and kind in MULTI_DEVICE_MANUAL_CAPTURE_KINDS + self._combos = ( + RunConfigModel.build_multi_device_virtual_combos() + if self._is_multi_device + else RunConfigModel.build_full_combos(config.input_switch.positions, config.output_switch.positions) + ) if not self._combos: raise RuntimeError("No switch combinations available for capture") @@ -50,27 +59,38 @@ class SequentialCaptureSession: self._next_index = 0 self._opened = False - self._radar = LibreVnaService(serial=config.radar.serial or None) - self._input_switch = SwitchService( - name=config.input_switch.name, - positions=config.input_switch.positions, - mode=config.input_switch.driver_mode, - driver=config.input_switch.driver, - gpio_chip=config.input_switch.gpio_chip, - pin_a=config.input_switch.pin_a, - pin_b=config.input_switch.pin_b, - invert_logic=config.input_switch.invert_logic, - ) - self._output_switch = SwitchService( - name=config.output_switch.name, - positions=config.output_switch.positions, - mode=config.output_switch.driver_mode, - driver=config.output_switch.driver, - gpio_chip=config.output_switch.gpio_chip, - pin_a=config.output_switch.pin_a, - pin_b=config.output_switch.pin_b, - invert_logic=config.output_switch.invert_logic, - ) + if self._is_multi_device: + self._radar = MultiDeviceLibreVnaService( + master_serial=config.radar.serial, + slave_serials=list(config.radar.multi_device.slave_serials), + force_external_reference=config.radar.multi_device.force_external_reference, + recovery_attempts=config.radar.multi_device.recovery_attempts, + backend_mode=config.radar.driver_mode, + ) + self._input_switch = None + self._output_switch = None + else: + self._radar = LibreVnaService(serial=config.radar.serial or None) + self._input_switch = SwitchService( + name=config.input_switch.name, + positions=config.input_switch.positions, + mode=config.input_switch.driver_mode, + driver=config.input_switch.driver, + gpio_chip=config.input_switch.gpio_chip, + pin_a=config.input_switch.pin_a, + pin_b=config.input_switch.pin_b, + invert_logic=config.input_switch.invert_logic, + ) + self._output_switch = SwitchService( + name=config.output_switch.name, + positions=config.output_switch.positions, + mode=config.output_switch.driver_mode, + driver=config.output_switch.driver, + gpio_chip=config.output_switch.gpio_chip, + pin_a=config.output_switch.pin_a, + pin_b=config.output_switch.pin_b, + invert_logic=config.output_switch.invert_logic, + ) @property def kind(self) -> str: @@ -90,8 +110,10 @@ class SequentialCaptureSession: try: self._radar.open() self._radar.configure(self._config.radar.sweep) - self._input_switch.open() - self._output_switch.open() + if self._input_switch is not None: + self._input_switch.open() + if self._output_switch is not None: + self._output_switch.open() except Exception: self.close() raise @@ -99,9 +121,11 @@ class SequentialCaptureSession: def close(self) -> None: """Close all opened hardware resources.""" with suppress(Exception): - self._output_switch.close() + if self._output_switch is not None: + self._output_switch.close() with suppress(Exception): - self._input_switch.close() + if self._input_switch is not None: + self._input_switch.close() with suppress(Exception): self._radar.close() self._opened = False @@ -127,6 +151,22 @@ class SequentialCaptureSession: if combo is None: raise RuntimeError("Capture session is already complete") + if self._is_multi_device: + collection = self._radar.acquire_collection(collection_id=1) + if self._manual_multi_device_capture: + trace = select_trace_for_combo(collection, combo) + self._traces.append(trace) + self._next_index += 1 + return trace + + self._traces.extend(collection.traces) + self._next_index = len(self._combos) + if not collection.traces: + raise RuntimeError("Multi-device capture returned no traces") + return collection.traces[-1] + + assert self._input_switch is not None + assert self._output_switch is not None self._output_switch.switch_to(combo.output) self._input_switch.switch_to(combo.input) if self._config.runtime.settling_ms > 0: @@ -150,6 +190,14 @@ class SequentialCaptureSession: if not self._traces or self._next_index <= 0: raise RuntimeError("No captured combo is available to undo") + if self._is_multi_device and not self._manual_multi_device_capture: + if len(self._traces) != len(self._combos): + raise RuntimeError("Capture session state is inconsistent; multi-device trace matrix is incomplete") + removed_trace = self._traces[-1] + self._traces.clear() + self._next_index = 0 + return removed_trace + expected_combo = self._combos[self._next_index - 1] removed_trace = self._traces[-1] if ( @@ -193,6 +241,11 @@ class SequentialCaptureSession: sweep_points=self._config.radar.sweep.points, ifbw_hz=self._config.radar.sweep.if_bandwidth_hz, power_dbm=self._config.radar.sweep.power_dbm, + extra_serials=( + self._config.radar.multi_device.slave_serials + if self._config.is_multi_device + else None + ), ) store.save_set(self._kind, radar_key, self._set_name, collection) return radar_key, collection @@ -202,3 +255,14 @@ class SequentialCaptureSession: if self._next_index >= len(self._combos): return None return self._combos[self._next_index] + + +def select_trace_for_combo(collection: SweepCollection, combo: ComboModel) -> TraceData: + """Return the trace matching a virtual combo from a full multi-device capture.""" + for trace in collection.traces: + if ( + int(trace.combo.input_pos) == int(combo.input) + and int(trace.combo.output_pos) == int(combo.output) + ): + return trace + raise RuntimeError(f"Multi-device capture is missing trace for input={combo.input}, output={combo.output}") diff --git a/run_config.json b/run_config.json index 195fd4f..b763f27 100644 --- a/run_config.json +++ b/run_config.json @@ -1,9 +1,17 @@ { "radar": { - "model": "librevna", - "serial": "", + "model": "librevna_multi", + "serial": "207730885532", "driver_mode": "native", "mock_signal_hz": 5000000.0, + "multi_device": { + "slave_serials": [ + "20A1307D5532", + "2072306C5532" + ], + "force_external_reference": true, + "recovery_attempts": 3 + }, "sweep": { "start_hz": 1000000.0, "stop_hz": 6000000000.0, @@ -57,10 +65,10 @@ {"input": 1, "output": 0}, {"input": 2, "output": 0}, {"input": 3, "output": 0}, - {"input": 0, "output": 3}, - {"input": 1, "output": 3}, - {"input": 2, "output": 3}, - {"input": 3, "output": 3} + {"input": 0, "output": 1}, + {"input": 1, "output": 1}, + {"input": 2, "output": 1}, + {"input": 3, "output": 1} ] }, "preprocess": { @@ -97,11 +105,6 @@ "notch": { "enabled": true, "bands_hz": [ - [790000000.0, 850000000.0], - [1790000000.0, 1890000000.0], - [1970000000.0, 2050000000.0], - [2080000000.0, 2560000000.0], - [2650000000.0, 2700000000.0] ], "taper_width_hz": 40000000.0, "taper_type": "cosine" @@ -116,7 +119,7 @@ "x_m": 0.905 }, { - "output_pos": 3, + "output_pos": 1, "x_m": -0.905 } ],