Coverage for tests/test_visit_image.py: 71%
812 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-02 01:56 -0800
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_non_finite_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 non-finite float values in legacy metadata survive
553 conversion to a VisitImage, a FITS write, and conversion back.
554 """
555 from lsst.afw.detection import GaussianPsf
557 visit_image = VisitImage(
558 visit_image_components["image"],
559 variance=visit_image_components["variance"],
560 psf=PointSpreadFunction.from_legacy(GaussianPsf(33, 33, 2.5), bounds=Box.factory[0:1024, 0:1024]),
561 mask_schema=visit_image_components["mask_schema"],
562 sky_projection=visit_image_components["sky_projection"],
563 detector=visit_image_components["detector"],
564 obs_info=visit_image_components["obs_info"],
565 band="r",
566 )
567 legacy_exposure = visit_image.to_legacy()
568 legacy_metadata = legacy_exposure.getMetadata()
569 legacy_metadata.add("NANVAL", math.nan)
570 legacy_metadata.add("INFVAL", -math.inf)
572 converted = VisitImage.from_legacy(
573 legacy_exposure,
574 exposure_record=get_dp2_exposure_record(DimensionUniverse()),
575 )
576 path = tmp_path / "visit_image.fits"
577 images_fits.write(converted, path)
578 result = read_archive(path, VisitImage)
580 legacy_metadata = result.to_legacy().getMetadata()
581 assert math.isnan(legacy_metadata.getScalar("NANVAL"))
582 assert legacy_metadata.getScalar("INFVAL") == -math.inf
585def test_external_metadata_legacy_round_trip(
586 visit_image_components: dict[str, Any],
587 reset_afw_mask_planes: None, # noqa: F811
588 tmp_path: Path,
589) -> None:
590 """Verify that native metadata is written to a legacy file as
591 ``LSST IMAGES`` cards and is not visible as external metadata when the
592 file is read back.
593 """
594 from lsst.afw.detection import GaussianPsf
596 opaque_metadata = FitsOpaqueMetadata()
597 header = astropy.io.fits.Header()
598 header.append(("PLATFORM", "lsstcam"), end=True)
599 header.append(("BGMEAN", 1.5), end=True)
600 header.append(("BGMEAN", 2.5), end=True)
601 # An external ID card must neither block nor replace the native "id".
602 header.append(("ID", 99), end=True)
603 opaque_metadata.extract_legacy_primary_header(header)
604 visit_image = VisitImage(
605 visit_image_components["image"],
606 variance=visit_image_components["variance"],
607 psf=PointSpreadFunction.from_legacy(GaussianPsf(33, 33, 2.5), bounds=Box.factory[0:1024, 0:1024]),
608 mask_schema=visit_image_components["mask_schema"],
609 sky_projection=visit_image_components["sky_projection"],
610 detector=visit_image_components["detector"],
611 obs_info=visit_image_components["obs_info"],
612 band="r",
613 metadata={"native_key": 7, "MixedCase": "yes"},
614 )
615 visit_image._opaque_metadata = opaque_metadata
616 path = tmp_path / "legacy.fits"
617 visit_image.to_legacy().writeFits(str(path))
619 with astropy.io.fits.open(path) as hdu_list:
620 primary = hdu_list[0].header
621 native_cards = {}
622 n = 1
623 while f"LSST IMAGES KEY {n}" in primary:
624 native_cards[primary[f"LSST IMAGES KEY {n}"]] = primary[f"LSST IMAGES VALUE {n}"]
625 n += 1
626 assert native_cards == {"native_key": 7, "MixedCase": "yes"}
627 assert [card.value for card in primary.cards if card.keyword == "BGMEAN"] == [1.5, 2.5]
629 result = VisitImage.read_legacy(
630 str(path),
631 exposure_record=get_dp2_exposure_record(DimensionUniverse()),
632 )
633 assert result.metadata.native["native_key"] == 7
634 assert result.metadata.native["MixedCase"] == "yes"
635 assert "id" in result.metadata.native
636 assert not [key for key in result.metadata.external if key.startswith("LSST IMAGES")]
637 assert result.metadata.external.get_all("BGMEAN") == (1.5, 2.5)
638 assert result.metadata["platform"] == "lsstcam"
639 assert result.metadata.external["ID"] == 99
641 # Converting the result back must not duplicate the external cards.
642 legacy_metadata = result.to_legacy().getMetadata()
643 assert legacy_metadata.getArray("BGMEAN") == [1.5, 2.5]
644 assert legacy_metadata["LSST IMAGES KEY 1"] == "native_key"
647def test_gaussian_psf_legacy_round_trip(
648 visit_image_components: dict[str, Any],
649 reset_afw_mask_planes: None, # noqa: F811
650) -> None:
651 """Verify that a Gaussian PSF set after construction is attached to the
652 legacy Exposure that `VisitImage.to_legacy` returns.
653 """
654 from lsst.afw.detection import GaussianPsf
656 visit_image = make_simplest_visit_image(visit_image_components)
657 psf = GaussianPointSpreadFunction(3.25, stamp_size=27, bounds=Box.factory[0:1024, 0:1024])
658 visit_image.psf = psf
659 assert visit_image.psf is psf
661 legacy_psf = visit_image.to_legacy().getPsf()
662 assert isinstance(legacy_psf, GaussianPsf)
663 assert legacy_psf.getSigma() == psf.sigma
664 assert legacy_psf.computeBBox(legacy_psf.getAveragePosition()).getWidth() == 27
667@skip_no_h5py
668def test_round_trip_ndf(visit_image_components: dict[str, Any]) -> None:
669 """Verify NDF round-trip produces a VisitImage equal to the original."""
670 visit_image = make_visit_image(visit_image_components)
671 with RoundtripNdf(visit_image, "VisitImage") as roundtrip:
672 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False)
675@skip_no_h5py
676def test_fits_ndf_consistency(visit_image_components: dict[str, Any]) -> None:
677 """Verify FITS and NDF backends produce equal VisitImages on round-trip."""
678 visit_image = make_visit_image(visit_image_components)
679 with RoundtripFits(visit_image) as fits_rt, RoundtripNdf(visit_image) as ndf_rt:
680 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False)
681 assert_visit_images_equal(visit_image, ndf_rt.result, expect_view=False)
682 assert_visit_images_equal(fits_rt.result, ndf_rt.result, expect_view=False)
685def test_fits_json_consistency(visit_image_components: dict[str, Any]) -> None:
686 """Verify FITS and JSON backends produce equal VisitImages."""
687 visit_image = make_visit_image(visit_image_components)
688 with (
689 RoundtripFits(visit_image) as fits_rt,
690 RoundtripJson(visit_image) as json_rt,
691 ):
692 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False)
693 assert_visit_images_equal(visit_image, json_rt.result, expect_view=False)
694 assert_visit_images_equal(fits_rt.result, json_rt.result, expect_view=False)
697def test_serialized_schema_version(visit_image_components: dict[str, Any]) -> None:
698 """The written tree's schema_version matches its schema_url.
700 The image, mask, and variance subtrees are stamped with their own
701 versions too.
702 """
703 visit_image = make_visit_image(visit_image_components)
704 tree = json.write(visit_image)
705 tree_type = type(tree)
706 assert tree.schema_version == tree_type.SCHEMA_VERSION
707 assert tree.min_read_version == tree_type.MIN_READ_VERSION
708 assert tree.schema_url.endswith(f"-{tree.schema_version}")
709 assert tree.model_dump()["schema_version"] == tree_type.SCHEMA_VERSION
710 for name in ("image", "mask", "variance"):
711 component = getattr(tree, name)
712 assert component.schema_version == type(component).SCHEMA_VERSION, name
715def test_read_write(visit_image_components: dict[str, Any]) -> None:
716 """Verify a VisitImage round-trips through FITS with correct compression.
718 Checks compression headers, subimage reads, equality, and opaque
719 metadata. Contains only butler-free assertions; component reads live
720 in `test_read_write_components`.
721 """
722 visit_image = make_visit_image(visit_image_components)
723 with RoundtripFits(visit_image, "VisitImage") as roundtrip:
724 # Check that we're still using the right compression, and that we
725 # wrote WCSs.
726 fits = roundtrip.inspect()
727 assert fits[1].header["ZCMPTYPE"] == "GZIP_2"
728 assert fits[1].header["CTYPE1"] == "RA---TAN"
729 assert fits[2].header["ZCMPTYPE"] == "GZIP_2"
730 assert fits[2].header["CTYPE1"] == "RA---TAN"
731 assert fits[3].header["ZCMPTYPE"] == "GZIP_2"
732 assert fits[3].header["CTYPE1"] == "RA---TAN"
733 # Check a subimage read (no component arg — does not trigger a skip).
734 subbox = Box.factory[8:13, 9:30]
735 subimage = roundtrip.get(bbox=subbox)
736 assert_masked_images_equal(subimage, visit_image[subbox], expect_view=False)
738 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False)
739 # Check that the round-tripped headers are the same (up to card order).
740 assert len(roundtrip.result._opaque_metadata.headers[ExtensionKey()]) == 1
741 assert dict(visit_image._opaque_metadata.headers[ExtensionKey()]) == dict(
742 roundtrip.result._opaque_metadata.headers[ExtensionKey()]
743 )
744 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")]
745 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")]
746 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")]
747 # Spot-check the concrete background contents (names, field types,
748 # subtracted entry) against the known fixture, so the equality check
749 # above is not vacuously satisfied by empty background maps.
750 assert isinstance(roundtrip.result.backgrounds, BackgroundMap)
751 assert roundtrip.result.backgrounds.keys() == {"standard"}
752 assert isinstance(roundtrip.result.backgrounds["standard"].field, ChebyshevField)
753 assert roundtrip.result.backgrounds.subtracted.name == "standard"
754 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image."
757def test_read_bounds_and_unit_components(visit_image_components: dict[str, Any], tmp_path: Path) -> None:
758 """Verify that the bounds and unit components match a full read, both
759 when the bounds are a polygon and when they default to the bbox.
760 """
761 for n, visit_image in enumerate(
762 [make_visit_image(visit_image_components), make_simplest_visit_image(visit_image_components)]
763 ):
764 path = tmp_path / f"visit_image_{n}.fits"
765 images_fits.write(visit_image, path)
766 with open_archive(path) as reader:
767 assert reader.get_component("bounds") == visit_image.bounds
768 assert reader.get_component("unit") == visit_image.unit
769 assert isinstance(make_visit_image(visit_image_components).bounds, Polygon)
772def test_read_write_components(visit_image_components: dict[str, Any]) -> None:
773 """Verify component reads and storage-class overrides round-trip correctly.
775 Requires a butler; skips when `lsst.daf.butler` is absent.
776 Butler-free assertions live in `test_read_write`.
777 """
778 c = visit_image_components
779 visit_image = make_visit_image(c)
780 with RoundtripFits(visit_image, "VisitImage") as roundtrip:
781 subbox = Box.factory[8:13, 9:30]
782 subimage = roundtrip.get(bbox=subbox)
784 # Get an explicit masked image to compare with the subimage.
785 subimage_masked = roundtrip.get("masked_image", bbox=subbox)
786 assert_masked_images_equal(subimage_masked, subimage, expect_view=False)
788 # Get the same masked image in a multi-component get and ensure
789 # it is the same thing.
790 components = roundtrip.get("components", components=["masked_image", "psf"], bbox=subbox)
791 assert set(components) == {"masked_image", "psf"}
792 assert_masked_images_equal(components["masked_image"], subimage_masked, expect_view=False)
794 assert roundtrip.get("bbox") == visit_image.bbox
795 assert roundtrip.get("bounds") == visit_image.bounds
796 assert roundtrip.get("unit") == visit_image.unit
798 obs_info = roundtrip.get("obs_info")
799 assert isinstance(obs_info, ObservationInfo)
800 assert obs_info == visit_image.obs_info
802 summary_stats = roundtrip.get("summary_stats")
803 assert isinstance(summary_stats, ObservationSummaryStats)
804 assert summary_stats == visit_image.summary_stats
806 psf = roundtrip.get("psf")
807 assert isinstance(psf, GaussianPointSpreadFunction)
808 assert psf.kernel_bbox == c["gaussian_psf"].kernel_bbox
810 backgrounds = roundtrip.get("backgrounds")
811 assert isinstance(backgrounds, BackgroundMap)
812 assert backgrounds.keys() == {"standard"}
813 assert isinstance(backgrounds["standard"].field, ChebyshevField)
814 assert backgrounds.subtracted.name == "standard"
815 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image."
817 # Test some components get edge cases.
818 components = roundtrip.get("components", components="image")
819 assert isinstance(components["image"], Image)
821 components = roundtrip.get("components")
822 assert set(components) == {
823 "image",
824 "variance",
825 "psf",
826 "bbox",
827 "mask",
828 "obs_info",
829 "backgrounds",
830 "detector",
831 "aperture_corrections",
832 "sky_projection",
833 "summary_stats",
834 "photometric_scaling",
835 "bounds",
836 "unit",
837 }
839 # Butler morphs RuntimeError to ValueError.
840 with pytest.raises(ValueError, match="component nonexistent"):
841 roundtrip.get("components", components=["image", "nonexistent"])
843 with pytest.raises(ValueError, match="should not be specified"):
844 roundtrip.get("components", components=["image", "components"])
846 with pytest.raises(ValueError, match="empty request"):
847 roundtrip.get("components", components=[])
849 with pytest.raises(ValueError, match="not known to any of the requested components"):
850 # PSF does not know how to use bbox so this fails.
851 roundtrip.get("components", components="psf", bbox=subbox)
854def test_sum_background_round_trip_fits(visit_image_components: dict[str, Any]) -> None:
855 """Verify FITS backend keeps two same-named SumField operands as distinct
856 EXTVERs.
857 """
858 visit_image = make_visit_image(visit_image_components)
859 visit = _make_sum_background_visit_image(visit_image_components, visit_image)
860 with RoundtripFits(visit) as roundtrip:
861 _check_sum_background_round_trip(roundtrip.result, visit)
864@skip_no_h5py
865def test_sum_background_round_trip_ndf(visit_image_components: dict[str, Any]) -> None:
866 """Verify NDF backend disambiguates the repeated ``data`` leaf, just as
867 the FITS backend does.
868 """
869 visit_image = make_visit_image(visit_image_components)
870 visit = _make_sum_background_visit_image(visit_image_components, visit_image)
871 with RoundtripNdf(visit) as roundtrip:
872 _check_sum_background_round_trip(roundtrip.result, visit)
875@pytest.mark.parametrize(
876 ("scaling_unit", "operation"),
877 [(u.electron / u.nJy, "multiply"), (u.nJy / u.electron, "divide")],
878 ids=["multiply", "divide"],
879)
880def test_convert_unit_subimage(
881 visit_image_components: dict[str, Any],
882 scaling_unit: u.UnitBase,
883 operation: Literal["multiply", "divide"],
884) -> None:
885 """Verify that converting the units of a subimage applies only the portion
886 of the photometric scaling that overlaps the subimage.
888 A photometric scaling keeps the bounds it was modeled over when the image
889 is subset, so both branches of the conversion must render it over the
890 subimage's bbox rather than over its own bounds.
891 """
892 visit_image = make_visit_image(visit_image_components)
893 scaling = ChebyshevField(
894 visit_image.bbox,
895 np.array([[4.0, 0.5, 0.125], [0.25, 0.0625, 0.0], [0.03125, 0.0, 0.0]]),
896 unit=scaling_unit,
897 )
898 visit_image.photometric_scaling = scaling
899 # Trim a different number of pixels from each side so a scaling rendered
900 # over the wrong box cannot match by chance.
901 subbox = Box.factory[
902 visit_image.bbox.y.start + 7 : visit_image.bbox.y.stop - 13,
903 visit_image.bbox.x.start + 3 : visit_image.bbox.x.stop - 21,
904 ]
905 subimage = visit_image[subbox]
906 assert subimage.photometric_scaling.bounds.bbox == visit_image.bbox
908 converted = subimage.convert_unit(u.electron)
910 assert converted.unit == u.electron
911 assert converted.image.bbox == subbox
912 scaling_array = scaling.render(subbox, dtype=subimage.image.array.dtype).array
913 if operation == "divide":
914 scaling_array = 1.0 / scaling_array
915 assert_values_equal(converted.image.array, subimage.image.array * scaling_array, rtol=1e-5)
916 assert_values_equal(converted.variance.array, subimage.variance.array * scaling_array**2, rtol=1e-5)
919@dataclasses.dataclass
920class _LegacyTestData:
921 filename: str
922 plane_map: dict[str, MaskPlane] = dataclasses.field(default_factory=get_legacy_visit_image_mask_planes)
923 unit: u.Unit = u.nJy
924 storage_class: str = "VisitImage"
925 read_cls: type[VisitImage] = VisitImage
926 exposure_record: Any = None
927 legacy_exposure: LegacyExposure = dataclasses.field(init=False)
929 @classmethod
930 def get(
931 cls, which: Literal["visit_image", "preliminary_visit_image", "difference_image"]
932 ) -> _LegacyTestData:
933 if EXTERNAL_DATA_DIR is None: 933 ↛ 935line 933 didn't jump to line 935 because the condition on line 933 was always true
934 pytest.skip("TESTDATA_IMAGES is not set up.")
935 result = cls(
936 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", f"{which}.fits"),
937 )
938 match which:
939 case "preliminary_visit_image":
940 result.unit = u.electron
941 case "difference_image":
942 result.storage_class = "DifferenceImage"
943 result.read_cls = DifferenceImage
944 result.plane_map = get_legacy_difference_image_mask_planes()
945 case "visit_image":
946 pass
947 try:
948 from lsst.afw.image import ExposureFitsReader
950 result.legacy_exposure = ExposureFitsReader(result.filename).read()
951 except ImportError:
952 pytest.skip("lsst.afw.image is not available; cannot read legacy exposures")
953 # All three legacy files are the same DP2 exposure.
954 result.exposure_record = get_dp2_exposure_record(DimensionUniverse())
955 result.visit_image = result.read_cls.read_legacy(
956 result.filename,
957 preserve_quantization=True,
958 plane_map=result.plane_map,
959 exposure_record=result.exposure_record,
960 )
961 return result
964@pytest.fixture(params=["visit_image", "preliminary_visit_image", "difference_image"])
965def legacy_test_data(request: pytest.FixtureRequest, reset_afw_mask_planes: None) -> _LegacyTestData: # noqa: F811
966 """Return legacy test data.
968 Tests that depend on this parameterized fixture run on all of the legacy
969 test images.
970 """
971 return _LegacyTestData.get(request.param)
974@pytest.fixture(params=["visit_image", "difference_image"])
975def legacy_test_data_calibrated(
976 request: pytest.FixtureRequest,
977 reset_afw_mask_planes: None, # noqa: F811
978) -> _LegacyTestData:
979 """Return legacy test data for calibrated images only.
981 Tests that depend on this parameterized fixture do not run on
982 preliminary_visit_image, since that has 'electron' pixel units
983 """
984 return _LegacyTestData.get(request.param)
987def _check_legacy_obs_info(obs_info: ObservationInfo | None) -> None:
988 """Assert obs_info carries expected LSSTCam/DP2 field values."""
989 assert isinstance(obs_info, ObservationInfo)
990 assert obs_info.instrument == "LSSTCam"
991 assert obs_info.detector_num == 85, obs_info
992 assert obs_info.detector_unique_name == "R21_S11", obs_info
993 assert obs_info.physical_filter == "r_57", obs_info
996def test_legacy_optional_mask_planes(legacy_test_data_calibrated: _LegacyTestData) -> None:
997 """Verify that an optional source injection plane converts in both
998 directions.
999 """
1000 legacy = legacy_test_data_calibrated.legacy_exposure.clone()
1001 injected = np.zeros(legacy.mask.array.shape, dtype=bool)
1002 injected[4:9, 3:11] = True
1003 for old_name in ("INJECTED", "INJECTED_CORE"):
1004 legacy.mask.addMaskPlane(old_name)
1005 legacy.mask.array[injected] |= legacy.mask.getPlaneBitMask("INJECTED")
1006 plane_map = legacy_test_data_calibrated.plane_map
1008 image = legacy_test_data_calibrated.read_cls.from_legacy(
1009 legacy, plane_map=plane_map, exposure_record=legacy_test_data_calibrated.exposure_record
1010 )
1012 assert "INJECTED" in image.mask.schema.names
1013 assert_values_equal(image.mask.get("INJECTED"), injected)
1014 assert "INJECTED_CORE" not in image.mask.schema.names
1015 # Every plane the image really uses, mapped and optional alike.
1016 compare_mask_to_legacy(image.mask, legacy.mask, plane_map)
1018 round_tripped = image.to_legacy()
1020 assert "INJECTED" in round_tripped.mask.getMaskPlaneDict()
1021 assert_values_equal(
1022 (round_tripped.mask.array & round_tripped.mask.getPlaneBitMask("INJECTED")).astype(bool),
1023 injected,
1024 )
1025 compare_mask_to_legacy(image.mask, round_tripped.mask, plane_map)
1028def test_legacy_errors(legacy_test_data: _LegacyTestData) -> None:
1029 """Verify that from_legacy and read_legacy raise on bad arguments."""
1030 with pytest.raises(TypeError):
1031 # exposure_record is a required argument.
1032 VisitImage.from_legacy(legacy_test_data.legacy_exposure)
1033 with pytest.raises(ValueError, match="BUNIT value .* disagrees with given unit"):
1034 VisitImage.from_legacy(
1035 legacy_test_data.legacy_exposure,
1036 unit=u.mJy,
1037 exposure_record=legacy_test_data.exposure_record,
1038 )
1039 visit = VisitImage.from_legacy(
1040 legacy_test_data.legacy_exposure,
1041 unit=legacy_test_data.unit,
1042 exposure_record=legacy_test_data.exposure_record,
1043 )
1044 assert visit.unit == legacy_test_data.unit
1046 with pytest.raises(TypeError):
1047 legacy_test_data.read_cls.read_legacy(legacy_test_data.filename)
1050def test_component_reads(legacy_test_data: _LegacyTestData) -> None:
1051 """Verify that individual components can be read from a legacy FITS
1052 file.
1053 """
1054 visit = VisitImage.read_legacy(
1055 legacy_test_data.filename, exposure_record=legacy_test_data.exposure_record
1056 )
1057 proj = VisitImage.read_legacy(
1058 legacy_test_data.filename,
1059 exposure_record=legacy_test_data.exposure_record,
1060 component="sky_projection",
1061 )
1062 assert_sky_projections_equal(proj, visit.sky_projection, expect_identity=False)
1063 image = VisitImage.read_legacy(
1064 legacy_test_data.filename,
1065 exposure_record=legacy_test_data.exposure_record,
1066 component="image",
1067 )
1068 assert image == visit.image
1069 assert_sky_projections_equal(proj, image.sky_projection, expect_identity=False)
1070 variance = VisitImage.read_legacy(
1071 legacy_test_data.filename,
1072 exposure_record=legacy_test_data.exposure_record,
1073 component="variance",
1074 )
1075 assert variance == visit.variance
1076 assert_sky_projections_equal(proj, variance.sky_projection, expect_identity=False)
1077 mask = VisitImage.read_legacy(
1078 legacy_test_data.filename,
1079 exposure_record=legacy_test_data.exposure_record,
1080 component="mask",
1081 )
1082 assert mask == visit.mask
1083 assert_sky_projections_equal(proj, mask.sky_projection, expect_identity=False)
1084 psf = VisitImage.read_legacy(
1085 legacy_test_data.filename,
1086 exposure_record=legacy_test_data.exposure_record,
1087 component="psf",
1088 )
1089 assert isinstance(psf, PointSpreadFunction)
1090 obs_info = VisitImage.read_legacy(
1091 legacy_test_data.filename,
1092 exposure_record=legacy_test_data.exposure_record,
1093 component="obs_info",
1094 )
1095 _check_legacy_obs_info(obs_info)
1096 summary_stats = VisitImage.read_legacy(
1097 legacy_test_data.filename,
1098 exposure_record=legacy_test_data.exposure_record,
1099 component="summary_stats",
1100 )
1101 assert isinstance(summary_stats, ObservationSummaryStats)
1102 assert summary_stats.nPsfStar == legacy_test_data.legacy_exposure.info.getSummaryStats().nPsfStar
1103 compare_aperture_corrections_to_legacy(
1104 VisitImage.read_legacy(
1105 legacy_test_data.filename,
1106 exposure_record=legacy_test_data.exposure_record,
1107 component="aperture_corrections",
1108 ),
1109 legacy_test_data.legacy_exposure.info.getApCorrMap(),
1110 visit.bbox,
1111 )
1112 detector = VisitImage.read_legacy(
1113 legacy_test_data.filename,
1114 exposure_record=legacy_test_data.exposure_record,
1115 component="detector",
1116 )
1117 compare_detector_to_legacy(
1118 detector, legacy_test_data.legacy_exposure.getDetector(), is_raw_assembled=True
1119 )
1120 photometric_scaling = VisitImage.read_legacy(
1121 legacy_test_data.filename,
1122 exposure_record=legacy_test_data.exposure_record,
1123 component="photometric_scaling",
1124 )
1125 compare_photo_calib_to_legacy(
1126 photometric_scaling,
1127 legacy_test_data.legacy_exposure.getPhotoCalib(),
1128 subimage_bbox=visit.bbox,
1129 )
1130 bounds = VisitImage.read_legacy(
1131 legacy_test_data.filename,
1132 exposure_record=legacy_test_data.exposure_record,
1133 component="bounds",
1134 )
1135 assert bounds == visit.bounds
1136 unit = VisitImage.read_legacy(
1137 legacy_test_data.filename,
1138 exposure_record=legacy_test_data.exposure_record,
1139 component="unit",
1140 )
1141 assert unit == visit.unit
1144def test_legacy_obs_info(legacy_test_data: _LegacyTestData) -> None:
1145 """Verify that ObservationInfo is constructed correctly from a legacy
1146 exposure.
1147 """
1148 legacy = VisitImage.from_legacy(
1149 legacy_test_data.legacy_exposure,
1150 plane_map=legacy_test_data.plane_map,
1151 exposure_record=legacy_test_data.exposure_record,
1152 )
1153 assert legacy.obs_info is not None
1154 assert legacy.obs_info == legacy_test_data.visit_image.obs_info
1155 assert legacy.obs_info is not None # for mypy
1156 assert legacy.obs_info.instrument == "LSSTCam"
1157 assert legacy.obs_info.detector_num == 85, legacy.obs_info
1158 assert legacy.obs_info.detector_unique_name == "R21_S11", legacy.obs_info
1159 assert legacy.obs_info.physical_filter == "r_57", legacy.obs_info
1162def test_aperture_corrections_to_legacy(legacy_test_data: _LegacyTestData) -> None:
1163 """Verify that aperture corrections round-trip through a legacy
1164 ApCorrMap.
1165 """
1166 ap_corrections = legacy_test_data.visit_image.aperture_corrections
1167 legacy_ap_corr_map = aperture_corrections_to_legacy(ap_corrections)
1168 compare_aperture_corrections_to_legacy(
1169 ap_corrections,
1170 legacy_ap_corr_map,
1171 legacy_test_data.visit_image.bbox,
1172 )
1175def _check_legacy_headers(visit_image: VisitImage) -> None:
1176 """Assert that primary and extension headers are stripped correctly."""
1177 header = visit_image._opaque_metadata.headers[ExtensionKey()]
1178 assert "EXPTIME" in header
1179 assert header["PLATFORM"] == "lsstcam"
1180 assert "LSST BUTLER ID" not in header
1181 assert "AR HDU" not in header
1182 assert "A_ORDER" not in header
1183 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("IMAGE"), astropy.io.fits.Header())
1184 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("MASK"), astropy.io.fits.Header())
1185 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("VARIANCE"), astropy.io.fits.Header())
1188def test_read_legacy_headers(legacy_test_data: _LegacyTestData) -> None:
1189 """Verify that headers were stripped and interpreted correctly in
1190 read_legacy.
1191 """
1192 assert legacy_test_data.visit_image.unit == legacy_test_data.unit
1193 _check_legacy_headers(legacy_test_data.visit_image)
1196def test_from_legacy_headers(legacy_test_data: _LegacyTestData) -> None:
1197 """Verify that from_legacy handles primary and extension headers
1198 correctly.
1199 """
1200 legacy = VisitImage.from_legacy(
1201 legacy_test_data.legacy_exposure,
1202 plane_map=legacy_test_data.plane_map,
1203 exposure_record=legacy_test_data.exposure_record,
1204 )
1205 assert legacy.unit == legacy_test_data.unit
1206 _check_legacy_headers(legacy)
1209def test_rewrite(legacy_test_data: _LegacyTestData) -> None:
1210 """Verify that a legacy VisitImage can be rewritten and round-trips both
1211 pixel values and all components.
1212 """
1213 with RoundtripFits(legacy_test_data.visit_image, legacy_test_data.storage_class) as roundtrip:
1214 fits = roundtrip.inspect()
1215 assert fits[1].header["ZCMPTYPE"] == "RICE_1"
1216 assert fits[1].header["CTYPE1"] == "RA---TAN-SIP"
1217 assert fits[2].header["ZCMPTYPE"] == "GZIP_2"
1218 assert fits[2].header["CTYPE1"] == "RA---TAN-SIP"
1219 assert fits[3].header["ZCMPTYPE"] == "RICE_1"
1220 assert fits[3].header["CTYPE1"] == "RA---TAN-SIP"
1221 subbox = Box.factory[8:13, 9:30]
1222 subimage = roundtrip.get(bbox=subbox)
1223 assert_masked_images_equal(subimage, legacy_test_data.visit_image[subbox], expect_view=False)
1224 alternates: dict[str, Any] = {}
1225 assert roundtrip.get("bbox") == legacy_test_data.visit_image.bbox
1226 alternates = {
1227 k: roundtrip.get(k)
1228 for k in [
1229 "sky_projection",
1230 "image",
1231 "mask",
1232 "variance",
1233 "psf",
1234 "obs_info",
1235 "summary_stats",
1236 "aperture_corrections",
1237 "detector",
1238 "photometric_scaling",
1239 ]
1240 }
1241 legacy_exposure = roundtrip.get(storageClass="Exposure")
1242 assert isinstance(legacy_exposure, LegacyExposure)
1243 compare_visit_image_to_legacy(
1244 legacy_test_data.visit_image,
1245 legacy_exposure,
1246 expect_view=False,
1247 plane_map=legacy_test_data.plane_map,
1248 **MINIMAL_VISIT_DATA_ID,
1249 )
1250 if legacy_test_data.visit_image.unit == u.nJy:
1251 assert legacy_exposure.getPhotoCalib()._isConstant
1252 assert legacy_exposure.getPhotoCalib().getCalibrationMean() == 1.0
1253 else:
1254 compare_photo_calib_to_legacy(
1255 legacy_test_data.visit_image.photometric_scaling,
1256 legacy_exposure.getPhotoCalib(),
1257 subimage_bbox=subbox,
1258 )
1259 assert legacy_exposure.info.getId() == legacy_test_data.legacy_exposure.info.getId()
1260 visit_info = roundtrip.get("obs_info", storageClass="VisitInfo")
1261 assert isinstance(visit_info, LegacyVisitInfo)
1262 assert visit_info.getInstrumentLabel() == "LSSTCam"
1264 assert_visit_images_equal(roundtrip.result, legacy_test_data.visit_image, expect_view=False)
1265 assert dict(legacy_test_data.visit_image._opaque_metadata.headers[ExtensionKey()]) == dict(
1266 roundtrip.result._opaque_metadata.headers[ExtensionKey()]
1267 )
1268 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")]
1269 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")]
1270 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")]
1271 assert roundtrip.result._opaque_metadata.headers[ExtensionKey()]["PLATFORM"] == "lsstcam"
1272 compare_visit_image_to_legacy(
1273 roundtrip.result,
1274 legacy_test_data.legacy_exposure,
1275 expect_view=False,
1276 plane_map=legacy_test_data.plane_map,
1277 **MINIMAL_VISIT_DATA_ID,
1278 alternates=alternates,
1279 )
1280 compare_visit_image_to_legacy(
1281 legacy_test_data.read_cls.from_legacy(
1282 legacy_test_data.legacy_exposure,
1283 plane_map=legacy_test_data.plane_map,
1284 exposure_record=legacy_test_data.exposure_record,
1285 ),
1286 legacy_test_data.legacy_exposure,
1287 expect_view=True,
1288 plane_map=legacy_test_data.plane_map,
1289 **MINIMAL_VISIT_DATA_ID,
1290 )
1293def test_butler_converters(legacy_test_data: _LegacyTestData) -> None:
1294 """Verify that a VisitImage can be read from a Butler dataset written as
1295 an Exposure.
1296 """
1297 try:
1298 from lsst.daf.butler import FileDataset
1299 except ImportError:
1300 pytest.skip("lsst.daf.butler could not be imported.")
1302 with TemporaryButler(legacy="ExposureF") as helper:
1303 helper.butler.ingest(
1304 FileDataset(path=legacy_test_data.filename, refs=[helper.legacy]), transfer="symlink"
1305 )
1306 visit_image_ref = helper.legacy.overrideStorageClass(legacy_test_data.storage_class)
1307 record_params = {"exposure_record": legacy_test_data.exposure_record}
1308 with warnings.catch_warnings():
1309 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
1310 visit_image = helper.butler.get(visit_image_ref, parameters=record_params)
1311 assert visit_image._opaque_metadata.precompressed.keys() == set()
1312 visit_image = helper.butler.get(
1313 visit_image_ref, parameters={"preserve_quantization": True, **record_params}
1314 )
1315 assert visit_image._opaque_metadata.precompressed.keys() == {"IMAGE", "VARIANCE"}
1316 bbox = helper.butler.get(visit_image_ref.makeComponentRef("bbox"), parameters=record_params)
1317 assert bbox == visit_image.bbox
1318 # bbox/psf/detector reads route through VisitImage and require the
1319 # exposure record; image/mask/variance route to MaskedImage, which
1320 # carries no ObservationInfo and must not be given it.
1321 alternates = {
1322 k: helper.butler.get(visit_image_ref.makeComponentRef(k)) for k in ["image", "mask", "variance"]
1323 } | {
1324 k: helper.butler.get(visit_image_ref.makeComponentRef(k), parameters=record_params)
1325 for k in ["bbox", "psf", "detector"]
1326 }
1327 compare_visit_image_to_legacy(
1328 visit_image,
1329 legacy_test_data.legacy_exposure,
1330 expect_view=False,
1331 plane_map=legacy_test_data.plane_map,
1332 alternates=alternates,
1333 **MINIMAL_VISIT_DATA_ID,
1334 )
1335 helper.butler.pruneDatasets([helper.legacy], purge=True, unstore=True, disassociate=True)
1336 visit_image.metadata["MixedCaseKey"] = 52
1337 helper.butler.put(visit_image, visit_image_ref)
1338 with warnings.catch_warnings():
1339 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
1340 legacy_exposure = helper.butler.get(helper.legacy)
1341 compare_visit_image_to_legacy(
1342 visit_image,
1343 legacy_exposure,
1344 expect_view=False,
1345 plane_map=legacy_test_data.plane_map,
1346 alternates=alternates,
1347 **MINIMAL_VISIT_DATA_ID,
1348 )
1349 visit_image_2 = helper.butler.get(visit_image_ref, parameters=record_params)
1350 compare_visit_image_to_legacy(
1351 visit_image_2,
1352 legacy_exposure,
1353 expect_view=False,
1354 plane_map=legacy_test_data.plane_map,
1355 alternates=alternates,
1356 **MINIMAL_VISIT_DATA_ID,
1357 )
1358 assert visit_image_2.metadata["MixedCaseKey"] == 52
1361def test_convert_unit(legacy_test_data_calibrated: _LegacyTestData) -> None:
1362 """Verify convert_unit round-trips between nJy, mJy, and electron via
1363 photometric_scaling.
1364 """
1365 from lsst.afw.table import ExposureCatalog
1367 legacy_test_data = legacy_test_data_calibrated
1368 original = legacy_test_data.visit_image.copy()
1369 with pytest.raises(u.UnitConversionError):
1370 original.convert_unit(u.electron)
1371 visit_image_nJy = original.convert_unit(u.nJy, copy=False)
1372 assert np.may_share_memory(visit_image_nJy.image.array, original.image.array)
1373 assert np.may_share_memory(visit_image_nJy.variance.array, original.variance.array)
1374 with pytest.raises(u.UnitConversionError):
1375 original.convert_unit(u.mJy, copy=False)
1376 visit_image_mJy = original.convert_unit(u.mJy, copy="as-needed")
1377 assert visit_image_mJy.unit == u.mJy
1378 assert_values_equal(visit_image_mJy.image.array, original.image.array * 1e-6, rtol=1e-5)
1379 assert np.may_share_memory(visit_image_nJy.mask.array, original.mask.array)
1380 assert_values_equal(visit_image_mJy.variance.array, original.variance.array * 1e-12, rtol=1e-5)
1381 legacy_exposure_mJy = visit_image_mJy.to_legacy()
1382 assert_values_equal(legacy_exposure_mJy.getPhotoCalib().getCalibrationMean(), 1e6, rtol=1e-14)
1383 legacy_masked_image_nJy = legacy_exposure_mJy.getPhotoCalib().calibrateImage(
1384 legacy_exposure_mJy.maskedImage
1385 )
1386 assert_values_equal(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array, rtol=1e-5)
1387 assert_values_equal(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array, rtol=1e-5)
1388 assert np.may_share_memory(visit_image_mJy.mask.array, original.mask.array)
1389 assert visit_image_mJy.sky_projection is original.sky_projection
1390 assert visit_image_mJy.obs_info is original.obs_info
1391 assert visit_image_mJy.summary_stats is original.summary_stats
1392 assert visit_image_mJy.psf is original.psf
1393 assert visit_image_mJy.detector is original.detector
1394 assert visit_image_mJy.bounds is original.bounds
1395 assert visit_image_mJy.aperture_corrections is original.aperture_corrections
1396 assert visit_image_mJy.photometric_scaling is original.photometric_scaling
1397 visit_summary = ExposureCatalog.readFits(
1398 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_summary.fits")
1399 )
1400 legacy_photo_calib = visit_summary.find(DP2_VISIT_DETECTOR_DATA_ID["detector"]).getPhotoCalib()
1401 visit_image_nJy.photometric_scaling = field_from_legacy_photo_calib(
1402 legacy_photo_calib, bounds=original.detector.bbox, instrumental_unit=u.electron
1403 )
1404 compare_photo_calib_to_legacy(
1405 visit_image_nJy.photometric_scaling,
1406 legacy_test_data.legacy_exposure.getPhotoCalib(),
1407 applied_legacy_photo_calib=legacy_photo_calib,
1408 subimage_bbox=visit_image_nJy.bbox,
1409 )
1410 with pytest.raises(u.UnitConversionError):
1411 visit_image_nJy.convert_unit(u.mm)
1412 with pytest.raises(u.UnitConversionError):
1413 visit_image_nJy.convert_unit(u.electron, copy=False)
1414 legacy_masked_image_e = legacy_photo_calib.uncalibrateImage(legacy_test_data.legacy_exposure.maskedImage)
1415 visit_image_e = visit_image_nJy.convert_unit(u.electron)
1416 assert_values_equal(visit_image_e.image.array, legacy_masked_image_e.image.array, rtol=1e-5)
1417 assert_values_equal(visit_image_e.variance.array, legacy_masked_image_e.variance.array, rtol=1e-5)
1418 assert not np.may_share_memory(visit_image_e.mask.array, visit_image_nJy.mask.array)
1419 visit_image_mJy.photometric_scaling = visit_image_nJy.photometric_scaling
1420 visit_image_e = visit_image_mJy.convert_unit(u.electron)
1421 assert_values_equal(visit_image_e.image.array, legacy_masked_image_e.image.array, rtol=1e-5)
1422 assert_values_equal(visit_image_e.variance.array, legacy_masked_image_e.variance.array, rtol=1e-5)
1423 visit_image_nJy_2 = visit_image_e.convert_unit(u.nJy)
1424 assert_values_equal(visit_image_nJy_2.image.array, visit_image_nJy.image.array, rtol=1e-5)
1425 assert_values_equal(visit_image_nJy_2.variance.array, original.variance.array, rtol=1e-5)
1426 visit_image_e.photometric_scaling = visit_image_nJy.photometric_scaling * (1e-6 * u.mJy / u.nJy)
1427 visit_image_nJy_3 = visit_image_e.convert_unit(u.nJy)
1428 assert_values_equal(visit_image_nJy_3.image.array, visit_image_nJy.image.array, rtol=1e-5)
1429 assert_values_equal(visit_image_nJy_3.variance.array, original.variance.array, rtol=1e-5)
1430 legacy_exposure_e = visit_image_e.to_legacy()
1431 assert_values_equal(
1432 legacy_exposure_e.getPhotoCalib().getCalibrationMean(),
1433 legacy_photo_calib.getCalibrationMean(),
1434 rtol=1e-5,
1435 )
1436 legacy_masked_image_nJy = legacy_exposure_e.getPhotoCalib().calibrateImage(legacy_exposure_e.maskedImage)
1437 assert_values_equal(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array, rtol=1e-5)
1438 assert_values_equal(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array, rtol=1e-5)
1441def test_background_map_describe() -> None:
1442 """An empty BackgroundMap._describe reports no backgrounds inline."""
1443 bg_map = BackgroundMap()
1444 assert isinstance(bg_map, DescribableMixin)
1445 report = bg_map._describe()
1446 assert isinstance(report, Report)
1447 assert report.type_name == "BackgroundMap"
1448 assert report.inline
1449 assert report.summary == "no backgrounds"
1450 assert report.children == {}
1453def test_background_map_with_entries_describe() -> None:
1454 """BackgroundMap._describe recurses into each background model."""
1455 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1456 bg_map = BackgroundMap(
1457 [Background("sky", cheby, "Sky model."), Background("fringe", cheby)],
1458 subtracted="sky",
1459 )
1460 report = bg_map._describe()
1461 assert report.type_name == "BackgroundMap"
1462 assert report.inline
1463 # The inline summary leads with the subtracted background and marks it,
1464 # since that is all a composite holding this map will show.
1465 # Only the subtracted background is marked, not the whole line.
1466 assert report.summary == "**sky [SUBTRACTED]**; also: fringe"
1467 assert report.emphasis_markup
1468 # str reads the summary as text, so the delimiters never reach a user.
1469 assert report.to_str() == "sky [SUBTRACTED]; also: fringe"
1470 # Standalone, each background is a child carrying its own model's report.
1471 assert set(report.children) == {"sky", "fringe"}
1472 sky = report.children["sky"]
1473 assert sky.type_name == "ChebyshevField"
1474 # The marker is on the heading, where it is read before the fields.
1475 assert sky.title == "ChebyshevField **[SUBTRACTED]**"
1476 assert sky.emphasis_markup
1477 sky_fields = {f.label: f.value for f in sky.fields}
1478 assert sky_fields["description"] == "Sky model."
1479 # The model's own fields survive alongside the background's attributes.
1480 assert "bounds" in sky_fields
1481 # Only the subtracted one is marked, and an absent description is omitted.
1482 fringe = report.children["fringe"]
1483 assert fringe.title is None
1484 assert fringe.emphasis_markup is False
1485 assert "description" not in {f.label for f in fringe.fields}
1488def test_background_map_describe_none_subtracted() -> None:
1489 """A map with no subtracted background says so rather than being silent."""
1490 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1491 bg_map = BackgroundMap([Background("sky", cheby), Background("fringe", cheby)])
1492 report = bg_map._describe()
1493 assert report.summary == "sky, fringe (none subtracted)"
1494 assert report.emphasis_markup is False
1495 assert not any(child.emphasis_markup for child in report.children.values())
1498def test_background_map_describe_leads_with_subtracted() -> None:
1499 """The subtracted background leads the summary wherever it sits in the
1500 map, and only its child is marked.
1501 """
1502 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1503 bg_map = BackgroundMap(
1504 [Background("sky", cheby), Background("skyCorr", cheby)],
1505 subtracted="skyCorr",
1506 )
1507 report = bg_map._describe()
1508 assert report.to_str() == "skyCorr [SUBTRACTED]; also: sky"
1509 assert report.children["sky"].emphasis_markup is False
1510 assert report.children["skyCorr"].emphasis_markup
1513def test_background_map_describe_brief_skips_children() -> None:
1514 """A brief background map report keeps the summary but not the models."""
1515 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]]))
1516 bg_map = BackgroundMap([Background("sky", cheby)], subtracted="sky")
1517 report = bg_map._describe(DescribeOptions(brief=True))
1518 assert report.to_str() == "sky [SUBTRACTED]"
1519 assert report.children == {}
1522def test_visit_image_repr_str_with_unreadable_psf() -> None:
1523 """Repr and str succeed even when the PSF stored an ArchiveReadError.
1525 An unreadable component is a supported state; repr and str read only the
1526 cheap fields and summary, so they do not build the child tree at all.
1527 """
1528 path = current_fixture_path(FIXTURE_DIR, "visit_image")
1529 visit_image = read_archive(path)
1530 visit_image._psf = ArchiveReadError("psf unreadable")
1531 assert repr(visit_image).startswith("VisitImage(")
1532 assert str(visit_image).startswith("VisitImage(")
1535def test_visit_image_describe_names_an_unreadable_psf() -> None:
1536 """A full report says the PSF could not be read, and describes the rest.
1538 A PSF model can need a package the reader does not have installed, so an
1539 unreadable PSF must not cost the report of everything else.
1540 """
1541 path = current_fixture_path(FIXTURE_DIR, "visit_image")
1542 visit_image = read_archive(path)
1543 visit_image._psf = ArchiveReadError("Failed to import piff.")
1544 report = visit_image.describe(detail=True)
1545 psf = report.children["psf"]
1546 assert psf.inline
1547 assert psf.to_str() == "unreadable (Failed to import piff.)"
1548 # Every other component is described as usual.
1549 assert "sky_projection" in report.children
1550 assert "detector" in report.children
1551 # Both renderers run over the report that contains it.
1552 assert "unreadable" in report._repr_html_()
1553 report.__rich__()
1556def test_visit_image_slice_preserves_unreadable_psf() -> None:
1557 """Slicing propagates a deferred PSF read failure without raising it."""
1558 path = current_fixture_path(FIXTURE_DIR, "visit_image")
1559 visit_image = read_archive(path)
1560 error = ArchiveReadError("psf unreadable")
1561 visit_image._psf = error
1563 sliced = visit_image[...]
1565 assert sliced._psf is error
1566 with pytest.raises(ArchiveReadError, match="psf unreadable"):
1567 sliced.psf
1570def test_observation_summary_stats_describe() -> None:
1571 """ObservationSummaryStats._describe reports the statistics that are set.
1573 Unset ones are omitted rather than shown as NaN.
1574 """
1575 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4, ra=180.0, dec=-30.0)
1576 assert isinstance(stats, DescribableMixin)
1577 report = stats._describe()
1578 assert isinstance(report, Report)
1579 assert report.type_name == "ObservationSummaryStats"
1580 # Scalars are packed into one group, ordered by name.
1581 (group,) = report.value_groups
1582 assert group.role is FieldRole.DERIVED
1583 assert [name for name, _ in group.values] == sorted(name for name, _ in group.values)
1584 values = dict(group.values)
1585 assert values["psfSigma"] == 2.5
1586 assert values["zeroPoint"] == 31.4
1587 assert values["ra"] == 180.0
1588 assert values["dec"] == -30.0
1589 # Unset statistics are omitted rather than reported as NaN, and the
1590 # serialization plumbing this class inherits never appears at all.
1591 assert "expTime" not in values
1592 assert "skyBg" not in values
1593 assert not {"metadata", "butler_info", "schema_version"} & set(values)
1596def test_observation_summary_stats_describe_omits_empty_sequences() -> None:
1597 """Sequence statistics appear only when they carry a value."""
1598 empty = ObservationSummaryStats(psfSigma=2.5)
1599 # raCorners defaults to all-NaN, which carries no more information than an
1600 # empty sequence does.
1601 assert all(math.isnan(v) for v in empty.raCorners)
1602 assert not any(f.label == "raCorners" for f in empty._describe().fields)
1604 filled = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
1605 corners = next(f for f in filled._describe().fields if f.label == "raCorners")
1606 assert corners.value == (5.2, 5.4, 5.4, 5.2)
1607 assert corners.role is FieldRole.DERIVED
1610def test_observation_summary_stats_describe_brief_counts() -> None:
1611 """A brief report gives the number set rather than listing them."""
1612 stats = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2))
1613 brief = stats._describe(DescribeOptions(brief=True))
1614 assert brief.type_name == "ObservationSummaryStats"
1615 assert brief.value_groups == []
1616 (field,) = brief.fields
1617 assert field.label == "statistics set"
1618 assert field.role is FieldRole.DERIVED
1620 # The count must agree with what the full report actually shows: two
1621 # scalars set by default plus psfSigma, and raCorners as a sequence.
1622 full = stats._describe()
1623 listed = len(full.fields) + sum(len(g.values) for g in full.value_groups)
1624 assert field.value.startswith(f"{listed} of ")
1625 assert brief.to_str() == full.to_str()
1626 assert f"{listed} of " in brief.to_str()
1629def test_observation_summary_stats_pydantic_repr() -> None:
1630 """ObservationSummaryStats uses pydantic's repr, not the mixin's."""
1631 stats = ObservationSummaryStats(psfSigma=2.5)
1632 r = repr(stats)
1633 assert r.startswith("ObservationSummaryStats(")
1634 assert "psfSigma=2.5" in r
1637def test_observation_summary_stats_str_is_the_report_summary() -> None:
1638 """The str output reports the count, not pydantic's field-by-field dump.
1640 The repr keeps the exhaustive form, since that is the one that
1641 round-trips.
1642 """
1643 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4)
1644 assert str(stats) == stats.describe().to_str()
1645 # The two set here, plus the integer counters, which default to a genuine
1646 # zero rather than to NaN.
1647 assert str(stats) == "ObservationSummaryStats(4 of 66 statistics set)"
1648 assert stats.nPsfStar == 0
1649 assert stats.nShapeletsStar == 0
1650 # The unset statistics reach repr but not str.
1651 assert "nan" not in str(stats)
1652 assert "nan" in repr(stats)
1655def test_archive_tree_repr_omits_schema_bookkeeping() -> None:
1656 """Schema version fields mirror class constants, so repr leaves them out.
1658 They are never passed on construction and say nothing the type does not.
1659 """
1660 stats = ObservationSummaryStats(psfSigma=2.5, indirect=[1])
1661 for name in ("schema_version", "min_read_version", "indirect"):
1662 assert name not in repr(stats), name
1663 # They are still real fields, and hiding them from repr does not hide them
1664 # from serialization.
1665 assert stats.schema_version == ObservationSummaryStats.SCHEMA_VERSION
1666 dumped = stats.model_dump()
1667 for name in ("schema_version", "min_read_version", "indirect"):
1668 assert name in dumped, name
1671def _match_detector_to_image(visit_image: VisitImage) -> VisitImage:
1672 """Return the visit image with its detector shrunk to the image bbox.
1674 Parameters
1675 ----------
1676 visit_image
1677 Image to adjust.
1679 Returns
1680 -------
1681 visit_image : `VisitImage`
1682 Image whose `~VisitImage.bbox` equals its detector's, as a full-frame
1683 image read from a butler has.
1684 """
1685 detector = visit_image.detector
1686 visit_image._detector = Detector(
1687 detector._attributes.model_copy(update={"bbox": visit_image.bbox}),
1688 detector.amplifiers,
1689 detector._frames,
1690 detector.visit,
1691 )
1692 return visit_image
1695def test_visit_image_describe_hides_summary_stat_corners_for_a_full_frame(
1696 visit_image_components: dict[str, Any],
1697) -> None:
1698 """The sky corners appear once, in the projection's readable table."""
1699 visit = _match_detector_to_image(make_visit_image(visit_image_components))
1700 visit.summary_stats.raCorners = (1.0, 2.0, 3.0, 4.0)
1701 visit.summary_stats.decCorners = (-1.0, -2.0, -3.0, -4.0)
1702 report = visit.describe()
1703 stats = report.children["summary_stats"]
1704 assert not {"raCorners", "decCorners"} & {field.label for field in stats.fields}
1705 # The projection is where they are read instead, and it still has them.
1706 assert any(table.title == "Corners" for table in report.children["sky_projection"].tables)
1709def test_visit_image_cutout_keeps_summary_stat_corners() -> None:
1710 """A cutout keeps the statistics' corners, which still describe the whole
1711 image the statistics were measured on.
1712 """
1713 # The DP2 variant is a cutout of a real image, so it carries the
1714 # statistics measured on the whole of it.
1715 path = current_fixture_path(FIXTURE_DIR, "visit_image", variant="dp2")
1716 visit_image = read_archive(path)
1717 assert visit_image.bbox != visit_image.detector.bbox
1718 stats = visit_image.describe().children["summary_stats"]
1719 assert {"raCorners", "decCorners"} <= {field.label for field in stats.fields}