Coverage for tests / test_formatters.py: 30%
245 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-16 00:52 -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 tempfile
15import unittest
16import warnings
18import numpy as np
20from lsst.images import (
21 Box,
22 Image,
23 MaskedImage,
24 MaskPlane,
25 MaskSchema,
26 VisitImage,
27 fits,
28)
29from lsst.images import json as images_json
30from lsst.images.fits import formatters as fits_shim
31from lsst.images.fits._common import PointerModel
32from lsst.images.fits._input_archive import FitsInputArchive
33from lsst.images.formatters import (
34 _BACKENDS,
35 GenericFormatter,
36 ImageFormatter,
37 MaskedImageFormatter,
38 VisitImageFormatter,
39)
40from lsst.images.json import formatters as json_shim
41from lsst.images.tests import make_test_formatter
42from lsst.resources import ResourcePath
44try:
45 from lsst.images import ndf
46 from lsst.images.ndf._common import NdfPointerModel
47 from lsst.images.ndf._input_archive import NdfInputArchive
49 HAVE_H5PY = True
50except ImportError:
51 HAVE_H5PY = False
54class BackendsTableTestCase(unittest.TestCase):
55 """The private _BACKENDS table wires extension -> read/write/archive."""
57 def test_table_keys(self):
58 expected = {".fits", ".json"}
59 if HAVE_H5PY:
60 expected.add(".sdf")
61 self.assertEqual(set(_BACKENDS), expected)
63 def test_fits_backend_wires_fits_read_write(self):
64 backend = _BACKENDS[".fits"]
65 self.assertIs(backend.read, fits.read)
66 self.assertIs(backend.write, fits.write)
67 self.assertIs(backend.input_archive, FitsInputArchive)
68 self.assertIs(backend.pointer_model, PointerModel)
70 @unittest.skipUnless(HAVE_H5PY, "h5py is not installed")
71 def test_sdf_backend_wires_ndf_read_write(self):
72 backend = _BACKENDS[".sdf"]
73 self.assertIs(backend.read, ndf.read)
74 self.assertIs(backend.write, ndf.write)
75 self.assertIs(backend.input_archive, NdfInputArchive)
76 self.assertIs(backend.pointer_model, NdfPointerModel)
78 def test_json_backend_wires_json_read_write_no_archive(self):
79 backend = _BACKENDS[".json"]
80 self.assertIs(backend.read, images_json.read)
81 self.assertIs(backend.write, images_json.write)
82 self.assertIsNone(backend.input_archive)
83 self.assertIsNone(backend.pointer_model)
86class GetWriteExtensionTestCase(unittest.TestCase):
87 """`get_write_extension` reads the `format` write parameter."""
89 def _make_formatter(self, write_parameters: dict[str, str] | None = None):
90 return make_test_formatter(GenericFormatter, Image, write_parameters=write_parameters)
92 def test_default_returns_fits(self):
93 formatter = self._make_formatter()
94 self.assertEqual(formatter.get_write_extension(), ".fits")
96 def test_explicit_fits(self):
97 formatter = self._make_formatter({"format": "fits"})
98 self.assertEqual(formatter.get_write_extension(), ".fits")
100 def test_explicit_json(self):
101 formatter = self._make_formatter({"format": "json"})
102 self.assertEqual(formatter.get_write_extension(), ".json")
104 @unittest.skipUnless(HAVE_H5PY, "h5py is not installed")
105 def test_explicit_sdf(self):
106 formatter = self._make_formatter({"format": "sdf"})
107 self.assertEqual(formatter.get_write_extension(), ".sdf")
109 def test_unknown_format_raises(self):
110 formatter = self._make_formatter({"format": "pickle"})
111 with self.assertRaisesRegex(RuntimeError, "is not supported"):
112 formatter.get_write_extension()
114 def test_recipe_with_non_fits_format_raises(self):
115 # `recipe` is FITS-only; using it with format=json must error.
116 formatter = self._make_formatter({"format": "json", "recipe": "default"})
117 with self.assertRaisesRegex(RuntimeError, "only valid for FITS"):
118 formatter._validate_write_parameters()
121class ExtensionFromUriTestCase(unittest.TestCase):
122 """`read_from_uri` routes based on `uri.getExtension()`."""
124 def _make_formatter(self):
125 return make_test_formatter(GenericFormatter, Image)
127 def test_fits(self):
128 formatter = self._make_formatter()
129 uri = ResourcePath("/tmp/x.fits")
130 self.assertEqual(formatter._extension_from_uri(uri), ".fits")
132 @unittest.skipUnless(HAVE_H5PY, "h5py is not installed")
133 def test_sdf(self):
134 formatter = self._make_formatter()
135 uri = ResourcePath("/tmp/x.sdf")
136 self.assertEqual(formatter._extension_from_uri(uri), ".sdf")
138 def test_json(self):
139 formatter = self._make_formatter()
140 uri = ResourcePath("/tmp/x.json")
141 self.assertEqual(formatter._extension_from_uri(uri), ".json")
143 def test_unknown(self):
144 formatter = self._make_formatter()
145 uri = ResourcePath("/tmp/x.pickle")
146 with self.assertRaisesRegex(RuntimeError, "unsupported extension"):
147 formatter._extension_from_uri(uri)
149 def test_compressed_fits_unsupported(self):
150 # We don't claim to handle .fits.gz; getExtension returns
151 # '.fits.gz' and the lookup misses.
152 formatter = self._make_formatter()
153 uri = ResourcePath("/tmp/x.fits.gz")
154 with self.assertRaisesRegex(RuntimeError, "unsupported extension"):
155 formatter._extension_from_uri(uri)
158class ImageFormatterFullReadTestCase(unittest.TestCase):
159 """`read_from_uri(component=None)` round-trips each backend."""
161 def _make_image(self):
162 return Image(
163 np.arange(20, dtype=np.float32).reshape(4, 5),
164 bbox=Box.factory[10:14, 20:25],
165 )
167 def test_fits_full_read(self):
168 image = self._make_image()
169 with tempfile.NamedTemporaryFile(suffix=".fits", delete_on_close=False) as tmp:
170 tmp.close()
171 fits.write(image, tmp.name)
172 formatter = make_test_formatter(ImageFormatter, Image)
173 result = formatter.read_from_uri(ResourcePath(tmp.name))
174 np.testing.assert_array_equal(result.array, image.array)
176 @unittest.skipUnless(HAVE_H5PY, "h5py is not installed")
177 def test_sdf_full_read(self):
178 image = self._make_image()
179 with tempfile.NamedTemporaryFile(suffix=".sdf", delete_on_close=False) as tmp:
180 tmp.close()
181 ndf.write(image, tmp.name)
182 formatter = make_test_formatter(ImageFormatter, Image)
183 result = formatter.read_from_uri(ResourcePath(tmp.name))
184 np.testing.assert_array_equal(result.array, image.array)
186 def test_json_full_read(self):
187 image = self._make_image()
188 with tempfile.NamedTemporaryFile(suffix=".json", delete_on_close=False) as tmp:
189 tmp.close()
190 images_json.write(image, tmp.name)
191 formatter = make_test_formatter(ImageFormatter, Image)
192 result = formatter.read_from_uri(ResourcePath(tmp.name))
193 np.testing.assert_array_equal(result.array, image.array)
196class ImageFormatterComponentReadTestCase(unittest.TestCase):
197 """ImageFormatter routes component reads per extension."""
199 def _make_image(self):
200 return Image(
201 np.arange(20, dtype=np.float32).reshape(4, 5),
202 bbox=Box.factory[10:14, 20:25],
203 )
205 def test_fits_bbox_component(self):
206 image = self._make_image()
207 with tempfile.NamedTemporaryFile(suffix=".fits", delete_on_close=False) as tmp:
208 tmp.close()
209 fits.write(image, tmp.name)
210 formatter = make_test_formatter(ImageFormatter, Image)
211 bbox = formatter._read_component_from_uri("bbox", ResourcePath(tmp.name))
212 self.assertEqual(bbox, image.bbox)
214 @unittest.skipUnless(HAVE_H5PY, "h5py is not installed")
215 def test_sdf_bbox_component(self):
216 image = self._make_image()
217 with tempfile.NamedTemporaryFile(suffix=".sdf", delete_on_close=False) as tmp:
218 tmp.close()
219 ndf.write(image, tmp.name)
220 formatter = make_test_formatter(ImageFormatter, Image)
221 bbox = formatter._read_component_from_uri("bbox", ResourcePath(tmp.name))
222 self.assertEqual(bbox, image.bbox)
224 def test_json_bbox_component_via_whole_object(self):
225 image = self._make_image()
226 with tempfile.NamedTemporaryFile(suffix=".json", delete_on_close=False) as tmp:
227 tmp.close()
228 images_json.write(image, tmp.name)
229 formatter = make_test_formatter(ImageFormatter, Image)
230 bbox = formatter._read_component_from_uri("bbox", ResourcePath(tmp.name))
231 self.assertEqual(bbox, image.bbox)
233 def test_json_unknown_component_raises(self):
234 image = self._make_image()
235 with tempfile.NamedTemporaryFile(suffix=".json", delete_on_close=False) as tmp:
236 tmp.close()
237 images_json.write(image, tmp.name)
238 formatter = make_test_formatter(ImageFormatter, Image)
239 with self.assertRaises(NotImplementedError):
240 formatter._read_component_from_uri("nonexistent", ResourcePath(tmp.name))
243class MaskedImageFormatterComponentReadTestCase(unittest.TestCase):
244 """MaskedImageFormatter routes image/mask/variance per extension."""
246 def _make_masked_image(self):
247 rng = np.random.default_rng(11)
248 return MaskedImage(
249 Image(rng.normal(100.0, 8.0, size=(10, 12)), start=(0, 0)),
250 mask_schema=MaskSchema([MaskPlane("BAD", "bad pixel")]),
251 )
253 def test_fits_image_component(self):
254 mi = self._make_masked_image()
255 with tempfile.NamedTemporaryFile(suffix=".fits", delete_on_close=False) as tmp:
256 tmp.close()
257 fits.write(mi, tmp.name)
258 formatter = make_test_formatter(MaskedImageFormatter, MaskedImage)
259 image = formatter._read_component_from_uri("image", ResourcePath(tmp.name))
260 self.assertEqual(image.bbox, mi.image.bbox)
262 @unittest.skipUnless(HAVE_H5PY, "h5py is not installed")
263 def test_sdf_mask_component(self):
264 mi = self._make_masked_image()
265 with tempfile.NamedTemporaryFile(suffix=".sdf", delete_on_close=False) as tmp:
266 tmp.close()
267 ndf.write(mi, tmp.name)
268 formatter = make_test_formatter(MaskedImageFormatter, MaskedImage)
269 mask = formatter._read_component_from_uri("mask", ResourcePath(tmp.name))
270 self.assertEqual(mask.bbox, mi.mask.bbox)
272 def test_json_variance_component_via_whole_object(self):
273 mi = self._make_masked_image()
274 with tempfile.NamedTemporaryFile(suffix=".json", delete_on_close=False) as tmp:
275 tmp.close()
276 images_json.write(mi, tmp.name)
277 formatter = make_test_formatter(MaskedImageFormatter, MaskedImage)
278 variance = formatter._read_component_from_uri("variance", ResourcePath(tmp.name))
279 self.assertEqual(variance.bbox, mi.variance.bbox)
282class VisitImageFormatterComponentReadTestCase(unittest.TestCase):
283 """VisitImageFormatter reads VisitImage-specific components."""
285 def _make_visit_image(self):
286 # Reuse the existing test helper from tests/test_visit_image.py.
287 # Pytest places the tests directory on sys.path, so import the
288 # sibling module by its bare name.
289 from test_visit_image import VisitImageTestCase # local import
291 VisitImageTestCase.setUpClass()
292 return VisitImageTestCase.visit_image
294 def test_fits_summary_stats_component(self):
295 vi = self._make_visit_image()
296 with tempfile.NamedTemporaryFile(suffix=".fits", delete_on_close=False) as tmp:
297 tmp.close()
298 fits.write(vi, tmp.name)
299 formatter = make_test_formatter(VisitImageFormatter, VisitImage)
300 summary = formatter._read_component_from_uri("summary_stats", ResourcePath(tmp.name))
301 self.assertEqual(summary, vi.summary_stats)
303 @unittest.skipUnless(HAVE_H5PY, "h5py is not installed")
304 def test_sdf_psf_component(self):
305 vi = self._make_visit_image()
306 with tempfile.NamedTemporaryFile(suffix=".sdf", delete_on_close=False) as tmp:
307 tmp.close()
308 ndf.write(vi, tmp.name)
309 formatter = make_test_formatter(VisitImageFormatter, VisitImage)
310 psf = formatter._read_component_from_uri("psf", ResourcePath(tmp.name))
311 self.assertEqual(type(psf), type(vi.psf))
313 def test_json_aperture_corrections_via_whole_object(self):
314 vi = self._make_visit_image()
315 with tempfile.NamedTemporaryFile(suffix=".json", delete_on_close=False) as tmp:
316 tmp.close()
317 images_json.write(vi, tmp.name)
318 formatter = make_test_formatter(VisitImageFormatter, VisitImage)
319 ap = formatter._read_component_from_uri("aperture_corrections", ResourcePath(tmp.name))
320 # ChebyshevField has no __eq__; compare keys and types.
321 self.assertEqual(ap.keys(), vi.aperture_corrections.keys())
322 for k, v in vi.aperture_corrections.items():
323 self.assertEqual(type(ap[k]), type(v))
326class FitsDeprecationShimTestCase(unittest.TestCase):
327 """lsst.images.fits.formatters is a deprecation shim."""
329 def test_image_formatter_warns(self):
330 with warnings.catch_warnings(record=True) as recorded:
331 warnings.simplefilter("always")
332 make_test_formatter(fits_shim.ImageFormatter, Image)
333 self.assertTrue(
334 any(
335 issubclass(w.category, DeprecationWarning)
336 and "fits.formatters.ImageFormatter is deprecated" in str(w.message)
337 for w in recorded
338 ),
339 f"No deprecation warning observed; got: {[str(w.message) for w in recorded]}",
340 )
342 def test_subclass_is_unified_class(self):
343 from lsst.images import formatters as unified
345 self.assertTrue(issubclass(fits_shim.GenericFormatter, unified.GenericFormatter))
346 self.assertTrue(issubclass(fits_shim.ImageFormatter, unified.ImageFormatter))
347 self.assertTrue(issubclass(fits_shim.MaskedImageFormatter, unified.MaskedImageFormatter))
348 self.assertTrue(issubclass(fits_shim.VisitImageFormatter, unified.VisitImageFormatter))
349 self.assertTrue(issubclass(fits_shim.CellCoaddFormatter, unified.CellCoaddFormatter))
352class JsonDeprecationShimTestCase(unittest.TestCase):
353 """lsst.images.json.formatters is a deprecation shim.
355 The shim defaults to ``.json`` output.
356 """
358 def test_generic_formatter_warns(self):
359 with warnings.catch_warnings(record=True) as recorded:
360 warnings.simplefilter("always")
361 make_test_formatter(json_shim.GenericFormatter, Image)
362 self.assertTrue(
363 any(
364 issubclass(w.category, DeprecationWarning)
365 and "json.formatters.GenericFormatter is deprecated" in str(w.message)
366 for w in recorded
367 )
368 )
370 def test_default_extension_is_json(self):
371 self.assertEqual(json_shim.GenericFormatter.default_extension, ".json")
373 def test_default_write_extension_is_json(self):
374 formatter = make_test_formatter(json_shim.GenericFormatter, Image)
375 self.assertEqual(formatter.get_write_extension(), ".json")
378if __name__ == "__main__":
379 unittest.main()