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