Mirrors/NawfalMotii79-PLFM_RADAR
1
0
Fork 0
mirror of https://github.com/NawfalMotii79/PLFM_RADAR.git synced 2026-06-10 23:41:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(usb): C-9 — GUI bulk-frame parser for FT2232H + clamp inert flag bits

Browse Source 
The GUI's radar_protocol.py parsed 11-byte legacy packets only. The
production board (50T, USB_MODE=1) emits ~35 KB bulk frames from
usb_data_interface_ft2232h.v, so the legacy parser saw a random walk
of false 11-byte boundaries through bulk data — no usable display on
production hardware.

Bulk parser added (radar_protocol.py):
- parse_bulk_frame validates header, reserved bits, n_range=512,
  n_doppler=32, footer-at-flag-derived-offset; unpacks range_profile
  / doppler_mag / cfar_dense per the format-flags byte.
- find_bulk_frame_boundaries is the bulk counterpart of
  find_packet_boundaries; status packets (0xBB) handled in the same
  stream since FT2232H emits them too.
- RadarAcquisition dispatches on isinstance(conn, FT2232HConnection):
  bulk path skips the per-sample state machine and fills RadarFrame
  in one shot. FT601 / 200T keeps legacy 11-byte (USB 3.0 has 50x
  bandwidth headroom; per-sample format is correct and already works).
- RadarFrame.mag_only flag carries the wire's mag_only bit so
  downstream consumers can skip I/Q panels cleanly.
- FT2232HConnection._mock_read now emits synthetic bulk frames
  (was misleading legacy 11-byte).

RTL alignment (AUDIT-C9 RTL stub option):
- usb_data_interface_ft2232h.v header no longer promises the
  unimplemented mag_only=0 (full-I/Q) and sparse_det=1 paths;
  explicit INERT FLAGS note distinguishes the two reasons:
    * Full-I/Q is constrained by hardware — needs ~28-BRAM18 I/Q
      buffer (50T currently 78% BRAM utilised after FFT IP) AND
      USB 2.0 bandwidth (12.21 MB/s vs 8 MB/s conservative budget).
    * Sparse-list is feasible — smaller than dense for typical
      scenes (<341 detections), ~1 BRAM18 cost. Just unimplemented
      RTL work (small list BRAM + new WR_DETECT_SPARSE state).
- New SIMULATION-only assertion fires if stream_mag_only ever
  becomes 0 or stream_sparse_det ever becomes 1 — backstop for
  any future regression that bypasses the host-register clamp.
- radar_system_top.v opcode 0x04 force-clamps mag_only=1 and
  sparse_det=0 in host_stream_control when USB_MODE=1, so a
  Custom-Command host write can't push the FPGA into a wire-format
  vs FSM divergence.

Bandwidth math (verified for 27c9c22+):
  Frame rate = 1 / (16x167 us + 175.4 us + 16x175 us) = ~178 fps
  Mag-only frame = 8+1024+32768+2048+1 = 35849 B = 6.38 MB/s
  FT2232H 245-Sync-FIFO sustained budget (FTDI AN_232B-04
  conservative): 8 MB/s. Headroom 20%.

Tests: test_GUI_V65_Tk.py TestBulkFrameParser — 18 new cases covering
round-trip per stream-flag combo, header/footer/n_range/n_doppler/
reserved-bit/truncation rejection, multi-frame boundaries, bulk+status
mixed streams, byte-drop resync, dispatch-by-connection-type,
ingest-to-RadarFrame end-to-end. GUI 117/117 PASS, v7 83/83 PASS,
FPGA quick regression 29/29 PASS, ruff clean.

Refs: AUDIT-C9 (GUI parses legacy 11-byte vs FT2232H bulk).
Follow-ups (separate patches):
  - Sparse-detection write FSM (~1 BRAM18 + ~100 RTL lines).
    Bandwidth- and memory-feasible; just unimplemented work.
  - Full-I/Q write FSM. Constrained: needs ~28-BRAM18 I/Q buffer
    AND USB 2.0 bandwidth headroom (50T post-FFT-IP at 78% BRAM).
This commit is contained in:
Jason
2026-04-29 15:12:04 +05:45
parent 24ef5e7251
commit 79a9353456
4 changed files with 652 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
9_Firmware/9_2_FPGA/radar_system_top.v
+18 -1
View File
@@ -1018,7 +1018,24 @@ always @(posedge clk_100m_buf or negedge sys_reset_n) begin
8'h01: host_radar_mode <= usb_cmd_value[1:0]; 8'h01: host_radar_mode <= usb_cmd_value[1:0];
8'h02: host_trigger_pulse <= 1'b1; 8'h02: host_trigger_pulse <= 1'b1;
8'h03: host_detect_threshold <= usb_cmd_value; 8'h03: host_detect_threshold <= usb_cmd_value;
8'h04: host_stream_control <= usb_cmd_value[5:0]; // AUDIT-C9: stream_control bits [3] (mag_only) and [4]
// (sparse_det) are documented in the FT2232H bulk-frame
// header but the write FSM does not implement the alternate
// encodings yet (see usb_data_interface_ft2232h.v "INERT
// FLAGS" note). Force-clamp them to the only encodings the
// FSM actually emits so a host write of 0x04 cannot create
// a wire-format vs FSM divergence on the production board.
8'h04: begin
if (USB_MODE == 1) begin
// FT2232H production: mag_only stuck at 1, sparse_det stuck at 0.
host_stream_control <= {usb_cmd_value[5],
1'b0, // sparse_det
1'b1, // mag_only
usb_cmd_value[2:0]}; // stream r/d/c
end else begin
host_stream_control <= usb_cmd_value[5:0];
end
end
// Gap 2: chirp timing configuration // Gap 2: chirp timing configuration
8'h10: host_long_chirp_cycles <= usb_cmd_value; 8'h10: host_long_chirp_cycles <= usb_cmd_value;
8'h11: host_long_listen_cycles <= usb_cmd_value; 8'h11: host_long_listen_cycles <= usb_cmd_value;
9_Firmware/9_2_FPGA/usb_data_interface_ft2232h.v
+59 -14
View File
@@ -20,21 +20,42 @@
* [If stream_range (bit 0):] * [If stream_range (bit 0):]
* Next 1024 bytes: Range profile, 512 × 16-bit magnitude, MSB first * Next 1024 bytes: Range profile, 512 × 16-bit magnitude, MSB first
* *
* [If stream_doppler (bit 1) AND mag_only (bit 3):] * [If stream_doppler (bit 1):]
* Next 32768 bytes: Doppler magnitude, 512×32 × 16-bit, row-major, MSB first * Next 32768 bytes: Doppler magnitude, 512×32 × 16-bit, row-major, MSB first
* *
* [If stream_doppler (bit 1) AND NOT mag_only:] * [If stream_cfar (bit 2):]
* Next 65536 bytes: Doppler I/Q, 512×32 × 32-bit (I16,Q16), row-major, MSB first
*
* [If stream_cfar (bit 2) AND NOT sparse_det (bit 4):]
* Next 2048 bytes: Detection flags, 512×32 bits packed into bytes, MSB-first bit order * Next 2048 bytes: Detection flags, 512×32 bits packed into bytes, MSB-first bit order
* *
* [If stream_cfar (bit 2) AND sparse_det (bit 4):]
* Next 2 bytes: Detection count N (16-bit, MSB first)
* Next N×6 bytes: Each = {range_bin[16], doppler_bin[16], magnitude[16]}, MSB first
*
* Last byte: 0x55 (frame end footer) * Last byte: 0x55 (frame end footer)
* *
* INERT FLAGS — mag_only (bit 3) and sparse_det (bit 4) (AUDIT-C9):
* The wire format byte 1 reserves these two bits for future encodings:
* - mag_only=0 was meant to switch the doppler section to 65536 B
* full-I/Q (16-bit I + 16-bit Q per cell, row-major, MSB first).
* - sparse_det=1 was meant to switch the CFAR section to a
* variable-length list: 2 B count N + N×6 B (range, doppler, mag).
* Neither encoding is implemented in the write FSM below — the FSM
* always emits 32768 B mag and 2048 B dense bitmap regardless of the
* flag bits. To eliminate the foot-gun, `radar_system_top.v` opcode
* 0x04 force-clamps mag_only=1 and sparse_det=0 in `host_stream_control`
* when USB_MODE=1. A SIMULATION-only assertion at the bottom of this
* module fires if either bit ever leaves its clamped value, in case a
* future patch adds a path that bypasses the host register clamp.
*
* Reasons differ between the two:
* - Full-I/Q is constrained by FPGA resources: it needs a new
* ~28-BRAM18 I/Q buffer (16384 cells × 32-bit) which may not fit
* on the 50T (currently ~78% BRAM18 utilisation after wiring the
* Xilinx FFT IP). USB 2.0 bandwidth is also tight: 12.21 MB/s vs
* the conservative 8 MB/s sustained budget. Both gating items.
* - Sparse-list is feasible — bandwidth-wise it's smaller than the
* dense bitmap for any frame with fewer than ~341 detections
* (typical scenes produce 10-200), and memory-wise it costs
* ~1 BRAM18 with MAX_DETECTIONS=256. The absence is just
* unimplemented RTL work (a small detection-list BRAM + a new
* WR_DETECT_SPARSE FSM state), not a hardware constraint.
* See the open-defects ledger for the follow-up work items.
*
* Status packet (FPGA→Host): 26 bytes (unchanged from legacy) * Status packet (FPGA→Host): 26 bytes (unchanged from legacy)
* Byte 0: 0xBB (status header) * Byte 0: 0xBB (status header)
* Bytes 1-24: 6 × 32-bit status words, MSB first * Bytes 1-24: 6 × 32-bit status words, MSB first
@@ -54,13 +75,13 @@
* Written in clk domain from range_valid events. * Written in clk domain from range_valid events.
* - Detection flag buffer: 512×32 = 16384 bits = 2048 bytes (~1 BRAM18) * - Detection flag buffer: 512×32 = 16384 bits = 2048 bytes (~1 BRAM18)
* Written in clk domain from cfar_valid events. * Written in clk domain from cfar_valid events.
* - Doppler I/Q BRAM: 16384 × 32-bit = 64 KB (~28 BRAM18) — only when mag_only=0
* NOTE: For 50T (75 BRAM18 total), I/Q mode may not fit alongside processing
* chain BRAMs. Default to mag_only=1. I/Q mode is a stretch goal.
* *
* BANDWIDTH BUDGET (mag_only=1, all streams): * BANDWIDTH BUDGET (current production: mag_only=1, all streams):
* Header: 8 B + Range: 1024 B + Doppler: 32768 B + CFAR: 2048 B + Footer: 1 B * Header: 8 B + Range: 1024 B + Doppler: 32768 B + CFAR: 2048 B + Footer: 1 B
* = 35,849 bytes/frame × 178 fps = 6.38 MB/s (80% of USB 2.0 Hi-Speed 8 MB/s) * = 35,849 bytes/frame × ~178 fps = 6.38 MB/s
* FT2232H 245-Sync-FIFO sustained budget ~8 MB/s conservative (FTDI
* AN_232B-04). 80% utilisation; full-I/Q (12.21 MB/s) would not fit at
* the conservative budget and is why mag_only is force-clamped to 1.
* *
* CDC STRATEGY: * CDC STRATEGY:
* - Frame data: Written to dual-port BRAM at 100 MHz, read at 60 MHz (inherently CDC-safe) * - Frame data: Written to dual-port BRAM at 100 MHz, read at 60 MHz (inherently CDC-safe)
@@ -1025,4 +1046,28 @@ always @(posedge ft_clk or negedge ft_reset_n) begin
end end
`endif `endif
// ============================================================================
// AUDIT-C9: inert-flag checker (simulation only)
//
// stream_mag_only and stream_sparse_det are documented in the wire format
// but the write FSM does not act on them — see the "INERT FLAGS" note in
// the module header. radar_system_top.v opcode 0x04 force-clamps these
// bits when USB_MODE=1 so production firmware cannot reach an unsupported
// state. This checker is the backstop: it fires `[ASSERT FAIL]` if either
// bit ever escapes its clamped value, catching any future patch that
// bypasses the host register clamp (e.g. a different opcode that writes
// stream_control directly, or a stream_control source other than the
// host). Synthesis-inert.
// ============================================================================
`ifdef SIMULATION
always @(posedge clk) begin
if (reset_n) begin
if (stream_mag_only !== 1'b1)
$display("[ASSERT FAIL] AUDIT-C9: stream_mag_only=0; full-I/Q write FSM not implemented");
if (stream_sparse_det !== 1'b0)
$display("[ASSERT FAIL] AUDIT-C9: stream_sparse_det=1; sparse-list write FSM not implemented");
end
end
`endif
endmodule endmodule
9_Firmware/9_3_GUI/radar_protocol.py
+322 -61
View File
@@ -5,15 +5,39 @@ AERIS-10 Radar Protocol Layer
Pure-logic module for USB packet parsing and command building. Pure-logic module for USB packet parsing and command building.
No GUI dependencies — safe to import from tests and headless scripts. No GUI dependencies — safe to import from tests and headless scripts.
USB Interface: FT2232H USB 2.0 (8-bit, 50T production board) via pyftdi USB transports + wire formats (these intentionally diverge):
FT601 USB 3.0 (32-bit, 200T premium board) via ftd3xx
USB Packet Protocol (11-byte): FT2232H USB 2.0 (50T production board, USB_MODE=1, default)
TX (FPGA→Host): Bulk per-frame format from `usb_data_interface_ft2232h.v`. One header
Data packet: [0xAA] [range_q 2B] [range_i 2B] [dop_re 2B] [dop_im 2B] [det 1B] [0x55] + variable-length sections + footer per Doppler frame. The bulk format
Status packet: [0xBB] [status 6x32b] [0x55] exists because USB 2.0's ~8 MB/s sustained ceiling cannot carry the
RX (Host→FPGA): production frame rate (~178 fps x 35849 B = 6.4 MB/s) at per-sample
Command: 4 bytes received sequentially {opcode, addr, value_hi, value_lo} granularity. Wire layout:
[0xAA][flags 1B][frame_num 2B][n_range 2B][n_doppler 2B]
[range_profile 1024 B if flags.stream_range]
[doppler_mag 32768 B if flags.stream_doppler] # mag_only=1 only
[cfar_dense 2048 B if flags.stream_cfar] # sparse_det=0 only
[0x55]
Production FPGA today only emits mag_only=1 + dense-bitmap CFAR; the
flag bits for full-I/Q (mag_only=0) and sparse-detection-list
(sparse_det=1) are reserved for a future RTL extension and currently
force-clamped to 1 and 0 respectively in `radar_system_top.v` opcode
0x04 handler when USB_MODE=1.
FT601 USB 3.0 (200T premium board, USB_MODE=0)
Per-sample 11-byte legacy format from `usb_data_interface.v`. USB 3.0
has ~50x the bandwidth headroom (~360 MB/s practical), so the lighter
per-sample format is fine and offers easier resync after byte drops.
Wire layout (per sample, 16384 samples per frame):
[0xAA][range_q 2B][range_i 2B][dop_re 2B][dop_im 2B][det 1B][0x55]
where det byte = {frame_start, 6'b0, cfar_detection}.
Status (both transports): [0xBB][6x32b status words][0x55] = 26 B.
RX (Host → FPGA, both transports)
4 bytes per command: {opcode[7:0], addr[7:0], value[15:8], value[7:0]}.
The GUI parser handles both formats; `RadarAcquisition` dispatches on
connection type (FT2232HConnection → bulk; FT601Connection → legacy).
""" """
import struct import struct
@@ -49,6 +73,25 @@ NUM_CELLS = NUM_RANGE_BINS * NUM_DOPPLER_BINS # 16384
WATERFALL_DEPTH = 64 WATERFALL_DEPTH = 64
# AUDIT-C9: FT2232H bulk-frame wire format constants. Mirrors
# usb_data_interface_ft2232h.v; if the RTL header changes, update both sides.
BULK_FRAME_HEADER_SIZE = 8 # AA + flags + fnum2 + nr2 + nd2
BULK_RANGE_SECTION_BYTES = NUM_RANGE_BINS * 2 # 512 x 2 = 1024
BULK_DOPPLER_MAG_BYTES = NUM_CELLS * 2 # 16384 x 2 = 32768
BULK_DETECT_DENSE_BYTES = NUM_CELLS // 8 # 16384 / 8 = 2048
BULK_FOOTER_SIZE = 1
BULK_FRAME_MIN_SIZE = BULK_FRAME_HEADER_SIZE + BULK_FOOTER_SIZE # 9
BULK_FRAME_MAX_SIZE = (BULK_FRAME_HEADER_SIZE + BULK_RANGE_SECTION_BYTES
+ BULK_DOPPLER_MAG_BYTES + BULK_DETECT_DENSE_BYTES
+ BULK_FOOTER_SIZE) # 35849
# Bulk-frame format flag bits (matches stream_ctrl_sync_1 layout in RTL).
BULK_FLAG_STREAM_RANGE = 0x01
BULK_FLAG_STREAM_DOPPLER = 0x02
BULK_FLAG_STREAM_CFAR = 0x04
BULK_FLAG_MAG_ONLY = 0x08 # Forced 1 by RTL; full-I/Q write FSM not implemented.
BULK_FLAG_SPARSE_DET = 0x10 # Forced 0 by RTL; sparse-list write FSM not implemented.
class Opcode(IntEnum): class Opcode(IntEnum):
"""Host register opcodes — must match radar_system_top.v case(usb_cmd_opcode). """Host register opcodes — must match radar_system_top.v case(usb_cmd_opcode).
@@ -134,6 +177,10 @@ class RadarFrame:
default_factory=lambda: np.zeros(NUM_RANGE_BINS, dtype=np.float64)) default_factory=lambda: np.zeros(NUM_RANGE_BINS, dtype=np.float64))
detection_count: int = 0 detection_count: int = 0
frame_number: int = 0 frame_number: int = 0
# AUDIT-C9: True when this frame came from FT2232H bulk format with
# mag_only=1 (the only mode FPGA emits today). I/Q arrays will be zero;
# `magnitude` carries the per-cell Manhattan magnitude from the FPGA.
mag_only: bool = False
@dataclass @dataclass
@@ -316,6 +363,164 @@ class RadarProtocol:
i += 1 i += 1
return packets return packets
# ----------------------------------------------------------------
# AUDIT-C9: FT2232H bulk-frame parsing (production board path)
# ----------------------------------------------------------------
@staticmethod
def _bulk_frame_size_from_flags(flags: int) -> int:
"""Compute the on-wire size of a bulk frame from its flags byte.
Tracks the FPGA write FSM in usb_data_interface_ft2232h.v: header
(8 B) + per-stream payload + footer (1 B). The mag_only and
sparse_det bits are documented in the wire format but the FPGA
write FSM does not implement the alternate encodings yet — it
always emits 32768 B mag and 2048 B dense bitmap. The host-side
register handler in radar_system_top.v force-clamps these flags
when USB_MODE=1, so any frame the parser sees in production will
have mag_only=1 and sparse_det=0.
"""
size = BULK_FRAME_HEADER_SIZE
if flags & BULK_FLAG_STREAM_RANGE:
size += BULK_RANGE_SECTION_BYTES
if flags & BULK_FLAG_STREAM_DOPPLER:
size += BULK_DOPPLER_MAG_BYTES
if flags & BULK_FLAG_STREAM_CFAR:
size += BULK_DETECT_DENSE_BYTES
size += BULK_FOOTER_SIZE
return size
@staticmethod
def parse_bulk_frame(raw: bytes, offset: int = 0) -> dict[str, Any] | None:
"""Parse one FT2232H bulk frame starting at `offset`.
Returns a dict with keys: frame_number, flags, n_range, n_doppler,
range_profile (np.ndarray | None), doppler_mag (np.ndarray | None,
shape n_rangexn_doppler), cfar_dense (np.ndarray | None, shape
n_rangexn_doppler, uint8 0/1), and frame_size (total bytes consumed
including header + sections + footer). Returns None on any
structural error (bad header/footer, wrong bin counts, reserved
bits set, unimplemented flag combo).
"""
n = len(raw)
if n - offset < BULK_FRAME_MIN_SIZE:
return None
if raw[offset] != HEADER_BYTE:
return None
flags = raw[offset + 1]
# Reserved high bits must be zero (FPGA emits {2'b00, 6-bit flags}).
if flags & 0xC0:
return None
# Production FPGA only emits mag_only=1 + dense bitmap. Any other
# encoding is either a corrupt frame or a future RTL revision the
# parser hasn't been updated for; reject so the caller can resync.
if (flags & BULK_FLAG_STREAM_DOPPLER) and not (flags & BULK_FLAG_MAG_ONLY):
return None
if (flags & BULK_FLAG_STREAM_CFAR) and (flags & BULK_FLAG_SPARSE_DET):
return None
frame_number = (raw[offset + 2] << 8) | raw[offset + 3]
n_range = (raw[offset + 4] << 8) | raw[offset + 5]
n_doppler = (raw[offset + 6] << 8) | raw[offset + 7]
if n_range != NUM_RANGE_BINS or n_doppler != NUM_DOPPLER_BINS:
return None
size = RadarProtocol._bulk_frame_size_from_flags(flags)
if n - offset < size:
return None
if raw[offset + size - 1] != FOOTER_BYTE:
return None
cursor = offset + BULK_FRAME_HEADER_SIZE
range_profile = None
doppler_mag = None
cfar_dense = None
if flags & BULK_FLAG_STREAM_RANGE:
range_profile = np.frombuffer(
raw, dtype=">u2", count=n_range, offset=cursor,
).astype(np.uint16, copy=True)
cursor += BULK_RANGE_SECTION_BYTES
if flags & BULK_FLAG_STREAM_DOPPLER:
doppler_mag = np.frombuffer(
raw, dtype=">u2", count=n_range * n_doppler, offset=cursor,
).astype(np.uint16, copy=True).reshape(n_range, n_doppler)
cursor += BULK_DOPPLER_MAG_BYTES
if flags & BULK_FLAG_STREAM_CFAR:
packed = np.frombuffer(
raw, dtype=np.uint8, count=BULK_DETECT_DENSE_BYTES, offset=cursor,
)
# Each byte holds 8 cells, MSB-first bit order (matches FPGA
# WR_DETECT_DATA emission). np.unpackbits keeps that order.
cfar_dense = np.unpackbits(packed).reshape(n_range, n_doppler)
cursor += BULK_DETECT_DENSE_BYTES
return {
"frame_number": frame_number,
"flags": flags,
"n_range": n_range,
"n_doppler": n_doppler,
"range_profile": range_profile,
"doppler_mag": doppler_mag,
"cfar_dense": cfar_dense,
"frame_size": size,
}
@staticmethod
def find_bulk_frame_boundaries(buf: bytes) -> list[tuple[int, int, str]]:
"""Scan a byte stream for FT2232H bulk frames and status packets.
Status packets (0xBB header, 26 B) are unchanged between transports
— the WR_STATUS_SEND state in usb_data_interface_ft2232h.v emits the
same layout as the legacy FT601 path. Bulk data frames (0xAA header)
are variable length per `_bulk_frame_size_from_flags`.
Returns a list of (start, end, ptype) tuples like
find_packet_boundaries, where ptype is "data" or "status". On a
false header (any structural mismatch) the cursor advances by 1 and
keeps scanning, mirroring the legacy parser's resync semantics.
"""
out: list[tuple[int, int, str]] = []
i = 0
n = len(buf)
while i < n:
b = buf[i]
if b == HEADER_BYTE:
# Need at least the 8-byte header to compute the frame size.
if n - i < BULK_FRAME_HEADER_SIZE:
break # partial header — caller keeps as residual
flags = buf[i + 1]
# Quick reject before the more expensive size compute. The
# full validation lives in parse_bulk_frame.
if flags & 0xC0:
i += 1
continue
size = RadarProtocol._bulk_frame_size_from_flags(flags)
if n - i < size:
break # partial frame — keep as residual
# Validate footer + bin counts before accepting the boundary.
if (buf[i + size - 1] == FOOTER_BYTE
and ((buf[i + 4] << 8) | buf[i + 5]) == NUM_RANGE_BINS
and ((buf[i + 6] << 8) | buf[i + 7]) == NUM_DOPPLER_BINS):
out.append((i, i + size, "data"))
i += size
else:
i += 1
elif b == STATUS_HEADER_BYTE:
end = i + STATUS_PACKET_SIZE
if end > n:
break
if buf[end - 1] == FOOTER_BYTE and buf[i + 1] == 0xFF:
out.append((i, end, "status"))
i = end
else:
i += 1
else:
i += 1
return out
# ============================================================================ # ============================================================================
# FT2232H USB 2.0 Connection (pyftdi, 245 Synchronous FIFO) # FT2232H USB 2.0 Connection (pyftdi, 245 Synchronous FIFO)
@@ -421,55 +626,57 @@ class FT2232HConnection:
return False return False
def _mock_read(self, size: int) -> bytes: def _mock_read(self, size: int) -> bytes:
""" """Generate one synthetic FT2232H bulk frame per call.
Generate synthetic 11-byte radar data packets for testing.
Emits packets in sequential FPGA order (range_bin 0..63, doppler_bin Mirrors `usb_data_interface_ft2232h.v` production behavior: mag-only
0..31 within each range bin) so that RadarAcquisition._ingest_sample() Doppler section + dense-bitmap CFAR, all three streams enabled
places them correctly. A target is injected near range bin 20, (matches `RP_STREAM_CTRL_DEFAULT = 6'b001_111`). A target is injected
Doppler bin 8. near range bin 20, Doppler bin 8 so dashboards have something to draw.
""" """
time.sleep(0.05) time.sleep(0.05)
self._mock_frame_num += 1 self._mock_frame_num += 1
flags = (BULK_FLAG_STREAM_RANGE | BULK_FLAG_STREAM_DOPPLER
| BULK_FLAG_STREAM_CFAR | BULK_FLAG_MAG_ONLY)
buf = bytearray() # Synthesize per-cell magnitudes once (vectorised).
num_packets = min(NUM_CELLS, size // DATA_PACKET_SIZE) rbins = np.arange(NUM_RANGE_BINS).reshape(-1, 1)
start_idx = getattr(self, '_mock_seq_idx', 0) dbins = np.arange(NUM_DOPPLER_BINS).reshape(1, -1)
noise = np.abs(self._mock_rng.normal(0, 50, size=(NUM_RANGE_BINS, NUM_DOPPLER_BINS)))
target_mask = (np.abs(rbins - 20) < 3) & (np.abs(dbins - 8) < 2)
mag = noise + target_mask * 12000.0
mag_u16 = np.clip(mag, 0, 65535).astype(np.uint16)
for n in range(num_packets): range_profile = np.clip(
idx = (start_idx + n) % NUM_CELLS np.abs(self._mock_rng.normal(0, 100, size=NUM_RANGE_BINS))
rbin = idx // NUM_DOPPLER_BINS + (np.abs(rbins.flatten() - 20) < 3) * 8000,
dbin = idx % NUM_DOPPLER_BINS 0, 65535,
).astype(np.uint16)
range_i = int(self._mock_rng.normal(0, 100)) det = (target_mask & (np.abs(dbins - 8) < 2) & (np.abs(rbins - 20) < 2)).astype(np.uint8)
range_q = int(self._mock_rng.normal(0, 100)) det_packed = np.packbits(det.flatten()) # MSB-first bit order matches FPGA
if abs(rbin - 20) < 3:
range_i += 5000
range_q += 3000
dop_i = int(self._mock_rng.normal(0, 50)) buf = bytearray(BULK_FRAME_MAX_SIZE)
dop_q = int(self._mock_rng.normal(0, 50)) buf[0] = HEADER_BYTE
if abs(rbin - 20) < 3 and abs(dbin - 8) < 2: buf[1] = flags & 0x3F # reserved high bits zero, matches RTL
dop_i += 8000 buf[2] = (self._mock_frame_num >> 8) & 0xFF
dop_q += 4000 buf[3] = self._mock_frame_num & 0xFF
buf[4] = (NUM_RANGE_BINS >> 8) & 0xFF
buf[5] = NUM_RANGE_BINS & 0xFF
buf[6] = (NUM_DOPPLER_BINS >> 8) & 0xFF
buf[7] = NUM_DOPPLER_BINS & 0xFF
cursor = BULK_FRAME_HEADER_SIZE
# Range profile (>u2 = big-endian uint16, matches FPGA MSB-first).
buf[cursor:cursor + BULK_RANGE_SECTION_BYTES] = range_profile.astype(">u2").tobytes()
cursor += BULK_RANGE_SECTION_BYTES
buf[cursor:cursor + BULK_DOPPLER_MAG_BYTES] = mag_u16.astype(">u2").tobytes()
cursor += BULK_DOPPLER_MAG_BYTES
buf[cursor:cursor + BULK_DETECT_DENSE_BYTES] = det_packed.tobytes()
cursor += BULK_DETECT_DENSE_BYTES
buf[cursor] = FOOTER_BYTE
detection = 1 if (abs(rbin - 20) < 2 and abs(dbin - 8) < 2) else 0 # `size` is the host's read budget; emit at most one frame per call
# (matches typical FT2232H driver semantics).
# Build compact 11-byte packet return bytes(buf[:min(size, BULK_FRAME_MAX_SIZE)])
pkt = bytearray()
pkt.append(HEADER_BYTE)
pkt += struct.pack(">h", np.clip(range_q, -32768, 32767))
pkt += struct.pack(">h", np.clip(range_i, -32768, 32767))
pkt += struct.pack(">h", np.clip(dop_i, -32768, 32767))
pkt += struct.pack(">h", np.clip(dop_q, -32768, 32767))
# Bit 7 = frame_start (sample_counter == 0), bit 0 = detection
det_byte = (detection & 0x01) | (0x80 if idx == 0 else 0x00)
pkt.append(det_byte)
pkt.append(FOOTER_BYTE)
buf += pkt
self._mock_seq_idx = (start_idx + num_packets) % NUM_CELLS
return bytes(buf)
# ============================================================================ # ============================================================================
@@ -744,9 +951,15 @@ class DataRecorder:
# ============================================================================ # ============================================================================
class RadarAcquisition(threading.Thread): class RadarAcquisition(threading.Thread):
""" """Background thread: reads USB bytes, parses frames, queues them.
Background thread: reads from USB (FT2232H), parses 11-byte packets,
assembles frames, and pushes complete frames to the display queue. Dispatches between two wire formats based on connection type:
- FT2232HConnection -> bulk per-frame format (parses 35 KB frames in
one shot via parse_bulk_frame; fills RadarFrame.magnitude directly).
- FT601Connection -> legacy 11-byte per-sample format (count-based
sample placement via _ingest_sample, the original behavior).
See module docstring for why both formats exist.
""" """
def __init__(self, connection, frame_queue: queue.Queue, def __init__(self, connection, frame_queue: queue.Queue,
@@ -761,37 +974,51 @@ class RadarAcquisition(threading.Thread):
self._frame = RadarFrame() self._frame = RadarFrame()
self._sample_idx = 0 self._sample_idx = 0
self._frame_num = 0 self._frame_num = 0
# AUDIT-C9: dispatch on connection type. The bulk path skips the
# per-sample state machine entirely.
self._is_bulk = isinstance(connection, FT2232HConnection)
self._read_chunk = (2 * BULK_FRAME_MAX_SIZE) if self._is_bulk else 4096
def stop(self): def stop(self):
self._stop_event.set() self._stop_event.set()
def run(self): def run(self):
log.info("Acquisition thread started") log.info(
"Acquisition thread started (%s wire format)",
"FT2232H bulk" if self._is_bulk else "FT601 legacy 11-byte",
)
residual = b"" residual = b""
while not self._stop_event.is_set(): while not self._stop_event.is_set():
chunk = self.conn.read(4096) chunk = self.conn.read(self._read_chunk)
if chunk is None or len(chunk) == 0: if chunk is None or len(chunk) == 0:
time.sleep(0.01) time.sleep(0.01)
continue continue
raw = residual + chunk raw = residual + chunk
if self._is_bulk:
packets = RadarProtocol.find_bulk_frame_boundaries(raw)
max_residual = BULK_FRAME_MAX_SIZE
else:
packets = RadarProtocol.find_packet_boundaries(raw) packets = RadarProtocol.find_packet_boundaries(raw)
max_residual = 2 * max(DATA_PACKET_SIZE, STATUS_PACKET_SIZE)
# Keep unparsed tail bytes for next iteration # Keep unparsed tail bytes for next iteration.
if packets: if packets:
last_end = packets[-1][1] last_end = packets[-1][1]
residual = raw[last_end:] residual = raw[last_end:]
else: else:
# No packets found — keep entire buffer as residual
# but cap at 2x max packet size to avoid unbounded growth
max_residual = 2 * max(DATA_PACKET_SIZE, STATUS_PACKET_SIZE)
residual = raw[-max_residual:] if len(raw) > max_residual else raw residual = raw[-max_residual:] if len(raw) > max_residual else raw
for start, end, ptype in packets: for start, end, ptype in packets:
if ptype == "data": if ptype == "data":
parsed = RadarProtocol.parse_data_packet( if self._is_bulk:
raw[start:end]) parsed = RadarProtocol.parse_bulk_frame(raw, offset=start)
if parsed is not None: if parsed is not None:
self._ingest_sample(parsed) self._ingest_bulk_frame(parsed)
else:
sample = RadarProtocol.parse_data_packet(raw[start:end])
if sample is not None:
self._ingest_sample(sample)
elif ptype == "status": elif ptype == "status":
status = RadarProtocol.parse_status_packet(raw[start:end]) status = RadarProtocol.parse_status_packet(raw[start:end])
if status is not None: if status is not None:
@@ -809,6 +1036,40 @@ class RadarAcquisition(threading.Thread):
log.info("Acquisition thread stopped") log.info("Acquisition thread stopped")
def _ingest_bulk_frame(self, parsed: dict):
"""Build a RadarFrame from one parsed bulk frame and emit it."""
frame = RadarFrame()
frame.timestamp = time.time()
frame.frame_number = parsed["frame_number"]
frame.mag_only = bool(parsed["flags"] & BULK_FLAG_MAG_ONLY)
rprof = parsed["range_profile"]
if rprof is not None:
# Wire is uint16; RadarFrame.range_profile is float64.
frame.range_profile[:] = rprof.astype(np.float64)
dmag = parsed["doppler_mag"]
if dmag is not None:
frame.magnitude[:] = dmag.astype(np.float64)
# I/Q arrays stay zero in mag-only mode (the only mode FPGA
# emits today). Future RTL may populate them; for now flag is
# the source of truth.
cdense = parsed["cfar_dense"]
if cdense is not None:
frame.detections[:] = cdense.astype(np.uint8)
frame.detection_count = int(cdense.sum())
try:
self.frame_queue.put_nowait(frame)
except queue.Full:
with contextlib.suppress(queue.Empty):
self.frame_queue.get_nowait()
self.frame_queue.put_nowait(frame)
if self.recorder and self.recorder.recording:
self.recorder.record_frame(frame)
def _ingest_sample(self, sample: dict): def _ingest_sample(self, sample: dict):
"""Place sample into current frame and emit when complete.""" """Place sample into current frame and emit when complete."""
# [GUI-C2 FIX] Use FPGA frame_start bit as the authoritative sync token. # [GUI-C2 FIX] Use FPGA frame_start bit as the authoritative sync token.
9_Firmware/9_3_GUI/test_GUI_V65_Tk.py
+251 -10
View File
@@ -21,6 +21,10 @@ from radar_protocol import (
HEADER_BYTE, FOOTER_BYTE, STATUS_HEADER_BYTE, HEADER_BYTE, FOOTER_BYTE, STATUS_HEADER_BYTE,
NUM_RANGE_BINS, NUM_DOPPLER_BINS, NUM_RANGE_BINS, NUM_DOPPLER_BINS,
DATA_PACKET_SIZE, DATA_PACKET_SIZE,
BULK_FRAME_HEADER_SIZE, BULK_FRAME_MAX_SIZE,
BULK_RANGE_SECTION_BYTES, BULK_FOOTER_SIZE,
BULK_FLAG_STREAM_RANGE, BULK_FLAG_STREAM_DOPPLER, BULK_FLAG_STREAM_CFAR,
BULK_FLAG_MAG_ONLY, BULK_FLAG_SPARSE_DET,
) )
from GUI_V65_Tk import DemoTarget, DemoSimulator, _ReplayController from GUI_V65_Tk import DemoTarget, DemoSimulator, _ReplayController
@@ -376,6 +380,240 @@ class TestRadarProtocol(unittest.TestCase):
self.assertIsNotNone(RadarProtocol.parse_data_packet(buf[start:end])) self.assertIsNotNone(RadarProtocol.parse_data_packet(buf[start:end]))
class TestBulkFrameParser(unittest.TestCase):
"""AUDIT-C9: parser for the FT2232H bulk per-frame wire format."""
def _build_bulk_frame(
self,
flags: int = (BULK_FLAG_STREAM_RANGE | BULK_FLAG_STREAM_DOPPLER
| BULK_FLAG_STREAM_CFAR | BULK_FLAG_MAG_ONLY),
frame_number: int = 0xBEEF,
n_range: int = NUM_RANGE_BINS,
n_doppler: int = NUM_DOPPLER_BINS,
range_seed: int = 1,
doppler_seed: int = 2,
cfar_seed: int = 3,
bad_footer: bool = False,
) -> tuple[bytes, np.ndarray, np.ndarray, np.ndarray]:
"""Synthesize a bulk frame matching usb_data_interface_ft2232h.v.
Returns (frame_bytes, range_profile, doppler_mag, cfar_dense). The
latter three are the source-of-truth arrays used to generate the
bytes; tests can assert round-trip equality.
"""
rng_r = np.random.RandomState(range_seed)
rng_d = np.random.RandomState(doppler_seed)
rng_c = np.random.RandomState(cfar_seed)
range_profile = (rng_r.randint(0, 65535, size=n_range)
.astype(np.uint16) if (flags & BULK_FLAG_STREAM_RANGE)
else None)
doppler_mag = (rng_d.randint(0, 65535, size=(n_range, n_doppler))
.astype(np.uint16) if (flags & BULK_FLAG_STREAM_DOPPLER)
else None)
cfar_dense = (rng_c.randint(0, 2, size=(n_range, n_doppler))
.astype(np.uint8) if (flags & BULK_FLAG_STREAM_CFAR)
else None)
out = bytearray()
out.append(HEADER_BYTE)
# Don't mask reserved bits here — the parser must reject any byte
# with bits [7:6] set, and the rejection test relies on those bits
# actually surviving into the synthesized frame.
out.append(flags & 0xFF)
out.append((frame_number >> 8) & 0xFF)
out.append(frame_number & 0xFF)
out.append((n_range >> 8) & 0xFF)
out.append(n_range & 0xFF)
out.append((n_doppler >> 8) & 0xFF)
out.append(n_doppler & 0xFF)
if range_profile is not None:
out += range_profile.astype(">u2").tobytes()
if doppler_mag is not None:
out += doppler_mag.astype(">u2").tobytes()
if cfar_dense is not None:
out += np.packbits(cfar_dense.flatten()).tobytes()
out.append(0x00 if bad_footer else FOOTER_BYTE)
return bytes(out), range_profile, doppler_mag, cfar_dense
def test_parse_full_frame_round_trip(self):
"""All-streams mag-only round trip: every cell exact."""
raw, rprof, dmag, cdense = self._build_bulk_frame()
parsed = RadarProtocol.parse_bulk_frame(raw)
self.assertIsNotNone(parsed)
self.assertEqual(parsed["frame_number"], 0xBEEF)
self.assertEqual(parsed["n_range"], NUM_RANGE_BINS)
self.assertEqual(parsed["n_doppler"], NUM_DOPPLER_BINS)
self.assertEqual(parsed["frame_size"], BULK_FRAME_MAX_SIZE)
np.testing.assert_array_equal(parsed["range_profile"], rprof)
np.testing.assert_array_equal(parsed["doppler_mag"], dmag)
np.testing.assert_array_equal(parsed["cfar_dense"], cdense)
def test_parse_range_only(self):
flags = BULK_FLAG_STREAM_RANGE | BULK_FLAG_MAG_ONLY
raw, rprof, _dmag, _cdense = self._build_bulk_frame(flags=flags)
parsed = RadarProtocol.parse_bulk_frame(raw)
self.assertIsNotNone(parsed)
np.testing.assert_array_equal(parsed["range_profile"], rprof)
self.assertIsNone(parsed["doppler_mag"])
self.assertIsNone(parsed["cfar_dense"])
self.assertEqual(
parsed["frame_size"],
BULK_FRAME_HEADER_SIZE + BULK_RANGE_SECTION_BYTES + BULK_FOOTER_SIZE,
)
def test_parse_doppler_only(self):
flags = BULK_FLAG_STREAM_DOPPLER | BULK_FLAG_MAG_ONLY
raw, _rprof, dmag, _cdense = self._build_bulk_frame(flags=flags)
parsed = RadarProtocol.parse_bulk_frame(raw)
self.assertIsNotNone(parsed)
self.assertIsNone(parsed["range_profile"])
np.testing.assert_array_equal(parsed["doppler_mag"], dmag)
self.assertIsNone(parsed["cfar_dense"])
def test_parse_cfar_only(self):
flags = BULK_FLAG_STREAM_CFAR | BULK_FLAG_MAG_ONLY
raw, _rprof, _dmag, cdense = self._build_bulk_frame(flags=flags)
parsed = RadarProtocol.parse_bulk_frame(raw)
self.assertIsNotNone(parsed)
np.testing.assert_array_equal(parsed["cfar_dense"], cdense)
def test_parse_no_streams(self):
"""Header + footer only (8 + 1 = 9 bytes)."""
flags = BULK_FLAG_MAG_ONLY # streams off, mag_only still set
raw, *_ = self._build_bulk_frame(flags=flags)
self.assertEqual(len(raw), 9)
parsed = RadarProtocol.parse_bulk_frame(raw)
self.assertIsNotNone(parsed)
self.assertEqual(parsed["frame_size"], 9)
def test_reject_full_iq_until_rtl_lands(self):
"""mag_only=0 must be rejected — FPGA write FSM doesn't emit it."""
flags = BULK_FLAG_STREAM_DOPPLER # NB: mag_only bit cleared
raw, *_ = self._build_bulk_frame(flags=flags)
# Frame is structurally consistent; rejection comes from the
# mag_only=0 + stream_doppler=1 combination check in parse_bulk_frame.
self.assertIsNone(RadarProtocol.parse_bulk_frame(raw))
def test_reject_sparse_det(self):
"""sparse_det=1 must be rejected — FPGA emits dense bitmap only."""
flags = (BULK_FLAG_STREAM_CFAR | BULK_FLAG_MAG_ONLY | BULK_FLAG_SPARSE_DET)
raw, *_ = self._build_bulk_frame(flags=flags)
self.assertIsNone(RadarProtocol.parse_bulk_frame(raw))
def test_reject_reserved_bits_set(self):
"""Reserved high bits in flags byte must be zero."""
raw, *_ = self._build_bulk_frame(flags=0x80 | BULK_FLAG_MAG_ONLY)
self.assertIsNone(RadarProtocol.parse_bulk_frame(raw))
def test_reject_wrong_n_range(self):
raw, *_ = self._build_bulk_frame(n_range=999)
# Note: the synthesized payload size is wrong too but the n_range
# check fires before any payload-size checks.
self.assertIsNone(RadarProtocol.parse_bulk_frame(raw))
def test_reject_wrong_n_doppler(self):
raw, *_ = self._build_bulk_frame(n_doppler=64)
self.assertIsNone(RadarProtocol.parse_bulk_frame(raw))
def test_reject_missing_footer(self):
raw, *_ = self._build_bulk_frame(bad_footer=True)
self.assertIsNone(RadarProtocol.parse_bulk_frame(raw))
def test_reject_wrong_header(self):
raw, *_ = self._build_bulk_frame()
bad = b"\x00" + raw[1:]
self.assertIsNone(RadarProtocol.parse_bulk_frame(bad))
def test_reject_truncated(self):
raw, *_ = self._build_bulk_frame()
self.assertIsNone(RadarProtocol.parse_bulk_frame(raw[:1000]))
def test_find_boundaries_two_frames(self):
f1, *_ = self._build_bulk_frame(frame_number=1)
f2, *_ = self._build_bulk_frame(frame_number=2)
buf = b"\x00\x12" + f1 + b"\x33" + f2 # garbage between/around frames
out = RadarProtocol.find_bulk_frame_boundaries(buf)
data = [(s, e, t) for (s, e, t) in out if t == "data"]
self.assertEqual(len(data), 2)
# Round-trip both
for s, _e, _t in data:
parsed = RadarProtocol.parse_bulk_frame(buf, offset=s)
self.assertIsNotNone(parsed)
def test_find_boundaries_with_status(self):
"""Status packets coexist with bulk frames in the same stream."""
f1, *_ = self._build_bulk_frame()
# Build a minimal valid status packet (byte 1 = 0xFF, footer = 0x55).
status = bytes([STATUS_HEADER_BYTE, 0xFF] + [0x00] * 23 + [FOOTER_BYTE])
buf = f1 + status
out = RadarProtocol.find_bulk_frame_boundaries(buf)
types = [t for _s, _e, t in out]
self.assertIn("data", types)
self.assertIn("status", types)
def test_find_boundaries_truncated_residual(self):
"""A partial frame at the buffer tail is not returned (kept as residual)."""
f1, *_ = self._build_bulk_frame()
# Cut off the last 100 bytes — find_bulk_frame_boundaries must not
# return this frame; the caller keeps the bytes for next iteration.
buf = f1[:-100]
out = RadarProtocol.find_bulk_frame_boundaries(buf)
self.assertEqual([t for _s, _e, t in out], [])
def test_resync_after_byte_drop(self):
"""A dropped byte at the head must not lock the parser onto false positives."""
f1, *_ = self._build_bulk_frame()
# Single garbage byte before the real frame.
buf = b"\x99" + f1
out = RadarProtocol.find_bulk_frame_boundaries(buf)
data = [(s, e, t) for (s, e, t) in out if t == "data"]
self.assertEqual(len(data), 1)
self.assertEqual(data[0][0], 1) # frame starts at offset 1
def test_acquisition_dispatches_bulk_for_ft2232h(self):
"""RadarAcquisition must select bulk format for FT2232H connections."""
ft = FT2232HConnection(mock=True)
ft.open()
q: queue.Queue = queue.Queue()
acq = RadarAcquisition(ft, q)
self.assertTrue(acq._is_bulk)
ft.close()
def test_acquisition_dispatches_legacy_for_ft601(self):
"""RadarAcquisition must select legacy format for FT601 connections."""
ft = FT601Connection(mock=True)
ft.open()
q: queue.Queue = queue.Queue()
acq = RadarAcquisition(ft, q)
self.assertFalse(acq._is_bulk)
ft.close()
def test_ingest_bulk_frame_populates_radarframe(self):
"""End-to-end: bulk parse → RadarFrame in queue with correct fields."""
ft = FT2232HConnection(mock=True)
ft.open()
q: queue.Queue = queue.Queue(maxsize=4)
acq = RadarAcquisition(ft, q)
# Drive one tick of run() manually instead of starting the thread.
chunk = ft.read(BULK_FRAME_MAX_SIZE * 2)
packets = RadarProtocol.find_bulk_frame_boundaries(chunk)
self.assertGreater(len(packets), 0)
for s, _e, t in packets:
if t == "data":
parsed = RadarProtocol.parse_bulk_frame(chunk, offset=s)
acq._ingest_bulk_frame(parsed)
ft.close()
frame = q.get_nowait()
self.assertIsInstance(frame, RadarFrame)
self.assertTrue(frame.mag_only)
# Mag-only mode: I/Q stay zero, magnitude carries data.
self.assertTrue((frame.range_doppler_i == 0).all())
self.assertTrue((frame.range_doppler_q == 0).all())
self.assertGreater(frame.magnitude.max(), 0)
class TestFT2232HConnection(unittest.TestCase): class TestFT2232HConnection(unittest.TestCase):
"""Test mock FT2232H connection.""" """Test mock FT2232H connection."""
@@ -395,16 +633,20 @@ class TestFT2232HConnection(unittest.TestCase):
conn.close() conn.close()
def test_mock_read_contains_valid_packets(self): def test_mock_read_contains_valid_packets(self):
"""Mock data should contain parseable data packets.""" """Mock data should contain a parseable bulk frame (AUDIT-C9)."""
conn = FT2232HConnection(mock=True) conn = FT2232HConnection(mock=True)
conn.open() conn.open()
raw = conn.read(4096) raw = conn.read(BULK_FRAME_MAX_SIZE * 2)
packets = RadarProtocol.find_packet_boundaries(raw) packets = RadarProtocol.find_bulk_frame_boundaries(raw)
self.assertGreater(len(packets), 0) self.assertGreater(len(packets), 0)
for start, end, ptype in packets: for start, _end, ptype in packets:
if ptype == "data": if ptype == "data":
result = RadarProtocol.parse_data_packet(raw[start:end]) parsed = RadarProtocol.parse_bulk_frame(raw, offset=start)
self.assertIsNotNone(result) self.assertIsNotNone(parsed)
self.assertEqual(parsed["n_range"], NUM_RANGE_BINS)
self.assertEqual(parsed["n_doppler"], NUM_DOPPLER_BINS)
self.assertTrue(parsed["flags"] & BULK_FLAG_MAG_ONLY)
self.assertFalse(parsed["flags"] & BULK_FLAG_SPARSE_DET)
conn.close() conn.close()
def test_mock_write(self): def test_mock_write(self):
@@ -578,8 +820,8 @@ class TestRadarAcquisition(unittest.TestCase):
acq = RadarAcquisition(conn, fq) acq = RadarAcquisition(conn, fq)
acq.start() acq.start()
# Wait for at least one frame (mock produces ~32 samples per read, # AUDIT-C9: FT2232H mock now emits one full bulk frame per read
# need 2048 for a full frame, so may take a few seconds) # (50 ms cadence in the mock), so a frame should land within ~100 ms.
frame = None frame = None
try: # noqa: SIM105 try: # noqa: SIM105
frame = fq.get(timeout=10) frame = fq.get(timeout=10)
@@ -590,12 +832,11 @@ class TestRadarAcquisition(unittest.TestCase):
acq.join(timeout=3) acq.join(timeout=3)
conn.close() conn.close()
# With mock data producing 32 packets per read at 50ms interval,
# a full frame (2048 samples) takes ~3.2s. Allow up to 10s.
if frame is not None: if frame is not None:
self.assertIsInstance(frame, RadarFrame) self.assertIsInstance(frame, RadarFrame)
self.assertEqual(frame.magnitude.shape, self.assertEqual(frame.magnitude.shape,
(NUM_RANGE_BINS, NUM_DOPPLER_BINS)) (NUM_RANGE_BINS, NUM_DOPPLER_BINS))
self.assertTrue(frame.mag_only)
# If no frame arrived in timeout, that's still OK for a fast CI run # If no frame arrived in timeout, that's still OK for a fast CI run
def test_acquisition_stop(self): def test_acquisition_stop(self):