From 3efe968dd1f7e31ca6371245bff5e6594b604702 Mon Sep 17 00:00:00 2001 From: Ayzen Date: Thu, 2 Jul 2026 17:44:24 +0300 Subject: [PATCH] some kamil_adc fixes --- Makefile | 39 ++++- .../include/switch_sequencer.h | 147 ++++++++++++++++++ .../kamil_adc_collector/src/main.cpp | 124 +++++++++++++++ .../hardware_full/kamil_adc/protocol.py | 31 +++- python_app/hardware_full/kamil_adc/service.py | 58 ++++++- .../hardware_full/kamil_adc/tty_reader.py | 56 ++++++- python_app/scripts/kamil_adc_raw_producer.py | 82 +++++++--- python_app/tests/test_kamil_adc_protocol.py | 39 +++++ python_app/tests/test_kamil_adc_service.py | 100 +++++++++++- .../workflows/sequential_capture_workflow.py | 10 ++ 10 files changed, 644 insertions(+), 42 deletions(-) create mode 100644 data_acq_and_processing/kamil_adc_collector/include/switch_sequencer.h diff --git a/Makefile b/Makefile index f071be8..5ff555c 100644 --- a/Makefile +++ b/Makefile @@ -65,17 +65,36 @@ PROCESSOR_SOURCES := \ # from the vendored headers alone and only needs those libraries present at run # time. Built on demand (not part of `all`) for radar.model == kamil_adc. KAMIL_COLLECTOR_DIR := data_acq_and_processing/kamil_adc_collector -KAMIL_COLLECTOR_INCLUDES := -I$(KAMIL_COLLECTOR_DIR)/include -I$(KAMIL_COLLECTOR_DIR)/vendor/lcard +# The collector now also drives the RF switches itself (switch-aware mode), so it +# reuses the orchestrator's switch drivers and the shared run_config loader, which +# pull in the common config/ipc/locator headers and the vendored nlohmann/json. +KAMIL_COLLECTOR_INCLUDES := \ + -I$(KAMIL_COLLECTOR_DIR)/include \ + -I$(KAMIL_COLLECTOR_DIR)/vendor/lcard \ + -Idata_acq_and_processing/common_cpp/ipc/include \ + -Idata_acq_and_processing/common_cpp/config/include \ + -Idata_acq_and_processing/sweep_orchestrator/device_drivers/interfaces \ + -Idata_acq_and_processing/sweep_orchestrator/device_drivers/switches \ + -Idata_acq_and_processing/processing/locator/include \ + -Idata_acq_and_processing/third_party KAMIL_COLLECTOR_LDFLAGS := -pthread -ldl -lutil +# Collector sources, including the shared drivers/config loader it reuses. These +# are compiled into a private build/kamil/ tree (see rule below) so they never +# collide with the orchestrator's objects, which use different compile flags. KAMIL_COLLECTOR_SOURCES := \ $(KAMIL_COLLECTOR_DIR)/src/main.cpp \ $(KAMIL_COLLECTOR_DIR)/src/tty_protocol_writer.cpp \ - $(KAMIL_COLLECTOR_DIR)/src/capture_file_writer.cpp + $(KAMIL_COLLECTOR_DIR)/src/capture_file_writer.cpp \ + data_acq_and_processing/common_cpp/config/src/run_config.cpp \ + data_acq_and_processing/common_cpp/ipc/src/shared_types.cpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.cpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.cpp 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)) -KAMIL_COLLECTOR_OBJS := $(addprefix $(BUILD_DIR)/,$(KAMIL_COLLECTOR_SOURCES:.cpp=.o)) +KAMIL_BUILD_DIR := $(BUILD_DIR)/kamil +KAMIL_COLLECTOR_OBJS := $(addprefix $(KAMIL_BUILD_DIR)/,$(notdir $(KAMIL_COLLECTOR_SOURCES:.cpp=.o))) DEPFILES := $(sort $(SWEEP_ORCH_OBJS:.o=.d) $(PREPROCESSOR_OBJS:.o=.d) $(DATA_PROCESSOR_OBJS:.o=.d) $(KAMIL_COLLECTOR_OBJS:.o=.d)) TARGETS := \ @@ -101,11 +120,15 @@ $(BIN_DIR)/data_processor: $(DATA_PROCESSOR_OBJS) @mkdir -p $(BIN_DIR) $(CXX) $(DATA_PROCESSOR_OBJS) -o $@ $(LDFLAGS) -# The collector is self-contained: compile its objects with only the vendored -# L-Card headers (no project/VISA includes) by overriding the generic rule's -# variables for these objects, then link with dlopen/openpty support. -$(KAMIL_COLLECTOR_OBJS): INCLUDES := $(KAMIL_COLLECTOR_INCLUDES) -$(KAMIL_COLLECTOR_OBJS): VISA_CXXFLAGS := +# The collector compiles into its own build/kamil/ tree with collector-only +# includes (no VISA), so the shared driver/config sources it reuses never collide +# with the orchestrator's objects of the same name. Basenames are unique, so a +# vpath lets one pattern rule find every source. +vpath %.cpp $(sort $(dir $(KAMIL_COLLECTOR_SOURCES))) + +$(KAMIL_BUILD_DIR)/%.o: %.cpp + @mkdir -p $(dir $@) + $(CXX) $(CXXFLAGS) $(KAMIL_COLLECTOR_INCLUDES) -c $< -o $@ $(BIN_DIR)/kamil_adc_collector: $(KAMIL_COLLECTOR_OBJS) @mkdir -p $(BIN_DIR) diff --git a/data_acq_and_processing/kamil_adc_collector/include/switch_sequencer.h b/data_acq_and_processing/kamil_adc_collector/include/switch_sequencer.h new file mode 100644 index 0000000..f2ec2b9 --- /dev/null +++ b/data_acq_and_processing/kamil_adc_collector/include/switch_sequencer.h @@ -0,0 +1,147 @@ +#pragma once + +#include +#include +#include +#include + +#include "shared_types.hpp" +#include "switch_driver.hpp" + +namespace radar::kamil { + +/** + * @brief Drives the input/output RF switches in lock-step with sweep boundaries. + * + * The E-502 stream is free-running: sweeps are delimited by DI_SYN2 edges with a + * hardware idle gap between them. On every completed sweep we step to the next + * switch combination *during that gap*, so the next sweep starts already settled + * in the new state and no samples are lost. + * + * The one failure mode of doing this purely in software is the readout backlog: + * if, at the moment we observe a sweep's end, we are running so far behind real + * time that the toggle could not have landed within the gap, the upcoming sweep + * straddles the transition. We flag that sweep *dirty*; the consumer drops it and + * we re-take the same combination on the following sweep (which is then clean, + * because no transition happens during its gap). A clean sweep costs nothing; a + * miss costs exactly one extra sweep period for that one combination. + * + * The sequencer owns the two switch drivers and exposes only what the collector + * needs at a sweep boundary: which combination the upcoming sweep belongs to and + * whether it is expected to be clean. All timing policy lives here so the recv + * loop stays readable. + */ +class SwitchSequencer { + public: + struct Settings { + /// Combinations to cycle through, one sweep each, in order. Empty disables + /// the sequencer entirely (the collector then streams a single passthrough + /// channel exactly as before). + std::vector combos{}; + /// Guaranteed minimum inter-sweep idle window, in milliseconds. This is a + /// hardware property of the sweep generator; treat it as a lower bound. + double gap_ms = 10.0; + /// RF settle time required after a position change, in milliseconds. + double settle_ms = 1.0; + }; + + SwitchSequencer( + Settings settings, + std::unique_ptr input_switch, + std::unique_ptr output_switch + ) + : settings_(std::move(settings)), + input_switch_(std::move(input_switch)), + output_switch_(std::move(output_switch)) {} + + [[nodiscard]] auto enabled() const -> bool { return !settings_.combos.empty(); } + + /// Open both switch drivers and park at the first combination. + void open() { + if (!enabled()) { + return; + } + output_switch_->open(); + input_switch_->open(); + index_ = 0; + applied_ = settings_.combos.front(); + apply(applied_); + upcoming_dirty_ = false; // the first sweep is captured in a settled state + } + + void close() { + if (!enabled()) { + return; + } + // Best-effort: teardown must never throw out of the collector's stop path. + try { + output_switch_->close(); + } catch (...) { + } + try { + input_switch_->close(); + } catch (...) { + } + } + + /// Combination the in-progress / upcoming sweep is captured under. + [[nodiscard]] auto current_combo() const -> radar::ipc::ComboKey { + return settings_.combos[index_]; + } + + /// Whether the in-progress / upcoming sweep is expected to straddle a switch + /// transition and must be dropped by the consumer. + [[nodiscard]] auto current_dirty() const -> bool { return upcoming_dirty_; } + + /** + * @brief Advance the sequence at a completed sweep's end (start of the gap). + * + * @param backlog_ms Estimated readout lag at this instant — how far behind + * real time we are, i.e. how late the toggle we issue now will land. + * + * If the sweep that just ended was clean we step to the next combination and + * toggle the switches now, inside the gap. If it was dirty we re-take the same + * combination (the switches are already there and long settled). The upcoming + * sweep is marked dirty only when a real transition happens *and* the backlog + * leaves no room for it to settle before the next sweep starts. + */ + void on_sweep_end(double backlog_ms) { + if (!enabled()) { + return; + } + + if (upcoming_dirty_) { + // The sweep that just ended straddled a transition: re-take the same + // combination. The switches are already applied and settled, so the + // next sweep is clean without touching the hardware again. + upcoming_dirty_ = false; + return; + } + + index_ = (index_ + 1U) % settings_.combos.size(); + const radar::ipc::ComboKey next = settings_.combos[index_]; + const bool position_changed = !(next == applied_); + if (position_changed) { + apply(next); + applied_ = next; + } + upcoming_dirty_ = position_changed && ((backlog_ms + settings_.settle_ms) > settings_.gap_ms); + } + + private: + void apply(const radar::ipc::ComboKey& combo) { + // Output first, then input — same order the sweep orchestrator uses, so + // both acquisition paths drive an identical hardware sequence. + output_switch_->switch_to(combo.output_pos); + input_switch_->switch_to(combo.input_pos); + } + + Settings settings_; + std::unique_ptr input_switch_; + std::unique_ptr output_switch_; + std::size_t index_ = 0; + radar::ipc::ComboKey applied_{}; + bool upcoming_dirty_ = false; +}; + +} // namespace radar::kamil diff --git a/data_acq_and_processing/kamil_adc_collector/src/main.cpp b/data_acq_and_processing/kamil_adc_collector/src/main.cpp index 0eb7cb5..ce954e2 100644 --- a/data_acq_and_processing/kamil_adc_collector/src/main.cpp +++ b/data_acq_and_processing/kamil_adc_collector/src/main.cpp @@ -17,6 +17,11 @@ #include "capture_file_writer.h" #include "tty_protocol_writer.h" +#include "switch_sequencer.h" +#include "run_config.hpp" +#include "h7992_minimal_driver.hpp" +#include "hmc349a_minimal_driver.hpp" + #include #include #include @@ -113,6 +118,12 @@ struct Config { std::string live_html_path = "live_plot.html"; std::string live_json_path = "live_plot.json"; std::optional tty_path; + // When set, the collector reads switch and combo configuration from this + // run_config.json and drives the RF switches itself, one combination per + // sweep, tagging each emitted sweep with its combo (see SwitchSequencer). + std::optional run_config_path; + double switch_gap_ms = 10.0; + double switch_settle_ms = 1.0; bool di1_group_average = false; bool do1_toggle_per_frame = false; bool do1_noise_subtract = false; @@ -479,6 +490,7 @@ void print_help(const char* exe_name) { << " [di1:zero|trace|ignore]\n" << " [duration_ms:100] [packet_limit:0] [csv:capture.csv] [svg:capture.svg]\n" << " [live_html:live_plot.html] [live_json:live_plot.json] [tty:/tmp/ttyADC_data] [di1_group_avg]\n" + << " [config:run_config.json] [switch_gap_ms:10] [switch_settle_ms:1]\n" << " [do1_toggle_per_frame] [do1_noise_subtract] [do1_raw_tty_marked] [do1_pair_subtract_avg] [noise_avg_steps:N]\n" << " [do8_freq_ref] [do8_cycle_period:10] [do8_threshold:X]\n" << " [recv_block:32768] [stats_period_ms:1000] [live_update_period_ms:1000] [svg_history_packets:50] [start_wait_ms:10000]\n" @@ -514,6 +526,13 @@ void print_help(const char* exe_name) { << " step_words:32768 -> input stream transfer step in 32-bit words\n" << " live_html/live_json -> live graph files updated as packets arrive outside tty fast stream-only modes\n" << " tty:/tmp/ttyADC_data -> write a continuous legacy 4-word CH1/CH2 stream; with channels:1, CH2 is 0\n" + << " config:run_config.json -> switch-aware mode: read the input/output switch and combo\n" + << " configuration from this run_config.json and drive the RF switches\n" + << " in lock-step with sweeps; each sweep is tagged with its combo via a\n" + << " 0x00C0 frame (input_pos, output_pos, dirty). Omit for standalone mode.\n" + << " switch_gap_ms:10 -> guaranteed minimum inter-sweep idle window; lower bound used to decide\n" + << " whether an in-gap switch lands before the next sweep starts\n" + << " switch_settle_ms:1 -> RF settle time required after a switch position change\n" << " di1_group_avg -> with tty + di1:trace, emit one averaged 4-word step per constant DI1 run\n" << " do1_toggle_per_frame -> hardware cyclic DO1 pattern in module memory:\n" << " DO1 outputs 00110011... continuously (toggle every 2 ADC ticks)\n" @@ -793,6 +812,18 @@ Config parse_args(int argc, char** argv) { cfg.tty_path = arg.substr(4); continue; } + if (starts_with(arg, "config:")) { + cfg.run_config_path = arg.substr(7); + continue; + } + if (starts_with(arg, "switch_gap_ms:")) { + cfg.switch_gap_ms = parse_double(arg.substr(14), "switch_gap_ms"); + continue; + } + if (starts_with(arg, "switch_settle_ms:")) { + cfg.switch_settle_ms = parse_double(arg.substr(17), "switch_settle_ms"); + continue; + } fail("Unknown argument: " + arg); } @@ -1203,6 +1234,10 @@ constexpr uint32_t kDo1TogglePeriodTicks = 2U; constexpr uint32_t kDo1CyclePatternWords = kDo1TogglePeriodTicks * 2U; constexpr uint32_t kDo8HighTicks = 2U; constexpr uint16_t kTtyMarkerDi8High = 0x00A8U; +// Combo tag frame: [0x00C0, input_pos, output_pos, dirty]. Emitted right after a +// sweep-boundary frame to tell the consumer which switch combination the upcoming +// sweep belongs to, and whether it straddled a switch transition (dirty != 0). +constexpr uint16_t kTtyMarkerComboTag = 0x00C0U; constexpr uint32_t kStreamInputAdcFlag = 0x80000000U; constexpr uint32_t kStreamInputCalibratedAdcFlag = 0x40000000U; @@ -1802,6 +1837,55 @@ void print_device_info(const t_x502_info& info) { << "MCU firmware: " << info.mcu_firmware_ver << "\n"; } +// Build one switch driver (native GPIO or mock) from a parsed switch config, +// reusing the exact driver implementations the sweep orchestrator uses so both +// acquisition paths drive identical hardware. +std::unique_ptr make_switch_driver(const radar::config::SwitchConfig& sc) { + if (sc.driver_kind == radar::config::SwitchDriverKind::HMC349A) { + return std::make_unique( + radar::drivers::HMC349AMinimalDriverSettings{ + .name = sc.name, + .mode = sc.driver_mode, + .positions = sc.positions, + .default_position = sc.default_position, + .gpio_chip = sc.gpio_chip, + .pin_a = sc.pin_a, + .invert_logic = sc.invert_logic, + }); + } + return std::make_unique( + radar::drivers::H7992MinimalDriverSettings{ + .name = sc.name, + .mode = sc.driver_mode, + .positions = sc.positions, + .default_position = sc.default_position, + .gpio_chip = sc.gpio_chip, + .pin_a = sc.pin_a, + .pin_b = sc.pin_b, + }); +} + +// Construct the switch sequencer from the run_config given via `config:`. +// Returns nullptr (switching disabled, single-channel passthrough) when no config +// was supplied or it declares no combinations. +std::unique_ptr make_switch_sequencer(const Config& cfg) { + if (!cfg.run_config_path.has_value()) { + return nullptr; + } + const radar::config::RunConfig rc = radar::config::load_run_config(*cfg.run_config_path); + if (rc.run_combos.empty()) { + return nullptr; + } + radar::kamil::SwitchSequencer::Settings settings; + settings.combos = rc.run_combos; + settings.gap_ms = cfg.switch_gap_ms; + settings.settle_ms = cfg.switch_settle_ms; + return std::make_unique( + std::move(settings), + make_switch_driver(rc.input_switch), + make_switch_driver(rc.output_switch)); +} + int run(const Config& cfg) { Api api; DeviceHandle device(api); @@ -2029,6 +2113,21 @@ int run(const Config& cfg) { tty_writer->emit_packet_start(tty_packet_start_marker); } } + // Switch sequencer (optional): drives the RF switches in lock-step with sweep + // boundaries and tags each sweep with its combination. Disabled (nullptr) when + // no run_config was passed, so the standalone/calibration use stays unchanged. + std::unique_ptr sequencer = make_switch_sequencer(cfg); + if (sequencer && sequencer->enabled()) { + sequencer->open(); + std::cout << "Switch sequencer enabled: cycling switch combinations per sweep " + << "(gap_ms=" << cfg.switch_gap_ms << ", settle_ms=" << cfg.switch_settle_ms << ")\n"; + } else { + sequencer.reset(); // normalize to nullptr so call sites can test the pointer + } + // Estimated readout backlog (ms) at the current loop iteration; how far behind + // real time we are, used to decide whether an in-gap switch lands in time. + double current_backlog_ms = 0.0; + std::unique_ptr writer; if (!fast_tty_avg_stream_mode) { writer = std::make_unique(cfg.csv_path, cfg.svg_path, cfg.live_html_path, cfg.live_json_path); @@ -2317,6 +2416,15 @@ int run(const Config& cfg) { auto append_tty_packet_start = [&]() { append_tty_frame(tty_packet_start_marker, 0xFFFF, 0xFFFF, 0xFFFF); + // Right after the boundary, tag the upcoming sweep with its switch combo so + // the consumer can group sweeps and drop the dirty ones. + if (sequencer) { + const radar::ipc::ComboKey combo = sequencer->current_combo(); + append_tty_frame(kTtyMarkerComboTag, + static_cast(combo.input_pos), + static_cast(combo.output_pos), + static_cast(sequencer->current_dirty() ? 1U : 0U)); + } }; auto append_tty_group_step = [&]() { @@ -2714,6 +2822,12 @@ int run(const Config& cfg) { std::cout << "\n"; } + // A real sweep just completed: step the switch sequence into the gap. A + // user-stop teardown is not a sweep boundary, so skip it. + if (sequencer && (frames != 0U) && (reason != PacketCloseReason::UserStop)) { + sequencer->on_sweep_end(current_backlog_ms); + } + packet_active = false; packet_avg_steps = 0; fast_packet_frames = 0; @@ -2768,6 +2882,11 @@ int run(const Config& cfg) { recv_request_words = std::min(ready_words, read_capacity_words); recv_timeout_ms = 0; } + // Backlog already queued in the driver = how far behind real time we are. + // Drives the switch sequencer's in-gap clean/dirty decision. + if ((ready_err == X502_ERR_OK) && (combined_input_rate_hz > 0.0)) { + current_backlog_ms = 1000.0 * static_cast(ready_words) / combined_input_rate_hz; + } const int32_t recvd = api.Recv(device.hnd, raw.data(), recv_request_words, recv_timeout_ms); if (recvd < 0) { @@ -3170,6 +3289,11 @@ int run(const Config& cfg) { expect_ok(api, api.StreamsStop(device.hnd), "Stop streams"); device.streams_started = false; + + // Park the switches in their safe default and release the GPIO lines. + if (sequencer) { + sequencer->close(); + } if (cfg.do1_toggle_per_frame) { const uint32_t clear_mask = kE502Do1Mask | kE502Do2Mask | (cfg.do8_freq_ref ? kE502Do8Mask : 0U); expect_ok(api, diff --git a/python_app/hardware_full/kamil_adc/protocol.py b/python_app/hardware_full/kamil_adc/protocol.py index 7415534..c6db4bb 100644 --- a/python_app/hardware_full/kamil_adc/protocol.py +++ b/python_app/hardware_full/kamil_adc/protocol.py @@ -29,6 +29,12 @@ import numpy as np FRAME_BYTES = 8 MAIN_MARKER = 0x000A REFERENCE_MARKER = 0x00A8 +# Combo tag — ``0x00C0, input_pos, output_pos, dirty``. Emitted by the switch-aware +# collector right after a sweep boundary to label the upcoming sweep with the RF +# switch combination it was captured under (``dirty != 0`` means the sweep straddled +# a switch transition and must be dropped). Absent in the standalone/calibration +# collector, where every sweep carries no combo (``combo is None``). +COMBO_MARKER = 0x00C0 _BOUNDARY_STEP = 0xFFFF # marker (u16), step (u16), ch1 (i16), ch2 (i16) — point frames carry signed I/Q. @@ -52,6 +58,8 @@ class RawSweep: steps: np.ndarray main: np.ndarray reference: np.ndarray + combo: tuple[int, int] | None = None + dirty: bool = False @property def size(self) -> int: @@ -66,13 +74,17 @@ class KamilAdcStreamParser: but holds no I/O and is cheap to unit-test. """ - __slots__ = ("_buffer", "_aligned", "_main", "_reference") + __slots__ = ("_buffer", "_aligned", "_main", "_reference", "_pending_combo", "_pending_dirty") def __init__(self) -> None: self._buffer = bytearray() self._aligned = False self._main: dict[int, complex] = {} self._reference: dict[int, complex] = {} + # Combo tag for the sweep currently being accumulated (set by the combo + # frame right after each boundary; ``None`` in non-switch collector modes). + self._pending_combo: tuple[int, int] | None = None + self._pending_dirty = False def feed(self, data: bytes) -> list[RawSweep]: """Append ``data`` and return any sweeps completed by it.""" @@ -95,6 +107,10 @@ class KamilAdcStreamParser: self._main[step] = complex(real, imag) elif marker == REFERENCE_MARKER: self._reference[step] = complex(real, imag) + elif marker == COMBO_MARKER: + # step = input_pos, real = output_pos, imag = dirty flag. + self._pending_combo = (int(step), int(real)) + self._pending_dirty = imag != 0 else: raise ValueError( f"Kamil ADC protocol violation: unexpected frame marker 0x{marker:04x}" @@ -107,6 +123,8 @@ class KamilAdcStreamParser: self._aligned = False self._main.clear() self._reference.clear() + self._pending_combo = None + self._pending_dirty = False def _align(self) -> bool: """Discard pre-roll up to and including the first sweep boundary. @@ -122,6 +140,8 @@ class KamilAdcStreamParser: del self._buffer[: index + FRAME_BYTES] self._main.clear() self._reference.clear() + self._pending_combo = None + self._pending_dirty = False self._aligned = True return True @@ -130,12 +150,21 @@ class KamilAdcStreamParser: shared = sorted(self._main.keys() & self._reference.keys()) main = self._main reference = self._reference + combo = self._pending_combo + dirty = self._pending_dirty self._main = {} self._reference = {} + # The next sweep's combo is set by its own combo frame (right after this + # boundary); clear so a sweep without one reports combo=None rather than + # inheriting a stale tag. + self._pending_combo = None + self._pending_dirty = False if not shared: return None return RawSweep( steps=np.asarray(shared, dtype=np.int32), main=np.asarray([main[step] for step in shared], dtype=np.complex64), reference=np.asarray([reference[step] for step in shared], dtype=np.complex64), + combo=combo, + dirty=dirty, ) diff --git a/python_app/hardware_full/kamil_adc/service.py b/python_app/hardware_full/kamil_adc/service.py index 01804a8..6d1ffd8 100644 --- a/python_app/hardware_full/kamil_adc/service.py +++ b/python_app/hardware_full/kamil_adc/service.py @@ -47,6 +47,10 @@ _REJECT_LOG_EVERY = 50 # brief window to release the device cleanly before escalating to SIGKILL. Caps # the configured stop_timeout_s so a stop can never hang. _STOP_KILL_GRACE_S = 0.5 +# Sweeps to skip after a Python-driven switch change before trusting a capture: one +# for the pre-switch sweep still in the mailbox, one for a possible transition +# straddler in flight. Calibration speed is not critical, so we err on safety. +_SWITCH_DRAIN_SWEEPS = 2 @dataclass(slots=True) @@ -54,6 +58,10 @@ class KamilAdcService: """Launch the external Kamil ADC collector and serve its processed sweeps.""" config: RunConfigModel + # When set, the collector is launched with ``config:`` so it drives the RF + # switches itself (switch-aware mode) from this run_config.json. ``None`` keeps + # the standalone collector that streams a single channel (calibration / mock). + switch_config_path: str | None = None _process: subprocess.Popen[bytes] | None = field(init=False, default=None, repr=False) _reader: KamilAdcTtyReader | None = field(init=False, default=None, repr=False) _processor: KamilAdcSweepProcessor | None = field(init=False, default=None, repr=False) @@ -64,9 +72,18 @@ class KamilAdcService: @property def command(self) -> list[str]: - """External collector command, including the generated ``tty:`` argument.""" + """External collector command, including the generated ``tty:`` argument. + + In switch-aware mode (``switch_config_path`` set) the collector also gets + ``config:`` so it reads the switch/combo configuration and drives the + switches in lock-step with the sweeps. + """ adc = self.config.radar.kamil_adc - return [str(self._resolve_executable()), *adc.args, f"tty:{adc.tty_path}"] + cmd = [str(self._resolve_executable()), *adc.args] + if self.switch_config_path is not None: + cmd.append(f"config:{self.switch_config_path}") + cmd.append(f"tty:{adc.tty_path}") + return cmd def open(self, *, stop_event: threading.Event | None = None) -> None: """Launch the collector and start the TTY reader thread. @@ -122,12 +139,17 @@ class KamilAdcService: """Kamil ADC has no runtime-readable sweep-limit API.""" raise RuntimeError("Kamil ADC device limits are not available") - def acquire(self) -> SweepResult: + def acquire(self, combo: tuple[int, int] | None = None) -> SweepResult: """Return the next sweep that covers the band, as S21 on the fixed grid. Sweeps whose floated frequency range does not span the configured band are rejected and the next sweep is read, until one passes or the sweep timeout elapses (which then surfaces as a :class:`TimeoutError`). + + When ``combo`` is given (switch-aware mode), only the clean sweep captured + under that switch combination is returned; the collector drives the switches + and tags each sweep. When ``None`` (calibration / non-switch mode), the + single newest sweep is returned regardless of combination. """ if self._processor is None: raise RuntimeError("Kamil ADC service is not configured") @@ -147,7 +169,10 @@ class KamilAdcService: raise TimeoutError( "Timed out waiting for a Kamil ADC sweep covering the configured band" ) - raw = self._reader.read_sweep(timeout_s=remaining_s, process=process) + if combo is None: + raw = self._reader.read_sweep(timeout_s=remaining_s, process=process) + else: + raw = self._reader.read_sweep_for(combo, timeout_s=remaining_s, process=process) s21 = self._processor.process(raw.main, raw.reference) if s21 is not None: return SweepResult( @@ -159,6 +184,31 @@ class KamilAdcService: ) self._log_rejected_sweep(raw) + def drain_after_switch(self, sweeps: int = _SWITCH_DRAIN_SWEEPS) -> None: + """Discard sweeps captured before / across a just-applied switch change. + + The collector free-runs, so right after the RF switches move the reader + still holds a sweep captured in the *previous* combination, and a sweep that + straddles the transition may still be in flight. Without this, the next + :meth:`acquire` would return that stale data and the capture would be + attributed to the wrong combination (an off-by-one across the sequence). + + Block until ``sweeps`` freshly-published sweeps have gone by, so the next + :meth:`acquire` returns a sweep captured entirely in the new switch state. + Used by the Python-driven calibration capture, where the collector does not + tag sweeps; the switch-aware collector path handles this with combo tags + instead. No-op when the service is not open. + """ + if self._reader is None: + return + target = self._reader.published_count + max(1, int(sweeps)) + deadline = time.monotonic() + self.config.radar.kamil_adc.sweep_timeout_s + while self._reader.published_count < target: + if time.monotonic() > deadline: + raise TimeoutError("Timed out draining Kamil ADC sweeps after a switch change") + raise_if_process_exited(self._process) + time.sleep(0.005) + def read_raw_sweep(self) -> RawSweep: """Return the next raw (main, reference) sweep without any processing. diff --git a/python_app/hardware_full/kamil_adc/tty_reader.py b/python_app/hardware_full/kamil_adc/tty_reader.py index 9b617da..b88ba9c 100644 --- a/python_app/hardware_full/kamil_adc/tty_reader.py +++ b/python_app/hardware_full/kamil_adc/tty_reader.py @@ -52,6 +52,10 @@ class KamilAdcTtyReader: _stop_event: threading.Event = field(init=False, default_factory=threading.Event, repr=False) _mailbox_cv: threading.Condition = field(init=False, default_factory=threading.Condition, repr=False) _latest_sweep: RawSweep | None = field(init=False, default=None, repr=False) + # Latest clean sweep per switch combination, for the switch-aware collector. + # read_sweep() ignores this and serves the single newest sweep (calibration / + # non-switch mode); read_sweep_for() serves a specific combination. + _combo_slots: dict[tuple[int, int], RawSweep] = field(init=False, default_factory=dict, repr=False) _reader_error: Exception | None = field(init=False, default=None, repr=False) _published_count: int = field(init=False, default=0, repr=False) @@ -62,6 +66,7 @@ class KamilAdcTtyReader: self._fd = os.open(self.tty_path, os.O_RDONLY | os.O_NOCTTY | os.O_NONBLOCK) self._stop_event.clear() self._latest_sweep = None + self._combo_slots = {} self._reader_error = None self._published_count = 0 self._thread = threading.Thread( @@ -89,6 +94,7 @@ class KamilAdcTtyReader: finally: self._fd = None self._latest_sweep = None + self._combo_slots = {} self._reader_error = None @property @@ -131,6 +137,39 @@ class KamilAdcTtyReader: ) self._mailbox_cv.wait(timeout=min(_READ_POLL_INTERVAL_S, remaining_s)) + def read_sweep_for( + self, + combo: tuple[int, int], + *, + timeout_s: float, + process: subprocess.Popen[bytes] | None = None, + ) -> RawSweep: + """Wait for and return the latest clean sweep for ``combo``. + + Used in switch-aware mode, where the collector drives the switches and tags + each sweep with its combination. Only clean sweeps are delivered (the reader + thread drops the dirty ones); the slot is consumed on read so each caller + gets a fresh capture. Raises like :meth:`read_sweep`. + """ + if self._thread is None: + raise RuntimeError("Kamil ADC TTY reader is not open") + deadline = time.monotonic() + float(timeout_s) + with self._mailbox_cv: + while True: + sweep = self._combo_slots.pop(combo, None) + if sweep is not None: + return sweep + if self._reader_error is not None: + raise self._reader_error + raise_if_process_exited(process) + remaining_s = deadline - time.monotonic() + if remaining_s <= 0.0: + raise TimeoutError( + f"Timed out waiting for Kamil ADC sweep for combo {combo} " + f"after {float(timeout_s):.3f}s" + ) + self._mailbox_cv.wait(timeout=min(_READ_POLL_INTERVAL_S, remaining_s)) + # ------------------------------------------------------------------ # Reader-thread internals # ------------------------------------------------------------------ @@ -177,11 +216,22 @@ class KamilAdcTtyReader: return chunk def _publish_sweep(self, sweep: RawSweep) -> None: - """Store ``sweep`` as the latest mailbox value, overwriting any unread one.""" + """Publish a completed sweep to the mailbox(es), waking any waiter. + + Dirty sweeps (those that straddled a switch transition) are counted but not + delivered: the collector re-takes that combination on the next sweep. Clean + tagged sweeps go to their per-combo slot; untagged sweeps (non-switch mode) + only update the single newest-sweep mailbox that read_sweep() serves. + """ with self._mailbox_cv: - self._latest_sweep = sweep self._published_count += 1 - self._mailbox_cv.notify() + if sweep.dirty: + self._mailbox_cv.notify_all() + return + self._latest_sweep = sweep + if sweep.combo is not None: + self._combo_slots[sweep.combo] = sweep + self._mailbox_cv.notify_all() def _publish_error(self, exc: Exception) -> None: """Record ``exc`` as the reader fault and wake any waiter.""" diff --git a/python_app/scripts/kamil_adc_raw_producer.py b/python_app/scripts/kamil_adc_raw_producer.py index 318dfc0..c4280f9 100644 --- a/python_app/scripts/kamil_adc_raw_producer.py +++ b/python_app/scripts/kamil_adc_raw_producer.py @@ -36,8 +36,8 @@ _OPEN_RETRY_LOG_EVERY = 30 def _open_radar_with_retry( config: RunConfigModel, radar: KamilAdcService, - input_switch: SwitchService, - output_switch: SwitchService, + input_switch: SwitchService | None, + output_switch: SwitchService | None, stop_requested: threading.Event, ) -> bool: """Open+configure the radar and both switches, retrying forever until stop. @@ -48,14 +48,19 @@ def _open_radar_with_retry( relaunched collector starts clean. Returns ``True`` once everything is open, or ``False`` if a stop was requested before the device became available. Backoff is capped and every wait is interruptible by SIGTERM. + + ``input_switch``/``output_switch`` are ``None`` in switch-aware mode, where the + collector owns the GPIO lines and the producer must not open them. """ # Tear down any prior open first: open()/switch.open() are idempotent no-ops # while still "open", so a mid-run reconnect must close them to force a fresh # collector relaunch and TTY re-attach. - with suppress(Exception): - input_switch.close() - with suppress(Exception): - output_switch.close() + if input_switch is not None: + with suppress(Exception): + input_switch.close() + if output_switch is not None: + with suppress(Exception): + output_switch.close() with suppress(Exception): radar.close() @@ -65,15 +70,19 @@ def _open_radar_with_retry( try: radar.open(stop_event=stop_requested) radar.configure(config.radar.sweep) - output_switch.open() - input_switch.open() + if output_switch is not None: + output_switch.open() + if input_switch is not None: + input_switch.open() except Exception as exc: # noqa: BLE001 — waiting for the device is the point # Drop any partial open (collector process, TTY reader, switches) # before the next attempt so the relaunch starts from a clean state. - with suppress(Exception): - input_switch.close() - with suppress(Exception): - output_switch.close() + if input_switch is not None: + with suppress(Exception): + input_switch.close() + if output_switch is not None: + with suppress(Exception): + output_switch.close() with suppress(Exception): radar.close() attempt += 1 @@ -136,9 +145,26 @@ def main() -> int: "Opened SHM ring writers: raw=%s, raw_tap=%s", config.rings.raw.name, config.rings.raw_tap.name, ) - radar = KamilAdcService(config) - input_switch = SwitchService.from_model(config.input_switch) - output_switch = SwitchService.from_model(config.output_switch) + # Switch-aware mode: with native switches the collector drives the RF switches + # itself, in the hardware gap between sweeps, and tags each sweep with its combo + # — no sweep lost at a switch boundary. The producer then only reads tagged + # sweeps and must not touch the GPIO lines the collector owns. With mock switches + # (dev/tests) we keep the Python-driven path, which is fine where speed and the + # in-gap timing do not matter. + collector_driven = ( + config.input_switch.driver_mode == "native" + and config.output_switch.driver_mode == "native" + ) + radar = KamilAdcService( + config, + switch_config_path=str(args.config) if collector_driven else None, + ) + input_switch = None if collector_driven else SwitchService.from_model(config.input_switch) + output_switch = None if collector_driven else SwitchService.from_model(config.output_switch) + logger.info( + "Kamil ADC switch control: %s", + "collector-driven (in-gap, lossless)" if collector_driven else "producer-driven", + ) try: if not _open_radar_with_retry(config, radar, input_switch, output_switch, stop_requested): @@ -155,12 +181,16 @@ def main() -> int: for combo in config.combos: if stop_requested.is_set(): break - output_switch.switch_to(combo.output) - input_switch.switch_to(combo.input) - if config.runtime.settling_ms > 0: - time.sleep(config.runtime.settling_ms / 1000.0) - - sweep = radar.acquire() + if collector_driven: + # The collector already switched and tagged the sweep; just + # read the clean capture for this combination. + sweep = radar.acquire(combo=(combo.input, combo.output)) + else: + output_switch.switch_to(combo.output) + input_switch.switch_to(combo.input) + if config.runtime.settling_ms > 0: + time.sleep(config.runtime.settling_ms / 1000.0) + sweep = radar.acquire() traces.append( TraceData( combo=ComboKey(input=combo.input, output=combo.output), @@ -222,10 +252,12 @@ def main() -> int: logger.info("Kamil ADC collection %d acquired in %.3f s", collection_id, collection_duration_s) collection_id += 1 finally: - with suppress(Exception): - output_switch.close() - with suppress(Exception): - input_switch.close() + if output_switch is not None: + with suppress(Exception): + output_switch.close() + if input_switch is not None: + with suppress(Exception): + input_switch.close() with suppress(Exception): radar.close() raw_tap_writer.close() diff --git a/python_app/tests/test_kamil_adc_protocol.py b/python_app/tests/test_kamil_adc_protocol.py index 90e0f8b..cdd5eca 100644 --- a/python_app/tests/test_kamil_adc_protocol.py +++ b/python_app/tests/test_kamil_adc_protocol.py @@ -8,6 +8,7 @@ import unittest import numpy as np from python_app.hardware_full.kamil_adc.protocol import ( + COMBO_MARKER, MAIN_MARKER, REFERENCE_MARKER, KamilAdcStreamParser, @@ -18,6 +19,10 @@ def _boundary() -> bytes: return struct.pack(" bytes: + return struct.pack(" bytes: return struct.pack(" None: + parser = KamilAdcStreamParser() + (sweep,) = parser.feed(_boundary() + _main(1, 1, 0) + _reference(1, 9, 0) + _boundary()) + self.assertIsNone(sweep.combo) + self.assertFalse(sweep.dirty) + + def test_combo_tag_labels_following_sweep(self) -> None: + parser = KamilAdcStreamParser() + stream = ( + _boundary() + _combo(1, 2) + + _main(1, 10, 0) + _reference(1, 100, 0) + + _boundary() + _combo(3, 0, dirty=1) + + _main(1, 20, 0) + _reference(1, 200, 0) + + _boundary() + ) + first, second = parser.feed(stream) + self.assertEqual(first.combo, (1, 2)) + self.assertFalse(first.dirty) + self.assertEqual(second.combo, (3, 0)) + self.assertTrue(second.dirty) + + def test_combo_not_carried_into_untagged_sweep(self) -> None: + parser = KamilAdcStreamParser() + stream = ( + _boundary() + _combo(1, 1) + + _main(1, 1, 0) + _reference(1, 1, 0) + + _boundary() # next sweep has no combo frame + + _main(2, 2, 0) + _reference(2, 2, 0) + + _boundary() + ) + first, second = parser.feed(stream) + self.assertEqual(first.combo, (1, 1)) + self.assertIsNone(second.combo) + def test_dtypes(self) -> None: parser = KamilAdcStreamParser() (sweep,) = parser.feed(_boundary() + _main(1, 1, 2) + _reference(1, 3, 4) + _boundary()) diff --git a/python_app/tests/test_kamil_adc_service.py b/python_app/tests/test_kamil_adc_service.py index 21666c9..2a17899 100644 --- a/python_app/tests/test_kamil_adc_service.py +++ b/python_app/tests/test_kamil_adc_service.py @@ -16,7 +16,11 @@ import unittest from unittest import mock from python_app.hardware_full.kamil_adc import KamilAdcService, KamilAdcTtyReader -from python_app.hardware_full.kamil_adc.protocol import MAIN_MARKER, REFERENCE_MARKER +from python_app.hardware_full.kamil_adc.protocol import ( + COMBO_MARKER, + MAIN_MARKER, + REFERENCE_MARKER, +) from python_app.models.run_config_model import RunConfigModel from python_app.orchestration.process_supervisor import ProcessSupervisor @@ -33,6 +37,10 @@ def _reference(step: int, real: int, imag: int) -> bytes: return struct.pack(" bytes: + return struct.pack(" None: + master_fd, slave_fd, reader = self._open_pty_reader() + try: + os.write( + master_fd, + _boundary() + _combo(0, 0) + _main(1, 11, 0) + _reference(1, 1, 0) + + _boundary() + _combo(0, 1) + _main(1, 22, 0) + _reference(1, 1, 0) + + _boundary(), + ) + # Each combination is served from its own slot, regardless of order. + second = reader.read_sweep_for((0, 1), timeout_s=1.0) + self.assertEqual(second.main.real.tolist(), [22]) + self.assertEqual(second.combo, (0, 1)) + first = reader.read_sweep_for((0, 0), timeout_s=1.0) + self.assertEqual(first.main.real.tolist(), [11]) + finally: + self._close(master_fd, slave_fd, reader) + + def test_read_sweep_for_drops_dirty_and_takes_retake(self) -> None: + master_fd, slave_fd, reader = self._open_pty_reader() + try: + os.write( + master_fd, + # A dirty combo (0,1) sweep, then its clean re-take of the same combo. + _boundary() + _combo(0, 1, dirty=1) + _main(1, 99, 0) + _reference(1, 1, 0) + + _boundary() + _combo(0, 1) + _main(1, 42, 0) + _reference(1, 1, 0) + + _boundary(), + ) + sweep = reader.read_sweep_for((0, 1), timeout_s=1.0) + # The dirty sweep (99) is dropped; only the clean re-take (42) is served. + self.assertEqual(sweep.main.real.tolist(), [42]) + self.assertFalse(sweep.dirty) + finally: + self._close(master_fd, slave_fd, reader) + class KamilAdcConfigTest(unittest.TestCase): def test_config_round_trip_preserves_kamil_sections(self) -> None: @@ -205,6 +248,61 @@ class KamilAdcConfigTest(unittest.TestCase): with mock.patch("python_app.hardware_full.kamil_adc.service.os.killpg"): service.close() # must not raise + def test_drain_after_switch_waits_for_fresh_sweeps(self) -> None: + """After a switch change, drain must skip the configured number of freshly + published sweeps before returning, so the next capture is post-switch.""" + import types + + from python_app.hardware_full.kamil_adc import service as service_module + + with tempfile.TemporaryDirectory() as tmp_dir: + config = RunConfigModel.from_dict( + { + "radar": { + "model": "kamil_adc", + "driver_mode": "native", + "kamil_adc": { + "project_dir": tmp_dir, + "executable_path": "/bin/sh", + "tty_path": "/tmp/ttyADC_test", + "sweep_timeout_s": 5.0, + }, + }, + "switches": {"port1": {"positions": 1}, "port2": {"positions": 1}}, + } + ) + service = KamilAdcService(config) + service._reader = types.SimpleNamespace(published_count=10) # type: ignore[assignment] + service._process = types.SimpleNamespace(poll=lambda: None) # type: ignore[assignment] + + # Each poll-sleep advances the published count, as the reader thread would. + def _advance(_seconds: float) -> None: + service._reader.published_count += 1 + + with mock.patch.object(service_module.time, "sleep", _advance): + service.drain_after_switch(sweeps=3) + + # Started at 10, must have waited for at least 3 more sweeps. + self.assertGreaterEqual(service._reader.published_count, 13) + + def test_drain_after_switch_is_noop_when_not_open(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + config = RunConfigModel.from_dict( + { + "radar": { + "model": "kamil_adc", + "driver_mode": "native", + "kamil_adc": { + "project_dir": tmp_dir, + "executable_path": "/bin/sh", + "tty_path": "/tmp/ttyADC_test", + }, + }, + "switches": {"port1": {"positions": 1}, "port2": {"positions": 1}}, + } + ) + KamilAdcService(config).drain_after_switch() # no reader → must not raise + def test_supervisor_selects_kamil_adc_producer(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: config_path = Path(tmp_dir) / "run_config.json" diff --git a/python_app/workflows/sequential_capture_workflow.py b/python_app/workflows/sequential_capture_workflow.py index 1a1beb5..215f312 100644 --- a/python_app/workflows/sequential_capture_workflow.py +++ b/python_app/workflows/sequential_capture_workflow.py @@ -187,6 +187,16 @@ class SequentialCaptureSession: if self._config.runtime.settling_ms > 0: time.sleep(self._config.runtime.settling_ms / 1000.0) + # A free-running streaming radar (Kamil ADC) keeps a sweep captured in the + # previous combination buffered, and may have a transition-straddling sweep + # in flight. Drop those so this capture holds data from the new switch state + # — otherwise the trace is labelled with this combo but carries the previous + # one's data (an off-by-one across the sequence). Discrete radars (LibreVNA) + # acquire a fresh sweep per call and expose no such method, so skip them. + drain_after_switch = getattr(self._radar, "drain_after_switch", None) + if callable(drain_after_switch): + drain_after_switch() + sweep_traces: list[TraceData] = [] for _ in range(self._median_sweep_count): sweep = self._radar.acquire()