92 lines
5.3 KiB
Python
92 lines
5.3 KiB
Python
"""Phase-only calibration preserves acquisition settings and unrelated JSON bytes."""
|
|
|
|
from contextlib import redirect_stdout
|
|
from io import StringIO
|
|
import json
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import numpy as np
|
|
|
|
from python_app.models.run_config_schema import RunConfigModel
|
|
from python_app.scripts import rph_radar_calibrate as calibration
|
|
|
|
|
|
class RphCalibrationTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
logging_setup = patch.object(calibration.logging, "basicConfig")
|
|
logging_setup.start()
|
|
self.addCleanup(logging_setup.stop)
|
|
self.payload = {"radar": {"model": "rph_radar", "driver_mode": "native", "rph_radar": {
|
|
"executable_path": "/custom/collector", "args": ["duration_ms:123"],
|
|
"band": {"start_hz": 1.2e9, "stop_hz": 1.8e9, "points": 137},
|
|
"phase_calibration": {"phase0_rad": 3.456789, "phase1_rad": 9.456789,
|
|
"freq0_hz": 1e9, "freq1_hz": 2e9}}}, "note": "Пример: phase0_rad"}
|
|
self.config = RunConfigModel.from_dict(self.payload)
|
|
self.service = Mock()
|
|
|
|
def raw(self, phase: np.ndarray) -> SimpleNamespace:
|
|
return SimpleNamespace(reference=np.exp(1j * phase), dirty=False)
|
|
|
|
def test_apply_preserves_every_other_byte_and_uses_current_settings(self) -> None:
|
|
for indent, separators in ((None, (",", ":")), (2, (",", ": ")), ("\t", (",", " : "))):
|
|
original = json.dumps(self.payload, ensure_ascii=False, indent=indent, separators=separators).replace("\n", "\r\n").encode()
|
|
for apply in (False, True):
|
|
with self.subTest(indent=indent, apply=apply), TemporaryDirectory() as directory:
|
|
path = Path(directory) / "profile.json"
|
|
path.write_bytes(original)
|
|
self.service.read_raw_sweep.side_effect = [self.raw(np.array([0., 1.]))] + [
|
|
self.raw(np.linspace(start, start + 2., 20)) for start in (3.10, -3.10, 3.12)]
|
|
argv = ["calibrate", "--config", str(path), "--sweeps", "3", "--warmup", "1"]
|
|
with patch.object(calibration, "RphRadarService", return_value=self.service) as factory, \
|
|
patch("sys.argv", argv + (["--apply"] if apply else [])), redirect_stdout(StringIO()) as output:
|
|
self.assertEqual(calibration.main(), 0)
|
|
result = json.loads(output.getvalue())
|
|
self.assertAlmostEqual(result["phase0_rad"], 3.12)
|
|
self.assertAlmostEqual(result["phase1_rad"], 5.12)
|
|
expected = original
|
|
if apply:
|
|
expected = original.replace(b"3.456789", json.dumps(result["phase0_rad"]).encode()).replace(
|
|
b"9.456789", json.dumps(result["phase1_rad"]).encode())
|
|
self.assertEqual(path.read_bytes(), expected)
|
|
self.assertEqual(factory.call_args.args[0].to_dict(), self.config.to_dict())
|
|
self.assertEqual(self.service.read_raw_sweep.call_count, 4)
|
|
self.service.close.assert_called_once()
|
|
self.service.reset_mock()
|
|
|
|
def test_invalid_reference_never_produces_anchors_and_always_closes(self) -> None:
|
|
for reference in (np.array([]), np.array([1]), np.array([1, np.nan]),
|
|
np.array([1, np.inf]), np.array([0, 1]), np.ones(3), np.ones((2, 2))):
|
|
self.service.read_raw_sweep.return_value = SimpleNamespace(reference=reference, dirty=False)
|
|
with self.subTest(reference=reference), patch.object(calibration, "RphRadarService", return_value=self.service):
|
|
with self.assertRaises(ValueError):
|
|
calibration.calibrate(self.config, sweeps=1, warmup=0)
|
|
self.service.close.assert_called_once()
|
|
self.service.reset_mock()
|
|
|
|
def test_changed_profile_is_not_overwritten(self) -> None:
|
|
with TemporaryDirectory() as directory:
|
|
path = Path(directory) / "profile.json"
|
|
path.write_text(json.dumps(self.payload))
|
|
def change_profile(*_args):
|
|
path.write_bytes(b"external edit")
|
|
return {"phase0_rad": 3., "phase1_rad": 6.}
|
|
with patch.object(calibration, "calibrate", side_effect=change_profile), \
|
|
patch("sys.argv", ["calibrate", "--config", str(path), "--apply"]), redirect_stdout(StringIO()):
|
|
with self.assertRaisesRegex(RuntimeError, "changed during capture"):
|
|
calibration.main()
|
|
self.assertEqual(path.read_bytes(), b"external edit")
|
|
|
|
def test_missing_anchors_fail_before_capture(self) -> None:
|
|
del self.payload["radar"]["rph_radar"]["phase_calibration"]["phase0_rad"]
|
|
with TemporaryDirectory() as directory:
|
|
path = Path(directory) / "profile.json"
|
|
path.write_text(json.dumps(self.payload))
|
|
with patch.object(calibration, "calibrate") as capture, patch("sys.argv", ["calibrate", "--config", str(path)]):
|
|
with self.assertRaisesRegex(ValueError, "explicit.*phase0_rad"):
|
|
calibration.main()
|
|
capture.assert_not_called()
|