Coverage for tests/test_ndf_input_archive.py: 99%
282 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-16 15:24 -0700
1# This file is part of lsst-images.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12from __future__ import annotations
14import os
15from pathlib import Path
17import astropy.io.fits
18import astropy.units as u
19import numpy as np
20import pydantic
21import pytest
23from lsst.images import Box, Image, ImageSerializationModel, Mask, MaskedImage
24from lsst.images._transforms import FrameSet
25from lsst.images.fits import ExtensionKey, FitsOpaqueMetadata
26from lsst.images.serialization import (
27 ArchiveReadError,
28 ArrayReferenceModel,
29 InlineArrayModel,
30 NumberType,
31 read_archive,
32)
34try:
35 import h5py
37 from lsst.images.ndf import (
38 NdfInputArchive,
39 NdfOutputArchive,
40 NdfPointerModel,
41 _hds,
42 read_starlink,
43 write,
44 )
46 HAVE_H5PY = True
47except ImportError:
48 HAVE_H5PY = False
50skip_no_h5py = pytest.mark.skipif(not HAVE_H5PY, reason="h5py is not installed")
53@skip_no_h5py
54def test_open_round_trips_image_tree(tmp_path: Path) -> None:
55 """Verify NdfInputArchive.open round-trips an Image's ArchiveTree."""
56 image = Image(
57 np.arange(20, dtype=np.float32).reshape(4, 5),
58 bbox=Box.factory[10:14, 20:25],
59 )
60 path = tmp_path / "test.sdf"
61 written_tree = write(image, path)
62 with NdfInputArchive.open(path) as archive:
63 tree = archive.get_tree(type(written_tree))
64 assert tree.model_dump_json() == written_tree.model_dump_json()
67@skip_no_h5py
68def test_get_tree_raises_when_main_json_missing(tmp_path: Path) -> None:
69 """Verify get_tree raises ArchiveReadError when /MORE/LSST/JSON is
70 absent.
71 """
72 path = tmp_path / "test.sdf"
73 with h5py.File(path, "w") as f:
74 f["/"].attrs["CLASS"] = "NDF"
75 with NdfInputArchive.open(path) as archive:
76 model_type = ImageSerializationModel[NdfPointerModel]
77 with pytest.raises(ArchiveReadError):
78 archive.get_tree(model_type)
81@skip_no_h5py
82def test_get_array_reads_image_array(tmp_path: Path) -> None:
83 """Verify get_array returns the full image array from the NDF file."""
84 image = Image(np.arange(20, dtype=np.float32).reshape(4, 5))
85 path = tmp_path / "test.sdf"
86 tree = write(image, path)
87 with NdfInputArchive.open(path) as archive:
88 arr = archive.get_array(tree.data)
89 np.testing.assert_array_equal(arr, image.array)
92@skip_no_h5py
93def test_get_array_supports_slicing(tmp_path: Path) -> None:
94 """Verify get_array returns the correct sub-array when slices are given."""
95 image = Image(np.arange(20, dtype=np.float32).reshape(4, 5))
96 path = tmp_path / "test.sdf"
97 tree = write(image, path)
98 with NdfInputArchive.open(path) as archive:
99 arr = archive.get_array(tree.data, slices=(slice(0, 2), slice(1, 4)))
100 np.testing.assert_array_equal(arr, image.array[:2, 1:4])
103@skip_no_h5py
104def test_get_array_handles_inline_array(tmp_path: Path) -> None:
105 """Verify get_array converts an InlineArrayModel to a numpy array."""
106 inline = InlineArrayModel(data=[1.0, 2.0, 3.0], datatype=NumberType.float64)
107 image = Image(np.zeros((2, 2), dtype=np.float32))
108 path = tmp_path / "test.sdf"
109 write(image, path)
110 with NdfInputArchive.open(path) as archive:
111 arr = archive.get_array(inline)
112 np.testing.assert_array_equal(arr, np.array([1.0, 2.0, 3.0]))
115@skip_no_h5py
116def test_get_array_unrecognised_source_raises(tmp_path: Path) -> None:
117 """Verify get_array raises ArchiveReadError for an unrecognised source
118 scheme.
119 """
120 image = Image(np.zeros((2, 2), dtype=np.float32))
121 bogus = ArrayReferenceModel(source="fits:NOTUS", shape=[2, 2], datatype=NumberType.float32)
122 path = tmp_path / "test.sdf"
123 write(image, path)
124 with NdfInputArchive.open(path) as archive:
125 with pytest.raises(ArchiveReadError):
126 archive.get_array(bogus)
129@skip_no_h5py
130def test_deserialize_pointer_round_trips_subtree(tmp_path: Path) -> None:
131 """Verify deserialize_pointer reconstructs a hoisted sub-tree correctly."""
133 class TinyTree(pydantic.BaseModel):
134 name: str
136 path = tmp_path / "test.sdf"
137 with h5py.File(path, "w") as f:
138 arch = NdfOutputArchive(f)
139 ptr = arch.serialize_pointer("psf", lambda nested: TinyTree(name="hello"), key=("psf", 1))
140 with NdfInputArchive.open(path) as archive:
141 result = archive.deserialize_pointer(ptr, TinyTree, lambda m, _a: m)
142 assert result.name == "hello"
145@skip_no_h5py
146def test_deserialize_pointer_caches_by_ref(tmp_path: Path) -> None:
147 """Verify deserialize_pointer calls the deserializer only once for the same
148 ref.
149 """
151 class TinyTree(pydantic.BaseModel):
152 name: str
154 calls: list[TinyTree] = []
156 def deserializer(model: TinyTree, _archive: NdfInputArchive) -> TinyTree:
157 calls.append(model)
158 return model
160 path = tmp_path / "test.sdf"
161 with h5py.File(path, "w") as f:
162 arch = NdfOutputArchive(f)
163 ptr = arch.serialize_pointer("psf", lambda nested: TinyTree(name="x"), key=("psf", 1))
164 with NdfInputArchive.open(path) as archive:
165 first = archive.deserialize_pointer(ptr, TinyTree, deserializer)
166 second = archive.deserialize_pointer(ptr, TinyTree, deserializer)
167 assert first is second
168 assert len(calls) == 1
171@skip_no_h5py
172def test_deserialize_pointer_caches_frame_set_for_get_frame_set(tmp_path: Path) -> None:
173 """Verify a deserialized FrameSet is returned by get_frame_set via the
174 shared cache.
175 """
177 class TinyTree(pydantic.BaseModel):
178 name: str
180 class DummyFrameSet(FrameSet):
181 def __contains__(self, frame: object) -> bool:
182 return False
184 def __getitem__(self, key: object) -> object:
185 raise AssertionError("DummyFrameSet should not be indexed in this test.")
187 sentinel = DummyFrameSet()
189 path = tmp_path / "test.sdf"
190 with h5py.File(path, "w") as f:
191 arch = NdfOutputArchive(f)
192 ptr = arch.serialize_frame_set(
193 "frames",
194 sentinel,
195 lambda nested: TinyTree(name="frames"),
196 key=("frames", 1),
197 )
198 with NdfInputArchive.open(path) as archive:
199 result = archive.deserialize_pointer(ptr, TinyTree, lambda _m, _a: sentinel)
200 assert result is sentinel
201 assert archive.get_frame_set(ptr) is sentinel
204@skip_no_h5py
205def test_get_frame_set_returns_cached_value(tmp_path: Path) -> None:
206 """Verify get_frame_set returns a sentinel previously placed in the
207 cache.
208 """
209 sentinel = object()
210 path = tmp_path / "test.sdf"
211 write(Image(np.zeros((2, 2), dtype=np.float32)), path)
212 with NdfInputArchive.open(path) as archive:
213 archive._frame_set_cache["/MORE/LSST/PIXEL_TO_SKY"] = sentinel
214 pointer = NdfPointerModel(path="/MORE/LSST/PIXEL_TO_SKY")
215 assert archive.get_frame_set(pointer) is sentinel
218@skip_no_h5py
219def test_get_frame_set_raises_if_not_cached(tmp_path: Path) -> None:
220 """Verify get_frame_set raises AssertionError when the pointer is not in
221 the cache.
222 """
223 path = tmp_path / "test.sdf"
224 write(Image(np.zeros((2, 2), dtype=np.float32)), path)
225 with NdfInputArchive.open(path) as archive:
226 pointer = NdfPointerModel(path="/MORE/LSST/UNKNOWN")
227 with pytest.raises(AssertionError):
228 archive.get_frame_set(pointer)
231@skip_no_h5py
232def test_more_fits_round_trips_via_opaque_metadata(tmp_path: Path) -> None:
233 """Verify /MORE/FITS headers are recovered by get_opaque_metadata."""
234 image = Image(np.zeros((2, 2), dtype=np.float32))
235 primary = astropy.io.fits.Header()
236 primary["FOO"] = ("bar", "test card")
237 opaque = FitsOpaqueMetadata()
238 opaque.add_header(primary, name="", ver=1)
239 image._opaque_metadata = opaque
240 path = tmp_path / "test.sdf"
241 write(image, path)
242 with NdfInputArchive.open(path) as archive:
243 recovered = archive.get_opaque_metadata()
244 assert ExtensionKey() in recovered.headers
245 assert recovered.headers[ExtensionKey()]["FOO"] == "bar"
248@skip_no_h5py
249def test_get_opaque_metadata_empty_when_no_more_fits(tmp_path: Path) -> None:
250 """Verify get_opaque_metadata returns an empty FitsOpaqueMetadata when
251 /MORE/FITS is absent.
252 """
253 image = Image(np.zeros((2, 2), dtype=np.float32))
254 path = tmp_path / "test.sdf"
255 write(image, path)
256 with NdfInputArchive.open(path) as archive:
257 recovered = archive.get_opaque_metadata()
258 assert isinstance(recovered, FitsOpaqueMetadata)
259 assert not recovered.headers
262@skip_no_h5py
263def test_read_round_trips_image(tmp_path: Path) -> None:
264 """Verify read_archive() round-trips an Image through an NDF file."""
265 image = Image(
266 np.arange(20, dtype=np.float32).reshape(4, 5),
267 bbox=Box.factory[10:14, 20:25],
268 )
269 path = tmp_path / "test.sdf"
270 write(image, path)
271 result = read_archive(path, Image)
272 assert isinstance(result, Image)
273 np.testing.assert_array_equal(result.array, image.array)
274 assert result.bbox == image.bbox
277@skip_no_h5py
278def test_read_starlink_file_auto_detects_image(tmp_path: Path) -> None:
279 """Verify read_starlink auto-detects an Image from a schema-less NDF
280 fixture.
281 """
282 example_path = os.path.join(os.path.dirname(__file__), "data", "example-ndf.sdf")
283 result = read_starlink(Image, example_path)
284 assert isinstance(result, Image)
285 assert result.array.shape == (611, 609)
286 assert result.array.dtype == np.int16
287 assert result.sky_projection is not None
290@skip_no_h5py
291def test_read_starlink_file_recovers_opaque_fits_metadata(tmp_path: Path) -> None:
292 """Verify read_starlink recovers /MORE/FITS opaque metadata from a Starlink
293 NDF.
294 """
295 example_path = os.path.join(os.path.dirname(__file__), "data", "example-ndf.sdf")
296 result = read_starlink(Image, example_path)
297 opaque = result._opaque_metadata
298 assert ExtensionKey() in opaque.headers
299 primary = opaque.headers[ExtensionKey()]
300 assert "NAXIS" in primary
303@skip_no_h5py
304def test_read_auto_detects_nested_quality_array(tmp_path: Path) -> None:
305 """Verify read_starlink maps a QUALITY array to a MaskedImage mask."""
306 image_array = np.arange(6, dtype=np.float32).reshape(2, 3)
307 quality_array = np.array([[0, 1, 0], [1, 0, 1]], dtype=np.uint8)
309 path = tmp_path / "test.sdf"
310 with h5py.File(path, "w") as f:
311 _hds.set_root_name(f, "TEST", "NDF")
312 data_array = _hds.create_structure(f, "DATA_ARRAY", "ARRAY")
313 _hds.write_array(data_array, "DATA", image_array)
314 quality = _hds.create_structure(f, "QUALITY", "QUALITY")
315 quality_array_struct = _hds.create_structure(quality, "QUALITY", "ARRAY")
316 _hds.write_array(quality_array_struct, "DATA", quality_array)
317 _hds.write_array(quality_array_struct, "ORIGIN", np.array([0, 0], dtype=np.int32))
318 _hds.write_array(quality_array_struct, "BAD_PIXEL", np.array(False, dtype=np.bool_))
319 _hds.write_array(quality, "BADBITS", np.array(1, dtype=np.uint8))
320 result = read_starlink(MaskedImage, path)
321 assert isinstance(result, MaskedImage)
322 np.testing.assert_array_equal(result.mask.array[:, :, 0], quality_array)
323 assert set(result.mask.schema.names) == {f"MASK{i}" for i in range(8)}
324 image_result = read_starlink(Image, path)
325 assert isinstance(image_result, Image)
326 np.testing.assert_array_equal(image_result.array, image_array)
329@skip_no_h5py
330def test_read_auto_detect_preserves_quality_bits(tmp_path: Path) -> None:
331 """Verify read_starlink correctly maps BADBITS-selected quality planes."""
332 image_array = np.arange(6, dtype=np.float32).reshape(2, 3)
333 quality_array = np.array([[0, 2, 4], [2, 0, 6]], dtype=np.uint8)
334 expected_mask1 = np.array([[0, 1, 0], [1, 0, 1]], dtype=bool)
335 expected_mask2 = np.array([[0, 0, 1], [0, 0, 1]], dtype=bool)
337 path = tmp_path / "test.sdf"
338 with h5py.File(path, "w") as f:
339 _hds.set_root_name(f, "TEST", "NDF")
340 data_array = _hds.create_structure(f, "DATA_ARRAY", "ARRAY")
341 _hds.write_array(data_array, "DATA", image_array)
342 quality = _hds.create_structure(f, "QUALITY", "QUALITY")
343 quality_array_struct = _hds.create_structure(quality, "QUALITY", "ARRAY")
344 _hds.write_array(quality_array_struct, "DATA", quality_array)
345 _hds.write_array(quality_array_struct, "ORIGIN", np.array([0, 0], dtype=np.int32))
346 _hds.write_array(quality_array_struct, "BAD_PIXEL", np.array(False, dtype=np.bool_))
347 _hds.write_array(quality, "BADBITS", np.array(2, dtype=np.uint8))
348 result = read_starlink(MaskedImage, path)
349 assert isinstance(result, MaskedImage)
350 mask = result.mask
351 np.testing.assert_array_equal(mask.array[:, :, 0], quality_array)
352 np.testing.assert_array_equal(mask.get("MASK1"), expected_mask1)
353 np.testing.assert_array_equal(mask.get("MASK2"), expected_mask2)
354 assert "Selected by BADBITS" in mask.schema.descriptions["MASK1"]
355 assert "Selected by BADBITS" not in mask.schema.descriptions["MASK2"]
358@skip_no_h5py
359def test_read_auto_detected_data_only_as_masked_image_uses_defaults(tmp_path: Path) -> None:
360 """Verify read_starlink fills in default mask and variance when only
361 DATA_ARRAY is present.
362 """
363 image_array = np.arange(6, dtype=np.float32).reshape(2, 3)
365 path = tmp_path / "test.sdf"
366 with h5py.File(path, "w") as f:
367 _hds.set_root_name(f, "TEST", "NDF")
368 data_array = _hds.create_structure(f, "DATA_ARRAY", "ARRAY")
369 _hds.write_array(data_array, "DATA", image_array)
370 _hds.write_array(data_array, "ORIGIN", np.array([5, 4], dtype=np.int32))
371 result = read_starlink(MaskedImage, path)
372 assert isinstance(result, MaskedImage)
373 assert result.bbox == Box.factory[4:6, 5:8]
374 np.testing.assert_array_equal(result.image.array, image_array)
375 np.testing.assert_array_equal(result.mask.array, np.zeros((2, 3, 1), dtype=np.uint8))
376 np.testing.assert_array_equal(result.variance.array, np.ones((2, 3), dtype=np.float32))
379@skip_no_h5py
380def test_read_auto_detected_variance_as_masked_image_keeps_variance(tmp_path: Path) -> None:
381 """Verify read_starlink preserves a VARIANCE component in the
382 MaskedImage.
383 """
384 image_array = np.arange(6, dtype=np.float32).reshape(2, 3)
385 variance_array = np.full((2, 3), 2.5, dtype=np.float32)
387 path = tmp_path / "test.sdf"
388 with h5py.File(path, "w") as f:
389 _hds.set_root_name(f, "TEST", "NDF")
390 data_array = _hds.create_structure(f, "DATA_ARRAY", "ARRAY")
391 _hds.write_array(data_array, "DATA", image_array)
392 _hds.write_array(data_array, "ORIGIN", np.array([5, 4], dtype=np.int32))
393 variance = _hds.create_structure(f, "VARIANCE", "ARRAY")
394 _hds.write_array(variance, "DATA", variance_array)
395 _hds.write_array(variance, "ORIGIN", np.array([5, 4], dtype=np.int32))
396 result = read_starlink(MaskedImage, path)
397 assert isinstance(result, MaskedImage)
398 np.testing.assert_array_equal(result.variance.array, variance_array)
399 np.testing.assert_array_equal(result.mask.array, np.zeros((2, 3, 1), dtype=np.uint8))
402@skip_no_h5py
403def test_read_auto_detected_units_component(tmp_path: Path) -> None:
404 """Verify read_starlink maps a UNITS NDF component to the Image unit."""
405 image_array = np.arange(6, dtype=np.float32).reshape(2, 3)
407 path = tmp_path / "test.sdf"
408 with h5py.File(path, "w") as f:
409 _hds.set_root_name(f, "TEST", "NDF")
410 data_array = _hds.create_structure(f, "DATA_ARRAY", "ARRAY")
411 _hds.write_array(data_array, "DATA", image_array)
412 f.create_dataset("UNITS", data=np.bytes_("count"))
413 result = read_starlink(Image, path)
414 assert result.unit == u.ct
417@skip_no_h5py
418def test_read_missing_data_array_raises(tmp_path: Path) -> None:
419 """Verify read_starlink raises ArchiveReadError for an NDF with no
420 DATA_ARRAY or JSON.
421 """
422 path = tmp_path / "test.sdf"
423 with h5py.File(path, "w") as f:
424 f["/"].attrs["CLASS"] = "NDF"
425 with pytest.raises(ArchiveReadError):
426 read_starlink(Image, path)
429@skip_no_h5py
430def test_read_auto_detect_wrong_target_type_raises(tmp_path: Path) -> None:
431 """Verify read_starlink raises ArchiveReadError when the target type is
432 unsupported.
433 """
434 example_path = os.path.join(os.path.dirname(__file__), "data", "example-ndf.sdf")
435 with pytest.raises(ArchiveReadError):
436 read_starlink(Mask, example_path)