145 lines
7.4 KiB
Python
145 lines
7.4 KiB
Python
"""Collector packets preserve whole sweeps, source windows and pair identity."""
|
|
|
|
import struct
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from python_app.hardware_full.rph_radar.protocol import RfgSweepAssembler, RphRadarStreamParser
|
|
|
|
|
|
def packet(*, source_id=1, cycle_id=1, start=100, end=200, combo=(1, 2), dirty=0, points=None):
|
|
if points is None:
|
|
points = [(0xA, 1, 10, -1), (0xA8, 1, 100, 5)]
|
|
payload = b"".join(struct.pack("<HHhh", *point) for point in points)
|
|
return struct.pack("<4sIQQQQHHI", b"RPH1", len(payload), source_id, cycle_id,
|
|
start, end, *combo, dirty) + payload
|
|
|
|
|
|
class RphRadarStreamParserTest(unittest.TestCase):
|
|
def test_fragmented_packets_keep_source_window_and_identity(self):
|
|
parser = RphRadarStreamParser()
|
|
sweeps = []
|
|
for byte in b"preroll" + packet(source_id=73, cycle_id=9):
|
|
sweeps.extend(parser.feed(bytes([byte])))
|
|
self.assertEqual(len(sweeps), 1)
|
|
sweep = sweeps[0]
|
|
self.assertEqual((sweep.source_id, sweep.cycle_id, sweep.capture_start_ns, sweep.capture_end_ns),
|
|
(73, 9, 100, 200))
|
|
self.assertEqual(sweep.combo, (1, 2))
|
|
self.assertEqual(sweep.main.tolist(), [10-1j])
|
|
self.assertEqual(sweep.reference.tolist(), [100+5j])
|
|
self.assertEqual(sweep.main.dtype, np.complex64)
|
|
|
|
def test_whole_packet_loss_does_not_merge_pairs(self):
|
|
parser = RphRadarStreamParser()
|
|
first, last = parser.feed(packet(source_id=1) + packet(source_id=3, combo=(2, 3),
|
|
points=[(0xA, 2, 20, 0), (0xA8, 2, 200, 0)]))
|
|
self.assertEqual((first.combo, first.steps.tolist()), ((1, 2), [1]))
|
|
self.assertEqual((last.combo, last.steps.tolist()), ((2, 3), [2]))
|
|
self.assertEqual(last.source_id, 3)
|
|
|
|
def test_aligns_both_channels_and_keeps_only_shared_steps(self):
|
|
parser = RphRadarStreamParser()
|
|
(sweep,) = parser.feed(packet(points=[(0xA, 3, 3, 0), (0xA8, 3, 30, 0),
|
|
(0xA, 1, 1, 0), (0xA8, 1, 10, 0), (0xA, 2, 2, 0)]))
|
|
self.assertEqual(sweep.steps.tolist(), [1, 3])
|
|
self.assertEqual(sweep.main.real.tolist(), [1, 3])
|
|
self.assertEqual(sweep.reference.real.tolist(), [10, 30])
|
|
|
|
def test_bad_window_size_flags_and_point_marker_fail_fast(self):
|
|
for encoded in (packet(start=0), packet(start=200, end=100), packet(dirty=2),
|
|
packet(source_id=0), packet(cycle_id=0),
|
|
packet(points=[(0x1A, 1, 1, 1)]),
|
|
b"RPH1" + struct.pack("<I", 0xFFFFFFFF) + bytes(40)):
|
|
with self.subTest(encoded=encoded), self.assertRaisesRegex(ValueError, "protocol"):
|
|
RphRadarStreamParser().feed(encoded)
|
|
|
|
def test_reset_discards_partial_packet(self):
|
|
parser = RphRadarStreamParser()
|
|
self.assertEqual(parser.feed(packet()[:53]), [])
|
|
parser.reset()
|
|
(sweep,) = parser.feed(packet(source_id=2, dirty=1, combo=(65535, 65535)))
|
|
self.assertEqual(sweep.source_id, 2)
|
|
self.assertIsNone(sweep.combo)
|
|
self.assertTrue(sweep.dirty)
|
|
|
|
|
|
def rfg_datagram(sweep_id, *steps, values=(3, 1, 5, 4, 2, 6), q_shift=27):
|
|
"""RFG SWEEP_AVG datagram; ``values`` are in units of 2**(q_shift - 7)."""
|
|
return struct.pack("<HH", 6, q_shift) + b"".join(
|
|
struct.pack("<I6i", (sweep_id << 24) | step, *[value << (q_shift - 7) for value in values]) for step in steps)
|
|
|
|
|
|
class RfgSweepAssemblerTest(unittest.TestCase):
|
|
STEP_NS = 50_000
|
|
PERIOD_NS = 20_000_000
|
|
|
|
def feed(self, assembler, sweep_id, steps, start_ns, delay_ns=7_000):
|
|
sweeps = []
|
|
for step in steps:
|
|
sweeps += assembler.feed(rfg_datagram(sweep_id, step), start_ns + (step + 1) * self.STEP_NS + delay_ns)
|
|
return sweeps
|
|
|
|
def test_complete_sweep_maps_sync_detector_and_reference_and_its_true_window(self):
|
|
assembler = RfgSweepAssembler(steps=3, step_ns=self.STEP_NS, period_ns=self.PERIOD_NS)
|
|
self.feed(assembler, 9, [0, 1, 2], 1_000_000_000)
|
|
(sweep,) = assembler.feed(rfg_datagram(10, 0), 1_020_000_000)
|
|
self.assertFalse(sweep.dirty)
|
|
self.assertEqual(sweep.steps.tolist(), [0, 1, 2])
|
|
np.testing.assert_allclose(sweep.main, [(3 - 1 + 1j * (4 - 2)) / 128] * 3) # ch0 - ch1
|
|
np.testing.assert_allclose(sweep.reference, [(5 + 6j) / 128] * 3) # ch2
|
|
self.assertEqual(sweep.capture_start_ns, 1_000_007_000)
|
|
self.assertEqual(sweep.capture_end_ns - sweep.capture_start_ns, 3 * self.STEP_NS)
|
|
|
|
def test_window_takes_the_least_delayed_point_when_the_board_batches(self):
|
|
assembler = RfgSweepAssembler(steps=3, step_ns=self.STEP_NS, period_ns=self.PERIOD_NS)
|
|
start = 2_000_000_000
|
|
assembler.feed(rfg_datagram(1, 0, 1), start + 2 * self.STEP_NS + 10_000) # steps 0-1 batched
|
|
assembler.feed(rfg_datagram(1, 2), start + 3 * self.STEP_NS + 10_000)
|
|
(sweep,) = assembler.feed(rfg_datagram(2, 0), start + 30_000_000)
|
|
self.assertEqual(sweep.capture_start_ns, start + 10_000)
|
|
|
|
def test_points_of_older_sweeps_are_ignored_until_the_id_space_could_have_wrapped(self):
|
|
assembler = RfgSweepAssembler(steps=3, step_ns=self.STEP_NS, period_ns=self.PERIOD_NS)
|
|
sweeps = []
|
|
for sweep_id, start in ((1, 0), (3, 20_000_000), (2, 40_000_000), (4, 60_000_000), (3, 80_000_000)):
|
|
sweeps += self.feed(assembler, sweep_id, [0, 1, 2], start)
|
|
self.assertEqual([sweep.source_id for sweep in sweeps], [1, 2]) # late 2 and replayed 3 are dropped
|
|
sweeps = self.feed(assembler, 2, [0, 1, 2], 80_000_000 + 128 * self.PERIOD_NS) # 4 closes, 2 is new
|
|
sweeps += self.feed(assembler, 3, [0], 100_000_000 + 128 * self.PERIOD_NS)
|
|
self.assertEqual([(sweep.source_id, sweep.dirty) for sweep in sweeps], [(3, False), (4, False)])
|
|
|
|
def test_reset_turns_the_rest_of_the_sweep_in_progress_dirty(self):
|
|
assembler = RfgSweepAssembler(steps=3, step_ns=self.STEP_NS, period_ns=self.PERIOD_NS)
|
|
self.feed(assembler, 4, [0], 0)
|
|
assembler.reset()
|
|
self.feed(assembler, 4, [1, 2], 0)
|
|
(sweep,) = self.feed(assembler, 5, [0], 20_000_000)
|
|
self.assertTrue(sweep.dirty)
|
|
|
|
def test_partial_gapped_shortened_or_reordered_sweeps_are_dirty(self):
|
|
for steps in ([1, 2], [0, 2], [0, 1], [0, 2, 1], [0, 1, 1, 2]):
|
|
assembler = RfgSweepAssembler(steps=3, step_ns=self.STEP_NS, period_ns=self.PERIOD_NS)
|
|
self.feed(assembler, 255, steps, 0)
|
|
with self.subTest(steps=steps):
|
|
(sweep,) = assembler.feed(rfg_datagram(0, 0), 10_000_000) # 255 -> 0 wraps normally
|
|
self.assertTrue(sweep.dirty)
|
|
|
|
def test_a_reused_id_after_an_outage_does_not_splice_two_sweeps(self):
|
|
assembler = RfgSweepAssembler(steps=3, step_ns=self.STEP_NS, period_ns=self.PERIOD_NS)
|
|
self.feed(assembler, 5, [0], 0)
|
|
self.feed(assembler, 5, [1, 2], 4_920_000_000) # 256 sweep periods later
|
|
(sweep,) = self.feed(assembler, 6, [0], 4_940_000_000)
|
|
self.assertTrue(sweep.dirty)
|
|
|
|
def test_unexpected_layout_fails_fast(self):
|
|
for datagram in (b"\x06", struct.pack("<HH", 5, 27), struct.pack("<HH", 6, 31) + rfg_datagram(1, 0)[4:],
|
|
rfg_datagram(1, 0)[:-1]):
|
|
with self.subTest(datagram=datagram), self.assertRaisesRegex(ValueError, "protocol"):
|
|
RfgSweepAssembler(steps=3, step_ns=1, period_ns=1).feed(datagram, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|