added capture time for every sweep

.gitignore

@@ -21,8 +21,8 @@ dist/
 downloads/
 eggs/
 .eggs/
-lib/
+/lib/
-lib64/
+/lib64/
 parts/
 sdist/
 var/
@@ -227,5 +227,16 @@ python_app/runtime
 SHARE_INTERNET_TO_PI.md
 
 CLAUDE.md
-docs/
+/docs/
-test_end_2/
+test_end_2/
+
+# --- device_firmware: PlatformIO / STM32G431 cart remote ---
+# build output (regenerated by `pio run`)
+device_firmware/cart_firmware/.pio/
+# scope captures: raw .bin + .npz + preview .png, local-only
+device_firmware/cart_firmware/captures/
+# machine-specific, regenerated by the PlatformIO extension
+device_firmware/cart_firmware/.vscode/c_cpp_properties.json
+device_firmware/cart_firmware/.vscode/launch.json
+device_firmware/cart_firmware/.vscode/ipch/
+device_firmware/cart_firmware/.vscode/.browse.c_cpp.db*

data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp

@@ -38,6 +38,13 @@ struct SweepTraceBlock {
     std::vector<Complex32> s11{};
     // Complex S21 samples for matching frequency points.
     std::vector<Complex32> s21{};
+    // Monotonic window in which THIS trace's sweep was measured, excluding the
+    // switch drive and settling that preceded it. In a switched matrix the
+    // collection is assembled combo by combo over many milliseconds, so the
+    // collection-level window says nothing about when an individual combo was
+    // measured. Zero on both is a valid "unmeasured" sentinel.
+    std::uint64_t capture_start_ns = 0;
+    std::uint64_t capture_end_ns = 0;
 };
 
 struct RawSweepCollection {

data_acq_and_processing/common_cpp/ipc/src/shared_types.cpp

@@ -172,6 +172,10 @@ void require_count_fits(std::uint32_t count, std::size_t min_bytes_each, BinaryR
     return trace;
 }
 
+// The capture windows live in a TRAILER after the trace blocks rather than inside
+// them, so a reader built before they existed still decodes every trace and simply
+// stops early. The trailer grows the same way: collection window first, then the
+// per-trace window table (one pair per trace, in trace order).
 void write_trace_collection(BinaryWriter& writer, std::uint32_t magic, const RawSweepCollection& collection) {
     writer.write(magic);
     writer.write(collection.collection_id);
@@ -184,6 +188,12 @@ void write_trace_collection(BinaryWriter& writer, std::uint32_t magic, const Raw
 
     writer.write(collection.capture_start_ns);
     writer.write(collection.capture_end_ns);
+
+    writer.write(checked_count_to_u32(collection.traces.size(), "Trace capture window count"));
+    for (const auto& trace : collection.traces) {
+        writer.write(trace.capture_start_ns);
+        writer.write(trace.capture_end_ns);
+    }
 }
 
 [[nodiscard]] auto read_trace_collection(BinaryReader& reader, std::uint32_t expected_magic) -> RawSweepCollection {
@@ -204,16 +214,31 @@ void write_trace_collection(BinaryWriter& writer, std::uint32_t magic, const Raw
         collection.traces.push_back(read_trace_block(reader));
     }
 
+    // Each trailer stage is optional: a payload from an older producer stops after
+    // the trace blocks (or after the collection window) and leaves the rest zeroed.
     if (reader.remaining_bytes() == 0U) {
         return collection;
     }
-    if (reader.remaining_bytes() != (sizeof(std::uint64_t) * 2U)) {
+    if (reader.remaining_bytes() < (sizeof(std::uint64_t) * 2U)) {
-        throw std::runtime_error("Unexpected trailing bytes in trace collection");
+        throw std::runtime_error("Truncated capture window in trace collection");
     }
 
     collection.capture_start_ns = reader.read<std::uint64_t>();
     collection.capture_end_ns = reader.read<std::uint64_t>();
 
+    if (reader.remaining_bytes() == 0U) {
+        return collection;
+    }
+
+    const auto trace_time_count = reader.read<std::uint32_t>();
+    if (trace_time_count != collection.traces.size()) {
+        throw std::runtime_error("Per-trace capture window count does not match trace count");
+    }
+    for (auto& trace : collection.traces) {
+        trace.capture_start_ns = reader.read<std::uint64_t>();
+        trace.capture_end_ns = reader.read<std::uint64_t>();
+    }
+
     return collection;
 }
 

data_acq_and_processing/preprocessing/calibration_master/src/calibration_master.cpp

@@ -34,6 +34,9 @@ auto CalibrationMaster::apply_to_trace(const ipc::SweepTraceBlock& measured_trac
     output.frequency_hz = measured_trace.frequency_hz;
     output.s21 = apply_s21(measured_trace.combo, measured_trace.frequency_hz, measured_trace.s21);
     output.s11 = apply_s11(measured_trace.combo, measured_trace.frequency_hz, measured_trace.s11);
+    // Calibration reshapes the samples, not when they were measured.
+    output.capture_start_ns = measured_trace.capture_start_ns;
+    output.capture_end_ns = measured_trace.capture_end_ns;
     return output;
 }
 

data_acq_and_processing/preprocessing/reference_master/src/reference_master.cpp

@@ -132,6 +132,9 @@ auto ReferenceMaster::apply_to_trace(const ipc::SweepTraceBlock& calibrated_trac
     output.frequency_hz = calibrated_trace.frequency_hz;
     output.s21 = apply_s21(calibrated_trace.combo, calibrated_trace.frequency_hz, calibrated_trace.s21);
     output.s11 = apply_s11(calibrated_trace.combo, calibrated_trace.frequency_hz, calibrated_trace.s11);
+    // Reference subtraction reshapes the samples, not when they were measured.
+    output.capture_start_ns = calibrated_trace.capture_start_ns;
+    output.capture_end_ns = calibrated_trace.capture_end_ns;
     return output;
 }
 

data_acq_and_processing/sweep_orchestrator/src/sweep_orchestrator.cpp

@@ -147,14 +147,18 @@ class DriverLifecycleGuard {
 // Worst-case serialized size of a collection given the configured combo count and sweep
 // point count, using the trace wire format (see ipc::write_trace_collection/write_trace_block):
 //   collection header: magic(4) + collection_id(8) + monotonic_ns(8) + trace_count(4)
-//                     + capture_start_ns(8) + capture_end_ns(8) = 40 bytes
+//                     + capture_start_ns(8) + capture_end_ns(8)
+//                     + trace capture window count(4) = 44 bytes
 //   per trace block:   input_pos(4) + output_pos(4) + point_count(4) = 12 bytes
 //                     + per point: frequency(4) + s11(8) + s21(8) = 20 bytes
+//                     + trailer: capture_start_ns(8) + capture_end_ns(8) = 16 bytes
 [[nodiscard]] auto worst_case_serialized_bytes(std::size_t combo_count, std::uint32_t sweep_points) -> std::size_t {
-    constexpr std::size_t kCollectionHeaderBytes = 40U;
+    constexpr std::size_t kCollectionHeaderBytes = 44U;
     constexpr std::size_t kTraceHeaderBytes = 12U;
     constexpr std::size_t kBytesPerPoint = 20U;
-    const std::size_t per_trace = kTraceHeaderBytes + (static_cast<std::size_t>(sweep_points) * kBytesPerPoint);
+    constexpr std::size_t kTraceTrailerBytes = 16U;
+    const std::size_t per_trace =
+        kTraceHeaderBytes + (static_cast<std::size_t>(sweep_points) * kBytesPerPoint) + kTraceTrailerBytes;
     return kCollectionHeaderBytes + (combo_count * per_trace);
 }
 
@@ -290,7 +294,12 @@ auto SweepOrchestrator::acquire_one_collection(
         // Production drivers ignore this; mock drivers use it to give every
         // (input, output) pair its own synthetic response.
         radar_driver_.set_active_combo(combo);
+        // Stamp around the sweep only: the switch drive and settling above belong to
+        // neither the previous combo nor this one, so excluding them keeps the window
+        // an honest "when was this combo actually measured".
+        const auto sweep_start_ns = ipc::current_monotonic_ns();
         auto sweep = radar_driver_.acquire_sweep();
+        const auto sweep_end_ns = ipc::current_monotonic_ns();
         validate_sweep(sweep);
 
         ipc::SweepTraceBlock trace{};
@@ -298,6 +307,8 @@ auto SweepOrchestrator::acquire_one_collection(
         trace.frequency_hz = std::move(sweep.frequency_hz);
         trace.s11 = std::move(sweep.s11);
         trace.s21 = std::move(sweep.s21);
+        trace.capture_start_ns = sweep_start_ns;
+        trace.capture_end_ns = sweep_end_ns;
         collection.traces.push_back(std::move(trace));
     }
 

device_firmware/cart_firmware/.gitignore

@@ -0,0 +1,5 @@
+.pio
+.vscode/.browse.c_cpp.db*
+.vscode/c_cpp_properties.json
+.vscode/launch.json
+.vscode/ipch

device_firmware/cart_firmware/.vscode/extensions.json

@@ -0,0 +1,10 @@
+{
+    // See http://go.microsoft.com/fwlink/?LinkId=827846
+    // for the documentation about the extensions.json format
+    "recommendations": [
+        "platformio.platformio-ide"
+    ],
+    "unwantedRecommendations": [
+        "ms-vscode.cpptools-extension-pack"
+    ]
+}

device_firmware/cart_firmware/docs/protocol.md

@@ -0,0 +1,57 @@
+# Протокол старого пульта тележки (реверс-инжиниринг)
+
+Снято 2026-08-20 осциллографом Hantek DPO7204C с сигнального провода пульта
+(канал CH3). Инструменты: `tools/capture.py` (захват), `tools/decode.py`
+(декодер), сырые данные и картинки — в `captures/`.
+
+## Физический уровень
+
+- Один сигнальный провод, логика **3.3 В**.
+- **Инвертированный UART** (стандартная полярность SBUS): в покое линия
+  **низкая** (~0 В), импульсы вверх до ~3.3 В.
+- Скорость **100 000 бод**, формат **8E2** (8 бит данных, чётность even,
+  2 стоп-бита), биты LSB-first. Длительность бита 10 мкс.
+
+## Кадровый уровень — SBUS
+
+Стандартный кадр Futaba SBUS, 25 байт (3 мс на линии):
+
+| Смещение | Размер | Содержимое |
+|---|---|---|
+| 0 | 1 | Заголовок `0x0F` |
+| 1 | 22 | 16 каналов × 11 бит, упакованы подряд LSB-first |
+| 23 | 1 | Флаги: bit0=CH17, bit1=CH18, bit2=frame_lost, bit3=failsafe |
+| 24 | 1 | Футер `0x00` |
+
+Распаковка каналов: 22 байта складываются в 176-битное число LSB-first,
+канал N (N=0..15) = биты [11·N .. 11·N+10], диапазон значений 0–2047.
+
+- Кадры отправляются каждые **50 мс** (20 Гц). Это медленнее стандартного
+  SBUS (7/14 мс) — ответная часть с этим темпом работает.
+- Флаги во всех наблюдениях = `0x00`.
+
+## Карта каналов (нумерация с 1)
+
+| Управление | Канал | Мин | Нейтраль | Макс |
+|---|---|---|---|---|
+| Стик вперёд/назад | **2** | 433 (назад) | 1024 | 1643 (вперёд) |
+| Стик влево/вправо | **4** | 446 (влево) | 1024 | 1654 (вправо) |
+| Остальные 14 | — | всегда 1024 | | |
+
+Диапазон осей ~±600 от нейтрали (не полная шкала SBUS). Других органов
+управления на пульте нет.
+
+Эталонный кадр нейтрали (hex):
+
+```
+0F 00 04 20 00 01 08 40 00 02 10 80 00 04 20 00 01 08 40 00 02 10 80 00 00
+```
+
+## Воспроизведение на STM32G431
+
+- USART: 100000 бод, 8 бит + чётность even (в терминах STM32: M=1, 9-bit
+  с PCE=1), 2 стоп-бита, **TXINV=1** (аппаратная инверсия TX) — бит-бэнг
+  не нужен.
+- Отправлять 25-байтовый кадр по таймеру каждые 50 мс.
+- Нейтраль: все каналы 1024; управление — каналы 2 и 4 в измеренных
+  диапазонах.

device_firmware/cart_firmware/platformio.ini

@@ -0,0 +1,10 @@
+[env:weact_g431cb]
+platform = ststm32
+board = genericSTM32G431CB
+framework = arduino
+upload_protocol = stlink
+debug_tool = stlink
+monitor_speed = 115200
+build_flags =
+    -DUSBCON
+    -DUSBD_USE_CDC

device_firmware/cart_firmware/src/main.cpp

@@ -0,0 +1,166 @@
+// Пульт тележки: стик (АЦП PA0/PA1) -> SBUS на USART1 TX (PA9).
+// Протокол: инвертированный SBUS, 100000 бод 8E2, 25 байт каждые 50 мс
+// (см. docs/protocol.md). USB CDC (Serial) — отладочный вывод.
+#include <Arduino.h>
+
+// ---- калибровка стика (АЦП 12 бит, замерено 2026-08-20) ----
+static const int X_FWD = 650, X_MID = 2014, X_BACK = 3378;   // PA0 (плечи равны: 2014-650 = 3378-2014 = 1364)
+static const int Y_RIGHT = 479, Y_MID = 1981, Y_LEFT = 3586; // PA1
+static const int DEADZONE = 15; // отсечка дребезга вокруг нейтрали
+
+// ---- SBUS-значения старого пульта ----
+static const uint16_t SBUS_MID = 1024;
+static const uint16_t CH2_MIN = 433, CH2_MAX = 1643;  // назад..вперёд
+static const uint16_t CH4_MIN = 446, CH4_MAX = 1654;  // влево..вправо
+
+// ---- expo: 0 = линейно, 1 = максимально мягкая нейтраль ----
+static const float EXPO_K = 0.0f;
+// ---- общий масштаб выхода: 1.0 = диапазон старого пульта ----
+static const float RANGE_SCALE = 0.75f;
+
+// симметричные плечи: вперёд и назад дают одинаковый максимум
+static const uint16_t CH2_SPAN = 591; // min(1643-1024, 1024-433)
+static const uint16_t CH4_SPAN = 578; // min(1654-1024, 1024-446)
+
+// ---- профиль газа/поворота ----
+static const float MOVE_START = 0.15f; // старт движения, доля хода стика
+static const float POWER_FWD  = 0.60f; // потолок «вперёд»
+static const float POWER_BACK = 0.50f; // потолок «назад»
+static const float POWER_TURN = 0.70f; // потолок поворота
+static const uint16_t CH2_DB = 221;    // мёртвая зона приёмника тележки (ЗАМЕРИТЬ по монитору)
+static const uint16_t CH4_DB = 221;
+
+static const uint32_t FRAME_PERIOD_MS = 50;
+static const uint32_t PIN_X = PA0;
+static const uint32_t PIN_Y = PA1;
+
+static UART_HandleTypeDef s_sbusUart;
+
+// USART1 TX = PA9 (AF7), 100000 бод, 8E2, TX инвертирован
+static void sbusUartInit() {
+    __HAL_RCC_GPIOA_CLK_ENABLE();
+    __HAL_RCC_USART1_CLK_ENABLE();
+
+    GPIO_InitTypeDef gpio = {};
+    gpio.Pin = GPIO_PIN_9;
+    gpio.Mode = GPIO_MODE_AF_PP;
+    gpio.Pull = GPIO_NOPULL;
+    gpio.Speed = GPIO_SPEED_FREQ_LOW;
+    gpio.Alternate = GPIO_AF7_USART1;
+    HAL_GPIO_Init(GPIOA, &gpio);
+
+    s_sbusUart.Instance = USART1;
+    s_sbusUart.Init.BaudRate = 100000;
+    s_sbusUart.Init.WordLength = UART_WORDLENGTH_9B; // 8 данных + чётность
+    s_sbusUart.Init.StopBits = UART_STOPBITS_2;
+    s_sbusUart.Init.Parity = UART_PARITY_EVEN;
+    s_sbusUart.Init.Mode = UART_MODE_TX;
+    s_sbusUart.Init.HwFlowCtl = UART_HWCONTROL_NONE;
+    s_sbusUart.Init.OverSampling = UART_OVERSAMPLING_16;
+    s_sbusUart.AdvancedInit.AdvFeatureInit = UART_ADVFEATURE_TXINVERT_INIT;
+    s_sbusUart.AdvancedInit.TxPinLevelInvert = UART_ADVFEATURE_TXINV_ENABLE;
+    HAL_UART_Init(&s_sbusUart);
+}
+
+// нормализация одной оси в [-1..1] с мёртвой зоной и асимметричными плечами
+static float axisNorm(int adc, int lowEnd, int mid, int highEnd) {
+    float x;
+    if (adc < mid - DEADZONE) {
+        x = (float)(mid - adc) / (float)(mid - lowEnd); // к lowEnd -> +1
+    } else if (adc > mid + DEADZONE) {
+        x = -(float)(adc - mid) / (float)(highEnd - mid); // к highEnd -> -1
+    } else {
+        return 0.0f;
+    }
+    return constrain(x, -1.0f, 1.0f);
+}
+
+// expo-кривая: гасит чувствительность у нейтрали, сохраняет края
+static float expo(float x) {
+    return EXPO_K * x * x * x + (1.0f - EXPO_K) * x;
+}
+
+static uint16_t toSbus(float x, uint16_t span) {
+    return (uint16_t)(SBUS_MID + lroundf(x * span));
+}
+
+// стик [-1..1] -> SBUS-значение канала: ниже MOVE_START — нейтраль, выше —
+// линейно от порога срабатывания приёмника (db) до POWER_FRAC*span на полном стике
+static uint16_t driveToSbus(float x, uint16_t span, uint16_t db,
+                            float powerPos, float powerNeg) {
+    float mag = fabsf(x);
+    if (mag <= MOVE_START)
+        return SBUS_MID;
+    float outMax = (x > 0.0f ? powerPos : powerNeg) * span;
+    float t = (mag - MOVE_START) / (1.0f - MOVE_START);
+    long delta = lroundf(db + t * (outMax - db));
+    return (x > 0.0f) ? (uint16_t)(SBUS_MID + delta)
+                      : (uint16_t)(SBUS_MID - delta);
+}
+
+static void sbusPack(uint8_t out[25], const uint16_t ch[16]) {
+    out[0] = 0x0F;
+    memset(out + 1, 0, 22);
+    uint32_t bitpos = 0;
+    for (int n = 0; n < 16; n++) {
+        for (int b = 0; b < 11; b++) {
+            if (ch[n] & (1u << b))
+                out[1 + (bitpos >> 3)] |= 1u << (bitpos & 7);
+            bitpos++;
+        }
+    }
+    out[23] = 0x00; // флаги
+    out[24] = 0x00; // футер
+}
+
+static int readAvg(uint32_t pin) {
+    uint32_t acc = 0;
+    for (int i = 0; i < 16; i++)
+        acc += analogRead(pin);
+    return acc / 16;
+}
+
+void setup() {
+    Serial.begin(115200);
+    analogReadResolution(12);
+    pinMode(PIN_X, INPUT_ANALOG);
+    pinMode(PIN_Y, INPUT_ANALOG);
+    sbusUartInit();
+}
+
+void loop() {
+    static uint32_t next = 0;
+    uint32_t now = millis();
+    if (now < next)
+        return;
+    next = now + FRAME_PERIOD_MS;
+
+    int adcX = readAvg(PIN_X);
+    int adcY = readAvg(PIN_Y);
+    // вперёд = adcX к X_FWD (вниз) -> +1; вправо = adcY к Y_RIGHT -> +1
+    float fwd = axisNorm(adcX, X_FWD, X_MID, X_BACK);
+    float right = axisNorm(adcY, Y_RIGHT, Y_MID, Y_LEFT);
+
+    uint16_t ch[16];
+    for (int i = 0; i < 16; i++)
+        ch[i] = SBUS_MID;
+    ch[1] = driveToSbus(fwd, CH2_SPAN, CH2_DB, POWER_FWD, POWER_BACK);  // канал 2 — газ
+    ch[3] = driveToSbus(right, CH4_SPAN, CH4_DB, POWER_TURN, POWER_TURN); // канал 4 — поворот
+
+    uint8_t frame[25];
+    sbusPack(frame, ch);
+    HAL_UART_Transmit(&s_sbusUart, frame, sizeof(frame), 20);
+
+    Serial.print("adc=");
+    Serial.print(adcX);
+    Serial.print(",");
+    Serial.print(adcY);
+    Serial.print(" fwd=");
+    Serial.print(fwd, 3);
+    Serial.print(" right=");
+    Serial.print(right, 3);
+    Serial.print(" ch2=");
+    Serial.print(ch[1]);
+    Serial.print(" ch4=");
+    Serial.println(ch[3]);
+}

device_firmware/cart_firmware/tools/analyze.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""Analyze a captured frame: extract pulse timing structure.
+
+Usage: .venv/bin/python tools/analyze.py captures/<name>.npz
+"""
+import sys
+
+import numpy as np
+
+path = sys.argv[1]
+d = np.load(path)
+volts = d["volts"]
+srate = float(d["srate"])
+dt_us = 1e6 / srate
+
+# threshold midway between the two dominant plateaus
+hi_level = np.median(volts)  # idle dominates the frame -> median = idle (high)
+lo_level = np.percentile(volts, 10)
+thr = (hi_level + lo_level) / 2
+bits = (volts > thr).astype(np.int8)
+print(f"levels: high~{hi_level:.2f} low~{lo_level:.2f} thr={thr:.2f} (raw units)")
+
+# run-length encode
+edges = np.flatnonzero(np.diff(bits)) + 1
+starts = np.concatenate(([0], edges))
+ends = np.concatenate((edges, [len(bits)]))
+levels = bits[starts]
+dur_us = (ends - starts) * dt_us
+
+print(f"{len(levels)} runs total")
+print("\nidx level dur_us (first/last runs are idle padding)")
+for i, (lv, du) in enumerate(zip(levels, dur_us)):
+    tag = "H" if lv else "L"
+    print(f"{i:4d} {tag} {du:10.2f}")

device_firmware/cart_firmware/tools/capture.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""Capture one single-shot CH3 frame from Hantek DPO7204C, save raw + PNG.
+
+Flow: query settings -> :SINGle -> poll :TRIGger:STATus? until STOP ->
+WAVeform:DATA:ALL? CHANnel3 (drained completely, multi-packet aware).
+Scope is left in STOP so the on-screen frame matches the saved data.
+
+Usage: .venv/bin/python tools/capture.py <name> [device]
+Saves captures/<name>.bin, captures/<name>.npz, captures/<name>.png
+"""
+import os
+import sys
+import time
+
+import matplotlib
+
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import numpy as np
+
+DEV = sys.argv[2] if len(sys.argv) > 2 else "/dev/usbtmc2"
+NAME = sys.argv[1] if len(sys.argv) > 1 else "capture"
+OUTDIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "captures")
+os.makedirs(OUTDIR, exist_ok=True)
+
+FIRST_HDR = 11 + 117  # '#9'+9-digit len, then 18 bytes counters + 99 bytes info
+NEXT_HDR = 11 + 18    # follow-up packets: counters only
+
+
+def read_exact(fd, n):
+    buf = b""
+    while len(buf) < n:
+        chunk = os.read(fd, min(1 << 20, n - len(buf)))
+        if not chunk:
+            raise IOError("short read from scope")
+        buf += chunk
+    return buf
+
+
+def read_packet(fd):
+    head = read_exact(fd, 11)
+    assert head[:2] == b"#9", f"bad packet start: {head!r}"
+    pkt_len = int(head[2:11])
+    body = read_exact(fd, pkt_len)
+    return head + body
+
+
+def query(fd, cmd):
+    os.write(fd, cmd.encode() + b"\n")
+    return os.read(fd, 256).decode(errors="replace").strip()
+
+
+fd = os.open(DEV, os.O_RDWR)
+print("IDN:", query(fd, "*IDN?"))
+
+tdiv = float(query(fd, ":TIMebase:SCALe?"))
+vdiv = float(query(fd, ":CHANnel3:SCALe?"))
+voff = float(query(fd, ":CHANnel3:OFFSet?"))
+print(f"tdiv={tdiv} s/div  vdiv={vdiv} V/div  offset={voff} V")
+
+os.write(fd, b":SINGle\n")
+for _ in range(100):
+    time.sleep(0.1)
+    st = query(fd, ":TRIGger:STATus?")
+    if st == "STOP":
+        break
+else:
+    sys.exit(f"scope did not reach STOP (last status: {st})")
+print("status: STOP (frame captured)")
+
+srate = float(query(fd, ":ACQuire:SRATe?"))
+print(f"srate={srate:.3e} Sa/s")
+
+os.write(fd, b"WAVeform:DATA:ALL? CHANnel3\n")
+pkt = read_packet(fd)
+total_len = int(pkt[11:20])
+info_hdr = pkt[29:FIRST_HDR]
+data = pkt[FIRST_HDR:]
+raw_all = pkt
+while len(data) < total_len:
+    os.write(fd, b"WAVeform:DATA:ALL? CHANnel3\n")
+    p = read_packet(fd)
+    raw_all += p
+    data += p[NEXT_HDR:]
+print(f"received {len(data)} samples (declared {total_len})")
+print("info header hex:", info_hdr.hex(" "))
+os.close(fd)
+
+raw = np.frombuffer(data[:total_len], dtype=np.uint8).astype(np.float32)
+# unsigned 8-bit, 25.6 levels/div, mid-screen = code 128, offset shifts zero
+volts = (raw - 128.0) / 25.6 * vdiv - voff
+t = np.arange(len(volts)) / srate * 1e3  # ms
+
+base = os.path.join(OUTDIR, NAME)
+with open(base + ".bin", "wb") as f:
+    f.write(raw_all)
+np.savez(base + ".npz", volts=volts, srate=srate, vdiv=vdiv, voff=voff, tdiv=tdiv)
+
+fig, axes = plt.subplots(2, 1, figsize=(16, 8))
+axes[0].plot(t, volts, lw=0.5)
+axes[0].set_title(f"{NAME} — full frame ({srate:.0e} Sa/s, {vdiv} V/div, off {voff} V)")
+# zoom on activity: region where signal deviates from its median
+dev_idx = np.where(np.abs(volts - np.median(volts)) > 0.5)[0]
+if len(dev_idx):
+    lo = max(0, dev_idx[0] - int(0.05 * (dev_idx[-1] - dev_idx[0] + 1)) - 100)
+    hi = min(len(volts), dev_idx[-1] + int(0.05 * (dev_idx[-1] - dev_idx[0] + 1)) + 100)
+    axes[1].plot(t[lo:hi], volts[lo:hi], lw=0.7)
+    axes[1].set_title("zoom on activity")
+for ax in axes:
+    ax.set_xlabel("t, ms")
+    ax.set_ylabel("U, V")
+    ax.grid(True, alpha=0.3)
+fig.tight_layout()
+fig.savefig(base + ".png", dpi=110)
+print("saved:", base + ".png")
+print(f"range: min={volts.min():.3f} V  max={volts.max():.3f} V  median={np.median(volts):.3f} V")

device_firmware/cart_firmware/tools/compare.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""Rigorous comparison of two SBUS captures (old remote vs our firmware).
+
+Usage: .venv/bin/python tools/compare.py captures/a.npz captures/b.npz
+"""
+import sys
+
+import numpy as np
+
+
+def extract(path):
+    d = np.load(path)
+    v = d["volts"]
+    srate = float(d["srate"])
+    idle = np.median(v)
+    p10, p90 = np.percentile(v, [10, 90])
+    active = p10 if abs(p10 - idle) > abs(p90 - idle) else p90
+    thr = (idle + active) / 2
+    phys_hi = v > thr  # physical high (pulse)
+    # plateau levels: median of samples well inside each state
+    lvl_hi = np.median(v[phys_hi])
+    lvl_lo = np.median(v[~phys_hi])
+    # runs
+    sig = phys_hi.astype(np.int8)
+    edges = np.flatnonzero(np.diff(sig)) + 1
+    starts = np.concatenate(([0], edges))
+    ends = np.concatenate((edges, [len(sig)]))
+    levels = sig[starts]
+    dur_us = (ends - starts) * 1e6 / srate
+    # burst envelope: first to last physical-high sample
+    hi_idx = np.flatnonzero(phys_hi)
+    envelope_us = (hi_idx[-1] - hi_idx[0] + 1) * 1e6 / srate
+    # inner runs (drop leading/trailing idle)
+    runs = [(int(l), float(du)) for l, du in zip(levels[1:-1], dur_us[1:-1])]
+    # UART logic: logic1 == idle state; idle here is physical low
+    stream = []
+    for l, du in runs:
+        n = max(1, round(du / 10.0))
+        stream.extend([1 - l] * n)  # physical high -> logic 0
+    stream.extend([1] * 24)
+    frames = []
+    i = 0
+    while i + 12 <= len(stream):
+        if stream[i] == 1:
+            i += 1
+            continue
+        byte = sum(b << k for k, b in enumerate(stream[i + 1 : i + 9]))
+        frames.append(byte)
+        i += 12
+    # bit clock estimate: envelope should be 299 bits (last stop bits merge w/ idle)
+    n_units = round(envelope_us / 10.0)
+    bit_us = envelope_us / n_units
+    return {
+        "lvl_hi": lvl_hi,
+        "lvl_lo": lvl_lo,
+        "runs": runs,
+        "frames": frames,
+        "envelope_us": envelope_us,
+        "bit_us": bit_us,
+        "vmin": float(v.min()),
+        "vmax": float(v.max()),
+    }
+
+
+a_path, b_path = sys.argv[1], sys.argv[2]
+A, B = extract(a_path), extract(b_path)
+
+print(f"{'':24s} {'A: ' + a_path:>28s} {'B: ' + b_path:>28s}")
+print(f"{'bytes decoded':24s} {len(A['frames']):>28d} {len(B['frames']):>28d}")
+ha = " ".join(f"{x:02X}" for x in A["frames"])
+hb = " ".join(f"{x:02X}" for x in B["frames"])
+print(f"frames identical: {A['frames'] == B['frames']}")
+print(" A:", ha)
+print(" B:", hb)
+print(f"{'run count':24s} {len(A['runs']):>28d} {len(B['runs']):>28d}")
+qa = [round(du / 10) for _, du in A["runs"]]
+qb = [round(du / 10) for _, du in B["runs"]]
+la = [l for l, _ in A["runs"]]
+lb = [l for l, _ in B["runs"]]
+print(f"quantized run pattern identical: {qa == qb and la == lb}")
+print(f"{'envelope, us':24s} {A['envelope_us']:>28.2f} {B['envelope_us']:>28.2f}")
+print(f"{'bit time, us':24s} {A['bit_us']:>28.4f} {B['bit_us']:>28.4f}")
+print(f"{'-> baud':24s} {1e6/A['bit_us']:>28.1f} {1e6/B['bit_us']:>28.1f}")
+print(f"{'high plateau, V':24s} {A['lvl_hi']:>28.3f} {B['lvl_hi']:>28.3f}")
+print(f"{'low plateau, V':24s} {A['lvl_lo']:>28.3f} {B['lvl_lo']:>28.3f}")
+print(f"{'abs min/max, V':24s} {A['vmin']:>14.2f}/{A['vmax']:>12.2f} {B['vmin']:>14.2f}/{B['vmax']:>12.2f}")
+
+# worst run deviation from ideal 10us grid
+da = max(abs(du - 10 * round(du / 10)) for _, du in A["runs"])
+db = max(abs(du - 10 * round(du / 10)) for _, du in B["runs"])
+print(f"{'worst grid dev, us':24s} {da:>28.2f} {db:>28.2f}")

device_firmware/cart_firmware/tools/decode.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Decode a captured frame as SBUS: UART 100 kbit/s 8E2, 25-byte frame,
+16 channels x 11 bits.
+
+Usage: .venv/bin/python tools/decode.py captures/<name>.npz
+"""
+import sys
+
+import numpy as np
+
+BIT_US = 10.0
+
+path = sys.argv[1]
+d = np.load(path)
+volts = d["volts"]
+srate = float(d["srate"])
+dt_us = 1e6 / srate
+
+# idle level dominates the record; UART logic 1 == idle regardless of
+# physical polarity (this line is standard inverted SBUS: idle low)
+idle = np.median(volts)
+p10, p90 = np.percentile(volts, [10, 90])
+active = p10 if abs(p10 - idle) > abs(p90 - idle) else p90
+thr = (idle + active) / 2
+if active > idle:
+    sig = (volts < thr).astype(np.int8)  # pulses up -> logic 0
+else:
+    sig = (volts > thr).astype(np.int8)
+
+edges = np.flatnonzero(np.diff(sig)) + 1
+starts = np.concatenate(([0], edges))
+ends = np.concatenate((edges, [len(sig)]))
+levels = sig[starts]
+dur_us = (ends - starts) * dt_us
+
+stream = []
+for lv, du in zip(levels[1:-1], dur_us[1:-1]):
+    stream.extend([int(lv)] * max(1, round(du / BIT_US)))
+# trailing idle of last stop bits is trimmed by run cut; pad with idle-high
+stream.extend([1] * 24)
+
+# deframe 8E2: start=0, 8 data LSB-first, even parity, 2 stop=1
+i = 0
+frames = []
+errors = []
+while i + 12 <= len(stream):
+    if stream[i] == 1:
+        i += 1
+        continue
+    data = stream[i + 1 : i + 9]
+    par = stream[i + 9]
+    stops = stream[i + 10 : i + 12]
+    byte = sum(b << k for k, b in enumerate(data))
+    if par != (sum(data) & 1):
+        errors.append((len(frames), "parity"))
+    if stops != [1, 1]:
+        errors.append((len(frames), f"stop={stops}"))
+    frames.append(byte)
+    i += 12
+
+print(f"decoded {len(frames)} bytes, errors: {errors if errors else 'none'}")
+print("hex:", " ".join(f"{b:02X}" for b in frames))
+
+if len(frames) >= 25 and frames[0] == 0x0F:
+    payload = frames[1:23]
+    flags = frames[23]
+    footer = frames[24]
+    bits = 0
+    for k, b in enumerate(payload):
+        bits |= b << (8 * k)
+    ch = [(bits >> (11 * n)) & 0x7FF for n in range(16)]
+    print("\nSBUS frame OK" if footer == 0x00 else f"\nfooter unexpected: {footer:02X}")
+    print("channels:", ch)
+    print(f"flags: 0x{flags:02X} (bit0=ch17 bit1=ch18 bit2=frame_lost bit3=failsafe)")
+else:
+    print("not a valid SBUS frame (no 0x0F header)")

device_firmware/cart_firmware/tools/scope.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Minimal SCPI helper for Hantek DPO7204C over /dev/usbtmc2."""
+import os
+import sys
+import time
+
+DEV = "/dev/usbtmc2"
+
+
+class Scope:
+    def __init__(self, dev=DEV):
+        self.fd = os.open(dev, os.O_RDWR)
+
+    def write(self, cmd: str):
+        os.write(self.fd, cmd.encode() + b"\n")
+
+    def read(self, n=1 << 20, timeout=3.0) -> bytes:
+        # usbtmc read returns one transfer chunk; loop until short read
+        chunks = []
+        end = time.time() + timeout
+        while time.time() < end:
+            try:
+                data = os.read(self.fd, n)
+            except OSError:
+                break
+            chunks.append(data)
+            if not data or len(data) < n:
+                break
+        return b"".join(chunks)
+
+    def query(self, cmd: str, timeout=3.0) -> str:
+        self.write(cmd)
+        return self.read(timeout=timeout).decode(errors="replace").strip()
+
+    def query_raw(self, cmd: str, timeout=5.0) -> bytes:
+        self.write(cmd)
+        return self.read(timeout=timeout)
+
+    def close(self):
+        os.close(self.fd)
+
+
+if __name__ == "__main__":
+    s = Scope()
+    for cmd in sys.argv[1:]:
+        if cmd.endswith("?"):
+            print(f"{cmd:40s} -> {s.query(cmd)}")
+        else:
+            s.write(cmd)
+            print(f"{cmd:40s} [sent]")
+    s.close()

python_app/gui/app_window.py

@@ -15,9 +15,10 @@ import logging
 import os
 from pathlib import Path
 import sys
+import threading
 import traceback
 
-from PyQt6.QtCore import QObject, QTimer, pyqtSignal
+from PyQt6.QtCore import QTimer
 from PyQt6.QtGui import QTextCursor
 from PyQt6.QtWidgets import QApplication, QMainWindow, QMessageBox
 
@@ -58,33 +59,60 @@ def _panel_extra(details: str | None, once_key: str | None) -> dict[str, object]
     return {"panel_details": details, "panel_once_key": once_key}
 
 
-class _PanelLogBridge(QObject):
+class _PanelLogBuffer:
-    """Marshals log records from any thread onto the GUI thread for panel rendering.
+    """Thread-safe bounded buffer between logging handlers and the GUI flush timer.
 
-    A :class:`logging.Handler` can fire on a worker thread (readers, broadcaster),
+    A :class:`logging.Handler` can fire on a worker thread (readers, broadcaster)
-    but the log widget may only be touched on the GUI thread; emitting this queued
+    at a very high rate — e.g. the USB RX threads while the free-running sweep
-    signal hands the record across safely (the GPIO-button pattern).
+    streams. Posting one queued Qt event per record used to flood the GUI event
+    queue and keep the interface frozen long after a blocking operation finished
+    while the backlog rendered. Instead, records land in this bounded buffer and
+    a periodic GUI-side timer drains them in one batch; overflow drops the oldest
+    records and reports how many were lost.
     """
 
-    record = pyqtSignal(str, str, object, object)  # display level, message, details, once_key
+    _CAPACITY = 2000
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._entries: deque[tuple[str, str, str | None, str | None]] = deque(maxlen=self._CAPACITY)
+        self._dropped_count = 0
+
+    def append(self, level: str, text: str, details: str | None, once_key: str | None) -> None:
+        """Store one record, evicting the oldest when full (any thread)."""
+        with self._lock:
+            if len(self._entries) == self._CAPACITY:
+                self._dropped_count += 1
+            self._entries.append((level, text, details, once_key))
+
+    def drain(self) -> tuple[list[tuple[str, str, str | None, str | None]], int]:
+        """Return and clear all buffered records plus the overflow-drop count."""
+        with self._lock:
+            entries = list(self._entries)
+            self._entries.clear()
+            dropped_count = self._dropped_count
+            self._dropped_count = 0
+        return entries, dropped_count
 
 
 class _QtLogPanelHandler(logging.Handler):
     """Logging handler that forwards application log records to the GUI log panel."""
 
-    def __init__(self, bridge: _PanelLogBridge) -> None:
+    def __init__(self, buffer: _PanelLogBuffer) -> None:
         super().__init__()
-        self._bridge = bridge
+        self._buffer = buffer
 
     def emit(self, record: logging.LogRecord) -> None:
-        """Forward one record to the panel bridge, mapping WARNING to the short 'WARN'."""
+        """Buffer one record for the panel, mapping WARNING to the short 'WARN'."""
         try:
             display_level = "WARN" if record.levelname == "WARNING" else record.levelname
-            self._bridge.record.emit(
+            details = getattr(record, "panel_details", None)
+            once_key = getattr(record, "panel_once_key", None)
+            self._buffer.append(
                 display_level,
                 record.getMessage(),
-                getattr(record, "panel_details", None),
+                details if isinstance(details, str) else None,
-                getattr(record, "panel_once_key", None),
+                once_key if isinstance(once_key, str) else None,
             )
         except Exception:  # noqa: BLE001 - logging must never raise into the caller
             self.handleError(record)
@@ -145,23 +173,53 @@ class AppWindow(
         log_dir = self._project_root / "python_app/runtime/logs"
         configure_logging(level=DEFAULT_LOG_LEVEL, log_dir=log_dir, console=True)
         self._gui_logger = get_logger("gui")
-        self._log_panel_bridge = _PanelLogBridge()
+        self._log_panel_buffer = _PanelLogBuffer()
-        self._log_panel_bridge.record.connect(self._on_log_record)
 
     def _attach_log_panel(self) -> None:
         """Route application log records into the on-screen panel (widget now exists)."""
-        add_handler(_QtLogPanelHandler(self._log_panel_bridge))
+        add_handler(_QtLogPanelHandler(self._log_panel_buffer))
+        # One bounded flush per tick instead of one queued event per record: the
+        # panel can never flood the GUI event queue, no matter how chatty a
+        # DEBUG-level driver gets.
+        self._log_flush_timer = QTimer(self)
+        self._log_flush_timer.setInterval(100)
+        self._log_flush_timer.timeout.connect(self._flush_log_panel_buffer)
+        self._log_flush_timer.start()
 
-    def _on_log_record(self, level: str, text: str, details: object, once_key: object) -> None:
+    def _flush_log_panel_buffer(self) -> None:
-        """Render one forwarded log record in the panel (always on the GUI thread)."""
+        """Render every buffered log record into the panel as one batched insert."""
-        if not hasattr(self, "_log_box"):
+        entries, dropped_count = self._log_panel_buffer.drain()
+        if (not entries and not dropped_count) or not hasattr(self, "_log_box"):
             return
-        self._append_log_entry(
+
-            level,
+        entry_htmls: list[str] = []
-            text,
+        if dropped_count:
-            details=details if isinstance(details, str) else None,
+            entry_htmls.append(
-            once_key=once_key if isinstance(once_key, str) else None,
+                self._render_log_entry_html(
-        )
+                    "WARN",
+                    f"Log panel overflow: {dropped_count} record(s) dropped "
+                    "(they are still in the log file).",
+                )
+            )
+        error_seen = False
+        for level, text, details, once_key in entries:
+            if once_key is not None:
+                if once_key in self._logged_once_keys:
+                    continue
+                self._logged_once_keys.add(once_key)
+            entry_htmls.append(self._render_log_entry_html(level, text, details))
+            error_seen = error_seen or level.upper() == "ERROR"
+
+        if not entry_htmls:
+            return
+        cursor = self._log_box.textCursor()
+        cursor.movePosition(QTextCursor.MoveOperation.End)
+        self._log_box.setTextCursor(cursor)
+        self._log_box.insertHtml("".join(entry_htmls))
+        self._log_box.insertPlainText("\n")
+        self._log_box.ensureCursorVisible()
+        if error_seen and hasattr(self, "_status_label"):
+            self._status_label.setText("Status: error")
 
     def _init_runtime_services(self) -> None:
         """Initialize long-lived service objects used by mixins."""
@@ -266,6 +324,10 @@ class AppWindow(
     def _init_capture_state(self) -> None:
         """Initialize one-shot capture and sequence-control flags."""
         self._capture_session: SequentialCaptureSession | MultiRadarSequentialCaptureSession | None = None
+        # Guards the blocking per-combo capture against duplicate requests, and keeps
+        # the dialog's action buttons disabled until the post-capture input backlog
+        # is dropped (see AppWindowPreprocessMixin._begin/_end_preprocess_capture).
+        self._preprocess_capture_busy = False
         self._resume_pipeline_after_capture = False
         self._single_capture_active = False
         self._single_capture_start_ns: int | None = None
@@ -551,20 +613,8 @@ class AppWindow(
         """Return full chained traceback for error dialogs and log details."""
         return "".join(traceback.TracebackException.from_exception(exc).format(chain=True)).strip()
 
-    def _append_log_entry(
+    def _render_log_entry_html(self, level: str, text: str, details: str | None = None) -> str:
-        self,
+        """Render one log entry as the panel's HTML block."""
-        level: str,
-        text: str,
-        *,
-        details: str | None = None,
-        once_key: str | None = None,
-    ) -> None:
-        """Append formatted log entry with timestamp and optional details."""
-        if once_key is not None:
-            if once_key in self._logged_once_keys:
-                return
-            self._logged_once_keys.add(once_key)
-
         level_upper = level.upper()
         palette = {
             "DEBUG": ("#6c7b8d", "#52627a", "#8a97a8"),
@@ -586,16 +636,30 @@ class AppWindow(
                 "<pre style='margin:3px 0 0 16px; color:"
                 f"{detail_color};'>{html.escape(details)}</pre>"
             )
+        return "<div style='margin:0 0 6px 0;'>" + "".join(body_parts) + "</div>"
+
+    def _append_log_entry(
+        self,
+        level: str,
+        text: str,
+        *,
+        details: str | None = None,
+        once_key: str | None = None,
+    ) -> None:
+        """Append formatted log entry with timestamp and optional details."""
+        if once_key is not None:
+            if once_key in self._logged_once_keys:
+                return
+            self._logged_once_keys.add(once_key)
 
-        entry_html = "<div style='margin:0 0 6px 0;'>" + "".join(body_parts) + "</div>"
         cursor = self._log_box.textCursor()
         cursor.movePosition(QTextCursor.MoveOperation.End)
         self._log_box.setTextCursor(cursor)
-        self._log_box.insertHtml(entry_html)
+        self._log_box.insertHtml(self._render_log_entry_html(level, text, details))
         self._log_box.insertPlainText("\n")
         self._log_box.ensureCursorVisible()
 
-        if level_upper == "ERROR" and hasattr(self, "_status_label"):
+        if level.upper() == "ERROR" and hasattr(self, "_status_label"):
             self._status_label.setText("Status: error")
 
     def _on_log_level_selected(self, level_text: str) -> None:

python_app/gui/controllers/app_window_preprocess_mixin.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from PyQt6.QtCore import QTimer
+
 from python_app.gui.preprocess_dialog import PreprocessDialog
 from python_app.gui.trace_png_export import export_trace_png
 from python_app.orchestration.preprocess_assets import (
@@ -519,13 +521,23 @@ class AppWindowPreprocessMixin:
         if session is None:
             self._show_error("No active capture sequence")
             return
+        # The capture blocks the event loop, so clicks made during it are delivered
+        # only after it finishes. `_begin_preprocess_capture` disables the action
+        # buttons for that whole window (re-enabled via a posted event), so a queued
+        # click lands on a disabled button instead of silently starting — and
+        # advancing the combo cursor of — another capture.
+        if not self._begin_preprocess_capture():
+            return
 
         try:
-            capture_result = session.capture_current_combo()
+            try:
-        except Exception as exc:  # noqa: BLE001
+                capture_result = session.capture_current_combo()
-            self._on_capture_combo_failed(session, exc)
+            except Exception as exc:  # noqa: BLE001
-            return
+                self._on_capture_combo_failed(session, exc)
-        self._record_preprocess_capture(session, capture_result)
+                return
+            self._record_preprocess_capture(session, capture_result)
+        finally:
+            self._end_preprocess_capture()
 
     def _capture_all_remaining(self) -> None:
         """Capture all remaining combos for the active preprocess session."""
@@ -539,6 +551,8 @@ class AppWindowPreprocessMixin:
                 details=self._capture_state_details(),
             )
             return
+        if not self._begin_preprocess_capture():
+            return
 
         display_name = preprocess_asset_display_name(session.kind)
         dialog = self._ensure_preprocess_dialog()
@@ -547,13 +561,40 @@ class AppWindowPreprocessMixin:
             f"{display_name} batch capture started: remaining="
             f"{session.state().total_count - session.state().captured_count}"
         )
-        while not session.is_complete():
+        try:
-            try:
+            while not session.is_complete():
-                capture_result = session.capture_current_combo()
+                try:
-            except Exception as exc:  # noqa: BLE001
+                    capture_result = session.capture_current_combo()
-                self._on_capture_combo_failed(session, exc)
+                except Exception as exc:  # noqa: BLE001
-                return
+                    self._on_capture_combo_failed(session, exc)
-            self._record_preprocess_capture(session, capture_result)
+                    return
+                self._record_preprocess_capture(session, capture_result)
+        finally:
+            self._end_preprocess_capture()
+
+    def _begin_preprocess_capture(self) -> bool:
+        """Mark a blocking combo capture as running; refuse when one already is.
+
+        Returns False for a duplicate request (e.g. a click delivered while an
+        error dialog inside a capture pumps the event loop).
+        """
+        if self._preprocess_capture_busy:
+            self._log("Preprocess combo capture already in progress; ignoring duplicate request.")
+            return False
+        self._preprocess_capture_busy = True
+        # Disable the sequence action buttons for the whole blocked window.
+        self._update_capture_dialog_state()
+        return True
+
+    def _end_preprocess_capture(self) -> None:
+        """Re-enable capture actions after the pending input backlog is discarded.
+
+        The zero-delay timer fires only after Qt has dispatched the window-system
+        events queued while the capture blocked the loop; those clicks hit the
+        still-disabled buttons and are dropped, then the buttons come back.
+        """
+        self._preprocess_capture_busy = False
+        QTimer.singleShot(0, self._update_capture_dialog_state)
 
     def _on_capture_combo_failed(
         self,
@@ -817,6 +858,10 @@ class AppWindowPreprocessMixin:
                 and state.current_combo is not None
             ),
             variant_count=state.variant_count,
+            # While a blocking capture is executing, every action stays disabled no
+            # matter what the session state allows: clicks queued during the freeze
+            # must land on disabled buttons (see `_end_preprocess_capture`).
+            actions_enabled=not self._preprocess_capture_busy,
         )
 
     def _cleanup_capture_session(self) -> None:

python_app/gui/preprocess_dialog.py

@@ -354,8 +354,14 @@ class PreprocessDialog(QDialog):
         can_finalize: bool,
         can_capture_all: bool,
         variant_count: int = 1,
+        actions_enabled: bool = True,
     ) -> None:
-        """Update sequence progress/status widgets."""
+        """Update sequence progress/status widgets.
+
+        With ``actions_enabled=False`` the progress labels still update but every
+        sequence action button is kept disabled — used while a blocking capture
+        runs, so input queued during the freeze cannot trigger another action.
+        """
         if kind is None:
             self._active_kind_label.setText("<none>")
             self._progress_label.setText("0 / 0")
@@ -368,13 +374,14 @@ class PreprocessDialog(QDialog):
             self._capture_all_button.setText("Capture All Remaining")
             return
 
+        actions_enabled = bool(actions_enabled)
         active_label = preprocess_asset_display_name(kind) if kind in PREPROCESS_ASSET_SPECS else kind
         self._active_kind_label.setText(active_label)
         self._progress_label.setText(f"{captured_count} / {total_count}")
-        self._undo_last_button.setEnabled(bool(can_undo))
+        self._undo_last_button.setEnabled(bool(can_undo) and actions_enabled)
-        self._save_sequence_button.setEnabled(bool(can_finalize))
+        self._save_sequence_button.setEnabled(bool(can_finalize) and actions_enabled)
-        self._capture_all_button.setEnabled(bool(can_capture_all))
+        self._capture_all_button.setEnabled(bool(can_capture_all) and actions_enabled)
-        self._abort_button.setEnabled(True)
+        self._abort_button.setEnabled(actions_enabled)
         self._capture_all_button.setText("Capture All Remaining")
         if next_input is None or next_output is None:
             self._combo_label.setText("<complete>")
@@ -384,7 +391,7 @@ class PreprocessDialog(QDialog):
             if int(variant_count) > 1:
                 combo_text += f" | radar configs={int(variant_count)}"
             self._combo_label.setText(combo_text)
-            self._capture_next_button.setEnabled(True)
+            self._capture_next_button.setEnabled(actions_enabled)
 
     def set_available_sets(self, available_sets: dict[str, list[str]]) -> None:
         """Replace combo-box choices for all preprocess assets."""

python_app/hardware_full/librevna_driver/transport/usb.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from contextlib import suppress
 import logging
 import threading
+import time
 from typing import Callable
 
 from ..exceptions import DeviceDisconnectedError, TimeoutError
@@ -44,6 +45,10 @@ class USBTransport:
         self._rx_thread: threading.Thread | None = None
         self._stop_event = threading.Event()
         self._tx_lock = threading.Lock()
+        # Aggregation window for the RX debug trace (see `_rx_loop`).
+        self._rx_debug_bytes = 0
+        self._rx_debug_chunks = 0
+        self._rx_debug_window_start = 0.0
 
         self.connected_serial: str | None = None
 
@@ -270,7 +275,27 @@ class USBTransport:
 
             if data:
                 if logger.isEnabledFor(logging.DEBUG):
-                    logger.debug("USB RX %d bytes", len(data))
+                    # Aggregate: the free-running datapoint stream completes bulk
+                    # reads hundreds of times per second, and a log record per chunk
+                    # floods every handler (file, stderr, and the GUI panel, which
+                    # marshals each record onto the GUI thread). One summary per
+                    # second keeps the throughput trace without the flood.
+                    self._rx_debug_bytes += len(data)
+                    self._rx_debug_chunks += 1
+                    now = time.monotonic()
+                    if self._rx_debug_window_start == 0.0:
+                        self._rx_debug_window_start = now
+                    elif now - self._rx_debug_window_start >= 1.0:
+                        logger.debug(
+                            "USB RX %d bytes in %d chunks over %.2f s (serial=%s)",
+                            self._rx_debug_bytes,
+                            self._rx_debug_chunks,
+                            now - self._rx_debug_window_start,
+                            self.connected_serial,
+                        )
+                        self._rx_debug_bytes = 0
+                        self._rx_debug_chunks = 0
+                        self._rx_debug_window_start = now
                 self._on_data(bytes(data))
         logger.debug("USB RX thread stopped")
 

python_app/hardware_full/librevna_multi_device_driver/transport.py

@@ -15,6 +15,17 @@ from python_app.hardware_full.librevna_multi_device_driver.protocol import Packe
 
 logger = logging.getLogger(__name__)
 
+# The sweep free-runs by design, so devices stream datapoints continuously even
+# while no acquisition is consuming them (e.g. an operator pausing between manual
+# combo captures). An unbounded queue then grows without limit — hundreds of MB
+# over a few minutes — and the next acquisition's drain spends seconds discarding
+# the backlog on the GUI thread. Bound the queue and drop the OLDEST packet on
+# overflow: every acquisition drains stale packets before collecting anyway, and
+# whenever packets actually matter (ACK waits, cycle collection) a consumer is
+# already pulling, so the queue never approaches the bound. Sized to hold many
+# full sweeps of datapoints with a wide margin.
+_RECEIVED_PACKET_QUEUE_MAX = 32768
+
 
 class LibreVnaUsbBulkConnection:
     """Minimal packet transport for one LibreVNA device."""
@@ -29,7 +40,9 @@ class LibreVnaUsbBulkConnection:
             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._received_packets: queue.Queue[tuple[int, bytes]] = queue.Queue(
+            maxsize=_RECEIVED_PACKET_QUEUE_MAX
+        )
         self._fatal_error: Exception | None = None
         self._fatal_lock = threading.Lock()
         self._transport = USBTransport(
@@ -108,7 +121,20 @@ class LibreVnaUsbBulkConnection:
             logger.warning("Dropping unparseable USB chunk from %s: %s", self.serial_number, exc)
             return
         for packet in packets:
-            self._received_packets.put((int(packet.type), bytes(packet.payload)))
+            entry = (int(packet.type), bytes(packet.payload))
+            while True:
+                try:
+                    self._received_packets.put_nowait(entry)
+                    break
+                except queue.Full:
+                    # Blocking here would stall the USB read thread; discard the
+                    # oldest packet instead — stale data is what the pre-collect
+                    # drain throws away anyway. Racing a concurrent consumer just
+                    # means the queue already has room again.
+                    try:
+                        self._received_packets.get_nowait()
+                    except queue.Empty:
+                        pass
 
     def _on_disconnect(self, exc: Exception) -> None:
         """Record an asynchronous transport disconnect as the fatal error."""

python_app/hardware_full/multi_device_service.py

@@ -332,10 +332,15 @@ class MultiDeviceLibreVnaService:
         assert self._sweep_configuration is not None
 
         self._controller.configure_continuous_sweep(self._sweep_configuration)
+        # Bound the sweep itself rather than reusing `capture_start_ns`: the latter
+        # is taken before any retry/recovery, so it would overstate how long the
+        # traces below took to measure.
+        sweep_start_ns = time.monotonic_ns()
         result = self._controller.collect_running_sweep_cycles(
             1,
             datapoint_timeout_seconds=LIBREVNA_NATIVE_SWEEP_TIMEOUT_SECONDS,
         )
+        sweep_end_ns = time.monotonic_ns()
         normalized_s_parameters = {
             str(name).lower(): np.asarray(values, dtype=np.complex64)
             for name, values in result.s_parameters.items()
@@ -356,6 +361,11 @@ class MultiDeviceLibreVnaService:
                         frequency_hz=frequencies,
                         s11=reflection,
                         s21=self._required_s_parameter(normalized_s_parameters, s_parameter_name),
+                        # Every combo comes out of the same synchronized cycle, so
+                        # they all share one window — no combo was measured earlier
+                        # or later than another here.
+                        capture_start_ns=sweep_start_ns,
+                        capture_end_ns=sweep_end_ns,
                     )
                 )
 
@@ -368,6 +378,7 @@ class MultiDeviceLibreVnaService:
 
     def _acquire_mock_collection(self, collection_id: int, capture_start_ns: int) -> SweepCollection:
         assert self._sweep_configuration is not None
+        mock_sweep_start_ns = time.monotonic_ns()
         points = int(self._sweep_configuration.points)
         frequencies = np.linspace(
             self._sweep_configuration.start_hz,
@@ -393,6 +404,8 @@ class MultiDeviceLibreVnaService:
                         frequency_hz=frequencies,
                         s11=s11,
                         s21=s21,
+                        capture_start_ns=mock_sweep_start_ns,
+                        capture_end_ns=time.monotonic_ns(),
                     )
                 )
         self._mock_phase += 0.05

python_app/hardware_full/sn9000_service.py

@@ -183,7 +183,11 @@ class Sn9000Service:
         capture_start_ns = time.monotonic_ns()
 
         s_parameters = self._query_sweep_s_parameters(points)
-        traces = self._assemble_traces(s_parameters)
+        traces = self._assemble_traces(
+            s_parameters,
+            sweep_start_ns=capture_start_ns,
+            sweep_end_ns=time.monotonic_ns(),
+        )
 
         return SweepCollection(
             collection_id=int(collection_id),
@@ -256,7 +260,13 @@ class Sn9000Service:
     def _uses_pyvisa_py_backend(self) -> bool:
         return self.visa_library == "@py" or self.visa_library.endswith("@py")
 
-    def _assemble_traces(self, s_parameters: dict[str, np.ndarray]) -> list[TraceData]:
+    def _assemble_traces(
+        self,
+        s_parameters: dict[str, np.ndarray],
+        *,
+        sweep_start_ns: int,
+        sweep_end_ns: int,
+    ) -> list[TraceData]:
         frequency_hz = self._require_frequency_axis()
         traces: list[TraceData] = []
         for output_position, output_port in enumerate(_OUTPUT_PORT_BY_INDEX):
@@ -269,6 +279,10 @@ class Sn9000Service:
                         frequency_hz=frequency_hz,
                         s11=reflection,
                         s21=transmission,
+                        # One triggered sweep produces every port pair at once, so
+                        # all combos share the sweep's window.
+                        capture_start_ns=int(sweep_start_ns),
+                        capture_end_ns=int(sweep_end_ns),
                     )
                 )
         return traces

python_app/hardware_full/switched_matrix_radar_service.py

@@ -83,42 +83,8 @@ class SwitchedMatrixRadarService:
 
         for out_k in range(out_steps):
             for in_k in range(in_steps):
-                step_start_ns = time.monotonic_ns()
+                for trace in self._acquire_step_traces(out_k, in_k, collection_id):
-                if self.output_switch is not None:
+                    slots[trace.combo.output * total_inputs + trace.combo.input] = trace
-                    self.output_switch.switch_to(out_k)
-                if self.input_switch is not None:
-                    self.input_switch.switch_to(in_k)
-                switched_ns = time.monotonic_ns()
-                # Settle AFTER the last switch change and BEFORE collecting, so the
-                # cycle we anchor on starts with the RF path already stable.
-                if self.settling_ms > 0:
-                    time.sleep(self.settling_ms / 1000.0)
-                settled_ns = time.monotonic_ns()
-
-                sub = self.inner.acquire_collection(collection_id)
-                inner_end_ns = time.monotonic_ns()
-                logger.debug(
-                    "timing: collection %d step out=%d in=%d | gap_prev_collect_to_switch=%s ms, "
-                    "switch=%.3f ms, settle=%.2f ms, inner_collect=%.2f ms",
-                    collection_id,
-                    out_k,
-                    in_k,
-                    (
-                        f"{(step_start_ns - self._last_inner_end_ns) / 1e6:.2f}"
-                        if self._last_inner_end_ns
-                        else "n/a"
-                    ),
-                    (switched_ns - step_start_ns) / 1e6,
-                    (settled_ns - switched_ns) / 1e6,
-                    (inner_end_ns - settled_ns) / 1e6,
-                )
-                self._last_inner_end_ns = inner_end_ns
-                for trace in sub.traces:
-                    input_pos = in_k * self.inner_input_positions + int(trace.combo.input)
-                    output_pos = out_k * self.inner_output_positions + int(trace.combo.output)
-                    slots[output_pos * total_inputs + input_pos] = replace(
-                        trace, combo=ComboKey(input=input_pos, output=output_pos)
-                    )
 
         if any(trace is None for trace in slots):
             missing = sum(1 for trace in slots if trace is None)
@@ -134,6 +100,100 @@ class SwitchedMatrixRadarService:
             capture_end_ns=time.monotonic_ns(),
         )
 
+    def acquire_combo_collection(
+        self,
+        *,
+        input_pos: int,
+        output_pos: int,
+        collection_id: int = 1,
+    ) -> SweepCollection:
+        """Acquire only the physical switch step that carries one widened combo.
+
+        The per-combo capture workflows need a single trace at a time; sweeping
+        every switch position for that (a full ``acquire_collection``) multiplies
+        the capture time by the number of physical steps and freezes the caller
+        for the whole sweep. One widened combo lives entirely inside one
+        (out_k, in_k) step, so acquiring just that step is sufficient. The result
+        contains that step's traces with widened combo keys, including the
+        requested combo.
+        """
+        out_steps = self.output_switch.position_count() if self.output_switch is not None else 1
+        in_steps = self.input_switch.position_count() if self.input_switch is not None else 1
+        total_inputs = in_steps * self.inner_input_positions
+        total_outputs = out_steps * self.inner_output_positions
+        if not (0 <= int(input_pos) < total_inputs and 0 <= int(output_pos) < total_outputs):
+            raise ValueError(
+                f"Widened combo out of range: input={input_pos} (of {total_inputs}), "
+                f"output={output_pos} (of {total_outputs})"
+            )
+
+        capture_start_ns = time.monotonic_ns()
+        out_k = int(output_pos) // self.inner_output_positions
+        in_k = int(input_pos) // self.inner_input_positions
+        traces = self._acquire_step_traces(out_k, in_k, collection_id)
+        return SweepCollection(
+            collection_id=int(collection_id),
+            monotonic_ns=time.monotonic_ns(),
+            traces=traces,
+            capture_start_ns=capture_start_ns,
+            capture_end_ns=time.monotonic_ns(),
+        )
+
+    def _acquire_step_traces(self, out_k: int, in_k: int, collection_id: int) -> list[TraceData]:
+        """Drive both switches to one step, settle, and collect its widened traces.
+
+        Every returned trace carries the monotonic window of the inner collection
+        that produced it, so a consumer can tell when each combo of a switched
+        matrix was really measured instead of only when the whole cycle began and
+        ended. The switch drive and settling are deliberately outside the window.
+        """
+        step_start_ns = time.monotonic_ns()
+        if self.output_switch is not None:
+            self.output_switch.switch_to(out_k)
+        if self.input_switch is not None:
+            self.input_switch.switch_to(in_k)
+        switched_ns = time.monotonic_ns()
+        # Settle AFTER the last switch change and BEFORE collecting, so the
+        # cycle we anchor on starts with the RF path already stable.
+        if self.settling_ms > 0:
+            time.sleep(self.settling_ms / 1000.0)
+        settled_ns = time.monotonic_ns()
+
+        sub = self.inner.acquire_collection(collection_id)
+        inner_end_ns = time.monotonic_ns()
+        logger.debug(
+            "timing: collection %d step out=%d in=%d | gap_prev_collect_to_switch=%s ms, "
+            "switch=%.3f ms, settle=%.2f ms, inner_collect=%.2f ms",
+            collection_id,
+            out_k,
+            in_k,
+            (
+                f"{(step_start_ns - self._last_inner_end_ns) / 1e6:.2f}"
+                if self._last_inner_end_ns
+                else "n/a"
+            ),
+            (switched_ns - step_start_ns) / 1e6,
+            (settled_ns - switched_ns) / 1e6,
+            (inner_end_ns - settled_ns) / 1e6,
+        )
+        self._last_inner_end_ns = inner_end_ns
+        return [
+            replace(
+                trace,
+                combo=ComboKey(
+                    input=in_k * self.inner_input_positions + int(trace.combo.input),
+                    output=out_k * self.inner_output_positions + int(trace.combo.output),
+                ),
+                # Keep the inner service's own per-trace window when it reports one
+                # (it knows its internal port order better than this step does);
+                # otherwise fall back to the window of this inner collection.
+                capture_start_ns=int(trace.capture_start_ns) or settled_ns,
+                capture_end_ns=int(trace.capture_end_ns) or inner_end_ns,
+            )
+            for trace in sub.traces
+        ]
+
+
 def build_physical_switch(
     model: SwitchModel,
     physical_positions: int,

python_app/models/dataset_model.py

@@ -32,12 +32,22 @@ class ComboKey:
 
 @dataclass(slots=True)
 class TraceData:
-    """One frequency-domain trace set for a specific switch combination."""
+    """One frequency-domain trace set for a specific switch combination.
+
+    ``capture_start_ns``/``capture_end_ns`` bound the monotonic window in which
+    THIS trace's sweep was measured, excluding the switch drive and settling that
+    preceded it. In switched modes a collection is assembled combo by combo over
+    many milliseconds, so the collection-level window says nothing about when any
+    individual combo was measured — these do. Zero on both means the producer did
+    not report per-trace timing.
+    """
 
     combo: ComboKey
     frequency_hz: np.ndarray
     s11: np.ndarray
     s21: np.ndarray
+    capture_start_ns: int = 0
+    capture_end_ns: int = 0
 
 
 @dataclass(slots=True)

python_app/orchestration/shm/decoder.py

@@ -54,12 +54,27 @@ def decode_trace_collection(payload: bytes, expected_magic: int) -> SweepCollect
             )
         )
 
+    # Optional trailer, written after the trace blocks by newer producers: the
+    # collection capture window, then a per-trace window table. Both stages are
+    # optional so payloads from an older producer still decode (the timestamps
+    # simply stay zero).
     capture_start_ns = 0
     capture_end_ns = 0
-    if cursor.remaining_bytes() == 16:
+    if cursor.remaining_bytes() != 0:
+        if cursor.remaining_bytes() < 16:
+            raise ValueError("Truncated capture window in trace collection")
         capture_start_ns = cursor.read_u64()
         capture_end_ns = cursor.read_u64()
-    elif cursor.remaining_bytes() != 0:
+
+    if cursor.remaining_bytes() != 0:
+        trace_time_count = cursor.read_u32()
+        if trace_time_count != len(traces):
+            raise ValueError("Per-trace capture window count does not match trace count")
+        for trace in traces:
+            trace.capture_start_ns = cursor.read_u64()
+            trace.capture_end_ns = cursor.read_u64()
+
+    if cursor.remaining_bytes() != 0:
         raise ValueError("Unexpected trailing bytes in trace collection")
 
     return SweepCollection(

python_app/scripts/convert_snapshot_to_vna_history.py

@@ -23,6 +23,9 @@ class TraceRecord:
     stage_index: int
     frequency_hz: np.ndarray
     samples: np.ndarray
+    # End of this trace's own sweep, from the snapshot's per-trace metadata; 0 for a
+    # snapshot recorded before per-trace timing existed.
+    capture_end_ns: int = 0
 
 
 @dataclass(frozen=True)
@@ -128,6 +131,7 @@ def _load_stage_records(
                 stage_index=_parse_stage_index(collection_dir.name, fallback_idx),
                 frequency_hz=frequency_hz,
                 samples=samples,
+                capture_end_ns=int(trace_meta.get("capture_end_ns", 0)),
             )
         )
 
@@ -214,7 +218,15 @@ def _build_sweep_history(
 
         start_freq_hz = float(base.frequency_hz[0])
         stop_freq_hz = float(base.frequency_hz[-1])
-        timestamp_sec = float(base.monotonic_ns) / 1_000_000_000.0 if base.monotonic_ns > 0 else float(fallback_index)
+        # Prefer this trace's own sweep time: with a switching matrix the combos of
+        # one collection are measured milliseconds apart, so the collection
+        # timestamp misplaces every combo but the last.
+        if base.capture_end_ns > 0:
+            timestamp_sec = float(base.capture_end_ns) / 1_000_000_000.0
+        elif base.monotonic_ns > 0:
+            timestamp_sec = float(base.monotonic_ns) / 1_000_000_000.0
+        else:
+            timestamp_sec = float(fallback_index)
 
         history.append(
             {

python_app/scripts/kamil_adc_raw_producer.py

@@ -184,12 +184,16 @@ def main() -> int:
                     if collector_driven:
                         # The collector already switched and tagged the sweep; just
                         # read the clean capture for this combination.
+                        sweep_start_ns = time.monotonic_ns()
                         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)
+                        # Stamped after switching and settling so the window covers
+                        # the sweep alone, not the dead time before it.
+                        sweep_start_ns = time.monotonic_ns()
                         sweep = radar.acquire()
                     traces.append(
                         TraceData(
@@ -197,6 +201,8 @@ def main() -> int:
                             frequency_hz=np.asarray(sweep.x, dtype=np.float32),
                             s11=np.asarray(sweep.trace("s11"), dtype=np.complex64),
                             s21=np.asarray(sweep.trace("s21"), dtype=np.complex64),
+                            capture_start_ns=sweep_start_ns,
+                            capture_end_ns=time.monotonic_ns(),
                         )
                     )
             except Exception as exc:  # noqa: BLE001 — reconnect forever, never give up

python_app/storage/npz/serialize.py

@@ -22,7 +22,14 @@ def _write_interleaved_complex(buffer: bytearray, values: np.ndarray) -> None:
 
 
 def serialize_trace_collection(collection: SweepCollection, magic: int) -> bytes:
-    """Serialize one raw/preprocessed trace collection into ring-compatible binary format."""
+    """Serialize one raw/preprocessed trace collection into ring-compatible binary format.
+
+    The trailer is appended after the trace blocks so older readers, which stop at
+    the last block, still decode the traces: first the collection capture window,
+    then a per-trace window table (one ``(start_ns, end_ns)`` pair per trace, in
+    trace order). See :func:`python_app.orchestration.shm.decoder.decode_trace_collection`
+    and ``read_trace_collection`` in ``common_cpp/ipc/src/shared_types.cpp``.
+    """
     buffer = bytearray()
     buffer.extend(struct.pack("<IQQI", magic, collection.collection_id, collection.monotonic_ns, len(collection.traces)))
 
@@ -48,6 +55,11 @@ def serialize_trace_collection(collection: SweepCollection, magic: int) -> bytes
             int(collection.capture_end_ns),
         )
     )
+    buffer.extend(struct.pack("<I", len(collection.traces)))
+    for trace in collection.traces:
+        buffer.extend(
+            struct.pack("<QQ", int(trace.capture_start_ns), int(trace.capture_end_ns))
+        )
     return bytes(buffer)
 
 

python_app/storage/npz/snapshot_numpy.py

@@ -124,6 +124,17 @@ def save_trace_history_binary(stage_dir: Path, history: list[SweepCollection], m
                     "capture_start_ns": int(collection.capture_start_ns),
                     "capture_end_ns": int(collection.capture_end_ns),
                     "trace_count": len(collection.traces),
+                    # Also in the .bin trailer; repeated here so per-combo timing is
+                    # readable without decoding the binary payload.
+                    "traces": [
+                        {
+                            "input": int(trace.combo.input),
+                            "output": int(trace.combo.output),
+                            "capture_start_ns": int(trace.capture_start_ns),
+                            "capture_end_ns": int(trace.capture_end_ns),
+                        }
+                        for trace in collection.traces
+                    ],
                 },
                 indent=2,
             ),
@@ -182,6 +193,10 @@ def save_trace_history_numpy(
                     "input": int(trace.combo.input),
                     "output": int(trace.combo.output),
                     "points": int(freq.size),
+                    # When each combo was measured, which in a switched matrix is
+                    # spread across the collection window rather than aligned with it.
+                    "capture_start_ns": int(trace.capture_start_ns),
+                    "capture_end_ns": int(trace.capture_end_ns),
                     "freq_file": f"{tag}_freq.npy",
                     "s11_file": f"{tag}_s11.npy",
                     "s21_file": f"{tag}_s21.npy",

python_app/storage/npz/store.py

@@ -117,6 +117,8 @@ class NpzStore(StoreApi):
                 {
                     "input": trace.combo.input,
                     "output": trace.combo.output,
+                    "capture_start_ns": int(trace.capture_start_ns),
+                    "capture_end_ns": int(trace.capture_end_ns),
                     "freq_key": freq_key,
                     "s11_key": s11_key,
                     "s21_key": s21_key,
@@ -177,6 +179,8 @@ class NpzStore(StoreApi):
                         frequency_hz=freq,
                         s11=s11,
                         s21=s21,
+                        capture_start_ns=int(combo.get("capture_start_ns", 0)),
+                        capture_end_ns=int(combo.get("capture_end_ns", 0)),
                     )
                 )
 

python_app/storage/npz/vna_history_json.py

@@ -25,6 +25,10 @@ class TraceRecord:
     stage_index: int
     frequency_hz: np.ndarray
     samples: np.ndarray
+    # End of this trace's own sweep, or 0 when the producer reported no per-trace
+    # timing. Preferred over the collection timestamp for the exported sweep time:
+    # in a switched matrix each combo is measured at a different instant.
+    capture_end_ns: int = 0
 
 
 def _normalize_channel(channel: str) -> str:
@@ -85,6 +89,7 @@ def _build_stage_records(
                 stage_index=int(stage_index),
                 frequency_hz=frequency_hz,
                 samples=samples,
+                capture_end_ns=int(trace.capture_end_ns),
             )
         )
     return records
@@ -137,7 +142,15 @@ def _build_sweep_history(
 
         start_freq_hz = float(base.frequency_hz[0])
         stop_freq_hz = float(base.frequency_hz[-1])
-        timestamp_sec = float(base.monotonic_ns) / 1_000_000_000.0 if base.monotonic_ns > 0 else float(fallback_index)
+        # Prefer the exported trace's own sweep time: with a switching matrix the
+        # combos of one collection are measured milliseconds apart, so the
+        # collection timestamp misplaces every combo but the last.
+        if base.capture_end_ns > 0:
+            timestamp_sec = float(base.capture_end_ns) / 1_000_000_000.0
+        elif base.monotonic_ns > 0:
+            timestamp_sec = float(base.monotonic_ns) / 1_000_000_000.0
+        else:
+            timestamp_sec = float(fallback_index)
 
         history.append(
             {

python_app/tests/test_log_panel_buffer.py

@@ -0,0 +1,48 @@
+"""Unit tests for the bounded GUI log-panel buffer.
+
+The buffer decouples logging handlers (any thread, potentially very chatty at
+DEBUG) from the GUI: records are batched by a flush timer instead of posting one
+queued Qt event per record, and overflow drops the oldest records with a count.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from python_app.gui.app_window import _PanelLogBuffer
+
+
+class PanelLogBufferTest(unittest.TestCase):
+    """Bounded capacity, oldest-first eviction, and accurate drop accounting."""
+
+    def test_drain_returns_entries_in_order_and_clears(self) -> None:
+        buffer = _PanelLogBuffer()
+        buffer.append("INFO", "first", None, None)
+        buffer.append("WARN", "second", "details", "key")
+
+        entries, dropped_count = buffer.drain()
+
+        self.assertEqual(dropped_count, 0)
+        self.assertEqual(
+            entries,
+            [("INFO", "first", None, None), ("WARN", "second", "details", "key")],
+        )
+        self.assertEqual(buffer.drain(), ([], 0))
+
+    def test_overflow_drops_oldest_and_counts(self) -> None:
+        buffer = _PanelLogBuffer()
+        overflow = 100
+        total = _PanelLogBuffer._CAPACITY + overflow
+        for index in range(total):
+            buffer.append("DEBUG", f"m{index}", None, None)
+
+        entries, dropped_count = buffer.drain()
+
+        self.assertEqual(dropped_count, overflow)
+        self.assertEqual(len(entries), _PanelLogBuffer._CAPACITY)
+        self.assertEqual(entries[0][1], f"m{overflow}")
+        self.assertEqual(entries[-1][1], f"m{total - 1}")
+
+
+if __name__ == "__main__":
+    unittest.main()

python_app/tests/test_shm_ipc.py

@@ -39,12 +39,19 @@ from python_app.orchestration.shm.ring_writer import ShmRingWriter
 from python_app.storage.npz.serialize import serialize_result_collection, serialize_trace_collection
 
 
-def _trace(in_pos: int, out_pos: int, n: int) -> TraceData:
+def _trace(in_pos: int, out_pos: int, n: int, *, capture_ns: tuple[int, int] = (0, 0)) -> TraceData:
     """Build a trace with float32-exact data so round-trips compare exactly."""
     freq = np.arange(n, dtype=np.float32) + 1.0
     s11 = (np.arange(n, dtype=np.float32) + 0.5j * np.arange(n, dtype=np.float32)).astype(np.complex64)
     s21 = (-np.arange(n, dtype=np.float32) + 2.0j * np.arange(n, dtype=np.float32)).astype(np.complex64)
-    return TraceData(combo=ComboKey(input=in_pos, output=out_pos), frequency_hz=freq, s11=s11, s21=s21)
+    return TraceData(
+        combo=ComboKey(input=in_pos, output=out_pos),
+        frequency_hz=freq,
+        s11=s11,
+        s21=s21,
+        capture_start_ns=capture_ns[0],
+        capture_end_ns=capture_ns[1],
+    )
 
 
 class TraceCollectionRoundTripTest(unittest.TestCase):
@@ -52,7 +59,7 @@ class TraceCollectionRoundTripTest(unittest.TestCase):
         collection = SweepCollection(
             collection_id=7,
             monotonic_ns=123,
-            traces=[_trace(0, 0, 4), _trace(3, 1, 2)],
+            traces=[_trace(0, 0, 4, capture_ns=(11, 13)), _trace(3, 1, 2, capture_ns=(15, 19))],
             capture_start_ns=10,
             capture_end_ns=20,
         )
@@ -66,6 +73,10 @@ class TraceCollectionRoundTripTest(unittest.TestCase):
             self.assertTrue(np.array_equal(got.frequency_hz, original.frequency_hz))
             self.assertTrue(np.array_equal(got.s11, original.s11))
             self.assertTrue(np.array_equal(got.s21, original.s21))
+            self.assertEqual(
+                (got.capture_start_ns, got.capture_end_ns),
+                (original.capture_start_ns, original.capture_end_ns),
+            )
 
     def test_raw_round_trips(self) -> None:
         self._assert_round_trips(RAW_MAGIC)
@@ -78,6 +89,34 @@ class TraceCollectionRoundTripTest(unittest.TestCase):
         decoded = decode_trace_collection(serialize_trace_collection(collection, RAW_MAGIC), RAW_MAGIC)
         self.assertEqual(decoded.traces, [])
 
+    def test_payload_without_per_trace_window_table_still_decodes(self) -> None:
+        # A producer built before per-trace timing stops after the collection
+        # window; its traces must still decode, with the timestamps left at zero.
+        collection = SweepCollection(
+            collection_id=4,
+            monotonic_ns=5,
+            traces=[_trace(1, 0, 3, capture_ns=(7, 9))],
+            capture_start_ns=6,
+            capture_end_ns=10,
+        )
+        full = serialize_trace_collection(collection, RAW_MAGIC)
+        legacy = full[: -(4 + 16 * len(collection.traces))]
+
+        decoded = decode_trace_collection(legacy, RAW_MAGIC)
+        self.assertEqual((decoded.capture_start_ns, decoded.capture_end_ns), (6, 10))
+        self.assertEqual(len(decoded.traces), 1)
+        self.assertEqual((decoded.traces[0].capture_start_ns, decoded.traces[0].capture_end_ns), (0, 0))
+
+    def test_per_trace_window_count_mismatch_is_rejected(self) -> None:
+        collection = SweepCollection(
+            collection_id=4, monotonic_ns=5, traces=[_trace(1, 0, 3, capture_ns=(7, 9))]
+        )
+        payload = serialize_trace_collection(collection, RAW_MAGIC)
+        # Overwrite the window-table count (u32 before the single 16-byte pair).
+        corrupt = payload[:-20] + struct.pack("<I", 2) + payload[-16:]
+        with self.assertRaises(ValueError):
+            decode_trace_collection(corrupt, RAW_MAGIC)
+
 
 class ResultCollectionRoundTripTest(unittest.TestCase):
     def test_all_payload_kinds_round_trip(self) -> None:

python_app/tests/test_switched_matrix_capture.py

@@ -0,0 +1,217 @@
+"""Unit tests for switch-widened matrix capture.
+
+Cover the targeted single-step acquisition on ``SwitchedMatrixRadarService`` and
+verify the manual per-combo capture workflow uses it instead of sweeping the full
+widened matrix (the regression that froze the GUI for the whole matrix per click).
+"""
+
+from __future__ import annotations
+
+import time
+import unittest
+from unittest import mock
+
+import numpy as np
+
+from python_app.hardware_full.multi_device_service import MultiDeviceLibreVnaService
+from python_app.hardware_full.switched_matrix_radar_service import SwitchedMatrixRadarService
+from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData
+from python_app.models.run_config_model import RunConfigModel
+from python_app.workflows.sequential_capture_workflow import SequentialCaptureSession
+
+_INNER_INPUTS = 4
+_INNER_OUTPUTS = 2
+_POINTS = 8
+
+
+class _FakeInnerMatrixRadar:
+    """Matrix radar stub emitting the canonical 2x4 combo set per acquisition."""
+
+    def __init__(self) -> None:
+        self.acquire_count = 0
+
+    def open(self) -> None:
+        pass
+
+    def close(self) -> None:
+        pass
+
+    def configure(self, sweep) -> None:
+        pass
+
+    def recover(self) -> None:
+        pass
+
+    def acquire_collection(self, collection_id: int = 1) -> SweepCollection:
+        self.acquire_count += 1
+        frequency_hz = np.linspace(1e6, 2e6, _POINTS, dtype=np.float32)
+        traces = [
+            TraceData(
+                combo=ComboKey(input=input_pos, output=output_pos),
+                frequency_hz=frequency_hz,
+                s11=np.full(_POINTS, complex(self.acquire_count, 0), dtype=np.complex64),
+                s21=np.full(_POINTS, complex(input_pos, output_pos), dtype=np.complex64),
+            )
+            for output_pos in range(_INNER_OUTPUTS)
+            for input_pos in range(_INNER_INPUTS)
+        ]
+        return SweepCollection(
+            collection_id=int(collection_id),
+            monotonic_ns=time.monotonic_ns(),
+            traces=traces,
+        )
+
+
+class _FakeSwitch:
+    """Switch stub recording every position it is driven to."""
+
+    def __init__(self, positions: int) -> None:
+        self.positions = positions
+        self.switched_to: list[int] = []
+
+    def open(self) -> None:
+        pass
+
+    def close(self) -> None:
+        pass
+
+    def position_count(self) -> int:
+        return self.positions
+
+    def switch_to(self, position: int) -> None:
+        self.switched_to.append(int(position))
+
+
+def _switched_service(input_steps: int = 3) -> tuple[SwitchedMatrixRadarService, _FakeInnerMatrixRadar, _FakeSwitch]:
+    inner = _FakeInnerMatrixRadar()
+    input_switch = _FakeSwitch(input_steps)
+    service = SwitchedMatrixRadarService(
+        inner=inner,
+        output_switch=None,
+        input_switch=input_switch,
+        inner_output_positions=_INNER_OUTPUTS,
+        inner_input_positions=_INNER_INPUTS,
+        settling_ms=0,
+    )
+    return service, inner, input_switch
+
+
+class SwitchedMatrixComboAcquisitionTest(unittest.TestCase):
+    """acquire_combo_collection must acquire exactly one physical switch step."""
+
+    def test_acquires_only_the_step_containing_the_combo(self) -> None:
+        service, inner, input_switch = _switched_service(input_steps=3)
+
+        # Widened input 9 lives in physical step 9 // 4 = 2.
+        collection = service.acquire_combo_collection(input_pos=9, output_pos=1)
+
+        self.assertEqual(inner.acquire_count, 1)
+        self.assertEqual(input_switch.switched_to, [2])
+        self.assertEqual(len(collection.traces), _INNER_INPUTS * _INNER_OUTPUTS)
+        combos = {(trace.combo.input, trace.combo.output) for trace in collection.traces}
+        self.assertIn((9, 1), combos)
+        # Every trace of the step is remapped into the widened axis of that step.
+        self.assertEqual(
+            combos,
+            {(2 * _INNER_INPUTS + i, o) for i in range(_INNER_INPUTS) for o in range(_INNER_OUTPUTS)},
+        )
+
+    def test_rejects_out_of_range_combo(self) -> None:
+        service, _inner, _input_switch = _switched_service(input_steps=3)
+        with self.assertRaises(ValueError):
+            service.acquire_combo_collection(input_pos=12, output_pos=0)
+        with self.assertRaises(ValueError):
+            service.acquire_combo_collection(input_pos=0, output_pos=2)
+
+    def test_full_collection_still_covers_widened_matrix_in_canonical_order(self) -> None:
+        service, inner, input_switch = _switched_service(input_steps=3)
+
+        collection = service.acquire_collection(collection_id=7)
+
+        self.assertEqual(inner.acquire_count, 3)
+        self.assertEqual(input_switch.switched_to, [0, 1, 2])
+        expected_combos = [
+            (input_pos, output_pos)
+            for output_pos in range(_INNER_OUTPUTS)
+            for input_pos in range(3 * _INNER_INPUTS)
+        ]
+        self.assertEqual(
+            [(trace.combo.input, trace.combo.output) for trace in collection.traces],
+            expected_combos,
+        )
+
+    def test_each_switch_step_stamps_its_traces_with_its_own_capture_window(self) -> None:
+        # The whole point of per-trace timing: three switch steps are measured one
+        # after another, so their traces must NOT all share the collection window.
+        service, _inner, _input_switch = _switched_service(input_steps=3)
+
+        collection = service.acquire_collection(collection_id=7)
+
+        windows_by_step: dict[int, set[tuple[int, int]]] = {}
+        for trace in collection.traces:
+            step = int(trace.combo.input) // _INNER_INPUTS
+            windows_by_step.setdefault(step, set()).add(
+                (int(trace.capture_start_ns), int(trace.capture_end_ns))
+            )
+
+        self.assertEqual(sorted(windows_by_step), [0, 1, 2])
+        for step, windows in windows_by_step.items():
+            self.assertEqual(len(windows), 1, f"step {step} traces disagree on their window")
+            start_ns, end_ns = next(iter(windows))
+            self.assertGreater(start_ns, 0)
+            self.assertGreaterEqual(end_ns, start_ns)
+            # Each step's window sits inside the collection's.
+            self.assertGreaterEqual(start_ns, collection.capture_start_ns)
+            self.assertLessEqual(end_ns, collection.capture_end_ns)
+
+        # Steps are strictly ordered in time — the whole reason the collection-level
+        # window cannot stand in for a per-combo timestamp.
+        step_starts = [next(iter(windows_by_step[step]))[0] for step in sorted(windows_by_step)]
+        self.assertEqual(step_starts, sorted(step_starts))
+        self.assertGreater(len(set(step_starts)), 1)
+
+
+class ManualComboCaptureUsesTargetedAcquisitionTest(unittest.TestCase):
+    """The per-combo capture session must not sweep the full widened matrix."""
+
+    @staticmethod
+    def _switched_mock_config() -> RunConfigModel:
+        config = RunConfigModel()
+        config.radar.model = RunConfigModel.LIBREVNA_MULTI_MODEL
+        config.radar.driver_mode = "mock"
+        config.radar.multi_device.slave_serials = ["SLAVE_A", "SLAVE_B"]
+        config.radar.multi_device.input_switch_positions = 3
+        config.apply_device_model_constraints()
+        return config
+
+    def test_manual_capture_runs_one_inner_collection_per_median_sweep(self) -> None:
+        config = self._switched_mock_config()
+        session = SequentialCaptureSession(
+            config=config,
+            kind="s21_calibration",
+            set_name="targeted_test",
+            median_sweep_count=2,
+        )
+        with mock.patch.object(
+            MultiDeviceLibreVnaService,
+            "acquire_collection",
+            autospec=True,
+            side_effect=MultiDeviceLibreVnaService.acquire_collection,
+        ) as inner_acquire:
+            session.open()
+            try:
+                trace = session.capture_current_combo()
+            finally:
+                session.close()
+
+        first_combo = config.combos[0]
+        self.assertEqual(
+            (trace.combo.input, trace.combo.output),
+            (first_combo.input, first_combo.output),
+        )
+        # 2 median sweeps of ONE physical step — not 2 x 3 full-matrix steps.
+        self.assertEqual(inner_acquire.call_count, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

python_app/workflows/multi_radar_capture_workflow.py

@@ -19,6 +19,7 @@ from python_app.workflows.radar_config_variants import RadarConfigVariant
 from python_app.workflows.sequential_capture_workflow import (
     MATRIX_RADAR_MANUAL_CAPTURE_KINDS,
     SequentialCaptureState,
+    acquire_matrix_combo_collection,
     combine_collections_via_median,
     combine_traces_via_median,
     select_trace_for_combo,
@@ -192,20 +193,27 @@ class MultiRadarSequentialCaptureSession:
                 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)
-                collections: list[SweepCollection] = []
-                for _ in range(self._median_sweep_count):
-                    collection = self._radar.acquire_collection(collection_id=1)
-                    if not collection.traces:
-                        raise RuntimeError(
-                            f"Matrix radar variant {variant.display_name} returned no traces"
-                        )
-                    collections.append(collection)
                 if self._manual_matrix_radar_capture:
-                    per_sweep_traces = [select_trace_for_combo(collection, combo) for collection in collections]
+                    # Only this combo is kept, so acquire the smallest collection
+                    # that contains it instead of the full (switch-widened) matrix.
+                    per_sweep_traces = [
+                        select_trace_for_combo(
+                            acquire_matrix_combo_collection(self._radar, combo), combo
+                        )
+                        for _ in range(self._median_sweep_count)
+                    ]
                     trace = combine_traces_via_median(per_sweep_traces)
                     pending_traces_by_radar_key[variant.radar_key] = [trace]
                     display_traces.append(trace)
                 else:
+                    collections: list[SweepCollection] = []
+                    for _ in range(self._median_sweep_count):
+                        collection = self._radar.acquire_collection(collection_id=1)
+                        if not collection.traces:
+                            raise RuntimeError(
+                                f"Matrix radar variant {variant.display_name} returned no traces"
+                            )
+                        collections.append(collection)
                     combined_collection = combine_collections_via_median(collections)
                     pending_traces_by_radar_key[variant.radar_key] = list(combined_collection.traces)
                     display_traces.append(combined_collection.traces[-1])
@@ -224,6 +232,7 @@ class MultiRadarSequentialCaptureSession:
                     time.sleep(self._base_config.runtime.settling_ms / 1000.0)
                 sweep_traces: list[TraceData] = []
                 for _ in range(self._median_sweep_count):
+                    sweep_start_ns = time.monotonic_ns()
                     sweep = self._radar.acquire()
                     sweep_traces.append(
                         TraceData(
@@ -231,6 +240,8 @@ class MultiRadarSequentialCaptureSession:
                             frequency_hz=np.asarray(sweep.x, dtype=np.float32),
                             s11=np.asarray(sweep.trace("s11"), dtype=np.complex64),
                             s21=np.asarray(sweep.trace("s21"), dtype=np.complex64),
+                            capture_start_ns=sweep_start_ns,
+                            capture_end_ns=time.monotonic_ns(),
                         )
                     )
                 trace = combine_traces_via_median(sweep_traces)

python_app/workflows/sequential_capture_workflow.py

@@ -156,20 +156,27 @@ class SequentialCaptureSession:
             raise RuntimeError("Capture session is already complete")
 
         if self._is_matrix_radar:
-            collections: list[SweepCollection] = []
-            for _ in range(self._median_sweep_count):
-                collection = self._radar.acquire_collection(collection_id=1)
-                if not collection.traces:
-                    raise RuntimeError("Matrix radar capture returned no traces")
-                collections.append(collection)
             if self._manual_matrix_radar_capture:
-                per_sweep_traces = [select_trace_for_combo(collection, combo) for collection in collections]
+                # Only this combo is kept, so acquire the smallest collection that
+                # contains it instead of the full (switch-widened) matrix.
+                per_sweep_traces = [
+                    select_trace_for_combo(
+                        acquire_matrix_combo_collection(self._radar, combo), combo
+                    )
+                    for _ in range(self._median_sweep_count)
+                ]
                 trace = combine_traces_via_median(per_sweep_traces)
                 self._traces.append(trace)
                 self._next_index += 1
                 logger.debug("Captured matrix combo input=%d output=%d", combo.input, combo.output)
                 return trace
 
+            collections: list[SweepCollection] = []
+            for _ in range(self._median_sweep_count):
+                collection = self._radar.acquire_collection(collection_id=1)
+                if not collection.traces:
+                    raise RuntimeError("Matrix radar capture returned no traces")
+                collections.append(collection)
             combined_collection = combine_collections_via_median(collections)
             self._traces.extend(combined_collection.traces)
             self._next_index = len(self._combos)
@@ -195,6 +202,7 @@ class SequentialCaptureSession:
 
         sweep_traces: list[TraceData] = []
         for _ in range(self._median_sweep_count):
+            sweep_start_ns = time.monotonic_ns()
             sweep = self._radar.acquire()
             sweep_traces.append(
                 TraceData(
@@ -202,6 +210,8 @@ class SequentialCaptureSession:
                     frequency_hz=np.asarray(sweep.x, dtype=np.float32),
                     s11=np.asarray(sweep.trace("s11"), dtype=np.complex64),
                     s21=np.asarray(sweep.trace("s21"), dtype=np.complex64),
+                    capture_start_ns=sweep_start_ns,
+                    capture_end_ns=time.monotonic_ns(),
                 )
             )
         trace = combine_traces_via_median(sweep_traces)
@@ -301,6 +311,32 @@ class SequentialCaptureSession:
         return self._combos[self._next_index]
 
 
+def acquire_matrix_combo_collection(
+    radar: MatrixRadarService,
+    combo: ComboModel,
+    collection_id: int = 1,
+) -> SweepCollection:
+    """Acquire the smallest matrix collection that contains one combo.
+
+    A switch-widened matrix radar (``SwitchedMatrixRadarService``) can acquire just
+    the physical switch step carrying the combo, which is several times faster than
+    the full matrix and keeps the per-combo capture UI responsive. Plain matrix
+    radars expose only full-matrix acquisition, so they fall back to it.
+    """
+    acquire_combo = getattr(radar, "acquire_combo_collection", None)
+    if callable(acquire_combo):
+        collection = acquire_combo(
+            input_pos=int(combo.input),
+            output_pos=int(combo.output),
+            collection_id=collection_id,
+        )
+    else:
+        collection = radar.acquire_collection(collection_id=collection_id)
+    if not collection.traces:
+        raise RuntimeError("Matrix radar capture returned no traces")
+    return collection
+
+
 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:
@@ -352,11 +388,16 @@ def combine_traces_via_median(traces: list[TraceData]) -> TraceData:
     s21_median = (
         np.median(s21_stack.real, axis=0) + 1j * np.median(s21_stack.imag, axis=0)
     ).astype(np.complex64)
+    # The median is built from every input sweep, so its window spans all of them.
+    capture_starts = [int(t.capture_start_ns) for t in traces if int(t.capture_start_ns) > 0]
+    capture_ends = [int(t.capture_end_ns) for t in traces if int(t.capture_end_ns) > 0]
     return TraceData(
         combo=ComboKey(input=int(combo.input), output=int(combo.output)),
         frequency_hz=np.asarray(first.frequency_hz, dtype=np.float32),
         s11=s11_median,
         s21=s21_median,
+        capture_start_ns=min(capture_starts) if capture_starts else 0,
+        capture_end_ns=max(capture_ends) if capture_ends else 0,
     )