Coverage for tests/test_visit_image.py: 71%
798 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-13 09:31 +0000
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-13 09:31 +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 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.daf.butler import DimensionUniverse
29from lsst.images import (
30 Background,
31 BackgroundMap,
32 Box,
33 DetectorFrame,
34 DifferenceImage,
35 Image,
36 MaskPlane,
37 MaskSchema,
38 ObservationSummaryStats,
39 Polygon,
40 SkyProjectionAstropyView,
41 TractFrame,
42 VisitImage,
43 get_legacy_difference_image_mask_planes,
44 get_legacy_visit_image_mask_planes,
45 json,
46)
47from lsst.images import fits as images_fits
48from lsst.images.aperture_corrections import ApertureCorrectionMap, aperture_corrections_to_legacy
49from lsst.images.cameras import Detector
50from lsst.images.describe import DescribableMixin, DescribeOptions, FieldRole, Report
51from lsst.images.fields import ChebyshevField, SplineField, SumField, field_from_legacy_photo_calib
52from lsst.images.fits import ExtensionKey, FitsOpaqueMetadata
53from lsst.images.psfs import GaussianPointSpreadFunction, PointSpreadFunction
54from lsst.images.serialization import ArchiveReadError, open_archive, read_archive
55from lsst.images.tests import (
56 DP2_VISIT_DETECTOR_DATA_ID,
57 RoundtripFits,
58 RoundtripJson,
59 RoundtripNdf,
60 TemporaryButler,
61 assert_masked_images_equal,
62 assert_sky_projections_equal,
63 assert_values_equal,
64 assert_visit_images_equal,
65 compare_aperture_corrections_to_legacy,
66 compare_detector_to_legacy,
67 compare_mask_to_legacy,
68 compare_photo_calib_to_legacy,
69 compare_visit_image_to_legacy,
70 current_fixture_path,
71 get_dp2_exposure_record,
72 make_random_sky_projection,
73 reset_afw_mask_planes, # noqa: F401
74)
76try:
77 import h5py # noqa: F401
79 HAVE_H5PY = True
80except ImportError:
81 HAVE_H5PY = False
83try:
84 from lsst.afw.image import Exposure as LegacyExposure
85 from lsst.afw.image import VisitInfo as LegacyVisitInfo
86except ImportError:
87 type LegacyExposure = Any # type: ignore[no-redef]
88 type LegacyVisitInfo = Any # type: ignore[no-redef]
90EXTERNAL_DATA_DIR = os.environ.get("TESTDATA_IMAGES_DIR", None)
91LOCAL_DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
92FIXTURE_DIR = Path(__file__).parent / "data" / "schemas"
93MINIMAL_VISIT_DATA_ID = {key: DP2_VISIT_DETECTOR_DATA_ID[key] for key in ("instrument", "visit", "detector")}
95skip_no_h5py = pytest.mark.skipif(not HAVE_H5PY, reason="h5py is not installed")
98@pytest.fixture(scope="session")
99def visit_image_components() -> dict[str, Any]:
100 """Return a dictionary of VisitImage components."""
101 rng = np.random.default_rng(500)
102 det_frame = DetectorFrame(instrument="Inst", visit=1234, detector=1, bbox=Box.factory[1:4096, 1:4096])
103 mask_schema = MaskSchema([MaskPlane("M1", "D1")])
104 # The IDs must match the DP2 exposure record that legacy round-trip tests
105 # supply, because conversion validates the VisitInfo ID against it.
106 obs_info = ObservationInfo(
107 instrument="LSSTCam",
108 detector_num=4,
109 physical_filter="r1",
110 exposure_id=2025052000177,
111 observation_id="MC_O_20250520_000177",
112 )
113 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
114 gaussian_psf = GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13])
115 aperture_corrections: ApertureCorrectionMap = {
116 "flux1": ChebyshevField(det_frame.bbox, np.array([0.75])),
117 "flux2": ChebyshevField(det_frame.bbox, np.array([0.625])),
118 }
119 detector = read_archive(os.path.join(LOCAL_DATA_DIR, "detector.json"), Detector)
120 # Real visit images have float pixels, and some operations (e.g. rendering
121 # a photometric scaling) are only defined for floating-point images.
122 image = Image(42.0, shape=(1024, 1024), unit=u.nJy, dtype=np.float32)
123 variance = Image(5.0, shape=(1024, 1024), unit=u.nJy * u.nJy, dtype=np.float32)
124 # polygon is the lower triangle of the image.
125 polygon = Polygon(x_vertices=[-0.5, 1023.5, -0.5], y_vertices=[-0.5, -0.5, 1023.5])
126 sky_projection = make_random_sky_projection(rng, det_frame, det_frame.bbox)
127 return {
128 "mask_schema": mask_schema,
129 "obs_info": obs_info,
130 "summary_stats": summary_stats,
131 "gaussian_psf": gaussian_psf,
132 "aperture_corrections": aperture_corrections,
133 "detector": detector,
134 "image": image,
135 "variance": variance,
136 "polygon": polygon,
137 "sky_projection": sky_projection,
138 }
141def make_visit_image(components: dict[str, Any]) -> VisitImage:
142 """Construct a new VisitImage with most components populated."""
143 det_frame = components["sky_projection"].pixel_frame
144 opaque = FitsOpaqueMetadata()
145 hdr = astropy.io.fits.Header()
146 with warnings.catch_warnings():
147 # Silence warnings about long keys becoming HIERARCH.
148 warnings.simplefilter("ignore", category=astropy.io.fits.verify.VerifyWarning)
149 hdr.update({"PLATFORM": "lsstcam", "LSST BUTLER ID": "123456789"})
150 opaque.extract_legacy_primary_header(hdr)
151 # API signature suggests sky_projection and obs_info can be None but
152 # they are required (unless you pass them in via the image plane).
153 vi = VisitImage(
154 components["image"],
155 variance=components["variance"],
156 psf=GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13]),
157 mask_schema=components["mask_schema"],
158 sky_projection=components["sky_projection"],
159 obs_info=components["obs_info"],
160 summary_stats=components["summary_stats"],
161 detector=components["detector"],
162 bounds=components["polygon"],
163 aperture_corrections=components["aperture_corrections"],
164 band="r",
165 )
166 vi.backgrounds.add(
167 "standard",
168 ChebyshevField(det_frame.bbox, np.array([[2.0]])),
169 description="Background subtracted from the image.",
170 is_subtracted=True,
171 )
172 vi._opaque_metadata = opaque
173 return vi
176def make_simplest_visit_image(components: dict[str, Any]) -> VisitImage:
177 """Construct a VisitImage with the minimal set of components populated."""
178 return VisitImage(
179 components["image"],
180 psf=GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13]),
181 mask_schema=components["mask_schema"],
182 sky_projection=components["sky_projection"],
183 detector=components["detector"],
184 obs_info=components["obs_info"],
185 band="r",
186 )
189def _make_sum_background_visit_image(components: dict[str, Any], visit_image: VisitImage) -> VisitImage:
190 """Return a VisitImage whose subtracted background is a SumField."""
191 rng = np.random.default_rng(42)
192 bbox = visit_image.image.sky_projection.pixel_frame.bbox
193 bin_y = bbox.y.linspace(6)
194 bin_x = bbox.x.linspace(7)
195 spline_a = SplineField(
196 bbox,
197 rng.standard_normal(size=(bin_y.size, bin_x.size)),
198 y=bin_y,
199 x=bin_x,
200 )
201 spline_b = SplineField(
202 bbox,
203 rng.standard_normal(size=(bin_y.size, bin_x.size)),
204 y=bin_y,
205 x=bin_x,
206 )
207 sum_field = SumField([spline_a, spline_b])
208 bg_map = BackgroundMap()
209 bg_map.add(
210 "stacked",
211 sum_field,
212 description="Two-operand SumField subtracted background.",
213 is_subtracted=True,
214 )
215 return VisitImage(
216 components["image"],
217 variance=components["variance"],
218 psf=components["gaussian_psf"],
219 mask_schema=components["mask_schema"],
220 sky_projection=components["sky_projection"],
221 obs_info=components["obs_info"],
222 summary_stats=components["summary_stats"],
223 detector=components["detector"],
224 band="r",
225 backgrounds=bg_map,
226 )
229def _check_sum_background_round_trip(result: VisitImage, original: VisitImage) -> None:
230 """Assert that a round-tripped SumField background matches the original."""
231 subtracted = result.backgrounds.subtracted
232 assert subtracted is not None
233 assert isinstance(subtracted.field, SumField)
234 original_subtracted = original.backgrounds.subtracted
235 assert original_subtracted is not None
236 original_field = original_subtracted.field
237 assert isinstance(original_field, SumField)
238 round_field = subtracted.field
239 assert isinstance(round_field, SumField)
240 assert len(round_field.operands) == len(original_field.operands)
241 for round_op, orig_op in zip(round_field.operands, original_field.operands, strict=True):
242 assert round_op == orig_op
245def test_visit_image_repr_str_pinned(visit_image_components: dict[str, Any]) -> None:
246 """Pin the exact str and repr output of a VisitImage."""
247 visit = make_simplest_visit_image(visit_image_components)
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 )
256def test_basics(visit_image_components: dict[str, Any]) -> None:
257 """Verify VisitImage constructor patterns and required-argument checks."""
258 c = visit_image_components
259 # Test default fill of variance.
260 visit = make_simplest_visit_image(c)
261 assert visit.variance.array[0, 0] == 1.0
262 assert visit[...] is not visit
263 assert str(visit) == "VisitImage(Image([y=0:1024, x=0:1024], float32), ['M1'])"
264 assert repr(visit) == (
265 "VisitImage(Image(..., bbox=Box(y=Interval(start=0, stop=1024), x=Interval(start=0, stop=1024)),"
266 " dtype=dtype('float32')), mask_schema=MaskSchema([MaskPlane(name='M1', description='D1')],"
267 " dtype=dtype('uint8')))"
268 )
270 astropy_wcs = visit.astropy_wcs
271 assert isinstance(astropy_wcs, SkyProjectionAstropyView)
272 approx_wcs = visit.fits_wcs
273 assert isinstance(approx_wcs, astropy.wcs.WCS)
275 with pytest.raises(TypeError):
276 # Requires a PSF.
277 VisitImage(
278 c["image"],
279 mask_schema=c["mask_schema"],
280 sky_projection=c["sky_projection"],
281 obs_info=c["obs_info"],
282 detector=c["detector"],
283 band="r",
284 )
286 with pytest.raises(TypeError):
287 # Requires ObservationInfo.
288 VisitImage(
289 c["image"],
290 psf=c["gaussian_psf"],
291 mask_schema=c["mask_schema"],
292 sky_projection=c["sky_projection"],
293 detector=c["detector"],
294 band="r",
295 )
297 with pytest.raises(TypeError):
298 # Requires a sky_projection.
299 VisitImage(
300 c["image"],
301 psf=c["gaussian_psf"],
302 mask_schema=c["mask_schema"],
303 obs_info=c["obs_info"],
304 detector=c["detector"],
305 band="r",
306 )
308 with pytest.raises(TypeError):
309 # Requires a detector.
310 VisitImage(
311 c["image"],
312 psf=c["gaussian_psf"],
313 mask_schema=c["mask_schema"],
314 sky_projection=c["sky_projection"],
315 obs_info=c["obs_info"],
316 band="r",
317 )
319 with pytest.raises(TypeError):
320 # Requires some form of mask.
321 VisitImage(
322 c["image"],
323 psf=c["gaussian_psf"],
324 sky_projection=c["sky_projection"],
325 obs_info=c["obs_info"],
326 detector=c["detector"],
327 band="r",
328 )
330 with pytest.raises(TypeError):
331 VisitImage(
332 Image(42, shape=(5, 5)),
333 psf=c["gaussian_psf"],
334 mask_schema=c["mask_schema"],
335 sky_projection=c["sky_projection"],
336 obs_info=c["obs_info"],
337 detector=c["detector"],
338 band="r",
339 )
341 # Requires a DetectorFrame.
342 rng = np.random.default_rng(501)
343 tract_frame = TractFrame(skymap="Skymap", tract=1, bbox=Box.factory[1:10, 1:10])
344 tract_proj = make_random_sky_projection(rng, tract_frame, Box.factory[1:4096, 1:4096])
345 with pytest.raises(TypeError):
346 VisitImage(
347 c["image"],
348 sky_projection=tract_proj,
349 psf=c["gaussian_psf"],
350 mask_schema=c["mask_schema"],
351 obs_info=c["obs_info"],
352 detector=c["detector"],
353 band="r",
354 )
356 # Variance unit mismatch.
357 with pytest.raises(ValueError, match="should be the square of the image unit"):
358 VisitImage(
359 c["image"],
360 variance=c["image"],
361 psf=c["gaussian_psf"],
362 mask_schema=c["mask_schema"],
363 sky_projection=c["sky_projection"],
364 obs_info=c["obs_info"],
365 detector=c["detector"],
366 band="r",
367 )
370def test_copy_and_slice(visit_image_components: dict[str, Any]) -> None:
371 """Verify that copy deep-copies arrays and components while slice shares
372 them.
373 """
374 c = visit_image_components
375 visit_image = make_visit_image(c)
376 copy = visit_image.copy()
377 copy.image.array[0, 0] = 30.0
378 assert visit_image.image.array[0, 0] == 42.0
379 assert copy.image.array[0, 0] == 30.0
380 subvisit = visit_image[Box.factory[0:5, 0:5]]
381 # Check summary stats.
382 assert copy.summary_stats == visit_image.summary_stats
383 assert copy.summary_stats is not visit_image.summary_stats
384 assert subvisit.summary_stats == visit_image.summary_stats
385 assert subvisit.summary_stats is visit_image.summary_stats
386 # Check aperture corrections.
387 assert copy.aperture_corrections.keys() == visit_image.aperture_corrections.keys()
388 assert copy.aperture_corrections is not visit_image.aperture_corrections
389 assert subvisit.aperture_corrections.keys() == visit_image.aperture_corrections.keys()
390 assert subvisit.aperture_corrections is visit_image.aperture_corrections
391 # Check backgrounds.
392 assert copy.backgrounds.keys() == visit_image.backgrounds.keys()
393 assert copy.backgrounds is not visit_image.backgrounds
394 assert subvisit.backgrounds.keys() == visit_image.backgrounds.keys()
395 assert subvisit.backgrounds is visit_image.backgrounds
396 # Check bounds.
397 assert copy.bounds is c["polygon"]
398 assert subvisit.bounds == subvisit.bbox # original polygon wholly encloses subvisit.bbox
401def test_obs_info(visit_image_components: dict[str, Any]) -> None:
402 """Verify that ObservationInfo is present and carries the expected
403 instrument.
404 """
405 visit_image = make_visit_image(visit_image_components)
406 assert visit_image.obs_info is not None
407 assert visit_image.obs_info.instrument == "LSSTCam"
410def test_summary_stats(visit_image_components: dict[str, Any]) -> None:
411 """Verify ObservationSummaryStats equality and inequality comparisons."""
412 summary_stats = visit_image_components["summary_stats"]
413 assert summary_stats == ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
414 assert summary_stats != ObservationSummaryStats(psfSigma=2.5)
415 assert summary_stats != ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
418def test_summary_stats_to_legacy() -> None:
419 """Verify ObservationSummaryStats round-trips through the legacy
420 ExposureSummaryStats even when this package defines fields that the
421 installed afw does not.
422 """
423 try:
424 from lsst.afw.image import ExposureSummaryStats
425 except ImportError:
426 pytest.skip("lsst.afw.image is not available")
428 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
429 legacy = summary_stats.to_legacy()
430 assert isinstance(legacy, ExposureSummaryStats)
431 assert legacy.psfSigma == 2.5
432 assert legacy.zeroPoint == 31.4
433 # Empty (NaN) fields unknown to the legacy struct are dropped, so the
434 # round trip reproduces the original.
435 assert ObservationSummaryStats.from_legacy(legacy) == summary_stats
437 # A real value in a field unknown to the legacy struct cannot be
438 # represented and must raise rather than be silently dropped.
439 legacy_fields = {field.name for field in dataclasses.fields(ExposureSummaryStats)}
440 extra_float_fields = [
441 name
442 for name, info in ObservationSummaryStats.model_fields.items()
443 if name not in legacy_fields and info.annotation is float
444 ]
445 if extra_float_fields: 445 ↛ 446line 445 didn't jump to line 446 because the condition on line 445 was never true
446 with pytest.raises(ValueError, match="not supported by this version"):
447 ObservationSummaryStats(**{extra_float_fields[0]: 1.0}).to_legacy()
450def test_summary_stats_from_legacy_unknown_field() -> None:
451 """Verify from_legacy drops empty unknown fields but errors on set ones."""
453 @dataclasses.dataclass
454 class FakeLegacy:
455 psfSigma: float = 2.5
456 notARealField: float = math.nan
458 # An unknown field that is empty is dropped.
459 stats = ObservationSummaryStats.from_legacy(FakeLegacy())
460 assert stats.psfSigma == 2.5
462 # An unknown field that holds a real value cannot be represented.
463 with pytest.raises(ValueError, match="is not known to ObservationSummaryStats"):
464 ObservationSummaryStats.from_legacy(FakeLegacy(notARealField=1.0))
467def test_summary_stats_legacy_butler_read(reset_afw_mask_planes: None) -> None: # noqa: F811
468 """Test that a round-tripped ObservationSummaryStats can be read back as
469 a legacy afw ExposureSummaryStats via the converter registered on the
470 ExposureSummaryStats storage class.
471 """
472 from lsst.afw.image import ExposureSummaryStats as LegacyExposureSummaryStats
474 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
475 with RoundtripFits(summary_stats, storage_class="ObservationSummaryStats") as roundtrip:
476 legacy_stats = roundtrip.get(storageClass="ExposureSummaryStats")
477 assert isinstance(legacy_stats, LegacyExposureSummaryStats)
478 assert legacy_stats.psfSigma == 2.5
479 assert legacy_stats.zeroPoint == 31.4
480 assert ObservationSummaryStats.from_legacy(legacy_stats) == summary_stats
483def test_detector_legacy_butler_read(reset_afw_mask_planes: None) -> None: # noqa: F811
484 """Test that a round-tripped Detector (written under the DetectorV2
485 storage class) can be read back as a legacy lsst.afw.cameraGeom.Detector
486 via the converter registered on the Detector storage class.
487 """
488 from lsst.afw.cameraGeom import Detector as LegacyDetector
490 detector = read_archive(os.path.join(LOCAL_DATA_DIR, "detector.json"), Detector)
491 with RoundtripFits(detector, storage_class="DetectorV2") as roundtrip:
492 legacy_detector = roundtrip.get(storageClass="Detector")
493 assert isinstance(legacy_detector, LegacyDetector)
494 compare_detector_to_legacy(detector, legacy_detector, is_raw_assembled=True)
497def test_repeated_metadata_keys_legacy_round_trip(
498 visit_image_components: dict[str, Any],
499 reset_afw_mask_planes: None, # noqa: F811
500) -> None:
501 """Verify that a header key set more than once keeps all of its values
502 when a VisitImage is converted to a legacy Exposure and back.
503 """
504 from lsst.afw.detection import GaussianPsf
506 opaque_metadata = FitsOpaqueMetadata()
507 header = astropy.io.fits.Header()
508 header.append(("PLATFORM", "lsstcam"), end=True)
509 # SubtractBackgroundTask writes one BGMEAN card per background fit.
510 header.append(("BGMEAN", 1.5), end=True)
511 header.append(("BGMEAN", 2.5), end=True)
512 opaque_metadata.extract_legacy_primary_header(header)
513 visit_image = VisitImage(
514 visit_image_components["image"],
515 variance=visit_image_components["variance"],
516 # A legacy-backed PSF, so that to_legacy can attach it to the
517 # Exposure and from_legacy can read it back.
518 psf=PointSpreadFunction.from_legacy(GaussianPsf(33, 33, 2.5), bounds=Box.factory[0:1024, 0:1024]),
519 mask_schema=visit_image_components["mask_schema"],
520 sky_projection=visit_image_components["sky_projection"],
521 detector=visit_image_components["detector"],
522 obs_info=visit_image_components["obs_info"],
523 band="r",
524 )
525 visit_image._opaque_metadata = opaque_metadata
527 legacy_exposure = visit_image.to_legacy()
528 legacy_metadata = legacy_exposure.getMetadata()
529 assert legacy_metadata.getArray("BGMEAN") == [1.5, 2.5]
530 assert legacy_metadata["PLATFORM"] == "lsstcam"
531 # Add a random key directly
532 legacy_metadata.add("DUMMYVAR", 0.5)
533 legacy_metadata.add("DUMMYVAR", 1.5)
535 round_tripped = VisitImage.from_legacy(
536 legacy_exposure,
537 exposure_record=get_dp2_exposure_record(DimensionUniverse()),
538 )
539 round_tripped_header = round_tripped._opaque_metadata.headers[ExtensionKey()]
540 assert "DUMMYVAR" in round_tripped_header
541 assert "BGMEAN" in round_tripped_header
542 assert round_tripped_header["PLATFORM"] == "lsstcam"
543 assert [card.value for card in round_tripped_header.cards if card.keyword == "BGMEAN"] == [1.5, 2.5]
544 assert [card.value for card in round_tripped_header.cards if card.keyword == "DUMMYVAR"] == [0.5, 1.5]
547def test_external_metadata_legacy_round_trip(
548 visit_image_components: dict[str, Any],
549 reset_afw_mask_planes: None, # noqa: F811
550 tmp_path: Path,
551) -> None:
552 """Verify that native metadata is written to a legacy file as
553 ``LSST IMAGES`` cards and is not visible as external metadata when the
554 file is read back.
555 """
556 from lsst.afw.detection import GaussianPsf
558 opaque_metadata = FitsOpaqueMetadata()
559 header = astropy.io.fits.Header()
560 header.append(("PLATFORM", "lsstcam"), end=True)
561 header.append(("BGMEAN", 1.5), end=True)
562 header.append(("BGMEAN", 2.5), end=True)
563 # An external ID card must neither block nor replace the native "id".
564 header.append(("ID", 99), end=True)
565 opaque_metadata.extract_legacy_primary_header(header)
566 visit_image = VisitImage(
567 visit_image_components["image"],
568 variance=visit_image_components["variance"],
569 psf=PointSpreadFunction.from_legacy(GaussianPsf(33, 33, 2.5), bounds=Box.factory[0:1024, 0:1024]),
570 mask_schema=visit_image_components["mask_schema"],
571 sky_projection=visit_image_components["sky_projection"],
572 detector=visit_image_components["detector"],
573 obs_info=visit_image_components["obs_info"],
574 band="r",
575 metadata={"native_key": 7, "MixedCase": "yes"},
576 )
577 visit_image._opaque_metadata = opaque_metadata
578 path = tmp_path / "legacy.fits"
579 visit_image.to_legacy().writeFits(str(path))
581 with astropy.io.fits.open(path) as hdu_list:
582 primary = hdu_list[0].header
583 native_cards = {}
584 n = 1
585 while f"LSST IMAGES KEY {n}" in primary:
586 native_cards[primary[f"LSST IMAGES KEY {n}"]] = primary[f"LSST IMAGES VALUE {n}"]
587 n += 1
588 assert native_cards == {"native_key": 7, "MixedCase": "yes"}
589 assert [card.value for card in primary.cards if card.keyword == "BGMEAN"] == [1.5, 2.5]
591 result = VisitImage.read_legacy(
592 str(path),
593 exposure_record=get_dp2_exposure_record(DimensionUniverse()),
594 )
595 assert result.metadata.native["native_key"] == 7
596 assert result.metadata.native["MixedCase"] == "yes"
597 assert "id" in result.metadata.native
598 assert not [key for key in result.metadata.external if key.startswith("LSST IMAGES")]
599 assert result.metadata.external.get_all("BGMEAN") == (1.5, 2.5)
600 assert result.metadata["platform"] == "lsstcam"
601 assert result.metadata.external["ID"] == 99
603 # Converting the result back must not duplicate the external cards.
604 legacy_metadata = result.to_legacy().getMetadata()
605 assert legacy_metadata.getArray("BGMEAN") == [1.5, 2.5]
606 assert legacy_metadata["LSST IMAGES KEY 1"] == "native_key"
609def test_gaussian_psf_legacy_round_trip(
610 visit_image_components: dict[str, Any],
611 reset_afw_mask_planes: None, # noqa: F811
612) -> None:
613 """Verify that a Gaussian PSF set after construction is attached to the
614 legacy Exposure that `VisitImage.to_legacy` returns.
615 """
616 from lsst.afw.detection import GaussianPsf
618 visit_image = make_simplest_visit_image(visit_image_components)
619 psf = GaussianPointSpreadFunction(3.25, stamp_size=27, bounds=Box.factory[0:1024, 0:1024])
620 visit_image.psf = psf
621 assert visit_image.psf is psf
623 legacy_psf = visit_image.to_legacy().getPsf()
624 assert isinstance(legacy_psf, GaussianPsf)
625 assert legacy_psf.getSigma() == psf.sigma
626 assert legacy_psf.computeBBox(legacy_psf.getAveragePosition()).getWidth() == 27
629@skip_no_h5py
630def test_round_trip_ndf(visit_image_components: dict[str, Any]) -> None:
631 """Verify NDF round-trip produces a VisitImage equal to the original."""
632 visit_image = make_visit_image(visit_image_components)
633 with RoundtripNdf(visit_image, "VisitImage") as roundtrip:
634 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False)
637@skip_no_h5py
638def test_fits_ndf_consistency(visit_image_components: dict[str, Any]) -> None:
639 """Verify FITS and NDF backends produce equal VisitImages on round-trip."""
640 visit_image = make_visit_image(visit_image_components)
641 with RoundtripFits(visit_image) as fits_rt, RoundtripNdf(visit_image) as ndf_rt:
642 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False)
643 assert_visit_images_equal(visit_image, ndf_rt.result, expect_view=False)
644 assert_visit_images_equal(fits_rt.result, ndf_rt.result, expect_view=False)
647def test_fits_json_consistency(visit_image_components: dict[str, Any]) -> None:
648 """Verify FITS and JSON backends produce equal VisitImages."""
649 visit_image = make_visit_image(visit_image_components)
650 with (
651 RoundtripFits(visit_image) as fits_rt,
652 RoundtripJson(visit_image) as json_rt,
653 ):
654 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False)
655 assert_visit_images_equal(visit_image, json_rt.result, expect_view=False)
656 assert_visit_images_equal(fits_rt.result, json_rt.result, expect_view=False)
659def test_serialized_schema_version(visit_image_components: dict[str, Any]) -> None:
660 """The written tree's schema_version matches its schema_url.
662 The image, mask, and variance subtrees are stamped with their own
663 versions too.
664 """
665 visit_image = make_visit_image(visit_image_components)
666 tree = json.write(visit_image)
667 tree_type = type(tree)
668 assert tree.schema_version == tree_type.SCHEMA_VERSION
669 assert tree.min_read_version == tree_type.MIN_READ_VERSION
670 assert tree.schema_url.endswith(f"-{tree.schema_version}")
671 assert tree.model_dump()["schema_version"] == tree_type.SCHEMA_VERSION
672 for name in ("image", "mask", "variance"):
673 component = getattr(tree, name)
674 assert component.schema_version == type(component).SCHEMA_VERSION, name
677def test_read_write(visit_image_components: dict[str, Any]) -> None:
678 """Verify a VisitImage round-trips through FITS with correct compression.
680 Checks compression headers, subimage reads, equality, and opaque
681 metadata. Contains only butler-free assertions; component reads live
682 in `test_read_write_components`.
683 """
684 visit_image = make_visit_image(visit_image_components)
685 with RoundtripFits(visit_image, "VisitImage") as roundtrip:
686 # Check that we're still using the right compression, and that we
687 # wrote WCSs.
688 fits = roundtrip.inspect()
689 assert fits[1].header["ZCMPTYPE"] == "GZIP_2"
690 assert fits[1].header["CTYPE1"] == "RA---TAN"
691 assert fits[2].header["ZCMPTYPE"] == "GZIP_2"
692 assert fits[2].header["CTYPE1"] == "RA---TAN"
693 assert fits[3].header["ZCMPTYPE"] == "GZIP_2"
694 assert fits[3].header["CTYPE1"] == "RA---TAN"
695 # Check a subimage read (no component arg — does not trigger a skip).
696 subbox = Box.factory[8:13, 9:30]
697 subimage = roundtrip.get(bbox=subbox)
698 assert_masked_images_equal(subimage, visit_image[subbox], expect_view=False)
700 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False)
701 # Check that the round-tripped headers are the same (up to card order).
702 assert len(roundtrip.result._opaque_metadata.headers[ExtensionKey()]) == 1
703 assert dict(visit_image._opaque_metadata.headers[ExtensionKey()]) == dict(
704 roundtrip.result._opaque_metadata.headers[ExtensionKey()]
705 )
706 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")]
707 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")]
708 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")]
709 # Spot-check the concrete background contents (names, field types,
710 # subtracted entry) against the known fixture, so the equality check
711 # above is not vacuously satisfied by empty background maps.
712 assert isinstance(roundtrip.result.backgrounds, BackgroundMap)
713 assert roundtrip.result.backgrounds.keys() == {"standard"}
714 assert isinstance(roundtrip.result.backgrounds["standard"].field, ChebyshevField)
715 assert roundtrip.result.backgrounds.subtracted.name == "standard"
716 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image."
719def test_read_bounds_and_unit_components(visit_image_components: dict[str, Any], tmp_path: Path) -> None:
720 """Verify that the bounds and unit components match a full read, both
721 when the bounds are a polygon and when they default to the bbox.
722 """
723 for n, visit_image in enumerate(
724 [make_visit_image(visit_image_components), make_simplest_visit_image(visit_image_components)]
725 ):
726 path = tmp_path / f"visit_image_{n}.fits"
727 images_fits.write(visit_image, path)
728 with open_archive(path) as reader:
729 assert reader.get_component("bounds") == visit_image.bounds
730 assert reader.get_component("unit") == visit_image.unit
731 assert isinstance(make_visit_image(visit_image_components).bounds, Polygon)
734def test_read_write_components(visit_image_components: dict[str, Any]) -> None:
735 """Verify component reads and storage-class overrides round-trip correctly.
737 Requires a butler; skips when `lsst.daf.butler` is absent.
738 Butler-free assertions live in `test_read_write`.
739 """
740 c = visit_image_components
741 visit_image = make_visit_image(c)
742 with RoundtripFits(visit_image, "VisitImage") as roundtrip:
743 subbox = Box.factory[8:13, 9:30]
744 subimage = roundtrip.get(bbox=subbox)
746 # Get an explicit masked image to compare with the subimage.
747 subimage_masked = roundtrip.get("masked_image", bbox=subbox)
748 assert_masked_images_equal(subimage_masked, subimage, expect_view=False)
750 # Get the same masked image in a multi-component get and ensure
751 # it is the same thing.
752 components = roundtrip.get("components", components=["masked_image", "psf"], bbox=subbox)
753 assert set(components) == {"masked_image", "psf"}
754 assert_masked_images_equal(components["masked_image"], subimage_masked, expect_view=False)
756 assert roundtrip.get("bbox") == visit_image.bbox
757 assert roundtrip.get("bounds") == visit_image.bounds
758 assert roundtrip.get("unit") == visit_image.unit
760 obs_info = roundtrip.get("obs_info")
761 assert isinstance(obs_info, ObservationInfo)
762 assert obs_info == visit_image.obs_info
764 summary_stats = roundtrip.get("summary_stats")
765 assert isinstance(summary_stats, ObservationSummaryStats)
766 assert summary_stats == visit_image.summary_stats
768 psf = roundtrip.get("psf")
769 assert isinstance(psf, GaussianPointSpreadFunction)
770 assert psf.kernel_bbox == c["gaussian_psf"].kernel_bbox
772 backgrounds = roundtrip.get("backgrounds")
773 assert isinstance(backgrounds, BackgroundMap)
774 assert backgrounds.keys() == {"standard"}
775 assert isinstance(backgrounds["standard"].field, ChebyshevField)
776 assert backgrounds.subtracted.name == "standard"
777 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image."
779 # Test some components get edge cases.
780 components = roundtrip.get("components", components="image")
781 assert isinstance(components["image"], Image)
783 components = roundtrip.get("components")
784 assert set(components) == {
785 "image",
786 "variance",
787 "psf",
788 "bbox",
789 "mask",
790 "obs_info",
791 "backgrounds",
792 "detector",
793 "aperture_corrections",
794 "sky_projection",
795 "summary_stats",
796 "photometric_scaling",
797 "bounds",
798 "unit",
799 }
801 # Butler morphs RuntimeError to ValueError.
802 with pytest.raises(ValueError, match="component nonexistent"):
803 roundtrip.get("components", components=["image", "nonexistent"])
805 with pytest.raises(ValueError, match="should not be specified"):
806 roundtrip.get("components", components=["image", "components"])
808 with pytest.raises(ValueError, match="empty request"):
809 roundtrip.get("components", components=[])
811 with pytest.raises(ValueError, match="not known to any of the requested components"):
812 # PSF does not know how to use bbox so this fails.
813 roundtrip.get("components", components="psf", bbox=subbox)
816def test_sum_background_round_trip_fits(visit_image_components: dict[str, Any]) -> None:
817 """Verify FITS backend keeps two same-named SumField operands as distinct
818 EXTVERs.
819 """
820 visit_image = make_visit_image(visit_image_components)
821 visit = _make_sum_background_visit_image(visit_image_components, visit_image)
822 with RoundtripFits(visit) as roundtrip:
823 _check_sum_background_round_trip(roundtrip.result, visit)
826@skip_no_h5py
827def test_sum_background_round_trip_ndf(visit_image_components: dict[str, Any]) -> None:
828 """Verify NDF backend disambiguates the repeated ``data`` leaf, just as
829 the FITS backend does.
830 """
831 visit_image = make_visit_image(visit_image_components)
832 visit = _make_sum_background_visit_image(visit_image_components, visit_image)
833 with RoundtripNdf(visit) as roundtrip:
834 _check_sum_background_round_trip(roundtrip.result, visit)
837@pytest.mark.parametrize(
838 ("scaling_unit", "operation"),
839 [(u.electron / u.nJy, "multiply"), (u.nJy / u.electron, "divide")],
840 ids=["multiply", "divide"],
841)
842def test_convert_unit_subimage(
843 visit_image_components: dict[str, Any],
844 scaling_unit: u.UnitBase,
845 operation: Literal["multiply", "divide"],
846) -> None:
847 """Verify that converting the units of a subimage applies only the portion
848 of the photometric scaling that overlaps the subimage.
850 A photometric scaling keeps the bounds it was modeled over when the image
851 is subset, so both branches of the conversion must render it over the
852 subimage's bbox rather than over its own bounds.
853 """
854 visit_image = make_visit_image(visit_image_components)
855 scaling = ChebyshevField(
856 visit_image.bbox,
857 np.array([[4.0, 0.5, 0.125], [0.25, 0.0625, 0.0], [0.03125, 0.0, 0.0]]),
858 unit=scaling_unit,
859 )
860 visit_image.photometric_scaling = scaling
861 # Trim a different number of pixels from each side so a scaling rendered
862 # over the wrong box cannot match by chance.
863 subbox = Box.factory[
864 visit_image.bbox.y.start + 7 : visit_image.bbox.y.stop - 13,
865 visit_image.bbox.x.start + 3 : visit_image.bbox.x.stop - 21,
866 ]
867 subimage = visit_image[subbox]
868 assert subimage.photometric_scaling.bounds.bbox == visit_image.bbox
870 converted = subimage.convert_unit(u.electron)
872 assert converted.unit == u.electron
873 assert converted.image.bbox == subbox
874 scaling_array = scaling.render(subbox, dtype=subimage.image.array.dtype).array
875 if operation == "divide":
876 scaling_array = 1.0 / scaling_array
877 assert_values_equal(converted.image.array, subimage.image.array * scaling_array, rtol=1e-5)
878 assert_values_equal(converted.variance.array, subimage.variance.array * scaling_array**2, rtol=1e-5)
881@dataclasses.dataclass
882class _LegacyTestData:
883 filename: str
884 plane_map: dict[str, MaskPlane] = dataclasses.field(default_factory=get_legacy_visit_image_mask_planes)
885 unit: u.Unit = u.nJy
886 storage_class: str = "VisitImage"
887 read_cls: type[VisitImage] = VisitImage
888 exposure_record: Any = None
889 legacy_exposure: LegacyExposure = dataclasses.field(init=False)
891 @classmethod
892 def get(
893 cls, which: Literal["visit_image", "preliminary_visit_image", "difference_image"]
894 ) -> _LegacyTestData:
895 if EXTERNAL_DATA_DIR is None: 895 ↛ 897line 895 didn't jump to line 897 because the condition on line 895 was always true
896 pytest.skip("TESTDATA_IMAGES is not set up.")
897 result = cls(
898 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", f"{which}.fits"),
899 )
900 match which:
901 case "preliminary_visit_image":
902 result.unit = u.electron
903 case "difference_image":
904 result.storage_class = "DifferenceImage"
905 result.read_cls = DifferenceImage
906 result.plane_map = get_legacy_difference_image_mask_planes()
907 case "visit_image":
908 pass
909 try:
910 from lsst.afw.image import ExposureFitsReader
912 result.legacy_exposure = ExposureFitsReader(result.filename).read()
913 except ImportError:
914 pytest.skip("lsst.afw.image is not available; cannot read legacy exposures")
915 # All three legacy files are the same DP2 exposure.
916 result.exposure_record = get_dp2_exposure_record(DimensionUniverse())
917 result.visit_image = result.read_cls.read_legacy(
918 result.filename,
919 preserve_quantization=True,
920 plane_map=result.plane_map,
921 exposure_record=result.exposure_record,
922 )
923 return result
926@pytest.fixture(params=["visit_image", "preliminary_visit_image", "difference_image"])
927def legacy_test_data(request: pytest.FixtureRequest, reset_afw_mask_planes: None) -> _LegacyTestData: # noqa: F811
928 """Return legacy test data.
930 Tests that depend on this parameterized fixture run on all of the legacy
931 test images.
932 """
933 return _LegacyTestData.get(request.param)
936@pytest.fixture(params=["visit_image", "difference_image"])
937def legacy_test_data_calibrated(
938 request: pytest.FixtureRequest,
939 reset_afw_mask_planes: None, # noqa: F811
940) -> _LegacyTestData:
941 """Return legacy test data for calibrated images only.
943 Tests that depend on this parameterized fixture do not run on
944 preliminary_visit_image, since that has 'electron' pixel units
945 """
946 return _LegacyTestData.get(request.param)
949def _check_legacy_obs_info(obs_info: ObservationInfo | None) -> None:
950 """Assert obs_info carries expected LSSTCam/DP2 field values."""
951 assert isinstance(obs_info, ObservationInfo)
952 assert obs_info.instrument == "LSSTCam"
953 assert obs_info.detector_num == 85, obs_info
954 assert obs_info.detector_unique_name == "R21_S11", obs_info
955 assert obs_info.physical_filter == "r_57", obs_info
958def test_legacy_optional_mask_planes(legacy_test_data_calibrated: _LegacyTestData) -> None:
959 """Verify that an optional source injection plane converts in both
960 directions.
961 """
962 legacy = legacy_test_data_calibrated.legacy_exposure.clone()
963 injected = np.zeros(legacy.mask.array.shape, dtype=bool)
964 injected[4:9, 3:11] = True
965 for old_name in ("INJECTED", "INJECTED_CORE"):
966 legacy.mask.addMaskPlane(old_name)
967 legacy.mask.array[injected] |= legacy.mask.getPlaneBitMask("INJECTED")
968 plane_map = legacy_test_data_calibrated.plane_map
970 image = legacy_test_data_calibrated.read_cls.from_legacy(
971 legacy, plane_map=plane_map, exposure_record=legacy_test_data_calibrated.exposure_record
972 )
974 assert "INJECTED" in image.mask.schema.names
975 assert_values_equal(image.mask.get("INJECTED"), injected)
976 assert "INJECTED_CORE" not in image.mask.schema.names
977 # Every plane the image really uses, mapped and optional alike.
978 compare_mask_to_legacy(image.mask, legacy.mask, plane_map)
980 round_tripped = image.to_legacy()
982 assert "INJECTED" in round_tripped.mask.getMaskPlaneDict()
983 assert_values_equal(
984 (round_tripped.mask.array & round_tripped.mask.getPlaneBitMask("INJECTED")).astype(bool),
985 injected,
986 )
987 compare_mask_to_legacy(image.mask, round_tripped.mask, plane_map)
990def test_legacy_errors(legacy_test_data: _LegacyTestData) -> None:
991 """Verify that from_legacy and read_legacy raise on bad arguments."""
992 with pytest.raises(TypeError):
993 # exposure_record is a required argument.
994 VisitImage.from_legacy(legacy_test_data.legacy_exposure)
995 with pytest.raises(ValueError, match="BUNIT value .* disagrees with given unit"):
996 VisitImage.from_legacy(
997 legacy_test_data.legacy_exposure,
998 unit=u.mJy,
999 exposure_record=legacy_test_data.exposure_record,
1000 )
1001 visit = VisitImage.from_legacy(
1002 legacy_test_data.legacy_exposure,
1003 unit=legacy_test_data.unit,
1004 exposure_record=legacy_test_data.exposure_record,
1005 )
1006 assert visit.unit == legacy_test_data.unit
1008 with pytest.raises(TypeError):
1009 legacy_test_data.read_cls.read_legacy(legacy_test_data.filename)
1012def test_component_reads(legacy_test_data: _LegacyTestData) -> None:
1013 """Verify that individual components can be read from a legacy FITS
1014 file.
1015 """
1016 visit = VisitImage.read_legacy(
1017 legacy_test_data.filename, exposure_record=legacy_test_data.exposure_record
1018 )
1019 proj = VisitImage.read_legacy(
1020 legacy_test_data.filename,
1021 exposure_record=legacy_test_data.exposure_record,
1022 component="sky_projection",
1023 )
1024 assert_sky_projections_equal(proj, visit.sky_projection, expect_identity=False)
1025 image = VisitImage.read_legacy(
1026 legacy_test_data.filename,
1027 exposure_record=legacy_test_data.exposure_record,
1028 component="image",
1029 )
1030 assert image == visit.image
1031 assert_sky_projections_equal(proj, image.sky_projection, expect_identity=False)
1032 variance = VisitImage.read_legacy(
1033 legacy_test_data.filename,
1034 exposure_record=legacy_test_data.exposure_record,
1035 component="variance",
1036 )
1037 assert variance == visit.variance
1038 assert_sky_projections_equal(proj, variance.sky_projection, expect_identity=False)
1039 mask = VisitImage.read_legacy(
1040 legacy_test_data.filename,
1041 exposure_record=legacy_test_data.exposure_record,
1042 component="mask",
1043 )
1044 assert mask == visit.mask
1045 assert_sky_projections_equal(proj, mask.sky_projection, expect_identity=False)
1046 psf = VisitImage.read_legacy(
1047 legacy_test_data.filename,
1048 exposure_record=legacy_test_data.exposure_record,
1049 component="psf",
1050 )
1051 assert isinstance(psf, PointSpreadFunction)
1052 obs_info = VisitImage.read_legacy(
1053 legacy_test_data.filename,
1054 exposure_record=legacy_test_data.exposure_record,
1055 component="obs_info",
1056 )
1057 _check_legacy_obs_info(obs_info)
1058 summary_stats = VisitImage.read_legacy(
1059 legacy_test_data.filename,
1060 exposure_record=legacy_test_data.exposure_record,
1061 component="summary_stats",
1062 )
1063 assert isinstance(summary_stats, ObservationSummaryStats)
1064 assert summary_stats.nPsfStar == legacy_test_data.legacy_exposure.info.getSummaryStats().nPsfStar
1065 compare_aperture_corrections_to_legacy(
1066 VisitImage.read_legacy(
1067 legacy_test_data.filename,
1068 exposure_record=legacy_test_data.exposure_record,
1069 component="aperture_corrections",
1070 ),
1071 legacy_test_data.legacy_exposure.info.getApCorrMap(),
1072 visit.bbox,
1073 )
1074 detector = VisitImage.read_legacy(
1075 legacy_test_data.filename,
1076 exposure_record=legacy_test_data.exposure_record,
1077 component="detector",
1078 )
1079 compare_detector_to_legacy(
1080 detector, legacy_test_data.legacy_exposure.getDetector(), is_raw_assembled=True
1081 )
1082 photometric_scaling = VisitImage.read_legacy(
1083 legacy_test_data.filename,
1084 exposure_record=legacy_test_data.exposure_record,
1085 component="photometric_scaling",
1086 )
1087 compare_photo_calib_to_legacy(
1088 photometric_scaling,
1089 legacy_test_data.legacy_exposure.getPhotoCalib(),
1090 subimage_bbox=visit.bbox,
1091 )
1092 bounds = VisitImage.read_legacy(
1093 legacy_test_data.filename,
1094 exposure_record=legacy_test_data.exposure_record,
1095 component="bounds",
1096 )
1097 assert bounds == visit.bounds
1098 unit = VisitImage.read_legacy(
1099 legacy_test_data.filename,
1100 exposure_record=legacy_test_data.exposure_record,
1101 component="unit",
1102 )
1103 assert unit == visit.unit
1106def test_legacy_obs_info(legacy_test_data: _LegacyTestData) -> None:
1107 """Verify that ObservationInfo is constructed correctly from a legacy
1108 exposure.
1109 """
1110 legacy = VisitImage.from_legacy(
1111 legacy_test_data.legacy_exposure,
1112 plane_map=legacy_test_data.plane_map,
1113 exposure_record=legacy_test_data.exposure_record,
1114 )
1115 assert legacy.obs_info is not None
1116 assert legacy.obs_info == legacy_test_data.visit_image.obs_info
1117 assert legacy.obs_info is not None # for mypy
1118 assert legacy.obs_info.instrument == "LSSTCam"
1119 assert legacy.obs_info.detector_num == 85, legacy.obs_info
1120 assert legacy.obs_info.detector_unique_name == "R21_S11", legacy.obs_info
1121 assert legacy.obs_info.physical_filter == "r_57", legacy.obs_info
1124def test_aperture_corrections_to_legacy(legacy_test_data: _LegacyTestData) -> None:
1125 """Verify that aperture corrections round-trip through a legacy
1126 ApCorrMap.
1127 """
1128 ap_corrections = legacy_test_data.visit_image.aperture_corrections
1129 legacy_ap_corr_map = aperture_corrections_to_legacy(ap_corrections)
1130 compare_aperture_corrections_to_legacy(
1131 ap_corrections,
1132 legacy_ap_corr_map,
1133 legacy_test_data.visit_image.bbox,
1134 )
1137def _check_legacy_headers(visit_image: VisitImage) -> None:
1138 """Assert that primary and extension headers are stripped correctly."""
1139 header = visit_image._opaque_metadata.headers[ExtensionKey()]
1140 assert "EXPTIME" in header
1141 assert header["PLATFORM"] == "lsstcam"
1142 assert "LSST BUTLER ID" not in header
1143 assert "AR HDU" not in header
1144 assert "A_ORDER" not in header
1145 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("IMAGE"), astropy.io.fits.Header())
1146 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("MASK"), astropy.io.fits.Header())
1147 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("VARIANCE"), astropy.io.fits.Header())
1150def test_read_legacy_headers(legacy_test_data: _LegacyTestData) -> None:
1151 """Verify that headers were stripped and interpreted correctly in
1152 read_legacy.
1153 """
1154 assert legacy_test_data.visit_image.unit == legacy_test_data.unit
1155 _check_legacy_headers(legacy_test_data.visit_image)
1158def test_from_legacy_headers(legacy_test_data: _LegacyTestData) -> None:
1159 """Verify that from_legacy handles primary and extension headers
1160 correctly.
1161 """
1162 legacy = VisitImage.from_legacy(
1163 legacy_test_data.legacy_exposure,
1164 plane_map=legacy_test_data.plane_map,
1165 exposure_record=legacy_test_data.exposure_record,
1166 )
1167 assert legacy.unit == legacy_test_data.unit
1168 _check_legacy_headers(legacy)
1171def test_rewrite(legacy_test_data: _LegacyTestData) -> None:
1172 """Verify that a legacy VisitImage can be rewritten and round-trips both
1173 pixel values and all components.
1174 """
1175 with RoundtripFits(legacy_test_data.visit_image, legacy_test_data.storage_class) as roundtrip:
1176 fits = roundtrip.inspect()
1177 assert fits[1].header["ZCMPTYPE"] == "RICE_1"
1178 assert fits[1].header["CTYPE1"] == "RA---TAN-SIP"
1179 assert fits[2].header["ZCMPTYPE"] == "GZIP_2"
1180 assert fits[2].header["CTYPE1"] == "RA---TAN-SIP"
1181 assert fits[3].header["ZCMPTYPE"] == "RICE_1"
1182 assert fits[3].header["CTYPE1"] == "RA---TAN-SIP"
1183 subbox = Box.factory[8:13, 9:30]
1184 subimage = roundtrip.get(bbox=subbox)
1185 assert_masked_images_equal(subimage, legacy_test_data.visit_image[subbox], expect_view=False)
1186 alternates: dict[str, Any] = {}
1187 assert roundtrip.get("bbox") == legacy_test_data.visit_image.bbox
1188 alternates = {
1189 k: roundtrip.get(k)
1190 for k in [
1191 "sky_projection",
1192 "image",
1193 "mask",
1194 "variance",
1195 "psf",
1196 "obs_info",
1197 "summary_stats",
1198 "aperture_corrections",
1199 "detector",
1200 "photometric_scaling",
1201 ]
1202 }
1203 legacy_exposure = roundtrip.get(storageClass="Exposure")
1204 assert isinstance(legacy_exposure, LegacyExposure)
1205 compare_visit_image_to_legacy(
1206 legacy_test_data.visit_image,
1207 legacy_exposure,
1208 expect_view=False,
1209 plane_map=legacy_test_data.plane_map,
1210 **MINIMAL_VISIT_DATA_ID,
1211 )
1212 if legacy_test_data.visit_image.unit == u.nJy:
1213 assert legacy_exposure.getPhotoCalib()._isConstant
1214 assert legacy_exposure.getPhotoCalib().getCalibrationMean() == 1.0
1215 else:
1216 compare_photo_calib_to_legacy(
1217 legacy_test_data.visit_image.photometric_scaling,
1218 legacy_exposure.getPhotoCalib(),
1219 subimage_bbox=subbox,
1220 )
1221 assert legacy_exposure.info.getId() == legacy_test_data.legacy_exposure.info.getId()
1222 visit_info = roundtrip.get("obs_info", storageClass="VisitInfo")
1223 assert isinstance(visit_info, LegacyVisitInfo)
1224 assert visit_info.getInstrumentLabel() == "LSSTCam"
1226 assert_visit_images_equal(roundtrip.result, legacy_test_data.visit_image, expect_view=False)
1227 assert dict(legacy_test_data.visit_image._opaque_metadata.headers[ExtensionKey()]) == dict(
1228 roundtrip.result._opaque_metadata.headers[ExtensionKey()]
1229 )
1230 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")]
1231 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")]
1232 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")]
1233 assert roundtrip.result._opaque_metadata.headers[ExtensionKey()]["PLATFORM"] == "lsstcam"
1234 compare_visit_image_to_legacy(
1235 roundtrip.result,
1236 legacy_test_data.legacy_exposure,
1237 expect_view=False,
1238 plane_map=legacy_test_data.plane_map,
1239 **MINIMAL_VISIT_DATA_ID,
1240 alternates=alternates,
1241 )
1242 compare_visit_image_to_legacy(
1243 legacy_test_data.read_cls.from_legacy(
1244 legacy_test_data.legacy_exposure,
1245 plane_map=legacy_test_data.plane_map,
1246 exposure_record=legacy_test_data.exposure_record,
1247 ),
1248 legacy_test_data.legacy_exposure,
1249 expect_view=True,
1250 plane_map=legacy_test_data.plane_map,
1251 **MINIMAL_VISIT_DATA_ID,
1252 )
1255def test_butler_converters(legacy_test_data: _LegacyTestData) -> None:
1256 """Verify that a VisitImage can be read from a Butler dataset written as
1257 an Exposure.
1258 """
1259 try:
1260 from lsst.daf.butler import FileDataset
1261 except ImportError:
1262 pytest.skip("lsst.daf.butler could not be imported.")
1264 with TemporaryButler(legacy="ExposureF") as helper:
1265 helper.butler.ingest(
1266 FileDataset(path=legacy_test_data.filename, refs=[helper.legacy]), transfer="symlink"
1267 )
1268 visit_image_ref = helper.legacy.overrideStorageClass(legacy_test_data.storage_class)
1269 record_params = {"exposure_record": legacy_test_data.exposure_record}
1270 with warnings.catch_warnings():
1271 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
1272 visit_image = helper.butler.get(visit_image_ref, parameters=record_params)
1273 assert visit_image._opaque_metadata.precompressed.keys() == set()
1274 visit_image = helper.butler.get(
1275 visit_image_ref, parameters={"preserve_quantization": True, **record_params}
1276 )
1277 assert visit_image._opaque_metadata.precompressed.keys() == {"IMAGE", "VARIANCE"}
1278 bbox = helper.butler.get(visit_image_ref.makeComponentRef("bbox"), parameters=record_params)
1279 assert bbox == visit_image.bbox
1280 # bbox/psf/detector reads route through VisitImage and require the
1281 # exposure record; image/mask/variance route to MaskedImage, which
1282 # carries no ObservationInfo and must not be given it.
1283 alternates = {
1284 k: helper.butler.get(visit_image_ref.makeComponentRef(k)) for k in ["image", "mask", "variance"]
1285 } | {
1286 k: helper.butler.get(visit_image_ref.makeComponentRef(k), parameters=record_params)
1287 for k in ["bbox", "psf", "detector"]
1288 }
1289 compare_visit_image_to_legacy(
1290 visit_image,
1291 legacy_test_data.legacy_exposure,
1292 expect_view=False,
1293 plane_map=legacy_test_data.plane_map,
1294 alternates=alternates,
1295 **MINIMAL_VISIT_DATA_ID,
1296 )
1297 helper.butler.pruneDatasets([helper.legacy], purge=True, unstore=True, disassociate=True)
1298 visit_image.metadata["MixedCaseKey"] = 52
1299 helper.butler.put(visit_image, visit_image_ref)
1300 with warnings.catch_warnings():
1301 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
1302 legacy_exposure = helper.butler.get(helper.legacy)
1303 compare_visit_image_to_legacy(
1304 visit_image,
1305 legacy_exposure,
1306 expect_view=False,
1307 plane_map=legacy_test_data.plane_map,
1308 alternates=alternates,
1309 **MINIMAL_VISIT_DATA_ID,
1310 )
1311 visit_image_2 = helper.butler.get(visit_image_ref, parameters=record_params)
1312 compare_visit_image_to_legacy(
1313 visit_image_2,
1314 legacy_exposure,
1315 expect_view=False,
1316 plane_map=legacy_test_data.plane_map,
1317 alternates=alternates,
1318 **MINIMAL_VISIT_DATA_ID,
1319 )
1320 assert visit_image_2.metadata["MixedCaseKey"] == 52
1323def test_convert_unit(legacy_test_data_calibrated: _LegacyTestData) -> None:
1324 """Verify convert_unit round-trips between nJy, mJy, and electron via
1325 photometric_scaling.
1326 """
1327 from lsst.afw.table import ExposureCatalog
1329 legacy_test_data = legacy_test_data_calibrated
1330 original = legacy_test_data.visit_image.copy()
1331 with pytest.raises(u.UnitConversionError):
1332 original.convert_unit(u.electron)
1333 visit_image_nJy = original.convert_unit(u.nJy, copy=False)
1334 assert np.may_share_memory(visit_image_nJy.image.array, original.image.array)
1335 assert np.may_share_memory(visit_image_nJy.variance.array, original.variance.array)
1336 with pytest.raises(u.UnitConversionError):
1337 original.convert_unit(u.mJy, copy=False)
1338 visit_image_mJy = original.convert_unit(u.mJy, copy="as-needed")
1339 assert visit_image_mJy.unit == u.mJy
1340 assert_values_equal(visit_image_mJy.image.array, original.image.array * 1e-6, rtol=1e-5)
1341 assert np.may_share_memory(visit_image_nJy.mask.array, original.mask.array)
1342 assert_values_equal(visit_image_mJy.variance.array, original.variance.array * 1e-12, rtol=1e-5)
1343 legacy_exposure_mJy = visit_image_mJy.to_legacy()
1344 assert_values_equal(legacy_exposure_mJy.getPhotoCalib().getCalibrationMean(), 1e6, rtol=1e-14)
1345 legacy_masked_image_nJy = legacy_exposure_mJy.getPhotoCalib().calibrateImage(
1346 legacy_exposure_mJy.maskedImage
1347 )
1348 assert_values_equal(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array, rtol=1e-5)
1349 assert_values_equal(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array, rtol=1e-5)
1350 assert np.may_share_memory(visit_image_mJy.mask.array, original.mask.array)
1351 assert visit_image_mJy.sky_projection is original.sky_projection
1352 assert visit_image_mJy.obs_info is original.obs_info
1353 assert visit_image_mJy.summary_stats is original.summary_stats
1354 assert visit_image_mJy.psf is original.psf
1355 assert visit_image_mJy.detector is original.detector
1356 assert visit_image_mJy.bounds is original.bounds
1357 assert visit_image_mJy.aperture_corrections is original.aperture_corrections
1358 assert visit_image_mJy.photometric_scaling is original.photometric_scaling
1359 visit_summary = ExposureCatalog.readFits(
1360 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_summary.fits")
1361 )
1362 legacy_photo_calib = visit_summary.find(DP2_VISIT_DETECTOR_DATA_ID["detector"]).getPhotoCalib()
1363 visit_image_nJy.photometric_scaling = field_from_legacy_photo_calib(
1364 legacy_photo_calib, bounds=original.detector.bbox, instrumental_unit=u.electron
1365 )
1366 compare_photo_calib_to_legacy(
1367 visit_image_nJy.photometric_scaling,
1368 legacy_test_data.legacy_exposure.getPhotoCalib(),
1369 applied_legacy_photo_calib=legacy_photo_calib,
1370 subimage_bbox=visit_image_nJy.bbox,
1371 )
1372 with pytest.raises(u.UnitConversionError):
1373 visit_image_nJy.convert_unit(u.mm)
1374 with pytest.raises(u.UnitConversionError):
1375 visit_image_nJy.convert_unit(u.electron, copy=False)
1376 legacy_masked_image_e = legacy_photo_calib.uncalibrateImage(legacy_test_data.legacy_exposure.maskedImage)
1377 visit_image_e = visit_image_nJy.convert_unit(u.electron)
1378 assert_values_equal(visit_image_e.image.array, legacy_masked_image_e.image.array, rtol=1e-5)
1379 assert_values_equal(visit_image_e.variance.array, legacy_masked_image_e.variance.array, rtol=1e-5)
1380 assert not np.may_share_memory(visit_image_e.mask.array, visit_image_nJy.mask.array)
1381 visit_image_mJy.photometric_scaling = visit_image_nJy.photometric_scaling
1382 visit_image_e = visit_image_mJy.convert_unit(u.electron)
1383 assert_values_equal(visit_image_e.image.array, legacy_masked_image_e.image.array, rtol=1e-5)
1384 assert_values_equal(visit_image_e.variance.array, legacy_masked_image_e.variance.array, rtol=1e-5)
1385 visit_image_nJy_2 = visit_image_e.convert_unit(u.nJy)
1386 assert_values_equal(visit_image_nJy_2.image.array, visit_image_nJy.image.array, rtol=1e-5)
1387 assert_values_equal(visit_image_nJy_2.variance.array, original.variance.array, rtol=1e-5)
1388 visit_image_e.photometric_scaling = visit_image_nJy.photometric_scaling * (1e-6 * u.mJy / u.nJy)
1389 visit_image_nJy_3 = visit_image_e.convert_unit(u.nJy)
1390 assert_values_equal(visit_image_nJy_3.image.array, visit_image_nJy.image.array, rtol=1e-5)
1391 assert_values_equal(visit_image_nJy_3.variance.array, original.variance.array, rtol=1e-5)
1392 legacy_exposure_e = visit_image_e.to_legacy()
1393 assert_values_equal(
1394 legacy_exposure_e.getPhotoCalib().getCalibrationMean(),
1395 legacy_photo_calib.getCalibrationMean(),
1396 rtol=1e-5,
1397 )
1398 legacy_masked_image_nJy = legacy_exposure_e.getPhotoCalib().calibrateImage(legacy_exposure_e.maskedImage)
1399 assert_values_equal(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array, rtol=1e-5)
1400 assert_values_equal(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array, rtol=1e-5)
1403def test_background_map_describe() -> None:
1404 """An empty BackgroundMap._describe reports no backgrounds inline."""
1405 bg_map = BackgroundMap()
1406 assert isinstance(bg_map, DescribableMixin)
1407 report = bg_map._describe()
1408 assert isinstance(report, Report)
1409 assert report.type_name == "BackgroundMap"
1410 assert report.inline
1411 assert report.summary == "no backgrounds"
1412 assert report.children == {}
1415def test_background_map_with_entries_describe() -> None:
1416 """BackgroundMap._describe recurses into each background model."""
1417 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1418 bg_map = BackgroundMap(
1419 [Background("sky", cheby, "Sky model."), Background("fringe", cheby)],
1420 subtracted="sky",
1421 )
1422 report = bg_map._describe()
1423 assert report.type_name == "BackgroundMap"
1424 assert report.inline
1425 # The inline summary leads with the subtracted background and marks it,
1426 # since that is all a composite holding this map will show.
1427 # Only the subtracted background is marked, not the whole line.
1428 assert report.summary == "**sky [SUBTRACTED]**; also: fringe"
1429 assert report.emphasis_markup
1430 # str reads the summary as text, so the delimiters never reach a user.
1431 assert report.to_str() == "sky [SUBTRACTED]; also: fringe"
1432 # Standalone, each background is a child carrying its own model's report.
1433 assert set(report.children) == {"sky", "fringe"}
1434 sky = report.children["sky"]
1435 assert sky.type_name == "ChebyshevField"
1436 # The marker is on the heading, where it is read before the fields.
1437 assert sky.title == "ChebyshevField **[SUBTRACTED]**"
1438 assert sky.emphasis_markup
1439 sky_fields = {f.label: f.value for f in sky.fields}
1440 assert sky_fields["description"] == "Sky model."
1441 # The model's own fields survive alongside the background's attributes.
1442 assert "bounds" in sky_fields
1443 # Only the subtracted one is marked, and an absent description is omitted.
1444 fringe = report.children["fringe"]
1445 assert fringe.title is None
1446 assert fringe.emphasis_markup is False
1447 assert "description" not in {f.label for f in fringe.fields}
1450def test_background_map_describe_none_subtracted() -> None:
1451 """A map with no subtracted background says so rather than being silent."""
1452 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1453 bg_map = BackgroundMap([Background("sky", cheby), Background("fringe", cheby)])
1454 report = bg_map._describe()
1455 assert report.summary == "sky, fringe (none subtracted)"
1456 assert report.emphasis_markup is False
1457 assert not any(child.emphasis_markup for child in report.children.values())
1460def test_background_map_describe_leads_with_subtracted() -> None:
1461 """The subtracted background leads the summary wherever it sits in the
1462 map, and only its child is marked.
1463 """
1464 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1465 bg_map = BackgroundMap(
1466 [Background("sky", cheby), Background("skyCorr", cheby)],
1467 subtracted="skyCorr",
1468 )
1469 report = bg_map._describe()
1470 assert report.to_str() == "skyCorr [SUBTRACTED]; also: sky"
1471 assert report.children["sky"].emphasis_markup is False
1472 assert report.children["skyCorr"].emphasis_markup
1475def test_background_map_describe_brief_skips_children() -> None:
1476 """A brief background map report keeps the summary but not the models."""
1477 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1478 bg_map = BackgroundMap([Background("sky", cheby)], subtracted="sky")
1479 report = bg_map._describe(DescribeOptions(brief=True))
1480 assert report.to_str() == "sky [SUBTRACTED]"
1481 assert report.children == {}
1484def test_visit_image_repr_str_with_unreadable_psf() -> None:
1485 """Repr and str succeed even when the PSF stored an ArchiveReadError.
1487 An unreadable component is a supported state; repr and str read only the
1488 cheap fields and summary, so they do not build the child tree at all.
1489 """
1490 path = current_fixture_path(FIXTURE_DIR, "visit_image")
1491 visit_image = read_archive(path)
1492 visit_image._psf = ArchiveReadError("psf unreadable")
1493 assert repr(visit_image).startswith("VisitImage(")
1494 assert str(visit_image).startswith("VisitImage(")
1497def test_visit_image_describe_names_an_unreadable_psf() -> None:
1498 """A full report says the PSF could not be read, and describes the rest.
1500 A PSF model can need a package the reader does not have installed, so an
1501 unreadable PSF must not cost the report of everything else.
1502 """
1503 path = current_fixture_path(FIXTURE_DIR, "visit_image")
1504 visit_image = read_archive(path)
1505 visit_image._psf = ArchiveReadError("Failed to import piff.")
1506 report = visit_image.describe(detail=True)
1507 psf = report.children["psf"]
1508 assert psf.inline
1509 assert psf.to_str() == "unreadable (Failed to import piff.)"
1510 # Every other component is described as usual.
1511 assert "sky_projection" in report.children
1512 assert "detector" in report.children
1513 # Both renderers run over the report that contains it.
1514 assert "unreadable" in report._repr_html_()
1515 report.__rich__()
1518def test_visit_image_slice_preserves_unreadable_psf() -> None:
1519 """Slicing propagates a deferred PSF read failure without raising it."""
1520 path = current_fixture_path(FIXTURE_DIR, "visit_image")
1521 visit_image = read_archive(path)
1522 error = ArchiveReadError("psf unreadable")
1523 visit_image._psf = error
1525 sliced = visit_image[...]
1527 assert sliced._psf is error
1528 with pytest.raises(ArchiveReadError, match="psf unreadable"):
1529 sliced.psf
1532def test_observation_summary_stats_describe() -> None:
1533 """ObservationSummaryStats._describe reports the statistics that are set.
1535 Unset ones are omitted rather than shown as NaN.
1536 """
1537 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4, ra=180.0, dec=-30.0)
1538 assert isinstance(stats, DescribableMixin)
1539 report = stats._describe()
1540 assert isinstance(report, Report)
1541 assert report.type_name == "ObservationSummaryStats"
1542 # Scalars are packed into one group, ordered by name.
1543 (group,) = report.value_groups
1544 assert group.role is FieldRole.DERIVED
1545 assert [name for name, _ in group.values] == sorted(name for name, _ in group.values)
1546 values = dict(group.values)
1547 assert values["psfSigma"] == 2.5
1548 assert values["zeroPoint"] == 31.4
1549 assert values["ra"] == 180.0
1550 assert values["dec"] == -30.0
1551 # Unset statistics are omitted rather than reported as NaN, and the
1552 # serialization plumbing this class inherits never appears at all.
1553 assert "expTime" not in values
1554 assert "skyBg" not in values
1555 assert not {"metadata", "butler_info", "schema_version"} & set(values)
1558def test_observation_summary_stats_describe_omits_empty_sequences() -> None:
1559 """Sequence statistics appear only when they carry a value."""
1560 empty = ObservationSummaryStats(psfSigma=2.5)
1561 # raCorners defaults to all-NaN, which carries no more information than an
1562 # empty sequence does.
1563 assert all(math.isnan(v) for v in empty.raCorners)
1564 assert not any(f.label == "raCorners" for f in empty._describe().fields)
1566 filled = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
1567 corners = next(f for f in filled._describe().fields if f.label == "raCorners")
1568 assert corners.value == (5.2, 5.4, 5.4, 5.2)
1569 assert corners.role is FieldRole.DERIVED
1572def test_observation_summary_stats_describe_brief_counts() -> None:
1573 """A brief report gives the number set rather than listing them."""
1574 stats = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
1575 brief = stats._describe(DescribeOptions(brief=True))
1576 assert brief.type_name == "ObservationSummaryStats"
1577 assert brief.value_groups == []
1578 (field,) = brief.fields
1579 assert field.label == "statistics set"
1580 assert field.role is FieldRole.DERIVED
1582 # The count must agree with what the full report actually shows: two
1583 # scalars set by default plus psfSigma, and raCorners as a sequence.
1584 full = stats._describe()
1585 listed = len(full.fields) + sum(len(g.values) for g in full.value_groups)
1586 assert field.value.startswith(f"{listed} of ")
1587 assert brief.to_str() == full.to_str()
1588 assert f"{listed} of " in brief.to_str()
1591def test_observation_summary_stats_pydantic_repr() -> None:
1592 """ObservationSummaryStats uses pydantic's repr, not the mixin's."""
1593 stats = ObservationSummaryStats(psfSigma=2.5)
1594 r = repr(stats)
1595 assert r.startswith("ObservationSummaryStats(")
1596 assert "psfSigma=2.5" in r
1599def test_observation_summary_stats_str_is_the_report_summary() -> None:
1600 """The str output reports the count, not pydantic's field-by-field dump.
1602 The repr keeps the exhaustive form, since that is the one that
1603 round-trips.
1604 """
1605 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
1606 assert str(stats) == stats.describe().to_str()
1607 # The two set here, plus the integer counters, which default to a genuine
1608 # zero rather than to NaN.
1609 assert str(stats) == "ObservationSummaryStats(4 of 66 statistics set)"
1610 assert stats.nPsfStar == 0
1611 assert stats.nShapeletsStar == 0
1612 # The unset statistics reach repr but not str.
1613 assert "nan" not in str(stats)
1614 assert "nan" in repr(stats)
1617def test_archive_tree_repr_omits_schema_bookkeeping() -> None:
1618 """Schema version fields mirror class constants, so repr leaves them out.
1620 They are never passed on construction and say nothing the type does not.
1621 """
1622 stats = ObservationSummaryStats(psfSigma=2.5, indirect=[1])
1623 for name in ("schema_version", "min_read_version", "indirect"):
1624 assert name not in repr(stats), name
1625 # They are still real fields, and hiding them from repr does not hide them
1626 # from serialization.
1627 assert stats.schema_version == ObservationSummaryStats.SCHEMA_VERSION
1628 dumped = stats.model_dump()
1629 for name in ("schema_version", "min_read_version", "indirect"):
1630 assert name in dumped, name
1633def _match_detector_to_image(visit_image: VisitImage) -> VisitImage:
1634 """Return the visit image with its detector shrunk to the image bbox.
1636 Parameters
1637 ----------
1638 visit_image
1639 Image to adjust.
1641 Returns
1642 -------
1643 visit_image : `VisitImage`
1644 Image whose `~VisitImage.bbox` equals its detector's, as a full-frame
1645 image read from a butler has.
1646 """
1647 detector = visit_image.detector
1648 visit_image._detector = Detector(
1649 detector._attributes.model_copy(update={"bbox": visit_image.bbox}),
1650 detector.amplifiers,
1651 detector._frames,
1652 detector.visit,
1653 )
1654 return visit_image
1657def test_visit_image_describe_hides_summary_stat_corners_for_a_full_frame(
1658 visit_image_components: dict[str, Any],
1659) -> None:
1660 """The sky corners appear once, in the projection's readable table."""
1661 visit = _match_detector_to_image(make_visit_image(visit_image_components))
1662 visit.summary_stats.raCorners = (1.0, 2.0, 3.0, 4.0)
1663 visit.summary_stats.decCorners = (-1.0, -2.0, -3.0, -4.0)
1664 report = visit.describe()
1665 stats = report.children["summary_stats"]
1666 assert not {"raCorners", "decCorners"} & {field.label for field in stats.fields}
1667 # The projection is where they are read instead, and it still has them.
1668 assert any(table.title == "Corners" for table in report.children["sky_projection"].tables)
1671def test_visit_image_cutout_keeps_summary_stat_corners() -> None:
1672 """A cutout keeps the statistics' corners, which still describe the whole
1673 image the statistics were measured on.
1674 """
1675 # The DP2 variant is a cutout of a real image, so it carries the
1676 # statistics measured on the whole of it.
1677 path = current_fixture_path(FIXTURE_DIR, "visit_image", variant="dp2")
1678 visit_image = read_archive(path)
1679 assert visit_image.bbox != visit_image.detector.bbox
1680 stats = visit_image.describe().children["summary_stats"]
1681 assert {"raCorners", "decCorners"} <= {field.label for field in stats.fields}