82 lines
4.3 KiB
Python
82 lines
4.3 KiB
Python
"""Embedded preprocess metadata and compatibility with older NPZ sets."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData
|
|
from python_app.storage.npz.store import NpzStore
|
|
|
|
|
|
class SetMetadataTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.root = Path(temporary.name)
|
|
self.store = NpzStore(self.root)
|
|
self.kind = "s21_object_calibration"
|
|
self.collection = SweepCollection(7, 100, [TraceData(
|
|
ComboKey(2, 3), np.array([1e9, 2e9], dtype=np.float32),
|
|
np.array([1 + 2j, 3 - 4j], dtype=np.complex64),
|
|
capture_start_ns=80, capture_end_ns=90,
|
|
)], capture_start_ns=80, capture_end_ns=90)
|
|
self.metadata = {"radar_key": "radar", "geometry": [{"input_pos": 2, "x": 0.1}],
|
|
"through": {"name": "through", "hash": "abc"}, "collection_id": "user value"}
|
|
self.store.save_set(self.kind, "radar", "object", self.collection, metadata=self.metadata)
|
|
self.archive = self.root / self.kind / "radar/object.npz"
|
|
self.sidecar = self.archive.with_suffix(".json")
|
|
|
|
|
|
def test_metadata_and_collection_round_trip_through_generic_set_api(self) -> None:
|
|
loaded, metadata = self.store.load_set_with_metadata(self.kind, "radar", "object")
|
|
self.assertEqual(metadata, self.metadata)
|
|
self.assertEqual((loaded.collection_id, loaded.monotonic_ns), (7, 100))
|
|
self.assertEqual((loaded.capture_start_ns, loaded.capture_end_ns), (80, 90))
|
|
self.assertEqual((loaded.traces[0].capture_start_ns, loaded.traces[0].capture_end_ns), (80, 90))
|
|
np.testing.assert_array_equal(loaded.traces[0].s21, self.collection.traces[0].s21)
|
|
np.testing.assert_array_equal(loaded.traces[0].frequency_hz, self.collection.traces[0].frequency_hz)
|
|
self.assertEqual(self.store.load_set(self.kind, "radar", "object").collection_id, 7)
|
|
self.assertEqual(self.store.list_sets(self.kind, "radar"), ["object"])
|
|
self.assertTrue(self.store.has_combo_coverage(self.kind, "radar", "object", [ComboKey(2, 3)]))
|
|
self.assertFalse(self.store.has_combo_coverage(self.kind, "radar", "object", [ComboKey(0, 0)]))
|
|
|
|
|
|
def test_embedded_metadata_is_authoritative_when_sidecar_is_stale_or_missing(self) -> None:
|
|
for sidecar in ('{"metadata": {"radar_key": "wrong"}}', "broken JSON", None):
|
|
with self.subTest(sidecar=sidecar):
|
|
if sidecar is None:
|
|
self.sidecar.unlink()
|
|
else:
|
|
self.sidecar.write_text(sidecar, encoding="utf-8")
|
|
_, metadata = self.store.load_set_with_metadata(self.kind, "radar", "object")
|
|
self.assertEqual(metadata, self.metadata)
|
|
|
|
|
|
def test_old_sets_without_user_metadata_load_with_empty_dict(self) -> None:
|
|
with np.load(self.archive, allow_pickle=False) as stored:
|
|
arrays = {key: stored[key] for key in stored.files if key != "metadata_json"}
|
|
envelope = json.loads(self.sidecar.read_text(encoding="utf-8"))
|
|
del envelope["metadata"]
|
|
self.sidecar.write_text(json.dumps(envelope), encoding="utf-8")
|
|
for embedded in (True, False):
|
|
with self.subTest(embedded=embedded):
|
|
payload = dict(arrays)
|
|
if embedded:
|
|
payload["metadata_json"] = np.asarray(json.dumps(envelope))
|
|
np.savez(self.archive, **payload)
|
|
collection, metadata = self.store.load_set_with_metadata(self.kind, "radar", "object")
|
|
self.assertEqual(metadata, {})
|
|
np.testing.assert_array_equal(collection.traces[0].s21, self.collection.traces[0].s21)
|
|
|
|
|
|
def test_invalid_metadata_cannot_overwrite_saved_set(self) -> None:
|
|
before = (self.archive.read_bytes(), self.sidecar.read_bytes())
|
|
for metadata in ({"value": float("nan")}, {"value": [float("inf")]}, {"value": object()}):
|
|
with self.subTest(metadata=metadata), self.assertRaises((TypeError, ValueError)):
|
|
self.store.save_set(self.kind, "radar", "object", self.collection, metadata=metadata)
|
|
self.assertEqual((self.archive.read_bytes(), self.sidecar.read_bytes()), before)
|
|
|