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