Coverage for python / lsst / obs / base / _fitsRawFormatterBase.py: 28%
151 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-31 01:09 -0800
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 logging
27from abc import abstractmethod
28from typing import TYPE_CHECKING, Any, ClassVar, Self
30from astro_metadata_translator import ObservationInfo, fix_header
32import lsst.afw.fits
33import lsst.afw.geom
34import lsst.afw.image
35from lsst.daf.butler import FileDescriptor, FormatterNotImplementedError, Location, StorageClass
36from lsst.resources import ResourcePath
37from lsst.utils.classes import cached_getter
39from .formatters.fitsExposure import FitsImageFormatterBase, standardizeAmplifierParameters
40from .makeRawVisitInfoViaObsInfo import MakeRawVisitInfoViaObsInfo
41from .utils import InitialSkyWcsError, createInitialSkyWcsFromBoresight
43if TYPE_CHECKING:
44 import astro_metadata_translator
46 import lsst.daf.base
48 from .filters import FilterDefinitionCollection
50log = logging.getLogger(__name__)
53class FitsRawFormatterBase(FitsImageFormatterBase):
54 """Abstract base class for reading and writing raw data to and from
55 FITS files.
56 """
58 ReaderClass = lsst.afw.image.ImageFitsReader
60 wcsFlipX: ClassVar[bool] = False
61 """Control whether the WCS is flipped in the X-direction (`bool`).
62 """
64 def __init__(self, *args: Any, **kwargs: Any) -> None:
65 super().__init__(*args, **kwargs)
66 self._metadata: lsst.daf.base.PropertyList | None = None
67 self._observationInfo: ObservationInfo | None = None
69 @classmethod
70 def fromMetadata(
71 cls,
72 metadata: lsst.daf.base.PropertyList,
73 obsInfo: ObservationInfo | None = None,
74 storageClass: StorageClass | None = None,
75 location: Location | None = None,
76 ) -> Self:
77 """Construct a possibly-limited formatter from known metadata.
79 Parameters
80 ----------
81 metadata : `lsst.daf.base.PropertyList`
82 Raw header metadata, with any fixes (see
83 `astro_metadata_translator.fix_header`) applied but nothing
84 stripped.
85 obsInfo : `astro_metadata_translator.ObservationInfo`, optional
86 Structured information already extracted from ``metadata``.
87 If not provided, will be read from ``metadata`` on first use.
88 storageClass : `lsst.daf.butler.StorageClass`, optional
89 StorageClass for this file. If not provided, the formatter will
90 only support `makeWcs`, `makeVisitInfo`, `makeFilter`, and other
91 operations that operate purely on metadata and not the actual file.
92 location : `lsst.daf.butler.Location`, optional.
93 Location of the file. If not provided, the formatter will only
94 support `makeWcs`, `makeVisitInfo`, `makeFilter`, and other
95 operations that operate purely on metadata and not the actual file.
97 Returns
98 -------
99 formatter : `FitsRawFormatterBase`
100 An instance of ``cls``.
101 """
102 self = cls(
103 FileDescriptor(
104 location if location is not None else Location("", "<unspecified>"),
105 storageClass if storageClass is not None else StorageClass(),
106 )
107 )
108 self._metadata = metadata
109 self._observationInfo = obsInfo
110 return self
112 @property
113 @abstractmethod
114 def translatorClass(self) -> type[astro_metadata_translator.MetadataTranslator] | None:
115 """`~astro_metadata_translator.MetadataTranslator` to translate
116 metadata header to `~astro_metadata_translator.ObservationInfo`.
117 """
118 return None
120 @property
121 @abstractmethod
122 def filterDefinitions(self) -> FilterDefinitionCollection | None:
123 """`~lsst.obs.base.FilterDefinitionCollection`, defining the filters
124 for this instrument.
125 """
126 return None
128 @property
129 @cached_getter
130 def checked_parameters(self) -> dict[str, Any]:
131 # Docstring inherited.
132 parameters = super().checked_parameters
133 if "bbox" in parameters:
134 raise TypeError(
135 "Raw formatters do not support reading arbitrary subimages, as some "
136 "implementations may be assembled on-the-fly."
137 )
138 return parameters
140 def readImage(self) -> lsst.afw.image.Image:
141 """Read just the image component of the Exposure.
143 Returns
144 -------
145 image : `~lsst.afw.image.Image`
146 In-memory image component.
147 """
148 return lsst.afw.image.ImageU(self._reader_path)
150 def isOnSky(self) -> bool:
151 """Boolean to determine if the exposure is thought to be on the sky.
153 Returns
154 -------
155 onSky : `bool`
156 Returns `True` if the observation looks like it was taken on the
157 sky. Returns `False` if this observation looks like a calibration
158 observation.
160 Notes
161 -----
162 If there is tracking RA/Dec information associated with the
163 observation it is assumed that the observation is on sky.
164 Currently the observation type is not checked.
165 """
166 if self.observationInfo.tracking_radec is None:
167 return False
168 return True
170 @property
171 def metadata(self) -> lsst.daf.base.PropertyList:
172 """The metadata read from this file. It will be stripped as
173 components are extracted from it
174 (`lsst.daf.base.PropertyList`).
175 """
176 if self._metadata is None:
177 self._metadata = self.readMetadata()
178 return self._metadata
180 def readMetadata(self) -> lsst.daf.base.PropertyList:
181 """Read all header metadata directly into a PropertyList.
183 Returns
184 -------
185 metadata : `~lsst.daf.base.PropertyList`
186 Header metadata.
187 """
188 md = self.reader.readMetadata()
189 translatorClass = self.translatorClass
190 assert translatorClass is not None
191 fix_header(md, translator_class=translatorClass)
192 return md
194 def stripMetadata(self) -> None:
195 """Remove metadata entries that are parsed into components."""
196 try:
197 lsst.afw.geom.stripWcsMetadata(self.metadata)
198 except TypeError as e:
199 log.debug("Error caught and ignored while stripping metadata: %s", e.args[0])
201 def makeVisitInfo(self) -> lsst.afw.image.VisitInfo:
202 """Construct a VisitInfo from metadata.
204 Returns
205 -------
206 visitInfo : `~lsst.afw.image.VisitInfo`
207 Structured metadata about the observation.
208 """
209 visit_info = MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.observationInfo)
210 if self.data_id and "exposure" in self.data_id:
211 # Special case exposure existing for this dataset type.
212 # Want to ensure that the id stored in the visitInfo matches that
213 # from the dataId from butler, regardless of what may have come
214 # from the ObservationInfo. In some edge cases they might differ.
215 exposure_id = self.data_id["exposure"]
216 if exposure_id != visit_info.id:
217 visit_info = visit_info.copyWith(id=exposure_id)
219 return visit_info
221 @abstractmethod
222 def getDetector(self, id: int) -> lsst.afw.cameraGeom.Detector:
223 """Return the detector that acquired this raw exposure.
225 Parameters
226 ----------
227 id : `int`
228 The identifying number of the detector to get.
230 Returns
231 -------
232 detector : `~lsst.afw.cameraGeom.Detector`
233 The detector associated with that ``id``.
234 """
235 raise NotImplementedError("Must be implemented by subclasses.")
237 def makeWcs(
238 self, visitInfo: lsst.afw.image.VisitInfo, detector: lsst.afw.cameraGeom.Detector
239 ) -> lsst.afw.geom.SkyWcs:
240 """Create a SkyWcs from information about the exposure.
242 If VisitInfo is not None, use it and the detector to create a SkyWcs,
243 otherwise return the metadata-based SkyWcs (always created, so that
244 the relevant metadata keywords are stripped).
246 Parameters
247 ----------
248 visitInfo : `~lsst.afw.image.VisitInfo`
249 The information about the telescope boresight and camera
250 orientation angle for this exposure.
251 detector : `~lsst.afw.cameraGeom.Detector`
252 The detector used to acquire this exposure.
254 Returns
255 -------
256 skyWcs : `~lsst.afw.geom.SkyWcs`
257 Reversible mapping from pixel coordinates to sky coordinates.
259 Raises
260 ------
261 InitialSkyWcsError
262 Raised if there is an error generating the SkyWcs, chained from the
263 lower-level exception if available.
264 """
265 if not self.isOnSky():
266 # This is not an on-sky observation
267 return None
269 if visitInfo is None:
270 msg = "No VisitInfo; cannot access boresight information. Defaulting to metadata-based SkyWcs."
271 log.warning(msg)
272 skyWcs = self._createSkyWcsFromMetadata()
273 if skyWcs is None:
274 raise InitialSkyWcsError(
275 "Failed to create both metadata and boresight-based SkyWcs."
276 "See warnings in log messages for details."
277 )
278 return skyWcs
280 return self.makeRawSkyWcsFromBoresight(
281 visitInfo.getBoresightRaDec(), visitInfo.getBoresightRotAngle(), detector
282 )
284 @classmethod
285 def makeRawSkyWcsFromBoresight(
286 cls,
287 boresight: lsst.geom.SpherePoint,
288 orientation: lsst.geom.Angle,
289 detector: lsst.afw.cameraGeom.Detector,
290 ) -> lsst.afw.geom.SkyWcs:
291 """Class method to make a raw sky WCS from boresight and detector.
293 Parameters
294 ----------
295 boresight : `lsst.geom.SpherePoint`
296 The ICRS boresight RA/Dec
297 orientation : `lsst.geom.Angle`
298 The rotation angle of the focal plane on the sky.
299 detector : `lsst.afw.cameraGeom.Detector`
300 Where to get the camera geometry from.
302 Returns
303 -------
304 skyWcs : `~lsst.afw.geom.SkyWcs`
305 Reversible mapping from pixel coordinates to sky coordinates.
306 """
307 return createInitialSkyWcsFromBoresight(boresight, orientation, detector, flipX=cls.wcsFlipX)
309 def _createSkyWcsFromMetadata(self) -> lsst.afw.geom.SkyWcs | None:
310 """Create a SkyWcs from the FITS header metadata in an Exposure.
312 Returns
313 -------
314 skyWcs: `lsst.afw.geom.SkyWcs`, or None
315 The WCS that was created from ``self.metadata``, or None if that
316 creation fails due to invalid metadata.
317 """
318 if not self.isOnSky():
319 # This is not an on-sky observation
320 return None
322 try:
323 return lsst.afw.geom.makeSkyWcs(self.metadata, strip=True)
324 except TypeError as e:
325 log.warning("Cannot create a valid WCS from metadata: %s", e.args[0])
326 return None
328 def makeFilterLabel(self) -> lsst.afw.image.FilterLabel:
329 """Construct a FilterLabel from metadata.
331 Returns
332 -------
333 filter : `~lsst.afw.image.FilterLabel`
334 Object that identifies the filter for this image.
335 """
336 physical = self.observationInfo.physical_filter
337 filter_definitions = self.filterDefinitions
338 assert filter_definitions is not None
339 band = filter_definitions.physical_to_band[physical]
340 return lsst.afw.image.FilterLabel(physical=physical, band=band)
342 def readComponent(self, component: str) -> Any:
343 # Docstring inherited.
344 _ = self.checked_parameters # just for checking; no supported parameters.
345 if component == "image":
346 return self.readImage()
347 elif component == "filter":
348 return self.makeFilterLabel()
349 elif component == "visitInfo":
350 return self.makeVisitInfo()
351 elif component == "detector":
352 return self.getDetector(self.observationInfo.detector_num)
353 elif component == "wcs":
354 detector = self.getDetector(self.observationInfo.detector_num)
355 visitInfo = self.makeVisitInfo()
356 return self.makeWcs(visitInfo, detector)
357 elif component == "metadata":
358 self.stripMetadata()
359 return self.metadata
360 return None
362 def readFull(self) -> Any:
363 # Docstring inherited.
364 amplifier, detector, _ = standardizeAmplifierParameters(
365 self.checked_parameters,
366 self.getDetector(self.observationInfo.detector_num),
367 )
368 if amplifier is not None:
369 amplifier_isolator = lsst.afw.cameraGeom.AmplifierIsolator(
370 amplifier,
371 self.reader.readBBox(),
372 detector,
373 )
374 subimage = amplifier_isolator.transform_subimage(
375 self.reader.read(bbox=amplifier_isolator.subimage_bbox)
376 )
377 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(subimage))
378 exposure.setDetector(amplifier_isolator.make_detector())
379 else:
380 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(self.readImage()))
381 exposure.setDetector(detector)
382 self.attachComponentsFromMetadata(exposure)
383 return exposure
385 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
386 raise FormatterNotImplementedError("Raw data cannot be `put`.")
388 @property
389 def observationInfo(self) -> ObservationInfo:
390 """The `~astro_metadata_translator.ObservationInfo` extracted from
391 this file's metadata (`~astro_metadata_translator.ObservationInfo`,
392 read-only).
393 """
394 if self._observationInfo is None:
395 # Use the primary path rather than any local variant that the
396 # formatter might be using.
397 location = self.file_descriptor.location
398 path = location.path if location is not None else None
399 self._observationInfo = ObservationInfo(
400 self.metadata, translator_class=self.translatorClass, filename=path
401 )
402 return self._observationInfo
404 def attachComponentsFromMetadata(self, exposure: lsst.afw.image.Exposure) -> None:
405 """Attach all `lsst.afw.image.Exposure` components derived from
406 metadata (including the stripped metadata itself).
408 Parameters
409 ----------
410 exposure : `lsst.afw.image.Exposure`
411 Exposure to attach components to (modified in place). Must already
412 have a detector attached.
413 """
414 info = exposure.getInfo()
415 info.id = self.observationInfo.detector_exposure_id
416 info.setFilter(self.makeFilterLabel())
417 info.setVisitInfo(self.makeVisitInfo())
418 info.setWcs(self.makeWcs(info.getVisitInfo(), info.getDetector()))
420 self.stripMetadata()
421 exposure.setMetadata(self.metadata)