Coverage for tests/test_obs_info_from_legacy.py: 95%
141 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-13 09:31 +0000
1# This file is part of lsst-images.
2#
3# Developed for the LSST Data Management System.
4#
5# This product includes software developed by the LSST Project
6# (https://www.lsst.org).
7# See the COPYRIGHT file at the top-level directory of this distribution
8# for details of code ownership.
9#
10# Use of this source code is governed by a 3-clause BSD-style
11# license that can be found in the LICENSE file.
13"""Tests for ObservationInfo->VisitInfo->ObservationInfo round-trips.
15Neither ObservationInfo nor VisitInfo live in `lsst.images`, but these tests
16live here because the reconstruction logic they exercise lives in
17`lsst.images.VisitImage.from_legacy` and `~lsst.images.VisitImage.read_legacy`.
18"""
20from __future__ import annotations
22import dataclasses
23import warnings
24from functools import cached_property
25from pathlib import Path
26from typing import Any
28import astropy.coordinates as coords
29import astropy.io.fits
30import astropy.units as u
31import pytest
32from astro_metadata_translator import ObservationInfo
34from lsst.images import VisitImage
35from lsst.images.cameras import Detector
36from lsst.images.serialization import read_archive
37from lsst.images.tests import (
38 TemporaryButler,
39 annotate_errors,
40 assert_obs_metadata_fields_equal,
41)
43try:
44 from lsst.afw.cameraGeom import Detector as LegacyDetector
45 from lsst.afw.detection import GaussianPsf as LegacyGaussianPsf
46 from lsst.afw.geom import makeCdMatrix, makeSkyWcs
47 from lsst.afw.image import ExposureF as LegacyExposureF
48 from lsst.afw.image import FilterLabel, stripVisitInfoKeywords
49 from lsst.afw.image import VisitInfo as LegacyVisitInfo
50 from lsst.daf.base import PropertyList
51 from lsst.daf.butler import DimensionUniverse
52 from lsst.geom import Point2D, SpherePoint, arcseconds, degrees
53 from lsst.obs.base import MakeRawVisitInfoViaObsInfo, makeExposureRecordFromObsInfo
55 HAVE_LEGACY = True
56except ImportError:
57 type LegacyVisitInfo = Any # type: ignore[no-redef]
58 HAVE_LEGACY = False
60skip_no_legacy = pytest.mark.skipif(not HAVE_LEGACY, reason="afw etc could not be imported.")
62DATA_DIR = Path(__file__).parent / "data" / "headers"
63OBS_INFO_DIR = Path(__file__).parent / "data" / "obs_info"
64DETECTORS_DIR = Path(__file__).parent / "data" / "detectors"
67@dataclasses.dataclass(frozen=True)
68class InstrumentCase:
69 """Definition of a per-instrument test fixture and helpers for loading its
70 data.
71 """
73 label: str
74 test_data_filename: str
75 obs_info_filename: str
76 detector_filename: str
78 def get_full_header(self) -> astropy.io.fits.Header:
79 return astropy.io.fits.Header.fromtextfile(str(DATA_DIR / self.test_data_filename))
81 def get_stripped_header(self) -> astropy.io.fits.Header:
82 md = PropertyList.from_astropy_header(self.get_full_header())
83 stripVisitInfoKeywords(md)
84 for key in [ # extra cards stripped in ExposureFitsReader
85 "MJD-OBS",
86 "DATE-OBS",
87 "DETNAME",
88 "DETSER",
89 "EXPID",
90 ]:
91 md.pop(key, ...)
92 return md.to_astropy_header()
94 @cached_property
95 def true_obs_info(self) -> ObservationInfo:
96 """The ground-truth ObservationInfo (translated from raw)."""
97 return ObservationInfo.model_validate_json((OBS_INFO_DIR / self.obs_info_filename).read_text())
99 def make_record(self) -> Any:
100 """Build the `exposure` dimension record from the true
101 `ObservationInfo`.
102 """
103 return makeExposureRecordFromObsInfo(self.true_obs_info, DimensionUniverse())
105 @cached_property
106 def legacy_detector(self) -> LegacyDetector:
107 """The legacy `lsst.afw.cameraGeom.Detector` for this test case."""
108 assert self.true_obs_info.detector_num is not None
109 detector = read_archive(str(DETECTORS_DIR / self.detector_filename), Detector)
110 assert detector.id == self.true_obs_info.detector_num
111 return detector.to_legacy()
113 def make_stripped_exposure(self) -> LegacyExposureF:
114 """Build a test ``ExposureF`` with metadata content mimicking a legacy
115 visit image that has been roundtripped through FITS at least once.
116 """
117 header = self.get_stripped_header()
118 exposure = LegacyExposureF(16, 16)
119 if self.true_obs_info.detector_exposure_id is not None: 119 ↛ 121line 119 didn't jump to line 121 because the condition on line 119 was always true
120 exposure.info.setId(self.true_obs_info.detector_exposure_id)
121 exposure.setMetadata(PropertyList.from_astropy_header(header))
122 visit_info = MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.true_obs_info)
123 exposure.info.setVisitInfo(visit_info)
124 wcs = makeSkyWcs(
125 crpix=Point2D(8.0, 8.0),
126 crval=SpherePoint(42.0 * degrees, -11.0 * degrees), # arbitrary ICRS position; not asserted on
127 cdMatrix=makeCdMatrix(scale=0.2 * arcseconds),
128 )
129 exposure.setWcs(wcs)
130 exposure.setDetector(self.legacy_detector)
131 exposure.setPsf(LegacyGaussianPsf(11, 11, 2.0))
132 exposure.info.setFilter(
133 FilterLabel(band="g", physical=self.true_obs_info.physical_filter or "UNKNOWN")
134 )
135 return exposure
138# Fields that we can't recover from any of the inputs to obs_info_from_legacy,
139# mapped to what we get instead.
140UNRECOVERABLE_FIELDS = {
141 "translator_name": "VisitInfo",
142 "detector_name": None,
143 "detector_group": None,
144 "observing_day_offset": None,
145 "altaz_end": None,
146}
149def assert_obs_info_equal(
150 actual: ObservationInfo,
151 expected: ObservationInfo,
152 path: str,
153 *,
154 detector_serial: str | None,
155 detector_unique_name: str,
156) -> None:
157 """Assert that two ObservationInfo structs are as equal as they can be
158 after a from-legacy conversion.
159 """
160 for field in ObservationInfo.model_fields:
161 if field == "warnings": 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 continue
163 expected_value = getattr(expected, field)
164 if expected_value is None:
165 continue
166 actual_value = getattr(actual, field, None)
167 if field in UNRECOVERABLE_FIELDS:
168 if actual_value != UNRECOVERABLE_FIELDS[field]:
169 raise AssertionError(
170 f"{field} via {path}: expected fallback "
171 f"{UNRECOVERABLE_FIELDS[field]!r}, got {actual_value!r}"
172 )
173 continue
174 if field == "visit_id":
175 if actual_value != actual.exposure_id:
176 raise AssertionError(
177 f"visit_id via {path}: {actual_value!r} != exposure_id {actual.exposure_id!r}"
178 )
179 continue
180 if field == "boresight_rotation_coord":
181 if isinstance(expected_value, str) and isinstance(actual_value, str): 181 ↛ 187line 181 didn't jump to line 187 because the condition on line 181 was always true
182 if actual_value.casefold() != expected_value.casefold():
183 raise AssertionError(
184 f"{field} via {path}: {actual_value!r} != {expected_value!r} (ignoring case)"
185 )
186 continue
187 if field == "detector_serial":
188 expected_value = detector_serial
189 if field == "detector_unique_name":
190 expected_value = detector_unique_name
191 if actual_value is None:
192 raise AssertionError(f"{field} was lost via {path}: expected {expected_value}")
193 with annotate_errors(field):
194 assert_obs_metadata_fields_equal(actual_value, expected_value, label=field)
197def _point_to_astropy(point: Any) -> coords.SkyCoord:
198 """Convert an `lsst.geom.SpherePoint` to an Astropy ICRS `SkyCoord`."""
199 return coords.SkyCoord(
200 ra=point.getRa().asRadians() * u.rad,
201 dec=point.getDec().asRadians() * u.rad,
202 )
205def _point_to_altaz(point: Any) -> coords.AltAz:
206 """Convert an `lsst.geom.SpherePoint` conventionally holding az/alt to
207 an Astropy `AltAz` frame.
208 """
209 return coords.AltAz(
210 az=coords.Longitude(point.getLongitude().asRadians(), u.rad),
211 alt=coords.Latitude(point.getLatitude().asRadians(), u.rad),
212 )
215def assert_visit_info_equal(actual: LegacyVisitInfo, expected: LegacyVisitInfo) -> None:
216 """Assert that two VisitInfo structs are equal after a roundtrip
217 through `lsst.images.VisitImage`.
218 """
219 for name in (
220 "id",
221 "date",
222 "ut1",
223 "instrumentLabel",
224 "scienceProgram",
225 "observationType",
226 "observationReason",
227 "object",
228 "observatory",
229 "rotType",
230 "weather",
231 "exposureTime",
232 "darkTime",
233 "boresightAirmass",
234 "focusZ",
235 "era",
236 "localEra",
237 "boresightHourAngle",
238 "boresightParAngle",
239 "boresightRotAngle",
240 "hasSimulatedContent",
241 "isPersistable",
242 ):
243 with annotate_errors(name):
244 assert_obs_metadata_fields_equal(getattr(actual, name), getattr(expected, name), label=name)
245 with annotate_errors("boresightRaDec"):
246 assert_obs_metadata_fields_equal(
247 _point_to_astropy(actual.boresightRaDec),
248 _point_to_astropy(expected.boresightRaDec),
249 label="boresightRaDec",
250 )
251 with annotate_errors("boresightAzAlt"):
252 assert_obs_metadata_fields_equal(
253 _point_to_altaz(actual.boresightAzAlt),
254 _point_to_altaz(expected.boresightAzAlt),
255 label="boresightAzAlt",
256 )
259@skip_no_legacy
260@pytest.mark.parametrize(
261 "instrument_case",
262 [
263 InstrumentCase(
264 label="DECam",
265 test_data_filename="decam-c4d_150218_052850-N25.hdr",
266 obs_info_filename="decam-c4d_150218_052850-N25.json",
267 detector_filename="decam-c4d_150218_052850-N25.json",
268 ),
269 InstrumentCase(
270 label="HSC",
271 test_data_filename="hsc-HSCA90333426.hdr",
272 obs_info_filename="hsc-HSCA90333426.json",
273 detector_filename="hsc-HSCA90333426.json",
274 ),
275 InstrumentCase(
276 label="LSSTCam",
277 test_data_filename="lsstcam-MC_O_20250501_000292_R21_S11.hdr",
278 obs_info_filename="lsstcam-MC_O_20250501_000292_R21_S11.json",
279 detector_filename="lsstcam-MC_O_20250501_000292_R21_S11.json",
280 ),
281 ],
282 ids=["DECam", "HSC", "LSSTCam"],
283)
284def test_from_and_read_legacy_preserve_obs_info(instrument_case: InstrumentCase) -> None:
285 """ObservationInfo fields must survive from_legacy on a stripped header."""
286 assert instrument_case.true_obs_info.instrument == instrument_case.label
287 assert instrument_case.true_obs_info.exposure_id is not None
288 exposure = instrument_case.make_stripped_exposure()
289 exposure_record = instrument_case.make_record()
290 with warnings.catch_warnings():
291 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
292 result1 = VisitImage.from_legacy(exposure, exposure_record=exposure_record)
293 assert_obs_info_equal(
294 result1.obs_info,
295 instrument_case.true_obs_info,
296 "VisitImage.from_legacy on an in-memory stripped header",
297 detector_serial=instrument_case.legacy_detector.getSerial(),
298 detector_unique_name=instrument_case.legacy_detector.getName(),
299 )
300 roundtrip1 = result1.to_legacy()
301 assert_visit_info_equal(roundtrip1.info.getVisitInfo(), exposure.info.getVisitInfo())
302 with TemporaryButler(legacy="ExposureF") as helper:
303 helper.butler.put(exposure, helper.legacy)
304 visit_image_ref = helper.legacy.overrideStorageClass("VisitImage")
305 with warnings.catch_warnings():
306 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
307 try:
308 result2 = helper.butler.get(visit_image_ref, parameters={"exposure_record": exposure_record})
309 except Exception as err:
310 pytest.fail(
311 f"butler.get as VisitImage raised {type(err).__name__}: {err} "
312 "-- every ObservationInfo field was lost."
313 )
314 roundtrip2 = result2.to_legacy()
315 assert_visit_info_equal(roundtrip2.info.getVisitInfo(), exposure.info.getVisitInfo())
316 assert_obs_info_equal(
317 result2.obs_info,
318 instrument_case.true_obs_info,
319 "butler.get of an ExposureF dataset as VisitImage",
320 detector_serial=instrument_case.legacy_detector.getSerial(),
321 detector_unique_name=instrument_case.legacy_detector.getName(),
322 )
323 with TemporaryButler(legacy="ExposureF") as helper:
324 helper.butler.put(exposure, helper.legacy)
325 visit_image_ref = helper.legacy.overrideStorageClass("VisitImage")
326 with pytest.raises(ValueError, match="exposure_record"):
327 helper.butler.get(visit_image_ref)