"""Notebook mathematics and acquisition provenance of live object corrections.""" from __future__ import annotations import copy from pathlib import Path import tempfile import unittest from unittest import mock import numpy as np from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData from python_app.storage.npz.store import NpzStore from python_app.workflows import mimo_calibration as mc from python_app.workflows import object_calibration as oc class MimoCalibrationTest(unittest.TestCase): def setUp(self): self.settings = mc.Settings(input_idx=(2, 7), output_idx=(1, 5), sphere_radius_m=0., antenna_height_m=1., f_min=2e9, f_max=6e9, refine_axes=(), coarse_range_search_m=0.) self.freq = np.linspace(2e9, 6e9, 129) self.target = (.2, .25, 4.) self.tx = np.array([[-.6, 0., 0.], [.6, 0., 0.]]) self.rx = np.array([[-.3, .1, 0.], [.3, .1, 0.]]) self.config = { 'radar': {'model': 'librevna_multi', 'serial': 'master', 'multi_device': {'slave_serials': ['slave']}}, 'preprocess': {'s21': {'calibration': {'set_name': 'through'}, 'reference': {'set_name': 'empty'}}}, 'gpr': { 'tx_geometry': [dict(output_pos=port, x_m=xyz[0], y_m=xyz[1], z_m=xyz[2]) for port, xyz in zip((1, 5), self.tx)], 'rx_geometry': [dict(input_pos=port, x_m=xyz[0], y_m=xyz[1], z_m=xyz[2]) for port, xyz in zip((2, 7), self.rx)], }, 'run': {'combos': [dict(input=input_, output=output) for output in (1, 5) for input_ in (2, 7)]}, } def signal(self, target=None, *, gains=False): target = self.target if target is None else target pitch = np.radians(self.settings.pitch_deg) y0 = target[1] - self.settings.antenna_height_m radar_target = np.array([target[0], y0 * np.cos(pitch) + target[2] * np.sin(pitch), -y0 * np.sin(pitch) + target[2] * np.cos(pitch)]) vt, vr = self.tx - radar_target, self.rx - radar_target dt, dr = np.linalg.norm(vt, axis=1), np.linalg.norm(vr, axis=1) paths = dt[:, None] + dr[None, :] if self.settings.sphere_radius_m: cosine = (vt @ vr.T) / (dt[:, None] * dr[None, :]) paths -= 2 * self.settings.sphere_radius_m * np.sqrt((1 + cosine) / 2) tx_phase = np.array([0., .7])[:, None] - 2 * np.pi * self.freq * np.array([0., .02])[:, None] / mc.C_LIGHT rx_phase = np.array([0., -.5])[:, None] - 2 * np.pi * self.freq * np.array([0., -.013])[:, None] / mc.C_LIGHT signal = np.exp(-2j * np.pi * self.freq * paths[:, :, None] / mc.C_LIGHT) signal *= np.exp(1j * (tx_phase[:, None, :] + rx_phase[None, :, :] + 1.1)) if gains: signal *= np.array([1., 2.])[:, None, None] * np.array([1., .5])[None, :, None] return signal, paths def collection(self, samples=None, *, freq=None, identity=1): samples = self.signal()[0] if samples is None else samples freq = self.freq if freq is None else freq traces = [TraceData(ComboKey(input_, output), freq.copy(), samples[t, r].copy()) for t, output in enumerate((1, 5)) for r, input_ in enumerate((2, 7))] return SweepCollection(identity, identity, traces) def calibration(self, samples=None): through = self.collection(np.ones((2, 2, self.freq.size), complex)) provenance = oc.capture_provenance(self.config, 'captured-radar', through) return mc.compute_from_collections( [{'collections': [self.collection(samples)], 'target_scene_m': self.target}], self.config, settings=self.settings, provenance=provenance, ) def object_set(self): collection, metadata = self.calibration() return collection, metadata, self.collection(np.ones((2, 2, self.freq.size), complex)) def test_partial_geometry_calibrates_subset_and_preserves_full_capture(self): from PyQt6.QtWidgets import QApplication, QDoubleSpinBox from python_app.gui.object_calibration_dialog import ObjectCalibrationDialog app = QApplication.instance() or QApplication([]) self.assertIsNotNone(app) capture = self.collection() known = {trace.combo for trace in capture.traces} full = [ComboKey(i, o) for o in (1, 5, 6) for i in (2, 7, 8)] capture.traces.extend(TraceData(combo, self.freq.copy(), np.ones(self.freq.size, complex)) for combo in full if combo not in known) self.config['run']['combos'] = [dict(input=combo.input, output=combo.output) for combo in full] through = copy.deepcopy(capture) for trace in through.traces: trace.s21.fill(1) provenance = oc.capture_provenance(self.config, 'captured-radar', through) dialog = ObjectCalibrationDialog('partial') self.addCleanup(dialog.close) dialog.add_capture([capture], self.config, provenance) for column, value in enumerate(self.target, 2): item = dialog._table.item(0, column) assert item is not None item.setText(str(value)) for name in ('sphere_radius_m', 'antenna_height_m', 'coarse_range_search_m'): field = dialog._settings[name] assert isinstance(field, QDoubleSpinBox) field.setValue(getattr(self.settings, name)) for field in dialog._refine_axes.values(): field.setChecked(False) groups, config, settings, provenance = dialog.calculation_inputs() correction, metadata = mc.compute_from_collections( groups, config, settings=mc.Settings(**settings), provenance=provenance, ) self.assertEqual(metadata['rx_input_pos'], [2, 7]) self.assertEqual(metadata['tx_output_pos'], [1, 5]) self.assertEqual(len(groups[0]['collections'][0].traces), 9) self.assertEqual({trace.combo for trace in correction.traces}, set(full)) self.assertEqual(metadata['unchanged_combos'], [dict(input=combo.input, output=combo.output) for combo in full if combo not in known]) for trace in correction.traces: if trace.combo not in known: np.testing.assert_array_equal(trace.s21, np.ones(self.freq.size)) self.assertGreater(metadata['diagnostics'][0]['coherence_after'], .999) oc.validate_object_set(correction, metadata, config, 'captured-radar', through) def test_point_and_sphere_recover_relative_antenna_phases(self): for radius in (0., .16): with self.subTest(radius=radius): self.settings.sphere_radius_m = radius signal, paths = self.signal() collection, metadata = self.calibration() pair = np.array([trace.s21 for trace in collection.traces]).reshape(signal.shape) contributions = signal * pair * np.exp(2j * np.pi * self.freq * paths[:, :, None] / mc.C_LIGHT) np.testing.assert_allclose(np.angle(contributions.mean(axis=2)), np.full((2, 2), 1.1), atol=.004) np.testing.assert_allclose(np.abs(pair), 1., atol=2e-7) diagnostic = metadata['diagnostics'][0] self.assertGreater(diagnostic['coherence_after'], .9999) self.assertLess(diagnostic['coherence_before'], .98) np.testing.assert_allclose(diagnostic['phase_after_deg'], np.zeros((2, 2)), atol=.25) def test_phase_maps_match_notebook_cell5_with_separate_sum_references(self): before = np.array([[1j, 3], [-1j, 1]]) after = np.array([[-1, -2j], [-1j, -3]]) with mock.patch.object(mc._Estimator, 'broadband', side_effect=[before, after]): diagnostic = self.calibration()[1]['diagnostics'][0] np.testing.assert_allclose(diagnostic['phase_before_deg'], [[90, 0], [-90, 0]], atol=1e-10) np.testing.assert_allclose(diagnostic['phase_after_deg'], [[-36.869897645844, 53.130102354156], [53.130102354156, -36.869897645844]], atol=1e-10) self.assertAlmostEqual(diagnostic['phase_std_before_deg'], np.sqrt(4050)) self.assertAlmostEqual(diagnostic['phase_std_after_deg'], 45) def test_overflow_in_inverted_pair_correction_is_rejected(self): self.settings.apply_amplitude = True with mock.patch.object(mc._Estimator, 'to_full_grid', return_value=np.full(self.freq.size, 1e-25)), \ np.errstate(over='ignore', invalid='ignore'), \ self.assertRaisesRegex(ValueError, 'pair corrections.*non-finite'): self.calibration() def test_through_hash_tracks_only_canonical_pair_frequency_and_s21_content(self): through = self.collection() fingerprint = oc.through_fingerprint(through) reordered = copy.deepcopy(through) reordered.traces.reverse() reordered.collection_id += 1 reordered.monotonic_ns += 1000 for trace in reordered.traces: trace.frequency_hz = trace.frequency_hz.astype('>f4') trace.s21 = trace.s21.astype('>c8') trace.capture_start_ns = 99 self.assertEqual(oc.through_fingerprint(reordered), fingerprint) for field in ('frequency_hz', 's21'): changed = copy.deepcopy(through) getattr(changed.traces[0], field)[0] += 1024 if field == 'frequency_hz' else .1j with self.subTest(field=field): self.assertNotEqual(oc.through_fingerprint(changed), fingerprint) changed = copy.deepcopy(through) changed.traces[0].combo = ComboKey(999, 1) self.assertNotEqual(oc.through_fingerprint(changed), fingerprint) def test_object_set_accepts_combo_subset_and_reordered_geometry_with_new_reference(self): collection, metadata, through = self.object_set() config = copy.deepcopy(self.config) config['run']['combos'] = config['run']['combos'][1:2] config['gpr']['tx_geometry'].reverse() config['gpr']['rx_geometry'].reverse() config['preprocess']['s21']['reference']['set_name'] = 'new empty scene' oc.validate_object_set(collection, metadata, config, 'captured-radar', through) def test_object_set_rejects_changed_capture_dependencies_and_missing_provenance(self): collection, metadata, through = self.object_set() for dependency in ('radar', 'geometry', 'through_name', 'through_content', 'provenance'): config = copy.deepcopy(self.config) active_through = copy.deepcopy(through) active_metadata = copy.deepcopy(metadata) key = 'captured-radar' if dependency == 'radar': key = 'another-radar' elif dependency == 'geometry': config['gpr']['tx_geometry'][0]['z_m'] += .01 elif dependency == 'through_name': config['preprocess']['s21']['calibration']['set_name'] = 'another-through' elif dependency == 'through_content': active_through.traces[0].s21[0] += .1 else: active_metadata.clear() with self.subTest(dependency=dependency), self.assertRaisesRegex(ValueError, 'Object calibration'): oc.validate_object_set(collection, active_metadata, config, key, active_through) def test_object_set_rejects_missing_pairs_shifted_grids_and_invalid_factors(self): collection, metadata, through = self.object_set() for defect in ('missing', 'duplicate', 'grid', 'zero', 'nan'): corrupted = copy.deepcopy(collection) if defect == 'missing': corrupted.traces.pop() elif defect == 'duplicate': corrupted.traces.append(corrupted.traces[0]) elif defect == 'grid': corrupted.traces[0].frequency_hz = corrupted.traces[0].frequency_hz + 1024 else: corrupted.traces[0].s21[0] = 0 if defect == 'zero' else np.nan with self.subTest(defect=defect), self.assertRaises(ValueError): oc.validate_object_set(corrupted, metadata, self.config, 'captured-radar', through) def test_frequency_geometry_channel_and_signal_failures_are_explicit(self): cases = [] incomplete = self.collection(); incomplete.traces.pop(); cases.append(incomplete) duplicate = self.collection(); duplicate.traces.append(duplicate.traces[0]); cases.append(duplicate) mismatched = self.collection(); mismatched.traces[0].frequency_hz += 10000; cases.append(mismatched) nonuniform = self.collection(); nonuniform.traces[0].frequency_hz[20] += 1e6; cases.append(nonuniform) zero = self.collection(np.zeros((2, 2, self.freq.size), complex)); cases.append(zero) invalid = self.collection(); invalid.traces[0].s21[0] = np.nan; cases.append(invalid) for index, collection in enumerate(cases): with self.subTest(collection=index): with self.assertRaises(ValueError): mc.compute_from_collections([{'collections': [collection], 'target_scene_m': self.target}], self.config, settings=self.settings, provenance={}) with self.assertRaisesRegex(ValueError, 'at least one'): mc.compute_from_collections([], self.config, settings=self.settings, provenance={}) def test_cancel_is_checked_inside_refinement(self): self.settings.refine_axes = ('z',) state = {'estimating': False, 'checks': 0} def progress(message): state['estimating'] |= message.startswith('Estimating') def cancelled(): if state['estimating']: state['checks'] += 1 return state['checks'] >= 6 with self.assertRaises(InterruptedError): mc.compute_from_collections([{'collections': [self.collection()], 'target_scene_m': self.target}], self.config, settings=self.settings, provenance={}, progress=progress, cancelled=cancelled) self.assertEqual(state['checks'], 6) def test_original_scipy_refinement_finishes_with_valid_diagnostics(self): self.settings.refine_axes = ('z',) self.settings.coarse_range_search_m = .1 _, metadata = self.calibration() self.assertGreater(metadata['diagnostics'][0]['coherence_after'], .999) refined = metadata['targets'][0]['refined_scene_m'] self.assertTrue(np.all(np.isfinite(refined))) def test_common_phase_invariance_and_explicit_amplitude_equalization(self): signal, paths = self.signal(gains=True) _, original = self.calibration(signal) _, rotated = self.calibration(signal * np.exp(2.8j)) for key in ('phase_before_deg', 'phase_after_deg'): np.testing.assert_allclose(rotated['diagnostics'][0][key], original['diagnostics'][0][key], atol=1e-5) self.settings.apply_amplitude = True collection, _ = self.calibration(signal) pair = np.array([trace.s21 for trace in collection.traces]).reshape(signal.shape) levels = np.abs((signal * pair * np.exp(2j * np.pi * self.freq * paths[:, :, None] / mc.C_LIGHT)).mean(axis=2)) self.assertLess(levels.max() / levels.min(), 1.01) self.assertGreater(np.max(np.abs(pair)), 1.5) def test_estimated_npz_roundtrip_keeps_physical_pairs_full_grid_and_provenance(self): collection, metadata, through = self.object_set() with tempfile.TemporaryDirectory() as directory: store = NpzStore(Path(directory)) store.save_set('s21_object_calibration', 'captured-radar', 'sphere', collection, metadata=metadata) loaded, stored = store.load_set_with_metadata('s21_object_calibration', 'captured-radar', 'sphere') self.assertEqual(stored['provenance'], metadata['provenance']) self.assertEqual(stored['targets'][0]['captures'][0]['monotonic_ns'], 1) for before, after in zip(collection.traces, loaded.traces, strict=True): self.assertEqual(before.combo, after.combo) np.testing.assert_array_equal(after.frequency_hz, self.freq.astype(np.float32)) np.testing.assert_array_equal(after.s21, before.s21) oc.validate_object_set(loaded, stored, self.config, 'captured-radar', through)