Compare commits

..
Author SHA1 Message Date
Ayzen 7997abe2d9 added capture time for every sweep 2026-09-03 17:30:19 +03:00
Guriy 8a52431bd3 cart_firmware_settings_changed 2026-08-27 13:53:33 +03:00
Ayzen cc6d189d52 cart_firmware_added 2026-08-26 15:40:10 +03:00
Ayzen 6ada811c2f some fixes 2026-08-26 15:02:21 +03:00
39 changed files with 1508 additions and 135 deletions
+15 -4
View File
@@ -21,8 +21,8 @@ dist/
downloads/
eggs/
.eggs/
lib/
lib64/
/lib/
/lib64/
parts/
sdist/
var/
@@ -227,5 +227,16 @@ python_app/runtime
SHARE_INTERNET_TO_PI.md
CLAUDE.md
docs/
test_end_2/
/docs/
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*
@@ -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 {
@@ -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)) {
throw std::runtime_error("Unexpected trailing bytes in trace collection");
if (reader.remaining_bytes() < (sizeof(std::uint64_t) * 2U)) {
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;
}
@@ -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;
}
@@ -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;
}
@@ -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));
}
+5
View File
@@ -0,0 +1,5 @@
.pio
.vscode/.browse.c_cpp.db*
.vscode/c_cpp_properties.json
.vscode/launch.json
.vscode/ipch
+10
View File
@@ -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"
]
}
@@ -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 в измеренных
диапазонах.
@@ -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
+166
View File
@@ -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]);
}
@@ -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}")
@@ -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")
@@ -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}")
@@ -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)")
@@ -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()
+106 -42
View File
@@ -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):
"""Marshals log records from any thread onto the GUI thread for panel rendering.
class _PanelLogBuffer:
"""Thread-safe bounded buffer between logging handlers and the GUI flush timer.
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
signal hands the record across safely (the GPIO-button pattern).
A :class:`logging.Handler` can fire on a worker thread (readers, broadcaster)
at a very high rate — e.g. the USB RX threads while the free-running sweep
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),
getattr(record, "panel_once_key", None),
details if isinstance(details, str) else 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_bridge.record.connect(self._on_log_record)
self._log_panel_buffer = _PanelLogBuffer()
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:
"""Render one forwarded log record in the panel (always on the GUI thread)."""
if not hasattr(self, "_log_box"):
def _flush_log_panel_buffer(self) -> None:
"""Render every buffered log record into the panel as one batched insert."""
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,
text,
details=details if isinstance(details, str) else None,
once_key=once_key if isinstance(once_key, str) else None,
)
entry_htmls: list[str] = []
if dropped_count:
entry_htmls.append(
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(
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)
def _render_log_entry_html(self, level: str, text: str, details: str | None = None) -> str:
"""Render one log entry as the panel's HTML block."""
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:
@@ -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()
except Exception as exc: # noqa: BLE001
self._on_capture_combo_failed(session, exc)
return
self._record_preprocess_capture(session, capture_result)
try:
capture_result = session.capture_current_combo()
except Exception as exc: # noqa: BLE001
self._on_capture_combo_failed(session, exc)
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:
capture_result = session.capture_current_combo()
except Exception as exc: # noqa: BLE001
self._on_capture_combo_failed(session, exc)
return
self._record_preprocess_capture(session, capture_result)
try:
while not session.is_complete():
try:
capture_result = session.capture_current_combo()
except Exception as exc: # noqa: BLE001
self._on_capture_combo_failed(session, exc)
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:
+13 -6
View File
@@ -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._save_sequence_button.setEnabled(bool(can_finalize))
self._capture_all_button.setEnabled(bool(can_capture_all))
self._abort_button.setEnabled(True)
self._undo_last_button.setEnabled(bool(can_undo) and actions_enabled)
self._save_sequence_button.setEnabled(bool(can_finalize) and actions_enabled)
self._capture_all_button.setEnabled(bool(can_capture_all) and actions_enabled)
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."""
@@ -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")
@@ -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."""
@@ -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
+16 -2
View File
@@ -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
@@ -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()
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
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)
)
for trace in self._acquire_step_traces(out_k, in_k, collection_id):
slots[trace.combo.output * total_inputs + trace.combo.input] = trace
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,
+11 -1
View File
@@ -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)
+17 -2
View File
@@ -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(
@@ -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(
{
@@ -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
+13 -1
View File
@@ -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)
+15
View File
@@ -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",
+4
View File
@@ -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)),
)
)
+14 -1
View File
@@ -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(
{
+48
View File
@@ -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()
+42 -3
View File
@@ -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:
@@ -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()
@@ -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)
@@ -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,
)