mirror of
https://github.com/NawfalMotii79/PLFM_RADAR.git
synced 2026-06-10 23:41:18 +00:00
fix(gui): software_fpga revival post-e8b495c — port chain helpers to fpga_model
Restore SoftwareFPGA's process_chirps() pipeline by porting the missing
chain stages (MTI canceller, DC notch, CFAR, threshold detection) plus
thin wrappers (range FFT, decimator, Doppler FFT) to fpga_model.py and
swapping software_fpga.py's import target from the deleted
golden_reference.py to fpga_model.
History: golden_reference.py was deleted in e8b495c (the "dead golden
code cleanup") but software_fpga.py kept importing from it. The
ImportError was swallowed at v7/__init__.py:49-52 so package load
succeeded, but every direct `from v7.software_fpga import SoftwareFPGA`
hit the import-time failure — masking 21 broken tests as
"ModuleNotFoundError" instead of surfacing the real issue.
This was actively breaking the GUI replay-from-raw-IQ feature
(dashboard.py:1334-1347, 1577 + GUI_V65_Tk.py:271-300, 1106-1129):
opening a .npy SDR capture instantiates SoftwareFPGA + ReplayEngine;
the dashboard's opcode dual-dispatch routes spinbox changes to the
SoftwareFPGA setters so re-processing reflects live param tweaks.
With the import broken since April, that path silently dies.
fpga_model.py:
- New top-level constants: FFT_SIZE=2048, NUM_RANGE_BINS=512 (from
RangeBinDecimator.OUTPUT_BINS), DOPPLER_CHIRPS=48,
DOPPLER_TOTAL_BINS=48 (track current production: PR-O.6 / PR-F).
- run_range_fft(iq_i, iq_q, twiddle_file): N inferred from input
length; works for legacy 1024-pt and production 2048-pt callers.
- run_range_bin_decimator(range_i, range_q, mode): per-frame wrapper
over RangeBinDecimator.decimate (4x decim -> 512 bins).
- run_mti_canceller(decim_i, decim_q, enable): 2-pulse canceller,
ported verbatim from golden_reference @ commit 237e74c~1.
- run_doppler_fft(mti_i, mti_q): num_subframes inferred from chirp
count; RANGE_BINS overridden per input shape so legacy
2-sub-frame (32-chirp) and production 3-sub-frame (48-chirp)
callers both work.
- run_dc_notch(doppler_i, doppler_q, width): per-bin DC notch,
generalised to any sub-frame count.
- run_cfar_ca(...): CA / GO / SO modes with bit-accurate alpha-q44
threshold + 17-bit saturation, ported from golden_reference.
- run_detection(doppler_i, doppler_q, threshold): |I|+|Q| L1 magnitude
threshold detection.
software_fpga.py:
- _GOLDEN_REF_DIR (cosim/real_data/) -> _FPGA_COSIM_DIR (cosim/)
- `from golden_reference import (...)` -> `from fpga_model import (...)`
- TWIDDLE_1024 -> TWIDDLE_2048 (production 2048-pt range FFT).
- Stage 1 comment: "Range bin decimation (1024 -> 64)" ->
"(production 2048 -> 512)".
- Stage 1 twiddle path picks fft_twiddle_2048.mem only when
n_samples=2048 matches; otherwise None to fall back to math-
generated twiddles for legacy callers.
- Module docstring updated to reflect post-cleanup history.
test_v7.py — modernise three tests to current production dimensions:
- test_process_chirps_returns_radar_frame: pad input to 2048 samples;
assertions reference NUM_RANGE_BINS / NUM_DOPPLER_BINS from
radar_protocol; n_dop derived from input chirp count.
- test_cfar_enable_changes_detections: 48 chirps x 2048 samples;
output (NUM_RANGE_BINS, NUM_DOPPLER_BINS). No longer skips on
cosim absence — uses synthetic input.
- test_get_frame_raw_iq_synthetic: (2, 48, 2048) raw IQ;
(NUM_RANGE_BINS, NUM_DOPPLER_BINS) output.
- test_cosim_dir: also skip when doppler_map_*.npy absent (matches
_cosim_available pattern in TestSoftwareFPGASignalChain).
Local: test_v7 100/0/0 (9 graceful skips: optional deps + missing
cosim .npy data), test_GUI_V65_Tk 117/0/2. Down from 21 ERRORs.
This commit is contained in:
@@ -1383,6 +1383,218 @@ class SignalChain:
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# Frame-level chain helpers — restored post-e8b495c golden_reference deletion.
|
||||||
|
#
|
||||||
|
# These wrap the bit-accurate stage classes above (FFTEngine, RangeBinDecimator,
|
||||||
|
# DopplerProcessor) and add the missing stages (MTI, DC notch, CFAR, threshold
|
||||||
|
# detection) ported from golden_reference.py @ commit 237e74c~1. Used by
|
||||||
|
# v7.software_fpga (replay-from-raw-IQ in the GUIs).
|
||||||
|
#
|
||||||
|
# Dimensions track current production (PR-O.6 / PR-F):
|
||||||
|
# FFT_SIZE = 2048 (range FFT N)
|
||||||
|
# NUM_RANGE_BINS = 512 (after RangeBinDecimator 4x)
|
||||||
|
# DOPPLER_CHIRPS = 48 (3 sub-frames x 16 chirps)
|
||||||
|
# DOPPLER_TOTAL_BINS = 48 (3 sub-frames x 16-pt FFT)
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
import numpy as np # noqa: E402
|
||||||
|
|
||||||
|
FFT_SIZE = 2048
|
||||||
|
NUM_RANGE_BINS = RangeBinDecimator.OUTPUT_BINS # 512
|
||||||
|
DOPPLER_CHIRPS = DopplerProcessor.CHIRPS_PER_FRAME # 48
|
||||||
|
DOPPLER_TOTAL_BINS = (DopplerProcessor.NUM_SUBFRAMES
|
||||||
|
* DopplerProcessor.DOPPLER_FFT_SIZE) # 48
|
||||||
|
|
||||||
|
|
||||||
|
def run_range_fft(iq_i, iq_q, twiddle_file=None):
|
||||||
|
"""Per-chirp range FFT (wrapper around FFTEngine).
|
||||||
|
|
||||||
|
N is inferred from input length so legacy 1024-pt callers and
|
||||||
|
production 2048-pt callers both work.
|
||||||
|
"""
|
||||||
|
n = len(iq_i)
|
||||||
|
fft = FFTEngine(n=n, twiddle_file=twiddle_file)
|
||||||
|
out_re, out_im = fft.compute(list(iq_i), list(iq_q))
|
||||||
|
return np.array(out_re, dtype=np.int64), np.array(out_im, dtype=np.int64)
|
||||||
|
|
||||||
|
|
||||||
|
def run_range_bin_decimator(range_i, range_q, mode=1):
|
||||||
|
"""Per-frame range decimator (FFT_SIZE -> NUM_RANGE_BINS).
|
||||||
|
|
||||||
|
mode: 0=center, 1=peak, 2=average, 3=zero (matches RTL register 0x0E).
|
||||||
|
Output length is always RangeBinDecimator.OUTPUT_BINS (512 in production);
|
||||||
|
short inputs are zero-filled past the available source bins.
|
||||||
|
"""
|
||||||
|
n_chirps = range_i.shape[0]
|
||||||
|
decim_i = np.zeros((n_chirps, NUM_RANGE_BINS), dtype=np.int64)
|
||||||
|
decim_q = np.zeros((n_chirps, NUM_RANGE_BINS), dtype=np.int64)
|
||||||
|
for c in range(n_chirps):
|
||||||
|
out_re, out_im = RangeBinDecimator.decimate(
|
||||||
|
list(range_i[c]), list(range_q[c]), mode=mode,
|
||||||
|
)
|
||||||
|
decim_i[c, :] = out_re
|
||||||
|
decim_q[c, :] = out_im
|
||||||
|
return decim_i, decim_q
|
||||||
|
|
||||||
|
|
||||||
|
def run_mti_canceller(decim_i, decim_q, enable=True):
|
||||||
|
"""2-pulse MTI canceller (bit-accurate model of mti_canceller.v).
|
||||||
|
|
||||||
|
First chirp output is muted (no previous data); subsequent chirps:
|
||||||
|
out[c] = decim[c] - decim[c-1], saturated to 16-bit.
|
||||||
|
"""
|
||||||
|
if not enable:
|
||||||
|
return decim_i.copy(), decim_q.copy()
|
||||||
|
n_chirps, n_bins = decim_i.shape
|
||||||
|
mti_i = np.zeros_like(decim_i)
|
||||||
|
mti_q = np.zeros_like(decim_q)
|
||||||
|
for c in range(1, n_chirps):
|
||||||
|
for r in range(n_bins):
|
||||||
|
diff_i = int(decim_i[c, r]) - int(decim_i[c - 1, r])
|
||||||
|
diff_q = int(decim_q[c, r]) - int(decim_q[c - 1, r])
|
||||||
|
mti_i[c, r] = saturate(diff_i, 16)
|
||||||
|
mti_q[c, r] = saturate(diff_q, 16)
|
||||||
|
return mti_i, mti_q
|
||||||
|
|
||||||
|
|
||||||
|
def run_doppler_fft(mti_i, mti_q, twiddle_file_16=None):
|
||||||
|
"""Multi-sub-frame Doppler FFT (wrapper around DopplerProcessor.process_frame).
|
||||||
|
|
||||||
|
Sub-frame count and range-bin count are inferred from input shape so
|
||||||
|
legacy 2-sub-frame (32-chirp) and production 3-sub-frame (48-chirp)
|
||||||
|
callers both work.
|
||||||
|
|
||||||
|
Input : (n_chirps, n_rbins), 16-bit signed.
|
||||||
|
Output : (n_rbins, n_subframes * 16), 16-bit signed.
|
||||||
|
"""
|
||||||
|
n_chirps, n_rbins = mti_i.shape
|
||||||
|
chirps_per_sf = DopplerProcessor.CHIRPS_PER_SUBFRAME # 16
|
||||||
|
n_subframes = max(1, n_chirps // chirps_per_sf)
|
||||||
|
dp = DopplerProcessor(
|
||||||
|
twiddle_file_16=twiddle_file_16,
|
||||||
|
num_subframes=n_subframes,
|
||||||
|
)
|
||||||
|
dp.RANGE_BINS = n_rbins # override hardcoded production value
|
||||||
|
chirp_data_i = [list(mti_i[c]) for c in range(n_chirps)]
|
||||||
|
chirp_data_q = [list(mti_q[c]) for c in range(n_chirps)]
|
||||||
|
map_i, map_q = dp.process_frame(chirp_data_i, chirp_data_q)
|
||||||
|
return np.array(map_i, dtype=np.int64), np.array(map_q, dtype=np.int64)
|
||||||
|
|
||||||
|
|
||||||
|
def run_dc_notch(doppler_i, doppler_q, width=2):
|
||||||
|
"""Per-bin DC notch (bit-accurate model of radar_system_top.v inline filter).
|
||||||
|
|
||||||
|
bin_within_sf = dbin & 0xF; zero when
|
||||||
|
width != 0 and (bin_within_sf < width or bin_within_sf > 15-width+1).
|
||||||
|
Generalises to any number of sub-frames.
|
||||||
|
"""
|
||||||
|
notched_i = doppler_i.copy()
|
||||||
|
notched_q = doppler_q.copy()
|
||||||
|
if width == 0:
|
||||||
|
return notched_i, notched_q
|
||||||
|
n_doppler = doppler_i.shape[1]
|
||||||
|
for dbin in range(n_doppler):
|
||||||
|
bin_within_sf = dbin & 0xF
|
||||||
|
if bin_within_sf < width or bin_within_sf > (15 - width + 1):
|
||||||
|
notched_i[:, dbin] = 0
|
||||||
|
notched_q[:, dbin] = 0
|
||||||
|
return notched_i, notched_q
|
||||||
|
|
||||||
|
|
||||||
|
def run_cfar_ca(doppler_i, doppler_q, guard=2, train=8,
|
||||||
|
alpha_q44=0x30, mode='CA'):
|
||||||
|
"""CFAR detection — bit-accurate model of cfar_ca.v (CA / GO / SO modes).
|
||||||
|
|
||||||
|
Per Doppler column: |I|+|Q| L1 magnitude, then for each cell take
|
||||||
|
leading + lagging training cells (skipping ``guard`` cells),
|
||||||
|
threshold = (alpha_q44 * noise_sum) >> 4 saturated to 17 bits,
|
||||||
|
detect if magnitude > threshold.
|
||||||
|
|
||||||
|
Returns (detect_flags, magnitudes, thresholds) — each shape
|
||||||
|
(n_range, n_doppler).
|
||||||
|
"""
|
||||||
|
n_range, n_doppler = doppler_i.shape
|
||||||
|
ALPHA_FRAC_BITS = 4
|
||||||
|
if train == 0:
|
||||||
|
train = 1
|
||||||
|
|
||||||
|
magnitudes = np.zeros((n_range, n_doppler), dtype=np.int64)
|
||||||
|
for rbin in range(n_range):
|
||||||
|
for dbin in range(n_doppler):
|
||||||
|
i_val = int(doppler_i[rbin, dbin])
|
||||||
|
q_val = int(doppler_q[rbin, dbin])
|
||||||
|
abs_i = (-i_val) & 0xFFFF if i_val < 0 else i_val & 0xFFFF
|
||||||
|
abs_q = (-q_val) & 0xFFFF if q_val < 0 else q_val & 0xFFFF
|
||||||
|
magnitudes[rbin, dbin] = abs_i + abs_q
|
||||||
|
|
||||||
|
detect_flags = np.zeros((n_range, n_doppler), dtype=np.bool_)
|
||||||
|
thresholds = np.zeros((n_range, n_doppler), dtype=np.int64)
|
||||||
|
MAX_MAG = (1 << 17) - 1
|
||||||
|
|
||||||
|
for dbin in range(n_doppler):
|
||||||
|
col = magnitudes[:, dbin]
|
||||||
|
for cut_idx in range(n_range):
|
||||||
|
leading_sum = 0
|
||||||
|
leading_count = 0
|
||||||
|
for t in range(1, train + 1):
|
||||||
|
idx = cut_idx - guard - t
|
||||||
|
if 0 <= idx < n_range:
|
||||||
|
leading_sum += int(col[idx])
|
||||||
|
leading_count += 1
|
||||||
|
lagging_sum = 0
|
||||||
|
lagging_count = 0
|
||||||
|
for t in range(1, train + 1):
|
||||||
|
idx = cut_idx + guard + t
|
||||||
|
if 0 <= idx < n_range:
|
||||||
|
lagging_sum += int(col[idx])
|
||||||
|
lagging_count += 1
|
||||||
|
|
||||||
|
if mode in ('CA', 'CA-CFAR'):
|
||||||
|
noise_sum = leading_sum + lagging_sum
|
||||||
|
elif mode in ('GO', 'GO-CFAR'):
|
||||||
|
if leading_count > 0 and lagging_count > 0:
|
||||||
|
if leading_sum * lagging_count > lagging_sum * leading_count:
|
||||||
|
noise_sum = leading_sum
|
||||||
|
else:
|
||||||
|
noise_sum = lagging_sum
|
||||||
|
elif leading_count > 0:
|
||||||
|
noise_sum = leading_sum
|
||||||
|
else:
|
||||||
|
noise_sum = lagging_sum
|
||||||
|
elif mode in ('SO', 'SO-CFAR'):
|
||||||
|
if leading_count > 0 and lagging_count > 0:
|
||||||
|
if leading_sum * lagging_count < lagging_sum * leading_count:
|
||||||
|
noise_sum = leading_sum
|
||||||
|
else:
|
||||||
|
noise_sum = lagging_sum
|
||||||
|
elif leading_count > 0:
|
||||||
|
noise_sum = leading_sum
|
||||||
|
else:
|
||||||
|
noise_sum = lagging_sum
|
||||||
|
else:
|
||||||
|
noise_sum = leading_sum + lagging_sum
|
||||||
|
|
||||||
|
threshold_raw = (alpha_q44 * noise_sum) >> ALPHA_FRAC_BITS
|
||||||
|
threshold_val = MAX_MAG if threshold_raw > MAX_MAG else int(threshold_raw)
|
||||||
|
thresholds[cut_idx, dbin] = threshold_val
|
||||||
|
if int(col[cut_idx]) > threshold_val:
|
||||||
|
detect_flags[cut_idx, dbin] = True
|
||||||
|
|
||||||
|
return detect_flags, magnitudes, thresholds
|
||||||
|
|
||||||
|
|
||||||
|
def run_detection(doppler_i, doppler_q, threshold=10000):
|
||||||
|
"""Threshold detection — |I|+|Q| > threshold.
|
||||||
|
|
||||||
|
Returns (mag, det_indices) where det_indices is an (M, 2) array of
|
||||||
|
[rbin, dbin] cells exceeding the threshold.
|
||||||
|
"""
|
||||||
|
mag = np.abs(doppler_i.astype(np.int64)) + np.abs(doppler_q.astype(np.int64))
|
||||||
|
det_indices = np.argwhere(mag > threshold)
|
||||||
|
return mag, det_indices
|
||||||
|
|
||||||
|
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
# Self-test / Validation
|
# Self-test / Validation
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
|
|||||||
@@ -629,23 +629,23 @@ class TestSoftwareFPGASignalChain(unittest.TestCase):
|
|||||||
return os.path.isfile(os.path.join(self.COSIM_DIR, "doppler_map_i.npy"))
|
return os.path.isfile(os.path.join(self.COSIM_DIR, "doppler_map_i.npy"))
|
||||||
|
|
||||||
def test_process_chirps_returns_radar_frame(self):
|
def test_process_chirps_returns_radar_frame(self):
|
||||||
"""process_chirps produces a RadarFrame with correct shapes."""
|
"""process_chirps produces a RadarFrame with production shapes (PR-O.6 / PR-F)."""
|
||||||
if not self._cosim_available():
|
if not self._cosim_available():
|
||||||
self.skipTest("co-sim data not found")
|
self.skipTest("co-sim data not found")
|
||||||
from v7.software_fpga import SoftwareFPGA
|
from v7.software_fpga import SoftwareFPGA
|
||||||
from radar_protocol import RadarFrame
|
from radar_protocol import RadarFrame, NUM_RANGE_BINS, NUM_DOPPLER_BINS
|
||||||
|
|
||||||
# Load decimated range data as minimal input (32 chirps x 64 bins)
|
# Load decimated range data and pad up to current FPGA chirp width.
|
||||||
dec_i = np.load(os.path.join(self.COSIM_DIR, "decimated_range_i.npy"))
|
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"))
|
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)
|
# Production chirp width = FFT_SIZE = 2048 samples; pad with zeros.
|
||||||
n_chirps = dec_i.shape[0]
|
n_chirps = dec_i.shape[0]
|
||||||
iq_i = np.zeros((n_chirps, 1024), dtype=np.int64)
|
iq_i = np.zeros((n_chirps, 2048), dtype=np.int64)
|
||||||
iq_q = np.zeros((n_chirps, 1024), dtype=np.int64)
|
iq_q = np.zeros((n_chirps, 2048), dtype=np.int64)
|
||||||
# Put decimated data into first 64 bins so FFT has something
|
n_copy = min(2048, dec_i.shape[1])
|
||||||
iq_i[:, :dec_i.shape[1]] = dec_i
|
iq_i[:, :n_copy] = dec_i[:, :n_copy]
|
||||||
iq_q[:, :dec_q.shape[1]] = dec_q
|
iq_q[:, :n_copy] = dec_q[:, :n_copy]
|
||||||
|
|
||||||
fpga = SoftwareFPGA()
|
fpga = SoftwareFPGA()
|
||||||
frame = fpga.process_chirps(iq_i, iq_q, frame_number=42, timestamp=1.0)
|
frame = fpga.process_chirps(iq_i, iq_q, frame_number=42, timestamp=1.0)
|
||||||
@@ -653,22 +653,26 @@ class TestSoftwareFPGASignalChain(unittest.TestCase):
|
|||||||
self.assertIsInstance(frame, RadarFrame)
|
self.assertIsInstance(frame, RadarFrame)
|
||||||
self.assertEqual(frame.frame_number, 42)
|
self.assertEqual(frame.frame_number, 42)
|
||||||
self.assertAlmostEqual(frame.timestamp, 1.0)
|
self.assertAlmostEqual(frame.timestamp, 1.0)
|
||||||
self.assertEqual(frame.range_doppler_i.shape, (64, 32))
|
# Doppler width tracks input n_chirps (16-multiple → that-many sub-frames).
|
||||||
self.assertEqual(frame.range_doppler_q.shape, (64, 32))
|
n_dop = (n_chirps // 16) * 16
|
||||||
self.assertEqual(frame.magnitude.shape, (64, 32))
|
self.assertEqual(frame.range_doppler_i.shape, (NUM_RANGE_BINS, n_dop))
|
||||||
self.assertEqual(frame.detections.shape, (64, 32))
|
self.assertEqual(frame.range_doppler_q.shape, (NUM_RANGE_BINS, n_dop))
|
||||||
self.assertEqual(frame.range_profile.shape, (64,))
|
self.assertEqual(frame.magnitude.shape, (NUM_RANGE_BINS, n_dop))
|
||||||
|
self.assertEqual(frame.detections.shape, (NUM_RANGE_BINS, n_dop))
|
||||||
|
self.assertEqual(frame.range_profile.shape, (NUM_RANGE_BINS,))
|
||||||
self.assertEqual(frame.detection_count, int(frame.detections.sum()))
|
self.assertEqual(frame.detection_count, int(frame.detections.sum()))
|
||||||
|
# Sanity: NUM_DOPPLER_BINS is the production max (48).
|
||||||
|
self.assertLessEqual(n_dop, NUM_DOPPLER_BINS)
|
||||||
|
|
||||||
def test_cfar_enable_changes_detections(self):
|
def test_cfar_enable_changes_detections(self):
|
||||||
"""Enabling CFAR vs simple threshold should yield different detection counts."""
|
"""Enabling CFAR vs simple threshold should yield different detection counts."""
|
||||||
if not self._cosim_available():
|
|
||||||
self.skipTest("co-sim data not found")
|
|
||||||
from v7.software_fpga import SoftwareFPGA
|
from v7.software_fpga import SoftwareFPGA
|
||||||
|
from radar_protocol import NUM_RANGE_BINS, NUM_DOPPLER_BINS
|
||||||
|
|
||||||
iq_i = np.zeros((32, 1024), dtype=np.int64)
|
# Production dimensions: 48 chirps × 2048 samples.
|
||||||
iq_q = np.zeros((32, 1024), dtype=np.int64)
|
iq_i = np.zeros((NUM_DOPPLER_BINS, 2048), dtype=np.int64)
|
||||||
# Inject a single strong tone in bin 10 of every chirp
|
iq_q = np.zeros((NUM_DOPPLER_BINS, 2048), dtype=np.int64)
|
||||||
|
# Inject a single strong tone in bin 10 of every chirp.
|
||||||
iq_i[:, 10] = 5000
|
iq_i[:, 10] = 5000
|
||||||
iq_q[:, 10] = 3000
|
iq_q[:, 10] = 3000
|
||||||
|
|
||||||
@@ -678,14 +682,13 @@ class TestSoftwareFPGASignalChain(unittest.TestCase):
|
|||||||
|
|
||||||
fpga_cfar = SoftwareFPGA()
|
fpga_cfar = SoftwareFPGA()
|
||||||
fpga_cfar.set_cfar_enable(True)
|
fpga_cfar.set_cfar_enable(True)
|
||||||
fpga_cfar.set_cfar_alpha(0x10) # low alpha → more detections
|
fpga_cfar.set_cfar_alpha(0x10)
|
||||||
frame_cfar = fpga_cfar.process_chirps(iq_i, iq_q)
|
frame_cfar = fpga_cfar.process_chirps(iq_i, iq_q)
|
||||||
|
|
||||||
# Just verify both produce valid frames — exact counts depend on chain
|
|
||||||
self.assertIsNotNone(frame_thresh)
|
self.assertIsNotNone(frame_thresh)
|
||||||
self.assertIsNotNone(frame_cfar)
|
self.assertIsNotNone(frame_cfar)
|
||||||
self.assertEqual(frame_thresh.magnitude.shape, (64, 32))
|
self.assertEqual(frame_thresh.magnitude.shape, (NUM_RANGE_BINS, NUM_DOPPLER_BINS))
|
||||||
self.assertEqual(frame_cfar.magnitude.shape, (64, 32))
|
self.assertEqual(frame_cfar.magnitude.shape, (NUM_RANGE_BINS, NUM_DOPPLER_BINS))
|
||||||
|
|
||||||
|
|
||||||
class TestQuantizeRawIQ(unittest.TestCase):
|
class TestQuantizeRawIQ(unittest.TestCase):
|
||||||
@@ -741,6 +744,9 @@ class TestDetectFormat(unittest.TestCase):
|
|||||||
def test_cosim_dir(self):
|
def test_cosim_dir(self):
|
||||||
if not os.path.isdir(self.COSIM_DIR):
|
if not os.path.isdir(self.COSIM_DIR):
|
||||||
self.skipTest("co-sim dir not found")
|
self.skipTest("co-sim dir not found")
|
||||||
|
# COSIM_DIR format requires doppler_map_i/q.npy; skip if absent.
|
||||||
|
if not os.path.isfile(os.path.join(self.COSIM_DIR, "doppler_map_i.npy")):
|
||||||
|
self.skipTest("co-sim doppler_map .npy files not present")
|
||||||
from v7.replay import detect_format, ReplayFormat
|
from v7.replay import detect_format, ReplayFormat
|
||||||
self.assertEqual(detect_format(self.COSIM_DIR), ReplayFormat.COSIM_DIR)
|
self.assertEqual(detect_format(self.COSIM_DIR), ReplayFormat.COSIM_DIR)
|
||||||
|
|
||||||
@@ -841,9 +847,11 @@ class TestReplayEngineRawIQ(unittest.TestCase):
|
|||||||
import tempfile
|
import tempfile
|
||||||
from v7.replay import ReplayEngine
|
from v7.replay import ReplayEngine
|
||||||
from v7.software_fpga import SoftwareFPGA
|
from v7.software_fpga import SoftwareFPGA
|
||||||
from radar_protocol import RadarFrame
|
from radar_protocol import RadarFrame, NUM_RANGE_BINS, NUM_DOPPLER_BINS
|
||||||
|
|
||||||
raw = np.random.randn(2, 32, 1024) + 1j * np.random.randn(2, 32, 1024)
|
# Production dimensions: 48 chirps × 2048 samples per frame.
|
||||||
|
raw = (np.random.randn(2, NUM_DOPPLER_BINS, 2048)
|
||||||
|
+ 1j * np.random.randn(2, NUM_DOPPLER_BINS, 2048))
|
||||||
with tempfile.NamedTemporaryFile(suffix=".npy", delete=False) as f:
|
with tempfile.NamedTemporaryFile(suffix=".npy", delete=False) as f:
|
||||||
np.save(f, raw)
|
np.save(f, raw)
|
||||||
tmp = f.name
|
tmp = f.name
|
||||||
@@ -852,7 +860,7 @@ class TestReplayEngineRawIQ(unittest.TestCase):
|
|||||||
engine = ReplayEngine(tmp, software_fpga=fpga)
|
engine = ReplayEngine(tmp, software_fpga=fpga)
|
||||||
frame = engine.get_frame(0)
|
frame = engine.get_frame(0)
|
||||||
self.assertIsInstance(frame, RadarFrame)
|
self.assertIsInstance(frame, RadarFrame)
|
||||||
self.assertEqual(frame.range_doppler_i.shape, (64, 32))
|
self.assertEqual(frame.range_doppler_i.shape, (NUM_RANGE_BINS, NUM_DOPPLER_BINS))
|
||||||
self.assertEqual(frame.frame_number, 0)
|
self.assertEqual(frame.frame_number, 0)
|
||||||
finally:
|
finally:
|
||||||
os.unlink(tmp)
|
os.unlink(tmp)
|
||||||
|
|||||||
@@ -1,14 +1,22 @@
|
|||||||
"""
|
"""
|
||||||
v7.software_fpga — Bit-accurate software replica of the AERIS-10 FPGA signal chain.
|
v7.software_fpga — Bit-accurate software replica of the AERIS-10 FPGA signal chain.
|
||||||
|
|
||||||
Imports processing functions directly from golden_reference.py (Option A)
|
Imports processing functions directly from fpga_model.py to avoid code
|
||||||
to avoid code duplication. Every stage is toggleable via the same host
|
duplication. Every stage is toggleable via the same host register
|
||||||
register interface the real FPGA exposes, so the dashboard spinboxes can
|
interface the real FPGA exposes, so the dashboard spinboxes can drive
|
||||||
drive either backend transparently.
|
either backend transparently during replay-from-raw-IQ.
|
||||||
|
|
||||||
|
Signal chain order (matching RTL, post-PR-O.6 / PR-F dimensions —
|
||||||
|
2048-pt range FFT, 4x decimation -> 512 range bins, 48 chirps in
|
||||||
|
3 sub-frames -> 48 Doppler bins):
|
||||||
|
|
||||||
Signal chain order (matching RTL):
|
|
||||||
quantize → range_fft → decimator → MTI → doppler_fft → dc_notch → CFAR → RadarFrame
|
quantize → range_fft → decimator → MTI → doppler_fft → dc_notch → CFAR → RadarFrame
|
||||||
|
|
||||||
|
History: golden_reference.py was deleted in commit e8b495c (the "dead golden
|
||||||
|
code cleanup"). fpga_model.py is the surviving bit-accurate model and
|
||||||
|
holds the chain helpers via the run_* shims appended in the post-cleanup
|
||||||
|
revival.
|
||||||
|
|
||||||
Usage:
|
Usage:
|
||||||
fpga = SoftwareFPGA()
|
fpga = SoftwareFPGA()
|
||||||
fpga.set_cfar_enable(True)
|
fpga.set_cfar_enable(True)
|
||||||
@@ -25,16 +33,17 @@ from pathlib import Path
|
|||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Import golden_reference by adding the cosim path to sys.path
|
# Import chain helpers from fpga_model.py (cosim/) — was golden_reference.py
|
||||||
|
# under cosim/real_data/ before commit e8b495c.
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
_GOLDEN_REF_DIR = str(
|
_FPGA_COSIM_DIR = str(
|
||||||
Path(__file__).resolve().parents[2] # 9_Firmware/
|
Path(__file__).resolve().parents[2] # 9_Firmware/
|
||||||
/ "9_2_FPGA" / "tb" / "cosim" / "real_data"
|
/ "9_2_FPGA" / "tb" / "cosim"
|
||||||
)
|
)
|
||||||
if _GOLDEN_REF_DIR not in sys.path:
|
if _FPGA_COSIM_DIR not in sys.path:
|
||||||
sys.path.insert(0, _GOLDEN_REF_DIR)
|
sys.path.insert(0, _FPGA_COSIM_DIR)
|
||||||
|
|
||||||
from golden_reference import ( # noqa: E402
|
from fpga_model import ( # noqa: E402
|
||||||
run_range_fft,
|
run_range_fft,
|
||||||
run_range_bin_decimator,
|
run_range_bin_decimator,
|
||||||
run_mti_canceller,
|
run_mti_canceller,
|
||||||
@@ -53,10 +62,12 @@ from radar_protocol import RadarFrame # noqa: E402
|
|||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Twiddle factor file paths (relative to FPGA root)
|
# Twiddle factor file paths (relative to FPGA root). Production range FFT
|
||||||
|
# is 2048-pt (PR-O.6); fpga_model.load_twiddle_rom auto-falls back to
|
||||||
|
# math-generated twiddles when a path is None.
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
_FPGA_DIR = Path(__file__).resolve().parents[2] / "9_2_FPGA"
|
_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_16 = str(_FPGA_DIR / "fft_twiddle_16.mem")
|
TWIDDLE_16 = str(_FPGA_DIR / "fft_twiddle_16.mem")
|
||||||
|
|
||||||
# CFAR mode int→string mapping (FPGA register 0x24: 0=CA, 1=GO, 2=SO)
|
# CFAR mode int→string mapping (FPGA register 0x24: 0=CA, 1=GO, 2=SO)
|
||||||
@@ -176,18 +187,20 @@ class SoftwareFPGA:
|
|||||||
n_chirps = iq_i.shape[0]
|
n_chirps = iq_i.shape[0]
|
||||||
n_samples = iq_i.shape[1]
|
n_samples = iq_i.shape[1]
|
||||||
|
|
||||||
# --- Stage 1: Range FFT (per chirp) ---
|
# --- Stage 1: Range FFT (per chirp). N is inferred from input length;
|
||||||
|
# pass a twiddle file only when it matches the input N (defaults
|
||||||
|
# to math-generated twiddles otherwise).
|
||||||
range_i = np.zeros((n_chirps, n_samples), dtype=np.int64)
|
range_i = np.zeros((n_chirps, n_samples), dtype=np.int64)
|
||||||
range_q = 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
|
twiddle_path = TWIDDLE_2048 if (n_samples == 2048 and os.path.exists(TWIDDLE_2048)) else None
|
||||||
for c in range(n_chirps):
|
for c in range(n_chirps):
|
||||||
range_i[c], range_q[c] = run_range_fft(
|
range_i[c], range_q[c] = run_range_fft(
|
||||||
iq_i[c].astype(np.int64),
|
iq_i[c].astype(np.int64),
|
||||||
iq_q[c].astype(np.int64),
|
iq_q[c].astype(np.int64),
|
||||||
twiddle_file=twiddle_1024,
|
twiddle_file=twiddle_path,
|
||||||
)
|
)
|
||||||
|
|
||||||
# --- Stage 2: Range bin decimation (1024 → 64) ---
|
# --- Stage 2: Range bin decimation (production 2048 -> 512) ---
|
||||||
decim_i, decim_q = run_range_bin_decimator(range_i, range_q)
|
decim_i, decim_q = run_range_bin_decimator(range_i, range_q)
|
||||||
|
|
||||||
# --- Stage 3: MTI canceller (pre-Doppler, per-chirp) ---
|
# --- Stage 3: MTI canceller (pre-Doppler, per-chirp) ---
|
||||||
|
|||||||
Reference in New Issue
Block a user