Coverage for python/lsst/obs/base/_fitsRawFormatterBase.py: 71%
162 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 09:07 +0000
1# This file is part of obs_base.
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# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <http://www.gnu.org/licenses/>.
22from __future__ import annotations
24__all__ = ("FitsRawFormatterBase",)
26import copy
27import logging
28from abc import abstractmethod
29from typing import TYPE_CHECKING, Any, ClassVar, Self
31from astro_metadata_translator import FITS_FLAT_DICT_PREFIX, FITS_FLAT_DICT_SEP, ObservationInfo, fix_header
33import lsst.afw.fits
34import lsst.afw.geom
35import lsst.afw.image
36from lsst.daf.butler import FileDescriptor, FormatterNotImplementedError, Location, StorageClass
37from lsst.resources import ResourcePath
38from lsst.utils.classes import cached_getter
40from .formatters.fitsExposure import FitsImageFormatterBase, standardizeAmplifierParameters
41from .makeRawVisitInfoViaObsInfo import MakeRawVisitInfoViaObsInfo
42from .utils import InitialSkyWcsError, createInitialSkyWcsFromBoresight
44if TYPE_CHECKING:
45 import astro_metadata_translator
47 import lsst.daf.base
49 from .filters import FilterDefinitionCollection
51log = logging.getLogger(__name__)
54class FitsRawFormatterBase(FitsImageFormatterBase):
55 """Abstract base class for reading and writing raw data to and from
56 FITS files.
57 """
59 ReaderClass = lsst.afw.image.ImageFitsReader
61 wcsFlipX: ClassVar[bool] = False
62 """Control whether the WCS is flipped in the X-direction (`bool`).
63 """
65 def __init__(self, *args: Any, **kwargs: Any) -> None:
66 super().__init__(*args, **kwargs)
67 self._metadata: lsst.daf.base.PropertyList | None = None
68 self._observationInfo: ObservationInfo | None = None
70 @classmethod
71 def fromMetadata(
72 cls,
73 metadata: lsst.daf.base.PropertyList,
74 obsInfo: ObservationInfo | None = None,
75 storageClass: StorageClass | None = None,
76 location: Location | None = None,
77 ) -> Self:
78 """Construct a possibly-limited formatter from known metadata.
80 Parameters
81 ----------
82 metadata : `lsst.daf.base.PropertyList`
83 Raw header metadata, with any fixes (see
84 `astro_metadata_translator.fix_header`) applied but nothing
85 stripped.
86 obsInfo : `astro_metadata_translator.ObservationInfo`, optional
87 Structured information already extracted from ``metadata``.
88 If not provided, will be read from ``metadata`` on first use.
89 storageClass : `lsst.daf.butler.StorageClass`, optional
90 StorageClass for this file. If not provided, the formatter will
91 only support `makeWcs`, `makeVisitInfo`, `makeFilterLabel`, and
92 other operations that operate purely on metadata and not the actual
93 file.
94 location : `lsst.daf.butler.Location`, optional.
95 Location of the file. If not provided, the formatter will only
96 support `makeWcs`, `makeVisitInfo`, `makeFilterLabel`, and other
97 operations that operate purely on metadata and not the actual file.
99 Returns
100 -------
101 formatter : `FitsRawFormatterBase`
102 An instance of ``cls``.
103 """
104 self = cls(
105 FileDescriptor(
106 location if location is not None else Location("", "<unspecified>"),
107 storageClass if storageClass is not None else StorageClass(),
108 )
109 )
110 self._metadata = metadata
111 self._observationInfo = obsInfo
112 return self
114 @property
115 @abstractmethod
116 def translatorClass(self) -> type[astro_metadata_translator.MetadataTranslator] | None:
117 """`~astro_metadata_translator.MetadataTranslator` to translate
118 metadata header to `~astro_metadata_translator.ObservationInfo`.
119 """
120 return None
122 @property
123 @abstractmethod
124 def filterDefinitions(self) -> FilterDefinitionCollection | None:
125 """`~lsst.obs.base.FilterDefinitionCollection`, defining the filters
126 for this instrument.
127 """
128 return None
130 @property
131 @cached_getter
132 def checked_parameters(self) -> dict[str, Any]:
133 # Docstring inherited.
134 parameters = super().checked_parameters
135 if "bbox" in parameters: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true
136 raise TypeError(
137 "Raw formatters do not support reading arbitrary subimages, as some "
138 "implementations may be assembled on-the-fly."
139 )
140 return parameters
142 def readImage(self) -> lsst.afw.image.Image:
143 """Read just the image component of the Exposure.
145 Returns
146 -------
147 image : `~lsst.afw.image.Image`
148 In-memory image component.
149 """
150 return lsst.afw.image.ImageU(self._reader_path)
152 def isOnSky(self) -> bool:
153 """Boolean to determine if the exposure is thought to be on the sky.
155 Returns
156 -------
157 onSky : `bool`
158 Returns `True` if the observation looks like it was taken on the
159 sky. Returns `False` if this observation looks like a calibration
160 observation.
162 Notes
163 -----
164 If there is tracking RA/Dec information associated with the
165 observation it is assumed that the observation is on sky.
166 Currently the observation type is not checked.
167 """
168 if self.observationInfo.tracking_radec is None: 168 ↛ 169line 168 didn't jump to line 169 because the condition on line 168 was never true
169 return False
170 return True
172 @property
173 def metadata(self) -> lsst.daf.base.PropertyList:
174 """The metadata read from this file. It will be stripped as
175 components are extracted from it
176 (`lsst.daf.base.PropertyList`).
177 """
178 if self._metadata is None:
179 self._metadata = self.readMetadata()
180 return self._metadata
182 def readMetadata(self) -> lsst.daf.base.PropertyList:
183 """Read all header metadata directly into a PropertyList.
185 Returns
186 -------
187 metadata : `~lsst.daf.base.PropertyList`
188 Header metadata.
189 """
190 md = self.reader.readMetadata()
191 translatorClass = self.translatorClass
192 assert translatorClass is not None
193 fix_header(md, translator_class=translatorClass)
194 return md
196 def stripMetadata(self) -> None:
197 """Remove metadata entries that are parsed into components."""
198 try:
199 lsst.afw.geom.stripWcsMetadata(self.metadata)
200 except TypeError as e:
201 log.debug("Error caught and ignored while stripping metadata: %s", e.args[0])
203 def makeVisitInfo(self) -> lsst.afw.image.VisitInfo:
204 """Construct a VisitInfo from metadata.
206 Returns
207 -------
208 visitInfo : `~lsst.afw.image.VisitInfo`
209 Structured metadata about the observation.
210 """
211 visit_info = MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.observationInfo)
212 if self.data_id and "exposure" in self.data_id: 212 ↛ 221line 212 didn't jump to line 221 because the condition on line 212 was always true
213 # Special case exposure existing for this dataset type.
214 # Want to ensure that the id stored in the visitInfo matches that
215 # from the dataId from butler, regardless of what may have come
216 # from the ObservationInfo. In some edge cases they might differ.
217 exposure_id = self.data_id["exposure"]
218 if exposure_id != visit_info.id: 218 ↛ 221line 218 didn't jump to line 221 because the condition on line 218 was always true
219 visit_info = visit_info.copyWith(id=exposure_id)
221 return visit_info
223 @abstractmethod
224 def getDetector(self, id: int) -> lsst.afw.cameraGeom.Detector:
225 """Return the detector that acquired this raw exposure.
227 Parameters
228 ----------
229 id : `int`
230 The identifying number of the detector to get.
232 Returns
233 -------
234 detector : `~lsst.afw.cameraGeom.Detector`
235 The detector associated with that ``id``.
236 """
237 raise NotImplementedError("Must be implemented by subclasses.")
239 def makeWcs(
240 self, visitInfo: lsst.afw.image.VisitInfo, detector: lsst.afw.cameraGeom.Detector
241 ) -> lsst.afw.geom.SkyWcs:
242 """Create a SkyWcs from information about the exposure.
244 If VisitInfo is not None, use it and the detector to create a SkyWcs,
245 otherwise return the metadata-based SkyWcs (always created, so that
246 the relevant metadata keywords are stripped).
248 Parameters
249 ----------
250 visitInfo : `~lsst.afw.image.VisitInfo`
251 The information about the telescope boresight and camera
252 orientation angle for this exposure.
253 detector : `~lsst.afw.cameraGeom.Detector`
254 The detector used to acquire this exposure.
256 Returns
257 -------
258 skyWcs : `~lsst.afw.geom.SkyWcs`
259 Reversible mapping from pixel coordinates to sky coordinates.
261 Raises
262 ------
263 InitialSkyWcsError
264 Raised if there is an error generating the SkyWcs, chained from the
265 lower-level exception if available.
266 """
267 if not self.isOnSky(): 267 ↛ 269line 267 didn't jump to line 269 because the condition on line 267 was never true
268 # This is not an on-sky observation
269 return None
271 if visitInfo is None:
272 msg = "No VisitInfo; cannot access boresight information. Defaulting to metadata-based SkyWcs."
273 log.warning(msg)
274 skyWcs = self._createSkyWcsFromMetadata()
275 if skyWcs is None:
276 raise InitialSkyWcsError(
277 "Failed to create both metadata and boresight-based SkyWcs."
278 "See warnings in log messages for details."
279 )
280 return skyWcs
282 return self.makeRawSkyWcsFromBoresight(
283 visitInfo.getBoresightRaDec(), visitInfo.getBoresightRotAngle(), detector
284 )
286 @classmethod
287 def makeRawSkyWcsFromBoresight(
288 cls,
289 boresight: lsst.geom.SpherePoint,
290 orientation: lsst.geom.Angle,
291 detector: lsst.afw.cameraGeom.Detector,
292 ) -> lsst.afw.geom.SkyWcs:
293 """Class method to make a raw sky WCS from boresight and detector.
295 Parameters
296 ----------
297 boresight : `lsst.geom.SpherePoint`
298 The ICRS boresight RA/Dec
299 orientation : `lsst.geom.Angle`
300 The rotation angle of the focal plane on the sky.
301 detector : `lsst.afw.cameraGeom.Detector`
302 Where to get the camera geometry from.
304 Returns
305 -------
306 skyWcs : `~lsst.afw.geom.SkyWcs`
307 Reversible mapping from pixel coordinates to sky coordinates.
308 """
309 return createInitialSkyWcsFromBoresight(boresight, orientation, detector, flipX=cls.wcsFlipX)
311 def _createSkyWcsFromMetadata(self) -> lsst.afw.geom.SkyWcs | None:
312 """Create a SkyWcs from the FITS header metadata in an Exposure.
314 Returns
315 -------
316 skyWcs: `lsst.afw.geom.SkyWcs`, or None
317 The WCS that was created from ``self.metadata``, or None if that
318 creation fails due to invalid metadata.
319 """
320 if not self.isOnSky(): 320 ↛ 322line 320 didn't jump to line 322 because the condition on line 320 was never true
321 # This is not an on-sky observation
322 return None
324 try:
325 return lsst.afw.geom.makeSkyWcs(self.metadata, strip=True)
326 except TypeError as e:
327 log.warning("Cannot create a valid WCS from metadata: %s", e.args[0])
328 return None
330 def makeFilterLabel(self) -> lsst.afw.image.FilterLabel:
331 """Construct a FilterLabel from metadata.
333 Returns
334 -------
335 filter : `~lsst.afw.image.FilterLabel`
336 Object that identifies the filter for this image.
337 """
338 physical = self.observationInfo.physical_filter
339 assert physical is not None
340 filter_definitions = self.filterDefinitions
341 assert filter_definitions is not None
342 band = filter_definitions.physical_to_band[physical]
343 return lsst.afw.image.FilterLabel(physical=physical, band=band)
345 def readComponent(self, component: str) -> Any:
346 # Docstring inherited.
347 _ = self.checked_parameters # just for checking; no supported parameters.
348 if component == "image":
349 return self.readImage()
350 elif component == "filter":
351 return self.makeFilterLabel()
352 elif component == "visitInfo":
353 return self.makeVisitInfo()
354 elif component == "detector":
355 assert self.observationInfo.detector_num is not None
356 return self.getDetector(self.observationInfo.detector_num)
357 elif component == "wcs":
358 assert self.observationInfo.detector_num is not None
359 detector = self.getDetector(self.observationInfo.detector_num)
360 visitInfo = self.makeVisitInfo()
361 return self.makeWcs(visitInfo, detector)
362 elif component == "metadata":
363 # Translate the full header before stripping removes cards
364 # that the translator uses (such as RADESYS).
365 _ = self.observationInfo
366 self.stripMetadata()
367 return self.addObsInfoToMetadata()
368 return None
370 def readFull(self) -> Any:
371 # Docstring inherited.
372 assert self.observationInfo.detector_num is not None
373 amplifier, detector, _ = standardizeAmplifierParameters(
374 self.checked_parameters,
375 self.getDetector(self.observationInfo.detector_num),
376 )
377 if amplifier is not None: 377 ↛ 389line 377 didn't jump to line 389 because the condition on line 377 was always true
378 amplifier_isolator = lsst.afw.cameraGeom.AmplifierIsolator(
379 amplifier,
380 self.reader.readBBox(),
381 detector,
382 )
383 subimage = amplifier_isolator.transform_subimage(
384 self.reader.read(bbox=amplifier_isolator.subimage_bbox)
385 )
386 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(subimage))
387 exposure.setDetector(amplifier_isolator.make_detector())
388 else:
389 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(self.readImage()))
390 exposure.setDetector(detector)
391 self.attachComponentsFromMetadata(exposure)
392 return exposure
394 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
395 raise FormatterNotImplementedError("Raw data cannot be `put`.")
397 @property
398 def observationInfo(self) -> ObservationInfo:
399 """The `~astro_metadata_translator.ObservationInfo` extracted from
400 this file's metadata (`~astro_metadata_translator.ObservationInfo`,
401 read-only).
402 """
403 if self._observationInfo is None:
404 # Use the primary path rather than any local variant that the
405 # formatter might be using.
406 location = self.file_descriptor.location
407 path = location.path if location is not None else None
408 self._observationInfo = ObservationInfo(
409 self.metadata, translator_class=self.translatorClass, filename=path
410 )
411 return self._observationInfo
413 def addObsInfoToMetadata(self) -> lsst.daf.base.PropertyList:
414 """Add serialized ObservationInfo to metadata and return it.
416 Returns
417 -------
418 md : `lsst.daf.base.PropertyList`
419 Raw metadata with `~astro_metadata_translator.ObservationInfo`
420 headers embedded. This is a copy so the original metadata is
421 not affected.
422 """
423 obs_info_md = self.observationInfo.to_flat_dict(
424 prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP, encode_non_finite=True
425 )
426 md = copy.deepcopy(self.metadata)
427 md.update(obs_info_md)
428 return md
430 def attachComponentsFromMetadata(self, exposure: lsst.afw.image.Exposure) -> None:
431 """Attach all `lsst.afw.image.Exposure` components derived from
432 metadata (including the stripped metadata itself).
434 Parameters
435 ----------
436 exposure : `lsst.afw.image.Exposure`
437 Exposure to attach components to (modified in place). Must already
438 have a detector attached.
439 """
440 info = exposure.getInfo()
441 info.id = self.observationInfo.detector_exposure_id
442 info.setFilter(self.makeFilterLabel())
443 info.setVisitInfo(self.makeVisitInfo())
444 info.setWcs(self.makeWcs(info.getVisitInfo(), info.getDetector()))
446 self.stripMetadata()
447 exposure.setMetadata(self.addObsInfoToMetadata())