299 lines
16 KiB
Python
299 lines
16 KiB
Python
"""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)
|