Coverage for tests/test_visit_image.py: 65%
612 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 08:38 +0000
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 dataclasses
15import math
16import os
17import warnings
18from typing import Any, Literal
20import astropy.io.fits
21import astropy.units as u
22import astropy.wcs
23import numpy as np
24import pytest
25from astro_metadata_translator import ObservationInfo
27from lsst.images import (
28 Background,
29 BackgroundMap,
30 Box,
31 DetectorFrame,
32 DifferenceImage,
33 Image,
34 MaskPlane,
35 MaskSchema,
36 ObservationSummaryStats,
37 Polygon,
38 SkyProjectionAstropyView,
39 TractFrame,
40 VisitImage,
41 get_legacy_difference_image_mask_planes,
42 get_legacy_visit_image_mask_planes,
43)
44from lsst.images.aperture_corrections import ApertureCorrectionMap, aperture_corrections_to_legacy
45from lsst.images.cameras import Detector
46from lsst.images.describe import DescribableMixin, DescribeOptions, FieldRole, Report
47from lsst.images.fields import ChebyshevField, SplineField, SumField, field_from_legacy_photo_calib
48from lsst.images.fits import ExtensionKey, FitsOpaqueMetadata
49from lsst.images.psfs import GaussianPointSpreadFunction, PointSpreadFunction
50from lsst.images.serialization import ArchiveReadError, read_archive
51from lsst.images.tests import (
52 DP2_VISIT_DETECTOR_DATA_ID,
53 RoundtripFits,
54 RoundtripJson,
55 RoundtripNdf,
56 TemporaryButler,
57 assert_close,
58 assert_masked_images_equal,
59 assert_sky_projections_equal,
60 assert_visit_images_equal,
61 compare_aperture_corrections_to_legacy,
62 compare_detector_to_legacy,
63 compare_photo_calib_to_legacy,
64 compare_visit_image_to_legacy,
65 make_random_sky_projection,
66)
68try:
69 import h5py # noqa: F401
71 HAVE_H5PY = True
72except ImportError:
73 HAVE_H5PY = False
75try:
76 from lsst.afw.image import Exposure as LegacyExposure
77 from lsst.afw.image import VisitInfo as LegacyVisitInfo
78except ImportError:
79 type LegacyExposure = Any # type: ignore[no-redef]
80 type LegacyVisitInfo = Any # type: ignore[no-redef]
82EXTERNAL_DATA_DIR = os.environ.get("TESTDATA_IMAGES_DIR", None)
83LOCAL_DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
85skip_no_h5py = pytest.mark.skipif(not HAVE_H5PY, reason="h5py is not installed")
88@pytest.fixture(scope="session")
89def visit_image_components() -> dict[str, Any]:
90 """Return a dictionary of VisitImage components."""
91 rng = np.random.default_rng(500)
92 det_frame = DetectorFrame(instrument="Inst", visit=1234, detector=1, bbox=Box.factory[1:4096, 1:4096])
93 mask_schema = MaskSchema([MaskPlane("M1", "D1")])
94 obs_info = ObservationInfo(instrument="LSSTCam", detector_num=4, physical_filter="r1")
95 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
96 gaussian_psf = GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13])
97 aperture_corrections: ApertureCorrectionMap = {
98 "flux1": ChebyshevField(det_frame.bbox, np.array([0.75])),
99 "flux2": ChebyshevField(det_frame.bbox, np.array([0.625])),
100 }
101 detector = read_archive(os.path.join(LOCAL_DATA_DIR, "detector.json"), Detector)
102 # Real visit images have float pixels, and some operations (e.g. rendering
103 # a photometric scaling) are only defined for floating-point images.
104 image = Image(42.0, shape=(1024, 1024), unit=u.nJy, dtype=np.float32)
105 variance = Image(5.0, shape=(1024, 1024), unit=u.nJy * u.nJy, dtype=np.float32)
106 # polygon is the lower triangle of the image.
107 polygon = Polygon(x_vertices=[-0.5, 1023.5, -0.5], y_vertices=[-0.5, -0.5, 1023.5])
108 sky_projection = make_random_sky_projection(rng, det_frame, det_frame.bbox)
109 return {
110 "mask_schema": mask_schema,
111 "obs_info": obs_info,
112 "summary_stats": summary_stats,
113 "gaussian_psf": gaussian_psf,
114 "aperture_corrections": aperture_corrections,
115 "detector": detector,
116 "image": image,
117 "variance": variance,
118 "polygon": polygon,
119 "sky_projection": sky_projection,
120 }
123def make_visit_image(components: dict[str, Any]) -> VisitImage:
124 """Construct a new VisitImage with most components populated."""
125 det_frame = components["sky_projection"].pixel_frame
126 opaque = FitsOpaqueMetadata()
127 hdr = astropy.io.fits.Header()
128 with warnings.catch_warnings():
129 # Silence warnings about long keys becoming HIERARCH.
130 warnings.simplefilter("ignore", category=astropy.io.fits.verify.VerifyWarning)
131 hdr.update({"PLATFORM": "lsstcam", "LSST BUTLER ID": "123456789"})
132 opaque.extract_legacy_primary_header(hdr)
133 # API signature suggests sky_projection and obs_info can be None but
134 # they are required (unless you pass them in via the image plane).
135 vi = VisitImage(
136 components["image"],
137 variance=components["variance"],
138 psf=GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13]),
139 mask_schema=components["mask_schema"],
140 sky_projection=components["sky_projection"],
141 obs_info=components["obs_info"],
142 summary_stats=components["summary_stats"],
143 detector=components["detector"],
144 bounds=components["polygon"],
145 aperture_corrections=components["aperture_corrections"],
146 band="r",
147 )
148 vi.backgrounds.add(
149 "standard",
150 ChebyshevField(det_frame.bbox, np.array([[2.0]])),
151 description="Background subtracted from the image.",
152 is_subtracted=True,
153 )
154 vi._opaque_metadata = opaque
155 return vi
158def make_simplest_visit_image(components: dict[str, Any]) -> VisitImage:
159 """Construct a VisitImage with the minimal set of components populated."""
160 return VisitImage(
161 components["image"],
162 psf=GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13]),
163 mask_schema=components["mask_schema"],
164 sky_projection=components["sky_projection"],
165 detector=components["detector"],
166 obs_info=components["obs_info"],
167 band="r",
168 )
171def _make_sum_background_visit_image(components: dict[str, Any], visit_image: VisitImage) -> VisitImage:
172 """Return a VisitImage whose subtracted background is a SumField."""
173 rng = np.random.default_rng(42)
174 bbox = visit_image.image.sky_projection.pixel_frame.bbox
175 bin_y = bbox.y.linspace(6)
176 bin_x = bbox.x.linspace(7)
177 spline_a = SplineField(
178 bbox,
179 rng.standard_normal(size=(bin_y.size, bin_x.size)),
180 y=bin_y,
181 x=bin_x,
182 )
183 spline_b = SplineField(
184 bbox,
185 rng.standard_normal(size=(bin_y.size, bin_x.size)),
186 y=bin_y,
187 x=bin_x,
188 )
189 sum_field = SumField([spline_a, spline_b])
190 bg_map = BackgroundMap()
191 bg_map.add(
192 "stacked",
193 sum_field,
194 description="Two-operand SumField subtracted background.",
195 is_subtracted=True,
196 )
197 return VisitImage(
198 components["image"],
199 variance=components["variance"],
200 psf=components["gaussian_psf"],
201 mask_schema=components["mask_schema"],
202 sky_projection=components["sky_projection"],
203 obs_info=components["obs_info"],
204 summary_stats=components["summary_stats"],
205 detector=components["detector"],
206 band="r",
207 backgrounds=bg_map,
208 )
211def _check_sum_background_round_trip(result: VisitImage, original: VisitImage) -> None:
212 """Assert that a round-tripped SumField background matches the original."""
213 subtracted = result.backgrounds.subtracted
214 assert subtracted is not None
215 assert isinstance(subtracted.field, SumField)
216 original_subtracted = original.backgrounds.subtracted
217 assert original_subtracted is not None
218 original_field = original_subtracted.field
219 assert isinstance(original_field, SumField)
220 round_field = subtracted.field
221 assert isinstance(round_field, SumField)
222 assert len(round_field.operands) == len(original_field.operands)
223 for round_op, orig_op in zip(round_field.operands, original_field.operands, strict=True):
224 assert round_op == orig_op
227def test_visit_image_repr_str_pinned(visit_image_components: dict[str, Any]) -> None:
228 """Pin the exact str and repr output of a VisitImage."""
229 visit = make_simplest_visit_image(visit_image_components)
230 assert str(visit) == "VisitImage(Image([y=0:1024, x=0:1024], float32), ['M1'])"
231 assert repr(visit) == (
232 "VisitImage(Image(..., bbox=Box(y=Interval(start=0, stop=1024), x=Interval(start=0, stop=1024)),"
233 " dtype=dtype('float32')), mask_schema=MaskSchema([MaskPlane(name='M1', description='D1')],"
234 " dtype=dtype('uint8')))"
235 )
238def test_basics(visit_image_components: dict[str, Any]) -> None:
239 """Verify VisitImage constructor patterns and required-argument checks."""
240 c = visit_image_components
241 # Test default fill of variance.
242 visit = make_simplest_visit_image(c)
243 assert visit.variance.array[0, 0] == 1.0
244 assert visit[...] is not visit
245 assert str(visit) == "VisitImage(Image([y=0:1024, x=0:1024], float32), ['M1'])"
246 assert repr(visit) == (
247 "VisitImage(Image(..., bbox=Box(y=Interval(start=0, stop=1024), x=Interval(start=0, stop=1024)),"
248 " dtype=dtype('float32')), mask_schema=MaskSchema([MaskPlane(name='M1', description='D1')],"
249 " dtype=dtype('uint8')))"
250 )
252 astropy_wcs = visit.astropy_wcs
253 assert isinstance(astropy_wcs, SkyProjectionAstropyView)
254 approx_wcs = visit.fits_wcs
255 assert isinstance(approx_wcs, astropy.wcs.WCS)
257 with pytest.raises(TypeError):
258 # Requires a PSF.
259 VisitImage(
260 c["image"],
261 mask_schema=c["mask_schema"],
262 sky_projection=c["sky_projection"],
263 obs_info=c["obs_info"],
264 detector=c["detector"],
265 band="r",
266 )
268 with pytest.raises(TypeError):
269 # Requires ObservationInfo.
270 VisitImage(
271 c["image"],
272 psf=c["gaussian_psf"],
273 mask_schema=c["mask_schema"],
274 sky_projection=c["sky_projection"],
275 detector=c["detector"],
276 band="r",
277 )
279 with pytest.raises(TypeError):
280 # Requires a sky_projection.
281 VisitImage(
282 c["image"],
283 psf=c["gaussian_psf"],
284 mask_schema=c["mask_schema"],
285 obs_info=c["obs_info"],
286 detector=c["detector"],
287 band="r",
288 )
290 with pytest.raises(TypeError):
291 # Requires a detector.
292 VisitImage(
293 c["image"],
294 psf=c["gaussian_psf"],
295 mask_schema=c["mask_schema"],
296 sky_projection=c["sky_projection"],
297 obs_info=c["obs_info"],
298 band="r",
299 )
301 with pytest.raises(TypeError):
302 # Requires some form of mask.
303 VisitImage(
304 c["image"],
305 psf=c["gaussian_psf"],
306 sky_projection=c["sky_projection"],
307 obs_info=c["obs_info"],
308 detector=c["detector"],
309 band="r",
310 )
312 with pytest.raises(TypeError):
313 VisitImage(
314 Image(42, shape=(5, 5)),
315 psf=c["gaussian_psf"],
316 mask_schema=c["mask_schema"],
317 sky_projection=c["sky_projection"],
318 obs_info=c["obs_info"],
319 detector=c["detector"],
320 band="r",
321 )
323 # Requires a DetectorFrame.
324 rng = np.random.default_rng(501)
325 tract_frame = TractFrame(skymap="Skymap", tract=1, bbox=Box.factory[1:10, 1:10])
326 tract_proj = make_random_sky_projection(rng, tract_frame, Box.factory[1:4096, 1:4096])
327 with pytest.raises(TypeError):
328 VisitImage(
329 c["image"],
330 sky_projection=tract_proj,
331 psf=c["gaussian_psf"],
332 mask_schema=c["mask_schema"],
333 obs_info=c["obs_info"],
334 detector=c["detector"],
335 band="r",
336 )
338 # Variance unit mismatch.
339 with pytest.raises(ValueError):
340 VisitImage(
341 c["image"],
342 variance=c["image"],
343 psf=c["gaussian_psf"],
344 mask_schema=c["mask_schema"],
345 sky_projection=c["sky_projection"],
346 obs_info=c["obs_info"],
347 detector=c["detector"],
348 band="r",
349 )
352def test_copy_and_slice(visit_image_components: dict[str, Any]) -> None:
353 """Verify that copy deep-copies arrays and components while slice shares
354 them.
355 """
356 c = visit_image_components
357 visit_image = make_visit_image(c)
358 copy = visit_image.copy()
359 copy.image.array[0, 0] = 30.0
360 assert visit_image.image.array[0, 0] == 42.0
361 assert copy.image.array[0, 0] == 30.0
362 subvisit = visit_image[Box.factory[0:5, 0:5]]
363 # Check summary stats.
364 assert copy.summary_stats == visit_image.summary_stats
365 assert copy.summary_stats is not visit_image.summary_stats
366 assert subvisit.summary_stats == visit_image.summary_stats
367 assert subvisit.summary_stats is visit_image.summary_stats
368 # Check aperture corrections.
369 assert copy.aperture_corrections.keys() == visit_image.aperture_corrections.keys()
370 assert copy.aperture_corrections is not visit_image.aperture_corrections
371 assert subvisit.aperture_corrections.keys() == visit_image.aperture_corrections.keys()
372 assert subvisit.aperture_corrections is visit_image.aperture_corrections
373 # Check backgrounds.
374 assert copy.backgrounds.keys() == visit_image.backgrounds.keys()
375 assert copy.backgrounds is not visit_image.backgrounds
376 assert subvisit.backgrounds.keys() == visit_image.backgrounds.keys()
377 assert subvisit.backgrounds is visit_image.backgrounds
378 # Check bounds.
379 assert copy.bounds is c["polygon"]
380 assert subvisit.bounds == subvisit.bbox # original polygon wholly encloses subvisit.bbox
383def test_obs_info(visit_image_components: dict[str, Any]) -> None:
384 """Verify that ObservationInfo is present and carries the expected
385 instrument.
386 """
387 visit_image = make_visit_image(visit_image_components)
388 assert visit_image.obs_info is not None
389 assert visit_image.obs_info.instrument == "LSSTCam"
392def test_summary_stats(visit_image_components: dict[str, Any]) -> None:
393 """Verify ObservationSummaryStats equality and inequality comparisons."""
394 summary_stats = visit_image_components["summary_stats"]
395 assert summary_stats == ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
396 assert summary_stats != ObservationSummaryStats(psfSigma=2.5)
397 assert summary_stats != ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
400def test_summary_stats_to_legacy() -> None:
401 """Verify ObservationSummaryStats round-trips through the legacy
402 ExposureSummaryStats even when this package defines fields that the
403 installed afw does not.
404 """
405 try:
406 from lsst.afw.image import ExposureSummaryStats
407 except ImportError:
408 pytest.skip("lsst.afw.image is not available")
410 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
411 legacy = summary_stats.to_legacy()
412 assert isinstance(legacy, ExposureSummaryStats)
413 assert legacy.psfSigma == 2.5
414 assert legacy.zeroPoint == 31.4
415 # Empty (NaN) fields unknown to the legacy struct are dropped, so the
416 # round trip reproduces the original.
417 assert ObservationSummaryStats.from_legacy(legacy) == summary_stats
419 # A real value in a field unknown to the legacy struct cannot be
420 # represented and must raise rather than be silently dropped.
421 legacy_fields = {field.name for field in dataclasses.fields(ExposureSummaryStats)}
422 extra_float_fields = [
423 name
424 for name, info in ObservationSummaryStats.model_fields.items()
425 if name not in legacy_fields and info.annotation is float
426 ]
427 if extra_float_fields: 427 ↛ 428line 427 didn't jump to line 428 because the condition on line 427 was never true
428 with pytest.raises(ValueError):
429 ObservationSummaryStats(**{extra_float_fields[0]: 1.0}).to_legacy()
432def test_summary_stats_from_legacy_unknown_field() -> None:
433 """Verify from_legacy drops empty unknown fields but errors on set ones."""
435 @dataclasses.dataclass
436 class FakeLegacy:
437 psfSigma: float = 2.5
438 notARealField: float = math.nan
440 # An unknown field that is empty is dropped.
441 stats = ObservationSummaryStats.from_legacy(FakeLegacy())
442 assert stats.psfSigma == 2.5
444 # An unknown field that holds a real value cannot be represented.
445 with pytest.raises(ValueError):
446 ObservationSummaryStats.from_legacy(FakeLegacy(notARealField=1.0))
449@skip_no_h5py
450def test_round_trip_ndf(visit_image_components: dict[str, Any]) -> None:
451 """Verify NDF round-trip produces a VisitImage equal to the original."""
452 visit_image = make_visit_image(visit_image_components)
453 with RoundtripNdf(visit_image, "VisitImage") as roundtrip:
454 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False)
457@skip_no_h5py
458def test_fits_ndf_consistency(visit_image_components: dict[str, Any]) -> None:
459 """Verify FITS and NDF backends produce equal VisitImages on round-trip."""
460 visit_image = make_visit_image(visit_image_components)
461 with RoundtripFits(visit_image) as fits_rt, RoundtripNdf(visit_image) as ndf_rt:
462 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False)
463 assert_visit_images_equal(visit_image, ndf_rt.result, expect_view=False)
464 assert_visit_images_equal(fits_rt.result, ndf_rt.result, expect_view=False)
467def test_fits_json_consistency(visit_image_components: dict[str, Any]) -> None:
468 """Verify FITS and JSON backends produce equal VisitImages."""
469 visit_image = make_visit_image(visit_image_components)
470 with (
471 RoundtripFits(visit_image) as fits_rt,
472 RoundtripJson(visit_image) as json_rt,
473 ):
474 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False)
475 assert_visit_images_equal(visit_image, json_rt.result, expect_view=False)
476 assert_visit_images_equal(fits_rt.result, json_rt.result, expect_view=False)
479def test_read_write(visit_image_components: dict[str, Any]) -> None:
480 """Verify a VisitImage round-trips through FITS with correct compression.
482 Checks compression headers, subimage reads, equality, and opaque
483 metadata. Contains only butler-free assertions; component reads live
484 in `test_read_write_components`.
485 """
486 visit_image = make_visit_image(visit_image_components)
487 with RoundtripFits(visit_image, "VisitImage") as roundtrip:
488 # Check that we're still using the right compression, and that we
489 # wrote WCSs.
490 fits = roundtrip.inspect()
491 assert fits[1].header["ZCMPTYPE"] == "GZIP_2"
492 assert fits[1].header["CTYPE1"] == "RA---TAN"
493 assert fits[2].header["ZCMPTYPE"] == "GZIP_2"
494 assert fits[2].header["CTYPE1"] == "RA---TAN"
495 assert fits[3].header["ZCMPTYPE"] == "GZIP_2"
496 assert fits[3].header["CTYPE1"] == "RA---TAN"
497 # Check a subimage read (no component arg — does not trigger a skip).
498 subbox = Box.factory[8:13, 9:30]
499 subimage = roundtrip.get(bbox=subbox)
500 assert_masked_images_equal(subimage, visit_image[subbox], expect_view=False)
502 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False)
503 # Check that the round-tripped headers are the same (up to card order).
504 assert len(roundtrip.result._opaque_metadata.headers[ExtensionKey()]) == 1
505 assert dict(visit_image._opaque_metadata.headers[ExtensionKey()]) == dict(
506 roundtrip.result._opaque_metadata.headers[ExtensionKey()]
507 )
508 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")]
509 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")]
510 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")]
511 # Spot-check the concrete background contents (names, field types,
512 # subtracted entry) against the known fixture, so the equality check
513 # above is not vacuously satisfied by empty background maps.
514 assert isinstance(roundtrip.result.backgrounds, BackgroundMap)
515 assert roundtrip.result.backgrounds.keys() == {"standard"}
516 assert isinstance(roundtrip.result.backgrounds["standard"].field, ChebyshevField)
517 assert roundtrip.result.backgrounds.subtracted.name == "standard"
518 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image."
521def test_read_write_components(visit_image_components: dict[str, Any]) -> None:
522 """Verify component reads and storage-class overrides round-trip correctly.
524 Requires a butler; skips when `lsst.daf.butler` is absent.
525 Butler-free assertions live in `test_read_write`.
526 """
527 c = visit_image_components
528 visit_image = make_visit_image(c)
529 with RoundtripFits(visit_image, "VisitImage") as roundtrip:
530 subbox = Box.factory[8:13, 9:30]
531 subimage = roundtrip.get(bbox=subbox)
533 # Get an explicit masked image to compare with the subimage.
534 subimage_masked = roundtrip.get("masked_image", bbox=subbox)
535 assert_masked_images_equal(subimage_masked, subimage, expect_view=False)
537 # Get the same masked image in a multi-component get and ensure
538 # it is the same thing.
539 components = roundtrip.get("components", components=["masked_image", "psf"], bbox=subbox)
540 assert set(components) == {"masked_image", "psf"}
541 assert_masked_images_equal(components["masked_image"], subimage_masked, expect_view=False)
543 assert roundtrip.get("bbox") == visit_image.bbox
545 obs_info = roundtrip.get("obs_info")
546 assert isinstance(obs_info, ObservationInfo)
547 assert obs_info == visit_image.obs_info
549 summary_stats = roundtrip.get("summary_stats")
550 assert isinstance(summary_stats, ObservationSummaryStats)
551 assert summary_stats == visit_image.summary_stats
553 psf = roundtrip.get("psf")
554 assert isinstance(psf, GaussianPointSpreadFunction)
555 assert psf.kernel_bbox == c["gaussian_psf"].kernel_bbox
557 backgrounds = roundtrip.get("backgrounds")
558 assert isinstance(backgrounds, BackgroundMap)
559 assert backgrounds.keys() == {"standard"}
560 assert isinstance(backgrounds["standard"].field, ChebyshevField)
561 assert backgrounds.subtracted.name == "standard"
562 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image."
564 # Test some components get edge cases.
565 components = roundtrip.get("components", components="image")
566 assert isinstance(components["image"], Image)
568 components = roundtrip.get("components")
569 assert set(components) == {
570 "image",
571 "variance",
572 "psf",
573 "bbox",
574 "mask",
575 "obs_info",
576 "backgrounds",
577 "detector",
578 "aperture_corrections",
579 "sky_projection",
580 "summary_stats",
581 "photometric_scaling",
582 }
584 # Butler morphs RuntimeError to ValueError.
585 with pytest.raises(ValueError):
586 roundtrip.get("components", components=["image", "nonexistent"])
588 with pytest.raises(ValueError):
589 roundtrip.get("components", components=["image", "components"])
591 with pytest.raises(ValueError):
592 roundtrip.get("components", components=[])
594 with pytest.raises(ValueError):
595 # PSF does not know how to use bbox so this fails.
596 roundtrip.get("components", components="psf", bbox=subbox)
599def test_sum_background_round_trip_fits(visit_image_components: dict[str, Any]) -> None:
600 """Verify FITS backend keeps two same-named SumField operands as distinct
601 EXTVERs.
602 """
603 visit_image = make_visit_image(visit_image_components)
604 visit = _make_sum_background_visit_image(visit_image_components, visit_image)
605 with RoundtripFits(visit) as roundtrip:
606 _check_sum_background_round_trip(roundtrip.result, visit)
609@skip_no_h5py
610def test_sum_background_round_trip_ndf(visit_image_components: dict[str, Any]) -> None:
611 """Verify NDF backend disambiguates the repeated ``data`` leaf, just as
612 the FITS backend does.
613 """
614 visit_image = make_visit_image(visit_image_components)
615 visit = _make_sum_background_visit_image(visit_image_components, visit_image)
616 with RoundtripNdf(visit) as roundtrip:
617 _check_sum_background_round_trip(roundtrip.result, visit)
620@pytest.mark.parametrize(
621 "scaling_unit,operation",
622 [(u.electron / u.nJy, "multiply"), (u.nJy / u.electron, "divide")],
623 ids=["multiply", "divide"],
624)
625def test_convert_unit_subimage(
626 visit_image_components: dict[str, Any],
627 scaling_unit: u.UnitBase,
628 operation: Literal["multiply", "divide"],
629) -> None:
630 """Verify that converting the units of a subimage applies only the portion
631 of the photometric scaling that overlaps the subimage.
633 A photometric scaling keeps the bounds it was modeled over when the image
634 is subset, so both branches of the conversion must render it over the
635 subimage's bbox rather than over its own bounds.
636 """
637 visit_image = make_visit_image(visit_image_components)
638 scaling = ChebyshevField(
639 visit_image.bbox,
640 np.array([[4.0, 0.5, 0.125], [0.25, 0.0625, 0.0], [0.03125, 0.0, 0.0]]),
641 unit=scaling_unit,
642 )
643 visit_image.photometric_scaling = scaling
644 # Trim a different number of pixels from each side so a scaling rendered
645 # over the wrong box cannot match by chance.
646 subbox = Box.factory[
647 visit_image.bbox.y.start + 7 : visit_image.bbox.y.stop - 13,
648 visit_image.bbox.x.start + 3 : visit_image.bbox.x.stop - 21,
649 ]
650 subimage = visit_image[subbox]
651 assert subimage.photometric_scaling.bounds.bbox == visit_image.bbox
653 converted = subimage.convert_unit(u.electron)
655 assert converted.unit == u.electron
656 assert converted.image.bbox == subbox
657 scaling_array = scaling.render(subbox, dtype=subimage.image.array.dtype).array
658 if operation == "divide":
659 scaling_array = 1.0 / scaling_array
660 assert_close(converted.image.array, subimage.image.array * scaling_array)
661 assert_close(converted.variance.array, subimage.variance.array * scaling_array**2)
664@dataclasses.dataclass
665class _LegacyTestData:
666 filename: str
667 plane_map: dict[str, MaskPlane] = dataclasses.field(default_factory=get_legacy_visit_image_mask_planes)
668 unit: u.Unit = u.nJy
669 storage_class: str = "VisitImage"
670 read_cls: type[VisitImage] = VisitImage
671 legacy_exposure: LegacyExposure = dataclasses.field(init=False)
673 @classmethod
674 def get(
675 cls, which: Literal["visit_image", "preliminary_visit_image", "difference_image"]
676 ) -> _LegacyTestData:
677 if EXTERNAL_DATA_DIR is None: 677 ↛ 679line 677 didn't jump to line 679 because the condition on line 677 was always true
678 pytest.skip("TESTDATA_IMAGES is not set up.")
679 result = cls(
680 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", f"{which}.fits"),
681 )
682 match which:
683 case "preliminary_visit_image":
684 result.unit = u.electron
685 case "difference_image":
686 result.storage_class = "DifferenceImage"
687 result.read_cls = DifferenceImage
688 result.plane_map = get_legacy_difference_image_mask_planes()
689 case "visit_image":
690 pass
691 try:
692 from lsst.afw.image import ExposureFitsReader
694 result.legacy_exposure = ExposureFitsReader(result.filename).read()
695 except ImportError:
696 pytest.skip("lsst.afw.image is not available; cannot read legacy exposures")
697 result.visit_image = result.read_cls.read_legacy(
698 result.filename, preserve_quantization=True, plane_map=result.plane_map
699 )
700 return result
703@pytest.fixture(scope="session", params=["visit_image", "preliminary_visit_image", "difference_image"])
704def legacy_test_data(request: pytest.FixtureRequest) -> _LegacyTestData:
705 """Return legacy test data.
707 Tests that depend on this parameterized fixture run on all of the legacy
708 test images.
709 """
710 return _LegacyTestData.get(request.param)
713@pytest.fixture(scope="session", params=["visit_image", "difference_image"])
714def legacy_test_data_calibrated(request: pytest.FixtureRequest) -> _LegacyTestData:
715 """Return legacy test data for calibrated images only.
717 Tests that depend on this parameterized fixture do not run on
718 preliminary_visit_image, since that has 'electron' pixel units
719 """
720 return _LegacyTestData.get(request.param)
723def _check_legacy_obs_info(obs_info: ObservationInfo | None) -> None:
724 """Assert obs_info carries expected LSSTCam/DP2 field values."""
725 assert isinstance(obs_info, ObservationInfo)
726 assert obs_info.instrument == "LSSTCam"
727 assert obs_info.detector_num == 85, obs_info
728 assert obs_info.detector_unique_name == "R21_S11", obs_info
729 assert obs_info.physical_filter == "r_57", obs_info
732def test_legacy_errors(legacy_test_data: _LegacyTestData) -> None:
733 """Verify that from_legacy and read_legacy raise ValueError on
734 conflicting arguments.
735 """
736 with pytest.raises(ValueError):
737 VisitImage.from_legacy(legacy_test_data.legacy_exposure, instrument="HSC")
738 with pytest.raises(ValueError):
739 VisitImage.from_legacy(legacy_test_data.legacy_exposure, visit=123456)
740 with pytest.raises(ValueError):
741 VisitImage.from_legacy(legacy_test_data.legacy_exposure, unit=u.mJy)
742 visit = VisitImage.from_legacy(
743 legacy_test_data.legacy_exposure,
744 instrument="LSSTCam",
745 unit=legacy_test_data.unit,
746 visit=2025052000177,
747 )
748 assert visit.unit == legacy_test_data.unit
750 with pytest.raises(ValueError):
751 legacy_test_data.read_cls.read_legacy(legacy_test_data.filename, instrument="HSC")
752 with pytest.raises(ValueError):
753 legacy_test_data.read_cls.read_legacy(legacy_test_data.filename, visit=123456)
756def test_component_reads(legacy_test_data: _LegacyTestData) -> None:
757 """Verify that individual components can be read from a legacy FITS
758 file.
759 """
760 visit = VisitImage.read_legacy(legacy_test_data.filename)
761 proj = VisitImage.read_legacy(legacy_test_data.filename, component="sky_projection")
762 assert_sky_projections_equal(proj, visit.sky_projection, expect_identity=False)
763 image = VisitImage.read_legacy(legacy_test_data.filename, component="image")
764 assert image == visit.image
765 assert_sky_projections_equal(proj, image.sky_projection, expect_identity=False)
766 variance = VisitImage.read_legacy(legacy_test_data.filename, component="variance")
767 assert variance == visit.variance
768 assert_sky_projections_equal(proj, variance.sky_projection, expect_identity=False)
769 mask = VisitImage.read_legacy(legacy_test_data.filename, component="mask")
770 assert mask == visit.mask
771 assert_sky_projections_equal(proj, mask.sky_projection, expect_identity=False)
772 psf = VisitImage.read_legacy(legacy_test_data.filename, component="psf")
773 assert isinstance(psf, PointSpreadFunction)
774 obs_info = VisitImage.read_legacy(legacy_test_data.filename, component="obs_info")
775 _check_legacy_obs_info(obs_info)
776 summary_stats = VisitImage.read_legacy(legacy_test_data.filename, component="summary_stats")
777 assert isinstance(summary_stats, ObservationSummaryStats)
778 assert summary_stats.nPsfStar == legacy_test_data.legacy_exposure.info.getSummaryStats().nPsfStar
779 compare_aperture_corrections_to_legacy(
780 VisitImage.read_legacy(legacy_test_data.filename, component="aperture_corrections"),
781 legacy_test_data.legacy_exposure.info.getApCorrMap(),
782 visit.bbox,
783 )
784 detector = VisitImage.read_legacy(legacy_test_data.filename, component="detector")
785 compare_detector_to_legacy(
786 detector, legacy_test_data.legacy_exposure.getDetector(), is_raw_assembled=True
787 )
788 photometric_scaling = VisitImage.read_legacy(legacy_test_data.filename, component="photometric_scaling")
789 compare_photo_calib_to_legacy(
790 photometric_scaling,
791 legacy_test_data.legacy_exposure.getPhotoCalib(),
792 subimage_bbox=visit.bbox,
793 )
796def test_legacy_obs_info(legacy_test_data: _LegacyTestData) -> None:
797 """Verify that ObservationInfo is constructed correctly from a legacy
798 exposure.
799 """
800 legacy = VisitImage.from_legacy(legacy_test_data.legacy_exposure, plane_map=legacy_test_data.plane_map)
801 assert legacy.obs_info is not None
802 assert legacy.obs_info == legacy_test_data.visit_image.obs_info
803 assert legacy.obs_info is not None # for mypy
804 assert legacy.obs_info.instrument == "LSSTCam"
805 assert legacy.obs_info.detector_num == 85, legacy.obs_info
806 assert legacy.obs_info.detector_unique_name == "R21_S11", legacy.obs_info
807 assert legacy.obs_info.physical_filter == "r_57", legacy.obs_info
810def test_aperture_corrections_to_legacy(legacy_test_data: _LegacyTestData) -> None:
811 """Verify that aperture corrections round-trip through a legacy
812 ApCorrMap.
813 """
814 ap_corrections = legacy_test_data.visit_image.aperture_corrections
815 legacy_ap_corr_map = aperture_corrections_to_legacy(ap_corrections)
816 compare_aperture_corrections_to_legacy(
817 ap_corrections,
818 legacy_ap_corr_map,
819 legacy_test_data.visit_image.bbox,
820 )
823def _check_legacy_headers(visit_image: VisitImage) -> None:
824 """Assert that primary and extension headers are stripped correctly."""
825 header = visit_image._opaque_metadata.headers[ExtensionKey()]
826 assert "EXPTIME" in header
827 assert header["PLATFORM"] == "lsstcam"
828 assert "LSST BUTLER ID" not in header
829 assert "AR HDU" not in header
830 assert "A_ORDER" not in header
831 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("IMAGE"), astropy.io.fits.Header())
832 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("MASK"), astropy.io.fits.Header())
833 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("VARIANCE"), astropy.io.fits.Header())
836def test_read_legacy_headers(legacy_test_data: _LegacyTestData) -> None:
837 """Verify that headers were stripped and interpreted correctly in
838 read_legacy.
839 """
840 assert legacy_test_data.visit_image.unit == legacy_test_data.unit
841 _check_legacy_headers(legacy_test_data.visit_image)
844def test_from_legacy_headers(legacy_test_data: _LegacyTestData) -> None:
845 """Verify that from_legacy handles primary and extension headers
846 correctly.
847 """
848 legacy = VisitImage.from_legacy(legacy_test_data.legacy_exposure, plane_map=legacy_test_data.plane_map)
849 assert legacy.unit == legacy_test_data.unit
850 _check_legacy_headers(legacy)
853def test_rewrite(legacy_test_data: _LegacyTestData) -> None:
854 """Verify that a legacy VisitImage can be rewritten and round-trips both
855 pixel values and all components.
856 """
857 with RoundtripFits(legacy_test_data.visit_image, legacy_test_data.storage_class) as roundtrip:
858 fits = roundtrip.inspect()
859 assert fits[1].header["ZCMPTYPE"] == "RICE_1"
860 assert fits[1].header["CTYPE1"] == "RA---TAN-SIP"
861 assert fits[2].header["ZCMPTYPE"] == "GZIP_2"
862 assert fits[2].header["CTYPE1"] == "RA---TAN-SIP"
863 assert fits[3].header["ZCMPTYPE"] == "RICE_1"
864 assert fits[3].header["CTYPE1"] == "RA---TAN-SIP"
865 subbox = Box.factory[8:13, 9:30]
866 subimage = roundtrip.get(bbox=subbox)
867 assert_masked_images_equal(subimage, legacy_test_data.visit_image[subbox], expect_view=False)
868 alternates: dict[str, Any] = {}
869 assert roundtrip.get("bbox") == legacy_test_data.visit_image.bbox
870 alternates = {
871 k: roundtrip.get(k)
872 for k in [
873 "sky_projection",
874 "image",
875 "mask",
876 "variance",
877 "psf",
878 "obs_info",
879 "summary_stats",
880 "aperture_corrections",
881 "detector",
882 "photometric_scaling",
883 ]
884 }
885 legacy_exposure = roundtrip.get(storageClass="Exposure")
886 assert isinstance(legacy_exposure, LegacyExposure)
887 compare_visit_image_to_legacy(
888 legacy_test_data.visit_image,
889 legacy_exposure,
890 expect_view=False,
891 plane_map=legacy_test_data.plane_map,
892 **DP2_VISIT_DETECTOR_DATA_ID,
893 )
894 if legacy_test_data.visit_image.unit == u.nJy:
895 assert legacy_exposure.getPhotoCalib()._isConstant
896 assert legacy_exposure.getPhotoCalib().getCalibrationMean() == 1.0
897 else:
898 compare_photo_calib_to_legacy(
899 legacy_test_data.visit_image.photometric_scaling,
900 legacy_exposure.getPhotoCalib(),
901 subimage_bbox=subbox,
902 )
903 assert legacy_exposure.info.getId() == legacy_test_data.legacy_exposure.info.getId()
904 visit_info = roundtrip.get("obs_info", storageClass="VisitInfo")
905 assert isinstance(visit_info, LegacyVisitInfo)
906 assert visit_info.getInstrumentLabel() == "LSSTCam"
908 assert_visit_images_equal(roundtrip.result, legacy_test_data.visit_image, expect_view=False)
909 assert dict(legacy_test_data.visit_image._opaque_metadata.headers[ExtensionKey()]) == dict(
910 roundtrip.result._opaque_metadata.headers[ExtensionKey()]
911 )
912 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")]
913 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")]
914 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")]
915 assert roundtrip.result._opaque_metadata.headers[ExtensionKey()]["PLATFORM"] == "lsstcam"
916 compare_visit_image_to_legacy(
917 roundtrip.result,
918 legacy_test_data.legacy_exposure,
919 expect_view=False,
920 plane_map=legacy_test_data.plane_map,
921 **DP2_VISIT_DETECTOR_DATA_ID,
922 alternates=alternates,
923 )
924 compare_visit_image_to_legacy(
925 legacy_test_data.read_cls.from_legacy(
926 legacy_test_data.legacy_exposure, plane_map=legacy_test_data.plane_map
927 ),
928 legacy_test_data.legacy_exposure,
929 expect_view=True,
930 plane_map=legacy_test_data.plane_map,
931 **DP2_VISIT_DETECTOR_DATA_ID,
932 )
935def test_butler_converters(legacy_test_data: _LegacyTestData) -> None:
936 """Verify that a VisitImage can be read from a Butler dataset written as
937 an Exposure.
938 """
939 try:
940 from lsst.daf.butler import FileDataset
941 except ImportError:
942 pytest.skip("lsst.daf.butler could not be imported.")
944 with TemporaryButler(legacy="ExposureF") as helper:
945 helper.butler.ingest(
946 FileDataset(path=legacy_test_data.filename, refs=[helper.legacy]), transfer="symlink"
947 )
948 visit_image_ref = helper.legacy.overrideStorageClass(legacy_test_data.storage_class)
949 with warnings.catch_warnings():
950 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
951 visit_image = helper.butler.get(visit_image_ref)
952 assert visit_image._opaque_metadata.precompressed.keys() == set()
953 visit_image = helper.butler.get(visit_image_ref, parameters={"preserve_quantization": True})
954 assert visit_image._opaque_metadata.precompressed.keys() == {"IMAGE", "VARIANCE"}
955 bbox = helper.butler.get(visit_image_ref.makeComponentRef("bbox"))
956 assert bbox == visit_image.bbox
957 alternates = {
958 k: helper.butler.get(visit_image_ref.makeComponentRef(k))
959 for k in ["image", "mask", "variance", "bbox", "psf", "detector"]
960 }
961 compare_visit_image_to_legacy(
962 visit_image,
963 legacy_test_data.legacy_exposure,
964 expect_view=False,
965 plane_map=legacy_test_data.plane_map,
966 alternates=alternates,
967 **DP2_VISIT_DETECTOR_DATA_ID,
968 )
969 helper.butler.pruneDatasets([helper.legacy], purge=True, unstore=True, disassociate=True)
970 visit_image.metadata["MixedCaseKey"] = 52
971 helper.butler.put(visit_image, visit_image_ref)
972 with warnings.catch_warnings():
973 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
974 legacy_exposure = helper.butler.get(helper.legacy)
975 compare_visit_image_to_legacy(
976 visit_image,
977 legacy_exposure,
978 expect_view=False,
979 plane_map=legacy_test_data.plane_map,
980 alternates=alternates,
981 **DP2_VISIT_DETECTOR_DATA_ID,
982 )
983 visit_image_2 = helper.butler.get(visit_image_ref)
984 compare_visit_image_to_legacy(
985 visit_image_2,
986 legacy_exposure,
987 expect_view=False,
988 plane_map=legacy_test_data.plane_map,
989 alternates=alternates,
990 **DP2_VISIT_DETECTOR_DATA_ID,
991 )
992 assert visit_image_2.metadata["MixedCaseKey"] == 52
995def test_convert_unit(legacy_test_data_calibrated: _LegacyTestData) -> None:
996 """Verify convert_unit round-trips between nJy, mJy, and electron via
997 photometric_scaling.
998 """
999 from lsst.afw.table import ExposureCatalog
1001 legacy_test_data = legacy_test_data_calibrated
1002 original = legacy_test_data.visit_image.copy()
1003 with pytest.raises(u.UnitConversionError):
1004 original.convert_unit(u.electron)
1005 visit_image_nJy = original.convert_unit(u.nJy, copy=False)
1006 assert np.may_share_memory(visit_image_nJy.image.array, original.image.array)
1007 assert np.may_share_memory(visit_image_nJy.variance.array, original.variance.array)
1008 with pytest.raises(u.UnitConversionError):
1009 original.convert_unit(u.mJy, copy=False)
1010 visit_image_mJy = original.convert_unit(u.mJy, copy="as-needed")
1011 assert visit_image_mJy.unit == u.mJy
1012 assert_close(visit_image_mJy.image.array, original.image.array * 1e-6)
1013 assert np.may_share_memory(visit_image_nJy.mask.array, original.mask.array)
1014 assert_close(visit_image_mJy.variance.array, original.variance.array * 1e-12)
1015 legacy_exposure_mJy = visit_image_mJy.to_legacy()
1016 assert_close(legacy_exposure_mJy.getPhotoCalib().getCalibrationMean(), 1e6)
1017 legacy_masked_image_nJy = legacy_exposure_mJy.getPhotoCalib().calibrateImage(
1018 legacy_exposure_mJy.maskedImage
1019 )
1020 assert_close(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array)
1021 assert_close(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array)
1022 assert np.may_share_memory(visit_image_mJy.mask.array, original.mask.array)
1023 assert visit_image_mJy.sky_projection is original.sky_projection
1024 assert visit_image_mJy.obs_info is original.obs_info
1025 assert visit_image_mJy.summary_stats is original.summary_stats
1026 assert visit_image_mJy.psf is original.psf
1027 assert visit_image_mJy.detector is original.detector
1028 assert visit_image_mJy.bounds is original.bounds
1029 assert visit_image_mJy.aperture_corrections is original.aperture_corrections
1030 assert visit_image_mJy.photometric_scaling is original.photometric_scaling
1031 visit_summary = ExposureCatalog.readFits(
1032 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_summary.fits")
1033 )
1034 legacy_photo_calib = visit_summary.find(DP2_VISIT_DETECTOR_DATA_ID["detector"]).getPhotoCalib()
1035 visit_image_nJy.photometric_scaling = field_from_legacy_photo_calib(
1036 legacy_photo_calib, bounds=original.detector.bbox, instrumental_unit=u.electron
1037 )
1038 compare_photo_calib_to_legacy(
1039 visit_image_nJy.photometric_scaling,
1040 legacy_test_data.legacy_exposure.getPhotoCalib(),
1041 applied_legacy_photo_calib=legacy_photo_calib,
1042 subimage_bbox=visit_image_nJy.bbox,
1043 )
1044 with pytest.raises(u.UnitConversionError):
1045 visit_image_nJy.convert_unit(u.mm)
1046 with pytest.raises(u.UnitConversionError):
1047 visit_image_nJy.convert_unit(u.electron, copy=False)
1048 legacy_masked_image_e = legacy_photo_calib.uncalibrateImage(legacy_test_data.legacy_exposure.maskedImage)
1049 visit_image_e = visit_image_nJy.convert_unit(u.electron)
1050 assert_close(visit_image_e.image.array, legacy_masked_image_e.image.array)
1051 assert_close(visit_image_e.variance.array, legacy_masked_image_e.variance.array)
1052 assert not np.may_share_memory(visit_image_e.mask.array, visit_image_nJy.mask.array)
1053 visit_image_mJy.photometric_scaling = visit_image_nJy.photometric_scaling
1054 visit_image_e = visit_image_mJy.convert_unit(u.electron)
1055 assert_close(visit_image_e.image.array, legacy_masked_image_e.image.array)
1056 assert_close(visit_image_e.variance.array, legacy_masked_image_e.variance.array)
1057 visit_image_nJy_2 = visit_image_e.convert_unit(u.nJy)
1058 assert_close(visit_image_nJy_2.image.array, visit_image_nJy.image.array)
1059 assert_close(visit_image_nJy_2.variance.array, original.variance.array)
1060 visit_image_e.photometric_scaling = visit_image_nJy.photometric_scaling * (1e-6 * u.mJy / u.nJy)
1061 visit_image_nJy_3 = visit_image_e.convert_unit(u.nJy)
1062 assert_close(visit_image_nJy_3.image.array, visit_image_nJy.image.array)
1063 assert_close(visit_image_nJy_3.variance.array, original.variance.array)
1064 legacy_exposure_e = visit_image_e.to_legacy()
1065 assert_close(
1066 legacy_exposure_e.getPhotoCalib().getCalibrationMean(),
1067 legacy_photo_calib.getCalibrationMean(),
1068 )
1069 legacy_masked_image_nJy = legacy_exposure_e.getPhotoCalib().calibrateImage(legacy_exposure_e.maskedImage)
1070 assert_close(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array)
1071 assert_close(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array)
1074def test_background_map_describe() -> None:
1075 """An empty BackgroundMap._describe reports no backgrounds inline."""
1076 bg_map = BackgroundMap()
1077 assert isinstance(bg_map, DescribableMixin)
1078 report = bg_map._describe()
1079 assert isinstance(report, Report)
1080 assert report.type_name == "BackgroundMap"
1081 assert report.inline
1082 assert report.summary == "no backgrounds"
1083 assert report.children == {}
1086def test_background_map_with_entries_describe() -> None:
1087 """BackgroundMap._describe recurses into each background model."""
1088 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1089 bg_map = BackgroundMap(
1090 [Background("sky", cheby, "Sky model."), Background("fringe", cheby)],
1091 subtracted="sky",
1092 )
1093 report = bg_map._describe()
1094 assert report.type_name == "BackgroundMap"
1095 assert report.inline
1096 # The inline summary lists every background and marks the subtracted one,
1097 # since that is all a composite holding this map will show.
1098 assert report.summary == "sky (subtracted), fringe"
1099 # Standalone, each background is a child carrying its own model's report.
1100 assert set(report.children) == {"sky", "fringe"}
1101 sky = report.children["sky"]
1102 assert sky.type_name == "ChebyshevField"
1103 sky_fields = {f.label: f.value for f in sky.fields}
1104 assert sky_fields["subtracted"] == "yes"
1105 assert sky_fields["description"] == "Sky model."
1106 # The model's own fields survive alongside the background's attributes.
1107 assert "bounds" in sky_fields
1108 # Only the subtracted one is marked, and an absent description is omitted.
1109 fringe_fields = {f.label: f.value for f in report.children["fringe"].fields}
1110 assert "subtracted" not in fringe_fields
1111 assert "description" not in fringe_fields
1114def test_background_map_describe_brief_skips_children() -> None:
1115 """A brief background map report keeps the summary but not the models."""
1116 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1117 bg_map = BackgroundMap([Background("sky", cheby)], subtracted="sky")
1118 report = bg_map._describe(DescribeOptions(brief=True))
1119 assert report.summary == "sky (subtracted)"
1120 assert report.children == {}
1123def test_visit_image_repr_str_with_unreadable_psf() -> None:
1124 """Repr and str succeed even when the PSF stored an ArchiveReadError.
1126 An unreadable component is a supported state; repr and str read only the
1127 cheap fields and summary, so they must not build the child tree (which
1128 would raise when it accesses the PSF).
1129 """
1130 path = os.path.join(LOCAL_DATA_DIR, "schema_v1", "visit_image.json")
1131 visit_image = read_archive(path)
1132 visit_image._psf = ArchiveReadError("psf unreadable")
1133 assert repr(visit_image).startswith("VisitImage(")
1134 assert str(visit_image).startswith("VisitImage(")
1137def test_observation_summary_stats_describe() -> None:
1138 """ObservationSummaryStats._describe reports the statistics that are set.
1140 Unset ones are omitted rather than shown as NaN.
1141 """
1142 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4, ra=180.0, dec=-30.0)
1143 assert isinstance(stats, DescribableMixin)
1144 report = stats._describe()
1145 assert isinstance(report, Report)
1146 assert report.type_name == "ObservationSummaryStats"
1147 # Scalars are packed into one group, ordered by name.
1148 (group,) = report.value_groups
1149 assert group.role is FieldRole.DERIVED
1150 assert [name for name, _ in group.values] == sorted(name for name, _ in group.values)
1151 values = dict(group.values)
1152 assert values["psfSigma"] == 2.5
1153 assert values["zeroPoint"] == 31.4
1154 assert values["ra"] == 180.0
1155 assert values["dec"] == -30.0
1156 # Unset statistics are omitted rather than reported as NaN, and the
1157 # serialization plumbing this class inherits never appears at all.
1158 assert "expTime" not in values
1159 assert "skyBg" not in values
1160 assert not {"metadata", "butler_info", "schema_version"} & set(values)
1163def test_observation_summary_stats_describe_omits_empty_sequences() -> None:
1164 """Sequence statistics appear only when they carry a value."""
1165 empty = ObservationSummaryStats(psfSigma=2.5)
1166 # raCorners defaults to all-NaN, which carries no more information than an
1167 # empty sequence does.
1168 assert all(math.isnan(v) for v in empty.raCorners)
1169 assert not any(f.label == "raCorners" for f in empty._describe().fields)
1171 filled = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
1172 corners = next(f for f in filled._describe().fields if f.label == "raCorners")
1173 assert corners.value == (5.2, 5.4, 5.4, 5.2)
1174 assert corners.role is FieldRole.DERIVED
1177def test_observation_summary_stats_describe_brief_counts() -> None:
1178 """A brief report gives the number set rather than listing them."""
1179 stats = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
1180 brief = stats._describe(DescribeOptions(brief=True))
1181 assert brief.type_name == "ObservationSummaryStats"
1182 assert brief.value_groups == []
1183 (field,) = brief.fields
1184 assert field.label == "statistics set"
1185 assert field.role is FieldRole.DERIVED
1187 # The count must agree with what the full report actually shows: two
1188 # scalars set by default plus psfSigma, and raCorners as a sequence.
1189 full = stats._describe()
1190 listed = len(full.fields) + sum(len(g.values) for g in full.value_groups)
1191 assert field.value.startswith(f"{listed} of ")
1192 assert brief.to_str() == full.to_str()
1193 assert f"{listed} of " in brief.to_str()
1196def test_observation_summary_stats_pydantic_repr() -> None:
1197 """ObservationSummaryStats uses pydantic's repr, not the mixin's."""
1198 stats = ObservationSummaryStats(psfSigma=2.5)
1199 r = repr(stats)
1200 assert r.startswith("ObservationSummaryStats(")
1201 assert "psfSigma=2.5" in r
1204def test_observation_summary_stats_str_is_the_report_summary() -> None:
1205 """The str output reports the count, not pydantic's field-by-field dump.
1207 The repr keeps the exhaustive form, since that is the one that
1208 round-trips.
1209 """
1210 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
1211 assert str(stats) == stats.describe().to_str()
1212 # The two set here, plus the integer counters, which default to a genuine
1213 # zero rather than to NaN.
1214 assert str(stats) == "ObservationSummaryStats(4 of 66 statistics set)"
1215 assert stats.nPsfStar == 0 and stats.nShapeletsStar == 0
1216 # The unset statistics reach repr but not str.
1217 assert "nan" not in str(stats)
1218 assert "nan" in repr(stats)
1221def test_archive_tree_repr_omits_schema_bookkeeping() -> None:
1222 """Schema version fields mirror class constants, so repr leaves them out.
1224 They are never passed on construction and say nothing the type does not.
1225 """
1226 stats = ObservationSummaryStats(psfSigma=2.5, indirect=[1])
1227 for name in ("schema_version", "min_read_version", "indirect"):
1228 assert name not in repr(stats), name
1229 # They are still real fields, and hiding them from repr does not hide them
1230 # from serialization.
1231 assert stats.schema_version == ObservationSummaryStats.SCHEMA_VERSION
1232 dumped = stats.model_dump()
1233 for name in ("schema_version", "min_read_version", "indirect"):
1234 assert name in dumped, name