Coverage for tests/test_obs_info_from_legacy.py: 96%
204 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 09:26 +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 logging
24import warnings
25from functools import cached_property
26from pathlib import Path
27from typing import Any
29import astropy.coordinates as coords
30import astropy.io.fits
31import astropy.units as u
32import pytest
33from astro_metadata_translator import FITS_FLAT_DICT_PREFIX, FITS_FLAT_DICT_SEP, ObservationInfo
35from lsst.images import VisitImage
36from lsst.images.cameras import Detector
37from lsst.images.fits import ExtensionKey
38from lsst.images.serialization import read_archive
39from lsst.images.tests import (
40 TemporaryButler,
41 annotate_errors,
42 assert_obs_metadata_fields_equal,
43)
45try:
46 from lsst.afw.cameraGeom import Detector as LegacyDetector
47 from lsst.afw.detection import GaussianPsf as LegacyGaussianPsf
48 from lsst.afw.geom import makeCdMatrix, makeSkyWcs
49 from lsst.afw.image import ExposureF as LegacyExposureF
50 from lsst.afw.image import FilterLabel, stripVisitInfoKeywords
51 from lsst.afw.image import VisitInfo as LegacyVisitInfo
52 from lsst.daf.base import PropertyList
53 from lsst.daf.butler import DimensionUniverse
54 from lsst.geom import Point2D, SpherePoint, arcseconds, degrees
55 from lsst.obs.base import MakeRawVisitInfoViaObsInfo, makeExposureRecordFromObsInfo
57 HAVE_LEGACY = True
58except ImportError:
59 type LegacyVisitInfo = Any # type: ignore[no-redef]
60 HAVE_LEGACY = False
62skip_no_legacy = pytest.mark.skipif(not HAVE_LEGACY, reason="afw etc could not be imported.")
64DATA_DIR = Path(__file__).parent / "data" / "headers"
65OBS_INFO_DIR = Path(__file__).parent / "data" / "obs_info"
66DETECTORS_DIR = Path(__file__).parent / "data" / "detectors"
69@dataclasses.dataclass(frozen=True)
70class InstrumentCase:
71 """Definition of a per-instrument test fixture and helpers for loading its
72 data.
73 """
75 label: str
76 test_data_filename: str
77 obs_info_filename: str
78 detector_filename: str
80 def get_full_header(self) -> astropy.io.fits.Header:
81 return astropy.io.fits.Header.fromtextfile(str(DATA_DIR / self.test_data_filename))
83 def get_stripped_header(self) -> astropy.io.fits.Header:
84 md = PropertyList.from_astropy_header(self.get_full_header())
85 stripVisitInfoKeywords(md)
86 for key in [ # extra cards stripped in ExposureFitsReader
87 "MJD-OBS",
88 "DATE-OBS",
89 "DETNAME",
90 "DETSER",
91 "EXPID",
92 ]:
93 md.pop(key, ...)
94 return md.to_astropy_header()
96 @cached_property
97 def true_obs_info(self) -> ObservationInfo:
98 """The ground-truth ObservationInfo (translated from raw)."""
99 return ObservationInfo.model_validate_json((OBS_INFO_DIR / self.obs_info_filename).read_text())
101 def make_record(self) -> Any:
102 """Build the `exposure` dimension record from the true
103 `ObservationInfo`.
104 """
105 return makeExposureRecordFromObsInfo(self.true_obs_info, DimensionUniverse())
107 @cached_property
108 def legacy_detector(self) -> LegacyDetector:
109 """The legacy `lsst.afw.cameraGeom.Detector` for this test case."""
110 assert self.true_obs_info.detector_num is not None
111 detector = read_archive(str(DETECTORS_DIR / self.detector_filename), Detector)
112 assert detector.id == self.true_obs_info.detector_num
113 return detector.to_legacy()
115 def make_stripped_exposure(self) -> LegacyExposureF:
116 """Build a test ``ExposureF`` with metadata content mimicking a legacy
117 visit image that has been roundtripped through FITS at least once.
118 """
119 header = self.get_stripped_header()
120 exposure = LegacyExposureF(16, 16)
121 if self.true_obs_info.detector_exposure_id is not None: 121 ↛ 123line 121 didn't jump to line 123 because the condition on line 121 was always true
122 exposure.info.setId(self.true_obs_info.detector_exposure_id)
123 exposure.setMetadata(PropertyList.from_astropy_header(header))
124 visit_info = MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.true_obs_info)
125 exposure.info.setVisitInfo(visit_info)
126 wcs = makeSkyWcs(
127 crpix=Point2D(8.0, 8.0),
128 crval=SpherePoint(42.0 * degrees, -11.0 * degrees), # arbitrary ICRS position; not asserted on
129 cdMatrix=makeCdMatrix(scale=0.2 * arcseconds),
130 )
131 exposure.setWcs(wcs)
132 exposure.setDetector(self.legacy_detector)
133 exposure.setPsf(LegacyGaussianPsf(11, 11, 2.0))
134 exposure.info.setFilter(
135 FilterLabel(band="g", physical=self.true_obs_info.physical_filter or "UNKNOWN")
136 )
137 return exposure
139 def make_embedded_exposure(self) -> LegacyExposureF:
140 """Build a test ``ExposureF`` whose metadata carries the true
141 `ObservationInfo` as flat ``OBS_INFO`` cards, as written by the raw
142 formatter, in addition to the stripped legacy header content.
143 """
144 exposure = self.make_stripped_exposure()
145 md = exposure.getMetadata()
146 md.update(self.true_obs_info.to_flat_dict(prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP))
147 return exposure
150# Fields that we can't recover from any of the inputs to obs_info_from_legacy,
151# mapped to what we get instead.
152UNRECOVERABLE_FIELDS = {
153 "translator_name": "VisitInfo",
154 "detector_name": None,
155 "detector_group": None,
156 "observing_day_offset": None,
157 "altaz_end": None,
158}
161def assert_obs_info_equal(
162 actual: ObservationInfo,
163 expected: ObservationInfo,
164 path: str,
165 *,
166 detector_serial: str | None,
167 detector_unique_name: str,
168) -> None:
169 """Assert that two ObservationInfo structs are as equal as they can be
170 after a from-legacy conversion.
171 """
172 for field in ObservationInfo.model_fields:
173 if field == "warnings": 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 continue
175 expected_value = getattr(expected, field)
176 if expected_value is None:
177 continue
178 actual_value = getattr(actual, field, None)
179 if field in UNRECOVERABLE_FIELDS:
180 if actual_value != UNRECOVERABLE_FIELDS[field]:
181 raise AssertionError(
182 f"{field} via {path}: expected fallback "
183 f"{UNRECOVERABLE_FIELDS[field]!r}, got {actual_value!r}"
184 )
185 continue
186 if field == "visit_id":
187 if actual_value != actual.exposure_id:
188 raise AssertionError(
189 f"visit_id via {path}: {actual_value!r} != exposure_id {actual.exposure_id!r}"
190 )
191 continue
192 if field == "boresight_rotation_coord":
193 if isinstance(expected_value, str) and isinstance(actual_value, str): 193 ↛ 199line 193 didn't jump to line 199 because the condition on line 193 was always true
194 if actual_value.casefold() != expected_value.casefold():
195 raise AssertionError(
196 f"{field} via {path}: {actual_value!r} != {expected_value!r} (ignoring case)"
197 )
198 continue
199 if field == "detector_serial":
200 expected_value = detector_serial
201 if field == "detector_unique_name":
202 expected_value = detector_unique_name
203 if actual_value is None:
204 raise AssertionError(f"{field} was lost via {path}: expected {expected_value}")
205 with annotate_errors(field):
206 assert_obs_metadata_fields_equal(actual_value, expected_value, label=field)
209def _point_to_astropy(point: Any) -> coords.SkyCoord:
210 """Convert an `lsst.geom.SpherePoint` to an Astropy ICRS `SkyCoord`."""
211 return coords.SkyCoord(
212 ra=point.getRa().asRadians() * u.rad,
213 dec=point.getDec().asRadians() * u.rad,
214 )
217def _point_to_altaz(point: Any) -> coords.AltAz:
218 """Convert an `lsst.geom.SpherePoint` conventionally holding az/alt to
219 an Astropy `AltAz` frame.
220 """
221 return coords.AltAz(
222 az=coords.Longitude(point.getLongitude().asRadians(), u.rad),
223 alt=coords.Latitude(point.getLatitude().asRadians(), u.rad),
224 )
227def assert_visit_info_equal(actual: LegacyVisitInfo, expected: LegacyVisitInfo) -> None:
228 """Assert that two VisitInfo structs are equal after a roundtrip
229 through `lsst.images.VisitImage`.
230 """
231 for name in (
232 "id",
233 "date",
234 "ut1",
235 "instrumentLabel",
236 "scienceProgram",
237 "observationType",
238 "observationReason",
239 "object",
240 "observatory",
241 "rotType",
242 "weather",
243 "exposureTime",
244 "darkTime",
245 "boresightAirmass",
246 "focusZ",
247 "era",
248 "localEra",
249 "boresightHourAngle",
250 "boresightParAngle",
251 "boresightRotAngle",
252 "hasSimulatedContent",
253 "isPersistable",
254 ):
255 with annotate_errors(name):
256 assert_obs_metadata_fields_equal(getattr(actual, name), getattr(expected, name), label=name)
257 with annotate_errors("boresightRaDec"):
258 assert_obs_metadata_fields_equal(
259 _point_to_astropy(actual.boresightRaDec),
260 _point_to_astropy(expected.boresightRaDec),
261 label="boresightRaDec",
262 )
263 with annotate_errors("boresightAzAlt"):
264 assert_obs_metadata_fields_equal(
265 _point_to_altaz(actual.boresightAzAlt),
266 _point_to_altaz(expected.boresightAzAlt),
267 label="boresightAzAlt",
268 )
271INSTRUMENT_CASES = [
272 InstrumentCase(
273 label="DECam",
274 test_data_filename="decam-c4d_150218_052850-N25.hdr",
275 obs_info_filename="decam-c4d_150218_052850-N25.json",
276 detector_filename="decam-c4d_150218_052850-N25.json",
277 ),
278 InstrumentCase(
279 label="HSC",
280 test_data_filename="hsc-HSCA90333426.hdr",
281 obs_info_filename="hsc-HSCA90333426.json",
282 detector_filename="hsc-HSCA90333426.json",
283 ),
284 InstrumentCase(
285 label="LSSTCam",
286 test_data_filename="lsstcam-MC_O_20250501_000292_R21_S11.hdr",
287 obs_info_filename="lsstcam-MC_O_20250501_000292_R21_S11.json",
288 detector_filename="lsstcam-MC_O_20250501_000292_R21_S11.json",
289 ),
290]
291INSTRUMENT_IDS = [case.label for case in INSTRUMENT_CASES]
293parametrize_instruments = pytest.mark.parametrize("instrument_case", INSTRUMENT_CASES, ids=INSTRUMENT_IDS)
296@skip_no_legacy
297@parametrize_instruments
298def test_from_and_read_legacy_preserve_obs_info(instrument_case: InstrumentCase) -> None:
299 """ObservationInfo fields must survive from_legacy on a stripped header."""
300 assert instrument_case.true_obs_info.instrument == instrument_case.label
301 assert instrument_case.true_obs_info.exposure_id is not None
302 exposure = instrument_case.make_stripped_exposure()
303 exposure_record = instrument_case.make_record()
304 with warnings.catch_warnings():
305 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
306 result1 = VisitImage.from_legacy(exposure, exposure_record=exposure_record)
307 assert_obs_info_equal(
308 result1.obs_info,
309 instrument_case.true_obs_info,
310 "VisitImage.from_legacy on an in-memory stripped header",
311 detector_serial=instrument_case.legacy_detector.getSerial(),
312 detector_unique_name=instrument_case.legacy_detector.getName(),
313 )
314 roundtrip1 = result1.to_legacy()
315 assert_visit_info_equal(roundtrip1.info.getVisitInfo(), exposure.info.getVisitInfo())
316 with TemporaryButler(legacy="ExposureF") as helper:
317 helper.butler.put(exposure, helper.legacy)
318 visit_image_ref = helper.legacy.overrideStorageClass("VisitImage")
319 with warnings.catch_warnings():
320 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
321 try:
322 result2 = helper.butler.get(visit_image_ref, parameters={"exposure_record": exposure_record})
323 except Exception as err:
324 pytest.fail(
325 f"butler.get as VisitImage raised {type(err).__name__}: {err} "
326 "-- every ObservationInfo field was lost."
327 )
328 roundtrip2 = result2.to_legacy()
329 assert_visit_info_equal(roundtrip2.info.getVisitInfo(), exposure.info.getVisitInfo())
330 assert_obs_info_equal(
331 result2.obs_info,
332 instrument_case.true_obs_info,
333 "butler.get of an ExposureF dataset as VisitImage",
334 detector_serial=instrument_case.legacy_detector.getSerial(),
335 detector_unique_name=instrument_case.legacy_detector.getName(),
336 )
337 with TemporaryButler(legacy="ExposureF") as helper:
338 helper.butler.put(exposure, helper.legacy)
339 visit_image_ref = helper.legacy.overrideStorageClass("VisitImage")
340 with pytest.raises(ValueError, match="exposure_record"):
341 helper.butler.get(visit_image_ref)
344def _assert_obs_info_not_in_header(visit_image: VisitImage) -> None:
345 """Assert that no flat ``OBS_INFO`` cards leaked into the opaque primary
346 header attached to a converted image.
347 """
348 header = visit_image._opaque_metadata.headers[ExtensionKey()]
349 assert not ObservationInfo.has_flat_dict(header, prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP)
350 leaked = [key for key in header if key.startswith(FITS_FLAT_DICT_PREFIX)]
351 assert not leaked, leaked
354@skip_no_legacy
355@parametrize_instruments
356def test_from_legacy_uses_embedded_obs_info(
357 instrument_case: InstrumentCase, caplog: pytest.LogCaptureFixture
358) -> None:
359 """Embedded OBS_INFO cards are used verbatim, without a record, with no
360 degradation warning, and are stripped from the attached header.
361 """
362 exposure = instrument_case.make_embedded_exposure()
363 with caplog.at_level(logging.WARNING, logger="lsst.images"), warnings.catch_warnings():
364 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
365 result = VisitImage.from_legacy(exposure, exposure_record=None)
366 assert result.obs_info == instrument_case.true_obs_info
367 degraded = [r.getMessage() for r in caplog.records if "exposure dimension record" in r.getMessage()]
368 assert not degraded, degraded
369 _assert_obs_info_not_in_header(result)
370 # The exposure passed in is not modified.
371 assert ObservationInfo.has_flat_dict(
372 exposure.getMetadata(), prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP
373 )
376@skip_no_legacy
377@parametrize_instruments
378def test_read_legacy_uses_embedded_obs_info(instrument_case: InstrumentCase, tmp_path: Path) -> None:
379 """Embedded OBS_INFO cards survive a legacy FITS write and are used by
380 read_legacy, both for the full image and the obs_info component.
381 """
382 exposure = instrument_case.make_embedded_exposure()
383 filename = str(tmp_path / "embedded.fits")
384 exposure.writeFits(filename)
385 with warnings.catch_warnings():
386 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
387 result = VisitImage.read_legacy(filename, exposure_record=None)
388 component = VisitImage.read_legacy(filename, exposure_record=None, component="obs_info")
389 assert result.obs_info == instrument_case.true_obs_info
390 assert component == instrument_case.true_obs_info
391 _assert_obs_info_not_in_header(result)
394@skip_no_legacy
395@parametrize_instruments
396def test_butler_get_uses_embedded_obs_info(instrument_case: InstrumentCase) -> None:
397 """A butler read of an ExposureF dataset as VisitImage uses the embedded
398 OBS_INFO cards even though the exposure record must still be passed.
399 """
400 exposure = instrument_case.make_embedded_exposure()
401 exposure_record = instrument_case.make_record()
402 with TemporaryButler(legacy="ExposureF") as helper:
403 helper.butler.put(exposure, helper.legacy)
404 visit_image_ref = helper.legacy.overrideStorageClass("VisitImage")
405 with warnings.catch_warnings():
406 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning)
407 result = helper.butler.get(visit_image_ref, parameters={"exposure_record": exposure_record})
408 assert result.obs_info == instrument_case.true_obs_info
409 _assert_obs_info_not_in_header(result)
412@skip_no_legacy
413@parametrize_instruments
414def test_embedded_obs_info_rejects_mismatched_record(instrument_case: InstrumentCase) -> None:
415 """An exposure record for a different exposure than the embedded
416 ObservationInfo describes is an error.
417 """
418 exposure = instrument_case.make_embedded_exposure()
419 assert instrument_case.true_obs_info.exposure_id is not None
420 with instrument_case.true_obs_info.edit_copy() as other_obs_info:
421 other_obs_info.exposure_id = instrument_case.true_obs_info.exposure_id + 1
422 other_record = makeExposureRecordFromObsInfo(other_obs_info, DimensionUniverse())
423 with pytest.raises(ValueError, match="embedded ObservationInfo"):
424 VisitImage.from_legacy(exposure, exposure_record=other_record)