201 lines
11 KiB
Python
201 lines
11 KiB
Python
"""Analytical checks for one coherent moving reflector across mock radar modes."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from contextlib import ExitStack
|
||
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
import numpy as np
|
||
|
||
from python_app.hardware_full.matrix_radar_service import create_matrix_radar_service
|
||
from python_app.hardware_full.point_target_mock_service import PointTargetMockService
|
||
from python_app.hardware_full.single_radar_service import create_single_radar_service
|
||
from python_app.models.dataset_model import ComboKey
|
||
from python_app.models.run_config_schema import RunConfigModel
|
||
from python_app.models.run_config_schema import GprRxGeometryModel, GprTxGeometryModel
|
||
from python_app.workflows.radar_config_variants import RadarConfigVariant
|
||
from python_app.workflows.sequential_capture_workflow import SequentialCaptureSession
|
||
|
||
_EPOCH_NS = 10_000_000_000
|
||
_CLOCK = "python_app.hardware_full.point_target_mock_service.time.monotonic_ns"
|
||
_SLEEP = "python_app.hardware_full.point_target_mock_service.time.sleep"
|
||
|
||
|
||
def _config() -> RunConfigModel:
|
||
config = RunConfigModel()
|
||
config.radar.model = RunConfigModel.LIBREVNA_MODEL
|
||
config.radar.driver_mode = "mock"
|
||
config.radar.sweep.start_hz = 300e6
|
||
config.radar.sweep.stop_hz = 900e6
|
||
config.radar.sweep.points = 32
|
||
config.input_switch.positions = config.output_switch.positions = 2
|
||
config.input_switch.driver_mode = config.output_switch.driver_mode = "mock"
|
||
config.runtime.settling_ms = 0
|
||
config.combos = config.build_runtime_combos()
|
||
config.gpr.tx_geometry = [GprTxGeometryModel(0, -1), GprTxGeometryModel(1, 1)]
|
||
config.gpr.rx_geometry = [GprRxGeometryModel(0, -.4, .1, .2), GprRxGeometryModel(1, .5, -.1, .3)]
|
||
return config
|
||
|
||
|
||
class PointTargetMockTest(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
cleanup = ExitStack()
|
||
self.addCleanup(cleanup.close)
|
||
self.clock = cleanup.enter_context(mock.patch(_CLOCK, return_value=_EPOCH_NS))
|
||
self.sleep = cleanup.enter_context(mock.patch(_SLEEP))
|
||
self.config = _config()
|
||
self.service = PointTargetMockService(self.config)
|
||
self.service.configure(self.config.radar.sweep)
|
||
|
||
def test_all_channels_follow_bistatic_distance(self) -> None:
|
||
collection = self.service.acquire_collection(collection_id=43)
|
||
target = np.array([0, .25, 3.75])
|
||
self.assertEqual(collection.collection_id, 43)
|
||
self.assertEqual([(t.combo.input, t.combo.output) for t in collection.traces], [(0, 0), (1, 0), (0, 1), (1, 1)])
|
||
for trace in collection.traces:
|
||
tx = self.config.gpr.tx_geometry[trace.combo.output]
|
||
rx = self.config.gpr.rx_geometry[trace.combo.input]
|
||
r_tx = np.linalg.norm(target - [tx.x_m, tx.y_m, tx.z_m])
|
||
r_rx = np.linalg.norm(target - [rx.x_m, rx.y_m, rx.z_m])
|
||
phase_per_m = -2j * np.pi * trace.frequency_hz.astype(np.float64) / 3e8
|
||
np.testing.assert_allclose(trace.s21, .5 / (r_tx * r_rx) * np.exp(phase_per_m * (r_tx + r_rx)), rtol=1e-6)
|
||
self.assertEqual((trace.capture_start_ns, trace.capture_end_ns), (_EPOCH_NS, _EPOCH_NS))
|
||
self.assertEqual(trace.s21.dtype, np.dtype(np.complex64))
|
||
self.assertEqual((collection.capture_start_ns, collection.capture_end_ns, collection.monotonic_ns), (_EPOCH_NS,) * 3)
|
||
|
||
def test_coincident_antennas_have_identical_traces_regardless_of_channel_number(self) -> None:
|
||
self.config.gpr.tx_geometry = [GprTxGeometryModel(0, 1), GprTxGeometryModel(1, 1)]
|
||
self.config.gpr.rx_geometry = [GprRxGeometryModel(0, -.5), GprRxGeometryModel(1, -.5)]
|
||
traces = self.service.acquire_collection().traces
|
||
for trace in traces[1:]:
|
||
np.testing.assert_array_equal(trace.s21, traces[0].s21)
|
||
|
||
def test_geometry_changes_receiver_path(self) -> None:
|
||
before = self.service.acquire_collection().traces[0]
|
||
self.config.gpr.rx_geometry[0].z_m = 1
|
||
after = self.service.acquire_collection().traces[0]
|
||
self.assertFalse(np.array_equal(before.s21, after.s21))
|
||
|
||
def test_motion_uses_elapsed_time_and_returns_after_eight_seconds(self) -> None:
|
||
tx = {0: (-1., 0., 0.), 1: (1., 0., 0.)}
|
||
rx = {0: (-.4, .1, .2), 1: (.5, -.1, .3)}
|
||
for seconds, displacement in [(0, 0), (2, 1), (4, 0), (6, -1), (8, 0)]:
|
||
np.testing.assert_allclose(
|
||
self.service._target_position(_EPOCH_NS + seconds * 1_000_000_000, tx, rx),
|
||
[1.25 * displacement, .25, 3.75 + .5 * displacement], atol=1e-12,
|
||
)
|
||
first = self.service.acquire_collection(1)
|
||
repeated = self.service.acquire_collection(999)
|
||
for a, b in zip(first.traces, repeated.traces):
|
||
np.testing.assert_array_equal(a.s21, b.s21)
|
||
self.clock.return_value = _EPOCH_NS + 2_000_000_000
|
||
moved = self.service.acquire_collection(1)
|
||
self.assertFalse(np.array_equal(first.traces[0].s21, moved.traces[0].s21))
|
||
|
||
def test_pacing_excludes_wait_from_measurement_window(self) -> None:
|
||
current_ns = _EPOCH_NS
|
||
|
||
def sleep(seconds: float) -> None:
|
||
nonlocal current_ns
|
||
current_ns += round(seconds * 1e9)
|
||
|
||
self.clock.side_effect = lambda: current_ns
|
||
self.sleep.side_effect = sleep
|
||
first = self.service.acquire_collection()
|
||
current_ns += 20_000_000
|
||
second = self.service.acquire_collection()
|
||
self.sleep.assert_called_once()
|
||
self.assertAlmostEqual(self.sleep.call_args.args[0], .03)
|
||
self.assertEqual(second.capture_start_ns - first.capture_start_ns, 50_000_000)
|
||
self.assertEqual(second.capture_start_ns, second.capture_end_ns)
|
||
|
||
def test_full_motion_stays_inside_original_sweep_unambiguous_range(self) -> None:
|
||
# The existing 201-point 1 MHz–6 GHz profile aliases paths beyond about 5 m.
|
||
# Check its actual 4x4 geometry throughout the cycle, including both extremes.
|
||
self.config.input_switch.positions = self.config.output_switch.positions = 4
|
||
self.config.combos = self.config.build_runtime_combos()
|
||
tx = {i: (x, 0., 0.) for i, x in enumerate([.905, -.905, .905, -.905])}
|
||
rx = {i: (x, 0., 0.) for i, x in enumerate([-.18, .485, -.49, .185])}
|
||
self.config.gpr.tx_geometry = [GprTxGeometryModel(i, *position) for i, position in tx.items()]
|
||
self.config.gpr.rx_geometry = [GprRxGeometryModel(i, *position) for i, position in rx.items()]
|
||
service = PointTargetMockService(self.config)
|
||
unambiguous_range_m = 3e8 / (2 * ((6e9 - 1e6) / (201 - 1)))
|
||
ranges = []
|
||
for seconds in np.linspace(0, 8, 81):
|
||
target = service._target_position(_EPOCH_NS + round(seconds * 1e9), tx, rx)
|
||
for transmitter in tx.values():
|
||
for receiver in rx.values():
|
||
ranges.append((np.linalg.norm(target - transmitter) + np.linalg.norm(target - receiver)) / 2)
|
||
self.assertLess(max(ranges), unambiguous_range_m)
|
||
|
||
def test_single_manual_capture_selects_actual_antenna_and_keeps_capture_time(self) -> None:
|
||
radar = mock.Mock(wraps=self.service)
|
||
with mock.patch("python_app.workflows.sequential_capture_workflow.create_single_radar_service", return_value=radar):
|
||
session = SequentialCaptureSession(self.config, "s21_calibration", "mock_geometry")
|
||
session.open()
|
||
try:
|
||
first = session.capture_current_combo().display_trace
|
||
second = session.capture_current_combo().display_trace
|
||
finally:
|
||
session.close()
|
||
radar.configure.assert_called_once_with(self.config.radar.sweep)
|
||
self.assertEqual((first.combo, second.combo), (ComboKey(0, 0), ComboKey(1, 0)))
|
||
self.assertFalse(np.array_equal(first.s21, second.s21))
|
||
self.assertEqual((second.capture_start_ns, second.capture_end_ns), (_EPOCH_NS, _EPOCH_NS))
|
||
|
||
def test_factories_share_scene_and_mock_widened_matrix_needs_no_hardware(self) -> None:
|
||
single = create_single_radar_service(self.config)
|
||
self.assertIsInstance(single, PointTargetMockService)
|
||
single.configure(self.config.radar.sweep)
|
||
self.assertIsInstance(create_matrix_radar_service(self.config), PointTargetMockService)
|
||
self.config.radar.model = RunConfigModel.LIBREVNA_MULTI_MODEL
|
||
self.config.radar.multi_device.stimulus_switch_positions = 2
|
||
self.config.radar.multi_device.input_switch_positions = 2
|
||
self.config.radar.multi_device.output_switch_positions = 2
|
||
self.config.apply_device_model_constraints()
|
||
matrix = create_matrix_radar_service(self.config)
|
||
self.assertIsInstance(matrix, PointTargetMockService)
|
||
matrix.open()
|
||
matrix.configure(self.config.radar.sweep)
|
||
collection = matrix.acquire_collection()
|
||
matrix.close()
|
||
self.assertEqual(len(collection.traces), 64)
|
||
self.assertEqual([(t.combo.input, t.combo.output) for t in collection.traces], [(i, o) for o in range(8) for i in range(8)])
|
||
np.testing.assert_array_equal(collection.traces[0].s21, single.acquire().s21)
|
||
|
||
def test_multi_variant_single_capture_drains_each_variant_before_sampling(self) -> None:
|
||
radar = mock.Mock(wraps=self.service)
|
||
events = []
|
||
radar.prepare_combo.side_effect = lambda combo: (events.append("prepare"), self.service.prepare_combo(combo))
|
||
radar.acquire.side_effect = lambda: (events.append("acquire"), self.service.acquire())[1]
|
||
variants = [RadarConfigVariant(name, self.config, name) for name in ("a", "b")]
|
||
with mock.patch(
|
||
"python_app.workflows.sequential_capture_workflow.create_single_radar_service",
|
||
return_value=radar,
|
||
):
|
||
session = SequentialCaptureSession(
|
||
config=self.config, kind="s21_calibration", set_name="variants", radar_variants=variants,
|
||
)
|
||
session.open()
|
||
try:
|
||
session.capture_current_combo()
|
||
batch = session.capture_current_combo()
|
||
finally:
|
||
session.close()
|
||
self.assertEqual(events, ["prepare", "acquire"] * 4)
|
||
self.assertEqual([trace.combo for trace in batch.traces], [ComboKey(1, 0)] * 2)
|
||
for trace in batch.traces:
|
||
self.assertEqual((trace.capture_start_ns, trace.capture_end_ns), (_EPOCH_NS, _EPOCH_NS))
|
||
|
||
def test_invalid_or_unconfigured_mock_fails_clearly(self) -> None:
|
||
with self.assertRaisesRegex(ValueError, "outside the antenna matrix"):
|
||
self.service.acquire_collection(combo=ComboKey(2, 0))
|
||
with self.assertRaisesRegex(RuntimeError, "not configured"):
|
||
PointTargetMockService(self.config).acquire_collection()
|
||
self.config.input_switch.positions = 0
|
||
self.config.combos = []
|
||
with self.assertRaisesRegex(ValueError, "Switch positions must be positive integers"):
|
||
PointTargetMockService(self.config)
|