feat: 2048-pt FFT upgrade with decimation=4, 512 output bins, 6m spacing
Complete cross-layer upgrade from 1024-pt/64-bin to 2048-pt/512-bin FFT: FPGA RTL (14+ modules): - radar_params.vh: FFT_SIZE=2048, RANGE_BINS=512, 9-bit range, 6-bit stream - fft_engine.v: 2048-pt FFT with XPM BRAM - chirp_memory_loader_param.v: 2 segments x 2048 (was 4 x 1024) - matched_filter_multi_segment.v: BRAM inference for overlap_cache, explicit ov_waddr - mti_canceller.v: BRAM inference for prev_i/q arrays (was fabric FFs) - doppler_processor.v: 16384-deep memory, 14-bit addressing - cfar_ca.v: 512 rows, indentation fix - radar_receiver_final.v: rising-edge detector for frame_complete, 11-bit sample_addr - range_bin_decimator.v: 512 output bins - usb_data_interface_ft2232h.v: bulk per-frame with Manhattan magnitude - radar_mode_controller.v: XOR edge detector for toggle signals - rx_gain_control.v: updated for new bin count Python GUI + Protocol (8 files): - radar_protocol.py: 512-bin bulk frame parser, LSB-first bitmap - GUI_V65_Tk.py, v7/*.py: updated for 512 bins, 6m range resolution Golden data + tests: - All .hex/.csv/.npy golden references regenerated for 2048/512 - fft_twiddle_2048.mem added - Deleted stale seg2/seg3 chirp mem files - 9 new bulk frame cross-layer tests, deleted 6 stale per-sample tests - Deleted stale tb_cross_layer_ft2232h.v and dead contract_parser functions - Updated validate_mem_files.py for 2048/2-segment config MCU: RadarSettings.cpp max_distance/map_size 1536->3072 All 4 CI jobs pass: 285 tests, 0 failures, 0 skips
This commit is contained in:
Jason
@@ -7,7 +7,7 @@ via FT2232H USB 2.0 interface.
|
||||
|
||||
Features:
|
||||
- FT2232H USB reader with packet parsing (matches usb_data_interface_ft2232h.v)
|
||||
- Real-time range-Doppler magnitude heatmap (64x32)
|
||||
- Real-time range-Doppler magnitude heatmap (512x32)
|
||||
- CFAR detection overlay (flagged cells highlighted)
|
||||
- Range profile waterfall plot (range vs. time)
|
||||
- Host command sender (opcodes per radar_system_top.v:
|
||||
@@ -99,9 +99,9 @@ class DemoTarget:
|
||||
__slots__ = ("azimuth", "classification", "id", "range_m", "snr", "velocity")
|
||||
|
||||
# Physical range grid: matched-filter receiver, 100 MSPS post-DDC, 16:1 decimation
|
||||
# range_per_bin = c / (2 * 100e6) * 16 = 24.0 m
|
||||
_RANGE_PER_BIN: float = (3e8 / (2 * 100e6)) * 16 # 24.0 m
|
||||
_MAX_RANGE: float = _RANGE_PER_BIN * NUM_RANGE_BINS # 1536 m
|
||||
# range_per_bin = c / (2 * 100e6) * 4 = 6.0 m
|
||||
_RANGE_PER_BIN: float = (3e8 / (2 * 100e6)) * 4 # 6.0 m
|
||||
_MAX_RANGE: float = _RANGE_PER_BIN * NUM_RANGE_BINS # 3072 m
|
||||
|
||||
def __init__(self, tid: int):
|
||||
self.id = tid
|
||||
@@ -189,7 +189,7 @@ class DemoSimulator:
|
||||
det = np.zeros((NUM_RANGE_BINS, NUM_DOPPLER_BINS), dtype=np.uint8)
|
||||
|
||||
# Range/Doppler scaling -- matched-filter receiver, 100 MSPS, 16:1 decimation
|
||||
range_per_bin = (3e8 / (2 * 100e6)) * 16 # 24.0 m/bin
|
||||
range_per_bin = (3e8 / (2 * 100e6)) * 4 # 6.0 m/bin
|
||||
max_range = range_per_bin * NUM_RANGE_BINS
|
||||
vel_per_bin = 2.67 # m/s per Doppler bin (lam/(2*32*167us))
|
||||
|
||||
@@ -387,7 +387,7 @@ class RadarDashboard:
|
||||
|
||||
# Radar parameters used for range-axis scaling.
|
||||
# Matched-filter receiver: range_per_bin = c / (2 * fs_processing) * decimation
|
||||
# = 3e8 / (2 * 100e6) * 16 = 24.0 m/bin
|
||||
# = 3e8 / (2 * 100e6) * 4 = 6.0 m/bin
|
||||
BANDWIDTH = 20e6 # Hz — chirp bandwidth (for display/info only)
|
||||
C = 3e8 # m/s — speed of light
|
||||
|
||||
@@ -519,7 +519,7 @@ class RadarDashboard:
|
||||
def _build_display_tab(self, parent):
|
||||
# Compute physical axis limits -- matched-filter receiver
|
||||
# Range per bin: c / (2 * fs_processing) * decimation_factor = 24.0 m
|
||||
range_per_bin = self.C / (2.0 * 100e6) * 16 # 24.0 m
|
||||
range_per_bin = self.C / (2.0 * 100e6) * 4 # 6.0 m
|
||||
max_range = range_per_bin * NUM_RANGE_BINS
|
||||
|
||||
doppler_bin_lo = 0
|
||||
@@ -640,14 +640,14 @@ class RadarDashboard:
|
||||
sc_row = ttk.Frame(grp_op)
|
||||
sc_row.pack(fill="x", pady=2)
|
||||
ttk.Label(sc_row, text="Stream Control").pack(side="left")
|
||||
var_sc = tk.StringVar(value="7")
|
||||
var_sc = tk.StringVar(value="15")
|
||||
self._param_vars["4"] = var_sc
|
||||
ttk.Entry(sc_row, textvariable=var_sc, width=6).pack(side="left", padx=6)
|
||||
ttk.Label(sc_row, text="0-7", foreground=ACCENT,
|
||||
ttk.Label(sc_row, text="0-63", foreground=ACCENT,
|
||||
font=("Menlo", 9)).pack(side="left")
|
||||
ttk.Button(sc_row, text="Set",
|
||||
command=lambda: self._send_validated(
|
||||
0x04, var_sc, bits=3)).pack(side="right")
|
||||
0x04, var_sc, bits=6)).pack(side="right")
|
||||
|
||||
ttk.Button(grp_op, text="Request Status",
|
||||
command=lambda: self._send_cmd(0xFF, 0)).pack(fill="x", pady=2)
|
||||
|
||||
@@ -7,9 +7,11 @@ 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 Packet Protocol (11-byte):
|
||||
USB Packet Protocol:
|
||||
TX (FPGA→Host):
|
||||
Data packet: [0xAA] [range_q 2B] [range_i 2B] [dop_re 2B] [dop_im 2B] [det 1B] [0x55]
|
||||
Bulk frame (FT2232H):
|
||||
[0xAA] [flags 1B] [frame# 2B] [range_bins 2B] [doppler_bins 2B]
|
||||
[range profile (opt)] [doppler mag/IQ (opt)] [detect flags (opt)] [0x55]
|
||||
Status packet: [0xBB] [status 6x32b] [0x55]
|
||||
RX (Host→FPGA):
|
||||
Command: 4 bytes received sequentially {opcode, addr, value_hi, value_lo}
|
||||
@@ -39,12 +41,15 @@ FOOTER_BYTE = 0x55
|
||||
STATUS_HEADER_BYTE = 0xBB
|
||||
|
||||
# Packet sizes
|
||||
DATA_PACKET_SIZE = 11 # 1 + 4 + 2 + 2 + 1 + 1
|
||||
BULK_HEADER_SIZE = 8 # 1(AA)+1(flags)+2(frame#)+2(Rbins)+2(Dbins)
|
||||
STATUS_PACKET_SIZE = 26 # 1 + 24 + 1
|
||||
|
||||
NUM_RANGE_BINS = 64
|
||||
# Legacy per-sample protocol (FT601 USB 3.0 only)
|
||||
DATA_PACKET_SIZE = 11 # 1 + 4 + 2 + 2 + 1 + 1
|
||||
|
||||
NUM_RANGE_BINS = 512
|
||||
NUM_DOPPLER_BINS = 32
|
||||
NUM_CELLS = NUM_RANGE_BINS * NUM_DOPPLER_BINS # 2048
|
||||
NUM_CELLS = NUM_RANGE_BINS * NUM_DOPPLER_BINS # 16384
|
||||
|
||||
WATERFALL_DEPTH = 64
|
||||
|
||||
@@ -66,7 +71,7 @@ class Opcode(IntEnum):
|
||||
RADAR_MODE = 0x01 # 2-bit mode select
|
||||
TRIGGER_PULSE = 0x02 # self-clearing one-shot trigger
|
||||
DETECT_THRESHOLD = 0x03 # 16-bit detection threshold value
|
||||
STREAM_CONTROL = 0x04 # 3-bit stream enable mask
|
||||
STREAM_CONTROL = 0x04 # 6-bit stream/format control
|
||||
|
||||
# --- Digital gain (0x16) ---
|
||||
GAIN_SHIFT = 0x16 # 4-bit digital gain shift
|
||||
@@ -108,7 +113,7 @@ class Opcode(IntEnum):
|
||||
|
||||
@dataclass
|
||||
class RadarFrame:
|
||||
"""One complete radar frame (64 range x 32 Doppler)."""
|
||||
"""One complete radar frame (512 range x 32 Doppler)."""
|
||||
timestamp: float = 0.0
|
||||
range_doppler_i: np.ndarray = field(
|
||||
default_factory=lambda: np.zeros((NUM_RANGE_BINS, NUM_DOPPLER_BINS), dtype=np.int16))
|
||||
@@ -230,9 +235,9 @@ class RadarProtocol:
|
||||
return None
|
||||
|
||||
sr = StatusResponse()
|
||||
# Word 0: {0xFF[31:24], mode[23:22], stream[21:19], 3'b000[18:16], threshold[15:0]}
|
||||
# Word 0: {0xFF[31:24], mode[23:22], stream[21:16], threshold[15:0]}
|
||||
sr.cfar_threshold = words[0] & 0xFFFF
|
||||
sr.stream_ctrl = (words[0] >> 19) & 0x07
|
||||
sr.stream_ctrl = (words[0] >> 16) & 0x3F
|
||||
sr.radar_mode = (words[0] >> 22) & 0x03
|
||||
# Word 1: {long_chirp[31:16], long_listen[15:0]}
|
||||
sr.long_listen = words[1] & 0xFFFF
|
||||
@@ -257,24 +262,185 @@ class RadarProtocol:
|
||||
sr.self_test_busy = (words[5] >> 24) & 0x01
|
||||
return sr
|
||||
|
||||
@staticmethod
|
||||
def parse_bulk_frame(raw: bytes) -> RadarFrame | None:
|
||||
"""Parse a bulk per-frame transfer from the FT2232H USB interface.
|
||||
|
||||
Frame format (from usb_data_interface_ft2232h.v):
|
||||
Byte 0: 0xAA (header)
|
||||
Byte 1: Format flags {2'b0, sparse_det, mag_only,
|
||||
stream_cfar, stream_doppler, stream_range}
|
||||
Byte 2-3: Frame number (16-bit, MSB first)
|
||||
Byte 4-5: Range bin count (16-bit, MSB first)
|
||||
Byte 6-7: Doppler bin count (16-bit, MSB first)
|
||||
[variable payload based on flags]
|
||||
Last byte: 0x55 (footer)
|
||||
"""
|
||||
if len(raw) < BULK_HEADER_SIZE + 1: # header + footer minimum
|
||||
return None
|
||||
if raw[0] != HEADER_BYTE:
|
||||
return None
|
||||
|
||||
flags = raw[1]
|
||||
frame_num = struct.unpack_from(">H", raw, 2)[0]
|
||||
n_range = struct.unpack_from(">H", raw, 4)[0]
|
||||
n_doppler = struct.unpack_from(">H", raw, 6)[0]
|
||||
|
||||
stream_range = bool(flags & 0x01)
|
||||
stream_doppler = bool(flags & 0x02)
|
||||
stream_cfar = bool(flags & 0x04)
|
||||
mag_only = bool(flags & 0x08)
|
||||
sparse_det = bool(flags & 0x10)
|
||||
|
||||
offset = BULK_HEADER_SIZE
|
||||
frame = RadarFrame()
|
||||
frame.frame_number = frame_num
|
||||
frame.timestamp = time.time()
|
||||
|
||||
# --- Range profile section ---
|
||||
if stream_range:
|
||||
nbytes = n_range * 2
|
||||
if offset + nbytes > len(raw):
|
||||
return None
|
||||
range_data = np.frombuffer(raw[offset:offset + nbytes],
|
||||
dtype=">u2").astype(np.float64)
|
||||
frame.range_profile = range_data
|
||||
offset += nbytes
|
||||
|
||||
# --- Doppler magnitude or I/Q section ---
|
||||
if stream_doppler:
|
||||
if mag_only:
|
||||
nbytes = n_range * n_doppler * 2
|
||||
if offset + nbytes > len(raw):
|
||||
return None
|
||||
mag_flat = np.frombuffer(raw[offset:offset + nbytes],
|
||||
dtype=">u2").astype(np.float64)
|
||||
frame.magnitude = mag_flat.reshape(n_range, n_doppler)
|
||||
# No I/Q available in mag-only mode
|
||||
frame.range_doppler_i = np.zeros((n_range, n_doppler), dtype=np.int16)
|
||||
frame.range_doppler_q = np.zeros((n_range, n_doppler), dtype=np.int16)
|
||||
offset += nbytes
|
||||
else:
|
||||
# Full I/Q: 32-bit per cell (I16, Q16)
|
||||
nbytes = n_range * n_doppler * 4
|
||||
if offset + nbytes > len(raw):
|
||||
return None
|
||||
iq_data = np.frombuffer(raw[offset:offset + nbytes], dtype=">i2")
|
||||
iq_data = iq_data.reshape(n_range, n_doppler, 2)
|
||||
frame.range_doppler_i = iq_data[:, :, 0].astype(np.int16)
|
||||
frame.range_doppler_q = iq_data[:, :, 1].astype(np.int16)
|
||||
frame.magnitude = (
|
||||
np.abs(frame.range_doppler_i.astype(np.float64))
|
||||
+ np.abs(frame.range_doppler_q.astype(np.float64))
|
||||
)
|
||||
offset += nbytes
|
||||
|
||||
# --- Detection flags section ---
|
||||
if stream_cfar:
|
||||
if sparse_det:
|
||||
if offset + 2 > len(raw):
|
||||
return None
|
||||
det_count = struct.unpack_from(">H", raw, offset)[0]
|
||||
offset += 2
|
||||
nbytes = det_count * 6
|
||||
if offset + nbytes > len(raw):
|
||||
return None
|
||||
det_arr = np.zeros((n_range, n_doppler), dtype=np.uint8)
|
||||
for d in range(det_count):
|
||||
base = offset + d * 6
|
||||
rbin = struct.unpack_from(">H", raw, base)[0]
|
||||
dbin = struct.unpack_from(">H", raw, base + 2)[0]
|
||||
if rbin < n_range and dbin < n_doppler:
|
||||
det_arr[rbin, dbin] = 1
|
||||
frame.detections = det_arr
|
||||
frame.detection_count = det_count
|
||||
offset += nbytes
|
||||
else:
|
||||
# Packed bitmap: n_range * n_doppler bits, LSB-first per byte
|
||||
# RTL packs: byte_addr = {range_bin, doppler[4:3]}, bit = doppler[2:0]
|
||||
nbytes = (n_range * n_doppler + 7) // 8
|
||||
if offset + nbytes > len(raw):
|
||||
return None
|
||||
det_bytes = raw[offset:offset + nbytes]
|
||||
det_arr = np.zeros((n_range, n_doppler), dtype=np.uint8)
|
||||
for r in range(n_range):
|
||||
for db in range(n_doppler):
|
||||
byte_idx = r * (n_doppler // 8) + db // 8
|
||||
bit_pos = db % 8 # LSB-first: doppler[2:0] = bit position
|
||||
if byte_idx < len(det_bytes) and (det_bytes[byte_idx] >> bit_pos) & 1:
|
||||
det_arr[r, db] = 1
|
||||
frame.detections = det_arr
|
||||
frame.detection_count = int(det_arr.sum())
|
||||
offset += nbytes
|
||||
|
||||
# Footer check
|
||||
if offset >= len(raw) or raw[offset] != FOOTER_BYTE:
|
||||
return None
|
||||
|
||||
# Derive range_profile from magnitude if not streamed directly
|
||||
if not stream_range and stream_doppler:
|
||||
frame.range_profile = np.sum(frame.magnitude, axis=1)
|
||||
|
||||
return frame
|
||||
|
||||
@staticmethod
|
||||
def compute_bulk_frame_size(flags: int, n_range: int = NUM_RANGE_BINS,
|
||||
n_doppler: int = NUM_DOPPLER_BINS) -> int:
|
||||
"""Compute expected bulk frame size in bytes for given flags."""
|
||||
size = BULK_HEADER_SIZE # header
|
||||
if flags & 0x01: # stream_range
|
||||
size += n_range * 2
|
||||
if flags & 0x02: # stream_doppler
|
||||
if flags & 0x08: # mag_only
|
||||
size += n_range * n_doppler * 2
|
||||
else:
|
||||
size += n_range * n_doppler * 4
|
||||
if flags & 0x04: # stream_cfar
|
||||
if flags & 0x10: # sparse_det — variable, use bitmap estimate
|
||||
size += 2 # count field minimum
|
||||
else:
|
||||
size += (n_range * n_doppler + 7) // 8
|
||||
size += 1 # footer
|
||||
return size
|
||||
|
||||
@staticmethod
|
||||
def find_packet_boundaries(buf: bytes) -> list[tuple[int, int, str]]:
|
||||
"""
|
||||
Scan buffer for packet start markers (0xAA data, 0xBB status).
|
||||
Scan buffer for packet start markers.
|
||||
Supports bulk frames (0xAA with 8-byte header), status (0xBB),
|
||||
and legacy 11-byte data packets (0xAA with footer at offset 10).
|
||||
Returns list of (start_idx, expected_end_idx, packet_type).
|
||||
"""
|
||||
packets = []
|
||||
i = 0
|
||||
while i < len(buf):
|
||||
if buf[i] == HEADER_BYTE:
|
||||
# Try bulk frame first (8-byte header with range/doppler counts)
|
||||
if i + BULK_HEADER_SIZE <= len(buf):
|
||||
flags = buf[i + 1]
|
||||
n_range = struct.unpack_from(">H", buf, i + 4)[0]
|
||||
n_doppler = struct.unpack_from(">H", buf, i + 6)[0]
|
||||
# Sanity: valid bulk frame has reasonable dimensions
|
||||
if (1 <= n_range <= 2048 and 1 <= n_doppler <= 64
|
||||
and flags <= 0x3F):
|
||||
expected_size = RadarProtocol.compute_bulk_frame_size(
|
||||
flags, n_range, n_doppler)
|
||||
end = i + expected_size
|
||||
if end <= len(buf) and buf[end - 1] == FOOTER_BYTE:
|
||||
packets.append((i, end, "bulk"))
|
||||
i = end
|
||||
continue
|
||||
if end > len(buf):
|
||||
break # partial bulk frame
|
||||
# Fallback: legacy 11-byte per-sample packet (FT601)
|
||||
end = i + DATA_PACKET_SIZE
|
||||
if end <= len(buf) and buf[end - 1] == FOOTER_BYTE:
|
||||
packets.append((i, end, "data"))
|
||||
i = end
|
||||
else:
|
||||
if end > len(buf):
|
||||
break # partial packet at end — leave for residual
|
||||
i += 1 # footer mismatch — skip this false header
|
||||
break
|
||||
i += 1
|
||||
elif buf[i] == STATUS_HEADER_BYTE:
|
||||
end = i + STATUS_PACKET_SIZE
|
||||
if end <= len(buf) and buf[end - 1] == FOOTER_BYTE:
|
||||
@@ -392,53 +558,62 @@ class FT2232HConnection:
|
||||
log.error(f"FT2232H write error: {e}")
|
||||
return False
|
||||
|
||||
def _mock_read(self, size: int) -> bytes:
|
||||
def _mock_read(self, _size: int) -> bytes:
|
||||
"""
|
||||
Generate synthetic 11-byte radar data packets for testing.
|
||||
Emits packets in sequential FPGA order (range_bin 0..63, doppler_bin
|
||||
0..31 within each range bin) so that RadarAcquisition._ingest_sample()
|
||||
places them correctly. A target is injected near range bin 20,
|
||||
Doppler bin 8.
|
||||
Generate a synthetic bulk radar frame for testing.
|
||||
Matches the FT2232H sectioned transfer format: header (8B) →
|
||||
range profile → doppler magnitude → detection flags → footer.
|
||||
A target is injected near range bin 100, Doppler bin 8.
|
||||
"""
|
||||
time.sleep(0.05)
|
||||
self._mock_frame_num += 1
|
||||
|
||||
# Stream flags: range + doppler + cfar, mag-only, bitmap detections
|
||||
flags = 0x0F # stream_range | stream_doppler | stream_cfar | mag_only
|
||||
|
||||
buf = bytearray()
|
||||
num_packets = min(NUM_CELLS, size // DATA_PACKET_SIZE)
|
||||
start_idx = getattr(self, '_mock_seq_idx', 0)
|
||||
# --- Header (8 bytes) ---
|
||||
buf.append(HEADER_BYTE)
|
||||
buf.append(flags)
|
||||
buf += struct.pack(">H", self._mock_frame_num & 0xFFFF)
|
||||
buf += struct.pack(">H", NUM_RANGE_BINS)
|
||||
buf += struct.pack(">H", NUM_DOPPLER_BINS)
|
||||
|
||||
for n in range(num_packets):
|
||||
idx = (start_idx + n) % NUM_CELLS
|
||||
rbin = idx // NUM_DOPPLER_BINS
|
||||
dbin = idx % NUM_DOPPLER_BINS
|
||||
# --- Range profile (512 x 16-bit) ---
|
||||
range_profile = self._mock_rng.randint(50, 200, size=NUM_RANGE_BINS).astype(np.uint16)
|
||||
# Inject target peak at range bin ~100
|
||||
for rb in range(98, 103):
|
||||
range_profile[rb] = 8000
|
||||
buf += range_profile.astype(">u2").tobytes()
|
||||
|
||||
range_i = int(self._mock_rng.normal(0, 100))
|
||||
range_q = int(self._mock_rng.normal(0, 100))
|
||||
if abs(rbin - 20) < 3:
|
||||
range_i += 5000
|
||||
range_q += 3000
|
||||
# --- Doppler magnitude (512 x 32 x 16-bit, mag-only) ---
|
||||
mag = self._mock_rng.randint(
|
||||
10, 100, size=(NUM_RANGE_BINS, NUM_DOPPLER_BINS),
|
||||
).astype(np.uint16)
|
||||
for rb in range(98, 103):
|
||||
for db in range(7, 10):
|
||||
mag[rb, db] = 8000
|
||||
buf += mag.astype(">u2").tobytes()
|
||||
|
||||
dop_i = int(self._mock_rng.normal(0, 50))
|
||||
dop_q = int(self._mock_rng.normal(0, 50))
|
||||
if abs(rbin - 20) < 3 and abs(dbin - 8) < 2:
|
||||
dop_i += 8000
|
||||
dop_q += 4000
|
||||
# --- Detection flags (bitmap: 512*32/8 = 2048 bytes) ---
|
||||
det = np.zeros((NUM_RANGE_BINS, NUM_DOPPLER_BINS), dtype=np.uint8)
|
||||
for rb in range(99, 102):
|
||||
for db in range(7, 10):
|
||||
det[rb, db] = 1
|
||||
det_bytes = bytearray((NUM_RANGE_BINS * NUM_DOPPLER_BINS + 7) // 8)
|
||||
for r in range(NUM_RANGE_BINS):
|
||||
for d in range(NUM_DOPPLER_BINS):
|
||||
if det[r, d]:
|
||||
# LSB-first per byte, matching RTL: byte_addr = {range_bin, doppler[4:3]},
|
||||
# bit = doppler[2:0]. This matches the parser at line ~368.
|
||||
byte_idx = r * (NUM_DOPPLER_BINS // 8) + d // 8
|
||||
bit_pos = d % 8
|
||||
det_bytes[byte_idx] |= 1 << bit_pos
|
||||
buf += det_bytes
|
||||
|
||||
detection = 1 if (abs(rbin - 20) < 2 and abs(dbin - 8) < 2) else 0
|
||||
# --- Footer ---
|
||||
buf.append(FOOTER_BYTE)
|
||||
|
||||
# Build compact 11-byte packet
|
||||
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))
|
||||
pkt.append(detection & 0x01)
|
||||
pkt.append(FOOTER_BYTE)
|
||||
|
||||
buf += pkt
|
||||
|
||||
self._mock_seq_idx = (start_idx + num_packets) % NUM_CELLS
|
||||
return bytes(buf)
|
||||
|
||||
|
||||
@@ -523,8 +698,8 @@ class DataRecorder:
|
||||
|
||||
class RadarAcquisition(threading.Thread):
|
||||
"""
|
||||
Background thread: reads from USB (FT2232H), parses 11-byte packets,
|
||||
assembles frames, and pushes complete frames to the display queue.
|
||||
Background thread: reads from USB (FT2232H), parses bulk frames
|
||||
or legacy 11-byte packets, and pushes complete frames to the display queue.
|
||||
"""
|
||||
|
||||
def __init__(self, connection, frame_queue: queue.Queue,
|
||||
@@ -547,7 +722,7 @@ class RadarAcquisition(threading.Thread):
|
||||
log.info("Acquisition thread started")
|
||||
residual = b""
|
||||
while not self._stop_event.is_set():
|
||||
chunk = self.conn.read(4096)
|
||||
chunk = self.conn.read(65536)
|
||||
if chunk is None or len(chunk) == 0:
|
||||
time.sleep(0.01)
|
||||
continue
|
||||
@@ -561,11 +736,16 @@ class RadarAcquisition(threading.Thread):
|
||||
residual = raw[last_end:]
|
||||
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)
|
||||
# but cap to avoid unbounded growth
|
||||
max_residual = 65536
|
||||
residual = raw[-max_residual:] if len(raw) > max_residual else raw
|
||||
for start, end, ptype in packets:
|
||||
if ptype == "data":
|
||||
if ptype == "bulk":
|
||||
frame = RadarProtocol.parse_bulk_frame(raw[start:end])
|
||||
if frame is not None:
|
||||
self._emit_frame(frame)
|
||||
elif ptype == "data":
|
||||
# Legacy per-sample protocol (FT601)
|
||||
parsed = RadarProtocol.parse_data_packet(
|
||||
raw[start:end])
|
||||
if parsed is not None:
|
||||
@@ -587,8 +767,21 @@ class RadarAcquisition(threading.Thread):
|
||||
|
||||
log.info("Acquisition thread stopped")
|
||||
|
||||
def _emit_frame(self, frame: RadarFrame):
|
||||
"""Push a complete bulk frame to the display queue."""
|
||||
# Push to display queue (drop old if backed up)
|
||||
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):
|
||||
"""Place sample into current frame and emit when complete."""
|
||||
"""Place sample into current frame and emit when complete (legacy FT601 path)."""
|
||||
rbin = self._sample_idx // NUM_DOPPLER_BINS
|
||||
dbin = self._sample_idx % NUM_DOPPLER_BINS
|
||||
|
||||
@@ -607,22 +800,13 @@ class RadarAcquisition(threading.Thread):
|
||||
self._finalize_frame()
|
||||
|
||||
def _finalize_frame(self):
|
||||
"""Complete frame: compute range profile, push to queue, record."""
|
||||
"""Complete frame: compute range profile, push to queue, record (legacy path)."""
|
||||
self._frame.timestamp = time.time()
|
||||
self._frame.frame_number = self._frame_num
|
||||
# Range profile = sum of magnitude across Doppler bins
|
||||
self._frame.range_profile = np.sum(self._frame.magnitude, axis=1)
|
||||
|
||||
# Push to display queue (drop old if backed up)
|
||||
try:
|
||||
self.frame_queue.put_nowait(self._frame)
|
||||
except queue.Full:
|
||||
with contextlib.suppress(queue.Empty):
|
||||
self.frame_queue.get_nowait()
|
||||
self.frame_queue.put_nowait(self._frame)
|
||||
|
||||
if self.recorder and self.recorder.recording:
|
||||
self.recorder.record_frame(self._frame)
|
||||
self._emit_frame(self._frame)
|
||||
|
||||
self._frame_num += 1
|
||||
self._frame = RadarFrame()
|
||||
|
||||
@@ -127,12 +127,12 @@ class TestRadarProtocol(unittest.TestCase):
|
||||
short_listen=17450, chirps=32, range_mode=0,
|
||||
st_flags=0, st_detail=0, st_busy=0,
|
||||
agc_gain=0, agc_peak=0, agc_sat=0, agc_enable=0):
|
||||
"""Build a 26-byte status response matching FPGA format (Build 26)."""
|
||||
"""Build a 26-byte status response matching FPGA format."""
|
||||
pkt = bytearray()
|
||||
pkt.append(STATUS_HEADER_BYTE)
|
||||
|
||||
# Word 0: {0xFF[31:24], mode[23:22], stream[21:19], 3'b000[18:16], threshold[15:0]}
|
||||
w0 = (0xFF << 24) | ((mode & 0x03) << 22) | ((stream & 0x07) << 19) | (threshold & 0xFFFF)
|
||||
# Word 0: {0xFF[31:24], mode[23:22], stream[21:16], threshold[15:0]}
|
||||
w0 = (0xFF << 24) | ((mode & 0x03) << 22) | ((stream & 0x3F) << 16) | (threshold & 0xFFFF)
|
||||
pkt += struct.pack(">I", w0)
|
||||
|
||||
# Word 1: {long_chirp, long_listen}
|
||||
@@ -407,11 +407,11 @@ class TestRadarFrameDefaults(unittest.TestCase):
|
||||
|
||||
def test_default_shapes(self):
|
||||
f = RadarFrame()
|
||||
self.assertEqual(f.range_doppler_i.shape, (64, 32))
|
||||
self.assertEqual(f.range_doppler_q.shape, (64, 32))
|
||||
self.assertEqual(f.magnitude.shape, (64, 32))
|
||||
self.assertEqual(f.detections.shape, (64, 32))
|
||||
self.assertEqual(f.range_profile.shape, (64,))
|
||||
self.assertEqual(f.range_doppler_i.shape, (512, 32))
|
||||
self.assertEqual(f.range_doppler_q.shape, (512, 32))
|
||||
self.assertEqual(f.magnitude.shape, (512, 32))
|
||||
self.assertEqual(f.detections.shape, (512, 32))
|
||||
self.assertEqual(f.range_profile.shape, (512,))
|
||||
self.assertEqual(f.detection_count, 0)
|
||||
|
||||
def test_default_zeros(self):
|
||||
|
||||
@@ -66,8 +66,8 @@ class TestRadarSettings(unittest.TestCase):
|
||||
def test_defaults(self):
|
||||
s = _models().RadarSettings()
|
||||
self.assertEqual(s.system_frequency, 10.5e9)
|
||||
self.assertEqual(s.coverage_radius, 1536)
|
||||
self.assertEqual(s.max_distance, 1536)
|
||||
self.assertEqual(s.coverage_radius, 3072)
|
||||
self.assertEqual(s.max_distance, 3072)
|
||||
|
||||
|
||||
class TestGPSData(unittest.TestCase):
|
||||
@@ -430,16 +430,16 @@ class TestWaveformConfig(unittest.TestCase):
|
||||
self.assertEqual(wc.chirp_duration_s, 30e-6)
|
||||
self.assertEqual(wc.pri_s, 167e-6)
|
||||
self.assertEqual(wc.center_freq_hz, 10.5e9)
|
||||
self.assertEqual(wc.n_range_bins, 64)
|
||||
self.assertEqual(wc.n_range_bins, 512)
|
||||
self.assertEqual(wc.n_doppler_bins, 32)
|
||||
self.assertEqual(wc.fft_size, 1024)
|
||||
self.assertEqual(wc.decimation_factor, 16)
|
||||
self.assertEqual(wc.fft_size, 2048)
|
||||
self.assertEqual(wc.decimation_factor, 4)
|
||||
|
||||
def test_range_resolution(self):
|
||||
"""bin_spacing_m should be ~24.0 m/bin with PLFM defaults."""
|
||||
"""bin_spacing_m should be ~6.0 m/bin with PLFM defaults."""
|
||||
from v7.models import WaveformConfig
|
||||
wc = WaveformConfig()
|
||||
self.assertAlmostEqual(wc.bin_spacing_m, 23.98, places=1)
|
||||
self.assertAlmostEqual(wc.bin_spacing_m, 5.996, places=1)
|
||||
|
||||
def test_range_resolution_physical(self):
|
||||
"""range_resolution_m = c/(2*BW), ~7.5 m at 20 MHz BW."""
|
||||
@@ -460,7 +460,7 @@ class TestWaveformConfig(unittest.TestCase):
|
||||
"""max_range_m = bin_spacing * n_range_bins."""
|
||||
from v7.models import WaveformConfig
|
||||
wc = WaveformConfig()
|
||||
self.assertAlmostEqual(wc.max_range_m, wc.bin_spacing_m * 64, places=1)
|
||||
self.assertAlmostEqual(wc.max_range_m, wc.bin_spacing_m * 512, places=1)
|
||||
|
||||
def test_max_velocity(self):
|
||||
"""max_velocity_mps = velocity_resolution * n_doppler_bins / 2."""
|
||||
@@ -613,15 +613,15 @@ class TestSoftwareFPGASignalChain(unittest.TestCase):
|
||||
from v7.software_fpga import SoftwareFPGA
|
||||
from radar_protocol import RadarFrame
|
||||
|
||||
# Load decimated range data as minimal input (32 chirps x 64 bins)
|
||||
# Load decimated range data as minimal input (32 chirps x 512 bins)
|
||||
dec_i = np.load(os.path.join(self.COSIM_DIR, "decimated_range_i.npy"))
|
||||
dec_q = np.load(os.path.join(self.COSIM_DIR, "decimated_range_q.npy"))
|
||||
|
||||
# Build fake 1024-sample chirps from decimated data (pad with zeros)
|
||||
# Build fake 2048-sample chirps from decimated data (pad with zeros)
|
||||
n_chirps = dec_i.shape[0]
|
||||
iq_i = np.zeros((n_chirps, 1024), dtype=np.int64)
|
||||
iq_q = np.zeros((n_chirps, 1024), dtype=np.int64)
|
||||
# Put decimated data into first 64 bins so FFT has something
|
||||
iq_i = np.zeros((n_chirps, 2048), dtype=np.int64)
|
||||
iq_q = np.zeros((n_chirps, 2048), dtype=np.int64)
|
||||
# Put decimated data into first bins so FFT has something
|
||||
iq_i[:, :dec_i.shape[1]] = dec_i
|
||||
iq_q[:, :dec_q.shape[1]] = dec_q
|
||||
|
||||
@@ -631,11 +631,11 @@ class TestSoftwareFPGASignalChain(unittest.TestCase):
|
||||
self.assertIsInstance(frame, RadarFrame)
|
||||
self.assertEqual(frame.frame_number, 42)
|
||||
self.assertAlmostEqual(frame.timestamp, 1.0)
|
||||
self.assertEqual(frame.range_doppler_i.shape, (64, 32))
|
||||
self.assertEqual(frame.range_doppler_q.shape, (64, 32))
|
||||
self.assertEqual(frame.magnitude.shape, (64, 32))
|
||||
self.assertEqual(frame.detections.shape, (64, 32))
|
||||
self.assertEqual(frame.range_profile.shape, (64,))
|
||||
self.assertEqual(frame.range_doppler_i.shape, (512, 32))
|
||||
self.assertEqual(frame.range_doppler_q.shape, (512, 32))
|
||||
self.assertEqual(frame.magnitude.shape, (512, 32))
|
||||
self.assertEqual(frame.detections.shape, (512, 32))
|
||||
self.assertEqual(frame.range_profile.shape, (512,))
|
||||
self.assertEqual(frame.detection_count, int(frame.detections.sum()))
|
||||
|
||||
def test_cfar_enable_changes_detections(self):
|
||||
@@ -644,8 +644,8 @@ class TestSoftwareFPGASignalChain(unittest.TestCase):
|
||||
self.skipTest("co-sim data not found")
|
||||
from v7.software_fpga import SoftwareFPGA
|
||||
|
||||
iq_i = np.zeros((32, 1024), dtype=np.int64)
|
||||
iq_q = np.zeros((32, 1024), dtype=np.int64)
|
||||
iq_i = np.zeros((32, 2048), dtype=np.int64)
|
||||
iq_q = np.zeros((32, 2048), dtype=np.int64)
|
||||
# Inject a single strong tone in bin 10 of every chirp
|
||||
iq_i[:, 10] = 5000
|
||||
iq_q[:, 10] = 3000
|
||||
@@ -662,8 +662,8 @@ class TestSoftwareFPGASignalChain(unittest.TestCase):
|
||||
# Just verify both produce valid frames — exact counts depend on chain
|
||||
self.assertIsNotNone(frame_thresh)
|
||||
self.assertIsNotNone(frame_cfar)
|
||||
self.assertEqual(frame_thresh.magnitude.shape, (64, 32))
|
||||
self.assertEqual(frame_cfar.magnitude.shape, (64, 32))
|
||||
self.assertEqual(frame_thresh.magnitude.shape, (512, 32))
|
||||
self.assertEqual(frame_cfar.magnitude.shape, (512, 32))
|
||||
|
||||
|
||||
class TestGoldenReferenceReplayFixtures(unittest.TestCase):
|
||||
@@ -794,11 +794,11 @@ class TestGoldenReferenceReplayFixtures(unittest.TestCase):
|
||||
from v7.software_fpga import SoftwareFPGA
|
||||
from radar_protocol import RadarFrame
|
||||
|
||||
# Use decimated data padded to 1024 as input (so range FFT has content)
|
||||
# Use decimated data padded to 2048 as input (so range FFT has content)
|
||||
dec_i = self._load("decimated_range_i.npy")
|
||||
dec_q = self._load("decimated_range_q.npy")
|
||||
iq_i = np.zeros((32, 1024), dtype=np.int64)
|
||||
iq_q = np.zeros((32, 1024), dtype=np.int64)
|
||||
iq_i = np.zeros((32, 2048), dtype=np.int64)
|
||||
iq_q = np.zeros((32, 2048), dtype=np.int64)
|
||||
iq_i[:, :dec_i.shape[1]] = dec_i
|
||||
iq_q[:, :dec_q.shape[1]] = dec_q
|
||||
|
||||
@@ -809,10 +809,10 @@ class TestGoldenReferenceReplayFixtures(unittest.TestCase):
|
||||
frame = fpga.process_chirps(iq_i, iq_q, frame_number=99)
|
||||
|
||||
self.assertIsInstance(frame, RadarFrame)
|
||||
self.assertEqual(frame.range_doppler_i.shape, (64, 32))
|
||||
self.assertEqual(frame.magnitude.shape, (64, 32))
|
||||
self.assertEqual(frame.detections.shape, (64, 32))
|
||||
self.assertEqual(frame.range_profile.shape, (64,))
|
||||
self.assertEqual(frame.range_doppler_i.shape, (512, 32))
|
||||
self.assertEqual(frame.magnitude.shape, (512, 32))
|
||||
self.assertEqual(frame.detections.shape, (512, 32))
|
||||
self.assertEqual(frame.range_profile.shape, (512,))
|
||||
self.assertEqual(frame.frame_number, 99)
|
||||
|
||||
def test_software_fpga_wiring_matches_manual_chain(self):
|
||||
@@ -824,8 +824,8 @@ class TestGoldenReferenceReplayFixtures(unittest.TestCase):
|
||||
This catches wiring bugs: wrong stage order, wrong default params,
|
||||
missing DC notch, etc.
|
||||
"""
|
||||
from v7.software_fpga import SoftwareFPGA, TWIDDLE_1024, TWIDDLE_16
|
||||
if not os.path.exists(TWIDDLE_1024) or not os.path.exists(TWIDDLE_16):
|
||||
from v7.software_fpga import SoftwareFPGA, TWIDDLE_2048, TWIDDLE_16
|
||||
if not os.path.exists(TWIDDLE_2048) or not os.path.exists(TWIDDLE_16):
|
||||
self.skipTest("twiddle files not found")
|
||||
|
||||
import sys as _sys
|
||||
@@ -843,8 +843,8 @@ class TestGoldenReferenceReplayFixtures(unittest.TestCase):
|
||||
|
||||
# Deterministic synthetic input — small int16 values to avoid overflow
|
||||
rng = np.random.RandomState(42)
|
||||
iq_i = rng.randint(-500, 500, size=(32, 1024), dtype=np.int64)
|
||||
iq_q = rng.randint(-500, 500, size=(32, 1024), dtype=np.int64)
|
||||
iq_i = rng.randint(-500, 500, size=(32, 2048), dtype=np.int64)
|
||||
iq_q = rng.randint(-500, 500, size=(32, 2048), dtype=np.int64)
|
||||
|
||||
# --- SoftwareFPGA path (what we're testing) ---
|
||||
fpga = SoftwareFPGA()
|
||||
@@ -858,7 +858,7 @@ class TestGoldenReferenceReplayFixtures(unittest.TestCase):
|
||||
range_q = np.zeros_like(iq_q)
|
||||
for c in range(32):
|
||||
range_i[c], range_q[c] = run_range_fft(
|
||||
iq_i[c], iq_q[c], twiddle_file=TWIDDLE_1024,
|
||||
iq_i[c], iq_q[c], twiddle_file=TWIDDLE_2048,
|
||||
)
|
||||
dec_i, dec_q = run_range_bin_decimator(range_i, range_q)
|
||||
mti_i, mti_q = run_mti_canceller(dec_i, dec_q, enable=True)
|
||||
@@ -897,24 +897,24 @@ class TestQuantizeRawIQ(unittest.TestCase):
|
||||
def test_3d_input(self):
|
||||
"""3-D (frames, chirps, samples) → uses first frame."""
|
||||
from v7.software_fpga import quantize_raw_iq
|
||||
raw = np.random.randn(5, 32, 1024) + 1j * np.random.randn(5, 32, 1024)
|
||||
raw = np.random.randn(5, 32, 2048) + 1j * np.random.randn(5, 32, 2048)
|
||||
iq_i, iq_q = quantize_raw_iq(raw)
|
||||
self.assertEqual(iq_i.shape, (32, 1024))
|
||||
self.assertEqual(iq_q.shape, (32, 1024))
|
||||
self.assertEqual(iq_i.shape, (32, 2048))
|
||||
self.assertEqual(iq_q.shape, (32, 2048))
|
||||
self.assertTrue(np.all(np.abs(iq_i) <= 32767))
|
||||
self.assertTrue(np.all(np.abs(iq_q) <= 32767))
|
||||
|
||||
def test_2d_input(self):
|
||||
"""2-D (chirps, samples) → works directly."""
|
||||
from v7.software_fpga import quantize_raw_iq
|
||||
raw = np.random.randn(32, 1024) + 1j * np.random.randn(32, 1024)
|
||||
raw = np.random.randn(32, 2048) + 1j * np.random.randn(32, 2048)
|
||||
iq_i, _iq_q = quantize_raw_iq(raw)
|
||||
self.assertEqual(iq_i.shape, (32, 1024))
|
||||
self.assertEqual(iq_i.shape, (32, 2048))
|
||||
|
||||
def test_zero_input(self):
|
||||
"""All-zero complex input → all-zero output."""
|
||||
from v7.software_fpga import quantize_raw_iq
|
||||
raw = np.zeros((32, 1024), dtype=np.complex128)
|
||||
raw = np.zeros((32, 2048), dtype=np.complex128)
|
||||
iq_i, iq_q = quantize_raw_iq(raw)
|
||||
self.assertTrue(np.all(iq_i == 0))
|
||||
self.assertTrue(np.all(iq_q == 0))
|
||||
@@ -952,7 +952,7 @@ class TestDetectFormat(unittest.TestCase):
|
||||
from v7.replay import detect_format, ReplayFormat
|
||||
import tempfile
|
||||
with tempfile.NamedTemporaryFile(suffix=".npy", delete=False) as f:
|
||||
np.save(f, np.zeros((2, 32, 1024), dtype=np.complex128))
|
||||
np.save(f, np.zeros((2, 32, 2048), dtype=np.complex128))
|
||||
tmp = f.name
|
||||
try:
|
||||
self.assertEqual(detect_format(tmp), ReplayFormat.RAW_IQ_NPY)
|
||||
@@ -1004,8 +1004,8 @@ class TestReplayEngineCosim(unittest.TestCase):
|
||||
engine = ReplayEngine(self.COSIM_DIR)
|
||||
frame = engine.get_frame(0)
|
||||
self.assertIsInstance(frame, RadarFrame)
|
||||
self.assertEqual(frame.range_doppler_i.shape, (64, 32))
|
||||
self.assertEqual(frame.magnitude.shape, (64, 32))
|
||||
self.assertEqual(frame.range_doppler_i.shape, (512, 32))
|
||||
self.assertEqual(frame.magnitude.shape, (512, 32))
|
||||
|
||||
def test_get_frame_out_of_range(self):
|
||||
if not self._available():
|
||||
@@ -1055,7 +1055,7 @@ class TestReplayEngineRawIQ(unittest.TestCase):
|
||||
engine = ReplayEngine(tmp, software_fpga=fpga)
|
||||
frame = engine.get_frame(0)
|
||||
self.assertIsInstance(frame, RadarFrame)
|
||||
self.assertEqual(frame.range_doppler_i.shape, (64, 32))
|
||||
self.assertEqual(frame.range_doppler_i.shape, (512, 32))
|
||||
self.assertEqual(frame.frame_number, 0)
|
||||
finally:
|
||||
os.unlink(tmp)
|
||||
@@ -1100,7 +1100,7 @@ class TestReplayEngineHDF5(unittest.TestCase):
|
||||
try:
|
||||
with h5py.File(tmp, "w") as hf:
|
||||
hf.attrs["creator"] = "test"
|
||||
hf.attrs["range_bins"] = 64
|
||||
hf.attrs["range_bins"] = 512
|
||||
hf.attrs["doppler_bins"] = 32
|
||||
grp = hf.create_group("frames")
|
||||
for i in range(3):
|
||||
@@ -1109,15 +1109,15 @@ class TestReplayEngineHDF5(unittest.TestCase):
|
||||
fg.attrs["frame_number"] = i
|
||||
fg.attrs["detection_count"] = 0
|
||||
fg.create_dataset("range_doppler_i",
|
||||
data=np.zeros((64, 32), dtype=np.int16))
|
||||
data=np.zeros((512, 32), dtype=np.int16))
|
||||
fg.create_dataset("range_doppler_q",
|
||||
data=np.zeros((64, 32), dtype=np.int16))
|
||||
data=np.zeros((512, 32), dtype=np.int16))
|
||||
fg.create_dataset("magnitude",
|
||||
data=np.zeros((64, 32), dtype=np.float64))
|
||||
data=np.zeros((512, 32), dtype=np.float64))
|
||||
fg.create_dataset("detections",
|
||||
data=np.zeros((64, 32), dtype=np.uint8))
|
||||
data=np.zeros((512, 32), dtype=np.uint8))
|
||||
fg.create_dataset("range_profile",
|
||||
data=np.zeros(64, dtype=np.float64))
|
||||
data=np.zeros(512, dtype=np.float64))
|
||||
|
||||
engine = ReplayEngine(tmp)
|
||||
self.assertEqual(engine.fmt, ReplayFormat.HDF5)
|
||||
@@ -1126,7 +1126,7 @@ class TestReplayEngineHDF5(unittest.TestCase):
|
||||
frame = engine.get_frame(1)
|
||||
self.assertIsInstance(frame, RadarFrame)
|
||||
self.assertEqual(frame.frame_number, 1)
|
||||
self.assertEqual(frame.range_doppler_i.shape, (64, 32))
|
||||
self.assertEqual(frame.range_doppler_i.shape, (512, 32))
|
||||
engine.close()
|
||||
finally:
|
||||
os.unlink(tmp)
|
||||
@@ -1161,9 +1161,9 @@ class TestExtractTargetsFromFrame(unittest.TestCase):
|
||||
"""Detection at range bin 10 → range = 10 * range_resolution."""
|
||||
from v7.processing import extract_targets_from_frame
|
||||
frame = self._make_frame(det_cells=[(10, 16)]) # dbin=16 = center → vel=0
|
||||
targets = extract_targets_from_frame(frame, bin_spacing=23.98)
|
||||
targets = extract_targets_from_frame(frame, bin_spacing=5.996)
|
||||
self.assertEqual(len(targets), 1)
|
||||
self.assertAlmostEqual(targets[0].range, 10 * 23.98, places=1)
|
||||
self.assertAlmostEqual(targets[0].range, 10 * 5.996, places=1)
|
||||
self.assertAlmostEqual(targets[0].velocity, 0.0, places=2)
|
||||
|
||||
def test_velocity_sign(self):
|
||||
|
||||
@@ -72,7 +72,7 @@ if TYPE_CHECKING:
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Frame dimensions from FPGA
|
||||
NUM_RANGE_BINS = 64
|
||||
NUM_RANGE_BINS = 512
|
||||
NUM_DOPPLER_BINS = 32
|
||||
|
||||
# Force C locale (period as decimal separator) for all QDoubleSpinBox instances.
|
||||
@@ -92,7 +92,7 @@ def _make_dspin() -> QDoubleSpinBox:
|
||||
# =============================================================================
|
||||
|
||||
class RangeDopplerCanvas(FigureCanvasQTAgg):
|
||||
"""Matplotlib canvas showing the 64x32 Range-Doppler map with dark theme."""
|
||||
"""Matplotlib canvas showing the 512x32 Range-Doppler map with dark theme."""
|
||||
|
||||
def __init__(self, _parent=None):
|
||||
fig = Figure(figsize=(10, 6), facecolor=DARK_BG)
|
||||
@@ -104,7 +104,7 @@ class RangeDopplerCanvas(FigureCanvasQTAgg):
|
||||
extent=[0, NUM_DOPPLER_BINS, 0, NUM_RANGE_BINS], origin="lower",
|
||||
)
|
||||
|
||||
self.ax.set_title("Range-Doppler Map (64x32)", color=DARK_FG)
|
||||
self.ax.set_title("Range-Doppler Map (512x32)", color=DARK_FG)
|
||||
self.ax.set_xlabel("Doppler Bin", color=DARK_FG)
|
||||
self.ax.set_ylabel("Range Bin", color=DARK_FG)
|
||||
self.ax.tick_params(colors=DARK_FG)
|
||||
@@ -643,9 +643,9 @@ class RadarDashboard(QMainWindow):
|
||||
btn_trigger.clicked.connect(lambda: self._send_fpga_cmd(0x02, 1))
|
||||
op_layout.addWidget(btn_trigger)
|
||||
|
||||
# Stream Control (3-bit mask)
|
||||
self._add_fpga_param_row(op_layout, "Stream Control", 0x04, 7, 3,
|
||||
"0-7, 3-bit mask, rst=7")
|
||||
# Stream Control (6-bit)
|
||||
self._add_fpga_param_row(op_layout, "Stream Control", 0x04, 63, 6,
|
||||
"0-63, 6-bit stream/format, rst=15")
|
||||
|
||||
btn_status = QPushButton("Request Status")
|
||||
btn_status.clicked.connect(lambda: self._send_fpga_cmd(0xFF, 0))
|
||||
@@ -1639,7 +1639,7 @@ class RadarDashboard(QMainWindow):
|
||||
|
||||
@pyqtSlot(object)
|
||||
def _on_frame_ready(self, frame: RadarFrame):
|
||||
"""Handle a complete 64x32 radar frame from production acquisition."""
|
||||
"""Handle a complete 512x32 radar frame from production acquisition."""
|
||||
self._current_frame = frame
|
||||
self._frame_count += 1
|
||||
|
||||
|
||||
@@ -98,7 +98,7 @@ class RadarMapWidget(QWidget):
|
||||
)
|
||||
self._targets: list[RadarTarget] = []
|
||||
self._pending_targets: list[RadarTarget] | None = None
|
||||
self._coverage_radius = 1_536 # metres (64 bins x 24 m, 3 km mode)
|
||||
self._coverage_radius = 3_072 # metres (512 bins x 6 m, 3 km mode)
|
||||
self._tile_server = TileServer.OPENSTREETMAP
|
||||
self._show_coverage = True
|
||||
self._show_trails = False
|
||||
|
||||
@@ -109,11 +109,11 @@ class RadarSettings:
|
||||
the actual waveform parameters.
|
||||
"""
|
||||
system_frequency: float = 10.5e9 # Hz (PLFM TX LO, verified from ADF4382 config)
|
||||
range_bin_spacing: float = 24.0 # Meters per decimated range bin (c/(2*100MSPS)*16)
|
||||
range_bin_spacing: float = 6.0 # Meters per decimated range bin (c/(2*100MSPS)*4)
|
||||
velocity_resolution: float = 2.67 # m/s per Doppler bin (lam/(2*32*167us))
|
||||
max_distance: float = 1536 # Max detection range (m) -- 64 bins x 24 m (3 km mode)
|
||||
map_size: float = 1536 # Map display size (m)
|
||||
coverage_radius: float = 1536 # Map coverage radius (m)
|
||||
max_distance: float = 3072 # Max detection range (m) -- 512 bins x 6 m (3 km mode)
|
||||
map_size: float = 3072 # Map display size (m)
|
||||
coverage_radius: float = 3072 # Map coverage radius (m)
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -210,10 +210,10 @@ class WaveformConfig:
|
||||
chirp_duration_s: float = 30e-6 # Long chirp ramp (informational only)
|
||||
pri_s: float = 167e-6 # Pulse repetition interval (chirp + listen)
|
||||
center_freq_hz: float = 10.5e9 # TX LO carrier (verified: ADF4382 config)
|
||||
n_range_bins: int = 64 # After decimation (3 km mode)
|
||||
n_range_bins: int = 512 # After decimation (3 km mode)
|
||||
n_doppler_bins: int = 32 # After Doppler FFT
|
||||
fft_size: int = 1024 # Pre-decimation FFT length
|
||||
decimation_factor: int = 16 # 1024 → 64
|
||||
fft_size: int = 2048 # Pre-decimation FFT length
|
||||
decimation_factor: int = 4 # 2048 → 512
|
||||
|
||||
@property
|
||||
def bin_spacing_m(self) -> float:
|
||||
|
||||
@@ -56,7 +56,7 @@ class RadarProcessor:
|
||||
"""Full radar processing pipeline: fusion, clustering, association, tracking."""
|
||||
|
||||
def __init__(self):
|
||||
self.range_doppler_map = np.zeros((1024, 32))
|
||||
self.range_doppler_map = np.zeros((512, 32))
|
||||
self.detected_targets: list[RadarTarget] = []
|
||||
self.track_id_counter: int = 0
|
||||
self.tracks: dict[int, dict] = {}
|
||||
|
||||
@@ -56,7 +56,8 @@ log = logging.getLogger(__name__)
|
||||
# Twiddle factor file paths (relative to FPGA root)
|
||||
# ---------------------------------------------------------------------------
|
||||
_FPGA_DIR = Path(__file__).resolve().parents[2] / "9_2_FPGA"
|
||||
TWIDDLE_1024 = str(_FPGA_DIR / "fft_twiddle_1024.mem")
|
||||
TWIDDLE_2048 = str(_FPGA_DIR / "fft_twiddle_2048.mem")
|
||||
TWIDDLE_1024 = str(_FPGA_DIR / "fft_twiddle_1024.mem") # kept for reference
|
||||
TWIDDLE_16 = str(_FPGA_DIR / "fft_twiddle_16.mem")
|
||||
|
||||
# CFAR mode int→string mapping (FPGA register 0x24: 0=CA, 1=GO, 2=SO)
|
||||
@@ -179,15 +180,19 @@ class SoftwareFPGA:
|
||||
# --- Stage 1: Range FFT (per chirp) ---
|
||||
range_i = np.zeros((n_chirps, n_samples), dtype=np.int64)
|
||||
range_q = np.zeros((n_chirps, n_samples), dtype=np.int64)
|
||||
twiddle_1024 = TWIDDLE_1024 if os.path.exists(TWIDDLE_1024) else None
|
||||
# Select twiddle file matching input FFT size
|
||||
if n_samples >= 2048:
|
||||
twiddle = TWIDDLE_2048 if os.path.exists(TWIDDLE_2048) else None
|
||||
else:
|
||||
twiddle = TWIDDLE_1024 if os.path.exists(TWIDDLE_1024) else None
|
||||
for c in range(n_chirps):
|
||||
range_i[c], range_q[c] = run_range_fft(
|
||||
iq_i[c].astype(np.int64),
|
||||
iq_q[c].astype(np.int64),
|
||||
twiddle_file=twiddle_1024,
|
||||
twiddle_file=twiddle,
|
||||
)
|
||||
|
||||
# --- Stage 2: Range bin decimation (1024 → 64) ---
|
||||
# --- Stage 2: Range bin decimation (2048 → 512) ---
|
||||
decim_i, decim_q = run_range_bin_decimator(range_i, range_q)
|
||||
|
||||
# --- Stage 3: MTI canceller (pre-Doppler, per-chirp) ---
|
||||
@@ -230,6 +235,10 @@ class SoftwareFPGA:
|
||||
frame.range_doppler_q = np.clip(notch_q, -32768, 32767).astype(np.int16)
|
||||
frame.magnitude = mag
|
||||
frame.detections = det_mask
|
||||
# Range profile: magnitude at Doppler bin 0 (zero-velocity / stationary).
|
||||
# This differs from the FPGA USB stream which sends per-chirp decimated
|
||||
# Manhattan magnitude. The zero-Doppler slice is more useful for the
|
||||
# host-side display because it represents coherently integrated range energy.
|
||||
frame.range_profile = np.sqrt(
|
||||
notch_i[:, 0].astype(np.float64) ** 2
|
||||
+ notch_q[:, 0].astype(np.float64) ** 2
|
||||
@@ -257,7 +266,7 @@ def quantize_raw_iq(
|
||||
n_chirps : int
|
||||
Number of chirps to keep (default 32, matching FPGA).
|
||||
n_samples : int
|
||||
Number of samples per chirp to keep (default 1024, matching FFT).
|
||||
Number of samples per chirp to keep (default 2048, matching FFT).
|
||||
peak_target : int
|
||||
Target peak magnitude after scaling (default 200, matching
|
||||
golden_reference INPUT_PEAK_TARGET).
|
||||
@@ -270,8 +279,11 @@ def quantize_raw_iq(
|
||||
# (frames, chirps, samples) — take first frame
|
||||
raw_complex = raw_complex[0]
|
||||
|
||||
# Truncate to FPGA dimensions
|
||||
block = raw_complex[:n_chirps, :n_samples]
|
||||
# Truncate chirps, zero-pad samples if source is shorter than n_samples
|
||||
block = np.zeros((n_chirps, n_samples), dtype=raw_complex.dtype)
|
||||
avail_chirps = min(raw_complex.shape[0], n_chirps)
|
||||
avail_samples = min(raw_complex.shape[1], n_samples)
|
||||
block[:avail_chirps, :avail_samples] = raw_complex[:avail_chirps, :avail_samples]
|
||||
|
||||
max_abs = np.max(np.abs(block))
|
||||
if max_abs == 0:
|
||||
|
||||
@@ -3,7 +3,7 @@ v7.workers — QThread-based workers and demo target simulator.
|
||||
|
||||
Classes:
|
||||
- RadarDataWorker — reads from FT2232H via production RadarAcquisition,
|
||||
parses 0xAA/0xBB packets, assembles 64x32 frames,
|
||||
parses 0xAA/0xBB packets, assembles 512x32 frames,
|
||||
runs host-side DSP, emits PyQt signals.
|
||||
- GPSDataWorker — reads GPS frames from STM32 CDC, emits GPSData signals.
|
||||
- TargetSimulator — QTimer-based demo target generator.
|
||||
@@ -52,11 +52,11 @@ class RadarDataWorker(QThread):
|
||||
and emits PyQt signals with results.
|
||||
|
||||
Uses production radar_protocol.py for all packet parsing and frame
|
||||
assembly (11-byte 0xAA data packets → 64x32 RadarFrame).
|
||||
assembly (bulk frames or legacy 11-byte 0xAA data packets → 512x32 RadarFrame).
|
||||
For replay, use ReplayWorker instead.
|
||||
|
||||
Signals:
|
||||
frameReady(RadarFrame) — a complete 64x32 radar frame
|
||||
frameReady(RadarFrame) — a complete 512x32 radar frame
|
||||
statusReceived(object) — StatusResponse from FPGA
|
||||
targetsUpdated(list) — list of RadarTarget after host-side DSP
|
||||
errorOccurred(str) — error message
|
||||
@@ -368,7 +368,7 @@ class TargetSimulator(QObject):
|
||||
|
||||
for t in self._targets:
|
||||
new_range = t.range - t.velocity * 0.5
|
||||
if new_range < 50 or new_range > 1536:
|
||||
if new_range < 50 or new_range > 3072:
|
||||
continue # target exits coverage — drop it
|
||||
|
||||
new_vel = max(-150, min(150, t.velocity + random.uniform(-2, 2)))
|
||||
|
||||
Reference in New Issue
Block a user