Coverage for python / lsst / obs / base / _fitsRawFormatterBase.py: 28%
155 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-26 09:02 +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 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`, `makeFilterLabel`, and
91 other operations that operate purely on metadata and not the actual
92 file.
93 location : `lsst.daf.butler.Location`, optional.
94 Location of the file. If not provided, the formatter will only
95 support `makeWcs`, `makeVisitInfo`, `makeFilterLabel`, and other
96 operations that operate purely on metadata and not the actual file.
98 Returns
99 -------
100 formatter : `FitsRawFormatterBase`
101 An instance of ``cls``.
102 """
103 self = cls(
104 FileDescriptor(
105 location if location is not None else Location("", "<unspecified>"),
106 storageClass if storageClass is not None else StorageClass(),
107 )
108 )
109 self._metadata = metadata
110 self._observationInfo = obsInfo
111 return self
113 @property
114 @abstractmethod
115 def translatorClass(self) -> type[astro_metadata_translator.MetadataTranslator] | None:
116 """`~astro_metadata_translator.MetadataTranslator` to translate
117 metadata header to `~astro_metadata_translator.ObservationInfo`.
118 """
119 return None
121 @property
122 @abstractmethod
123 def filterDefinitions(self) -> FilterDefinitionCollection | None:
124 """`~lsst.obs.base.FilterDefinitionCollection`, defining the filters
125 for this instrument.
126 """
127 return None
129 @property
130 @cached_getter
131 def checked_parameters(self) -> dict[str, Any]:
132 # Docstring inherited.
133 parameters = super().checked_parameters
134 if "bbox" in parameters:
135 raise TypeError(
136 "Raw formatters do not support reading arbitrary subimages, as some "
137 "implementations may be assembled on-the-fly."
138 )
139 return parameters
141 def readImage(self) -> lsst.afw.image.Image:
142 """Read just the image component of the Exposure.
144 Returns
145 -------
146 image : `~lsst.afw.image.Image`
147 In-memory image component.
148 """
149 return lsst.afw.image.ImageU(self._reader_path)
151 def isOnSky(self) -> bool:
152 """Boolean to determine if the exposure is thought to be on the sky.
154 Returns
155 -------
156 onSky : `bool`
157 Returns `True` if the observation looks like it was taken on the
158 sky. Returns `False` if this observation looks like a calibration
159 observation.
161 Notes
162 -----
163 If there is tracking RA/Dec information associated with the
164 observation it is assumed that the observation is on sky.
165 Currently the observation type is not checked.
166 """
167 if self.observationInfo.tracking_radec is None:
168 return False
169 return True
171 @property
172 def metadata(self) -> lsst.daf.base.PropertyList:
173 """The metadata read from this file. It will be stripped as
174 components are extracted from it
175 (`lsst.daf.base.PropertyList`).
176 """
177 if self._metadata is None:
178 self._metadata = self.readMetadata()
179 return self._metadata
181 def readMetadata(self) -> lsst.daf.base.PropertyList:
182 """Read all header metadata directly into a PropertyList.
184 Returns
185 -------
186 metadata : `~lsst.daf.base.PropertyList`
187 Header metadata.
188 """
189 md = self.reader.readMetadata()
190 translatorClass = self.translatorClass
191 assert translatorClass is not None
192 fix_header(md, translator_class=translatorClass)
193 return md
195 def stripMetadata(self) -> None:
196 """Remove metadata entries that are parsed into components."""
197 try:
198 lsst.afw.geom.stripWcsMetadata(self.metadata)
199 except TypeError as e:
200 log.debug("Error caught and ignored while stripping metadata: %s", e.args[0])
202 def makeVisitInfo(self) -> lsst.afw.image.VisitInfo:
203 """Construct a VisitInfo from metadata.
205 Returns
206 -------
207 visitInfo : `~lsst.afw.image.VisitInfo`
208 Structured metadata about the observation.
209 """
210 visit_info = MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.observationInfo)
211 if self.data_id and "exposure" in self.data_id:
212 # Special case exposure existing for this dataset type.
213 # Want to ensure that the id stored in the visitInfo matches that
214 # from the dataId from butler, regardless of what may have come
215 # from the ObservationInfo. In some edge cases they might differ.
216 exposure_id = self.data_id["exposure"]
217 if exposure_id != visit_info.id:
218 visit_info = visit_info.copyWith(id=exposure_id)
220 return visit_info
222 @abstractmethod
223 def getDetector(self, id: int) -> lsst.afw.cameraGeom.Detector:
224 """Return the detector that acquired this raw exposure.
226 Parameters
227 ----------
228 id : `int`
229 The identifying number of the detector to get.
231 Returns
232 -------
233 detector : `~lsst.afw.cameraGeom.Detector`
234 The detector associated with that ``id``.
235 """
236 raise NotImplementedError("Must be implemented by subclasses.")
238 def makeWcs(
239 self, visitInfo: lsst.afw.image.VisitInfo, detector: lsst.afw.cameraGeom.Detector
240 ) -> lsst.afw.geom.SkyWcs:
241 """Create a SkyWcs from information about the exposure.
243 If VisitInfo is not None, use it and the detector to create a SkyWcs,
244 otherwise return the metadata-based SkyWcs (always created, so that
245 the relevant metadata keywords are stripped).
247 Parameters
248 ----------
249 visitInfo : `~lsst.afw.image.VisitInfo`
250 The information about the telescope boresight and camera
251 orientation angle for this exposure.
252 detector : `~lsst.afw.cameraGeom.Detector`
253 The detector used to acquire this exposure.
255 Returns
256 -------
257 skyWcs : `~lsst.afw.geom.SkyWcs`
258 Reversible mapping from pixel coordinates to sky coordinates.
260 Raises
261 ------
262 InitialSkyWcsError
263 Raised if there is an error generating the SkyWcs, chained from the
264 lower-level exception if available.
265 """
266 if not self.isOnSky():
267 # This is not an on-sky observation
268 return None
270 if visitInfo is None:
271 msg = "No VisitInfo; cannot access boresight information. Defaulting to metadata-based SkyWcs."
272 log.warning(msg)
273 skyWcs = self._createSkyWcsFromMetadata()
274 if skyWcs is None:
275 raise InitialSkyWcsError(
276 "Failed to create both metadata and boresight-based SkyWcs."
277 "See warnings in log messages for details."
278 )
279 return skyWcs
281 return self.makeRawSkyWcsFromBoresight(
282 visitInfo.getBoresightRaDec(), visitInfo.getBoresightRotAngle(), detector
283 )
285 @classmethod
286 def makeRawSkyWcsFromBoresight(
287 cls,
288 boresight: lsst.geom.SpherePoint,
289 orientation: lsst.geom.Angle,
290 detector: lsst.afw.cameraGeom.Detector,
291 ) -> lsst.afw.geom.SkyWcs:
292 """Class method to make a raw sky WCS from boresight and detector.
294 Parameters
295 ----------
296 boresight : `lsst.geom.SpherePoint`
297 The ICRS boresight RA/Dec
298 orientation : `lsst.geom.Angle`
299 The rotation angle of the focal plane on the sky.
300 detector : `lsst.afw.cameraGeom.Detector`
301 Where to get the camera geometry from.
303 Returns
304 -------
305 skyWcs : `~lsst.afw.geom.SkyWcs`
306 Reversible mapping from pixel coordinates to sky coordinates.
307 """
308 return createInitialSkyWcsFromBoresight(boresight, orientation, detector, flipX=cls.wcsFlipX)
310 def _createSkyWcsFromMetadata(self) -> lsst.afw.geom.SkyWcs | None:
311 """Create a SkyWcs from the FITS header metadata in an Exposure.
313 Returns
314 -------
315 skyWcs: `lsst.afw.geom.SkyWcs`, or None
316 The WCS that was created from ``self.metadata``, or None if that
317 creation fails due to invalid metadata.
318 """
319 if not self.isOnSky():
320 # This is not an on-sky observation
321 return None
323 try:
324 return lsst.afw.geom.makeSkyWcs(self.metadata, strip=True)
325 except TypeError as e:
326 log.warning("Cannot create a valid WCS from metadata: %s", e.args[0])
327 return None
329 def makeFilterLabel(self) -> lsst.afw.image.FilterLabel:
330 """Construct a FilterLabel from metadata.
332 Returns
333 -------
334 filter : `~lsst.afw.image.FilterLabel`
335 Object that identifies the filter for this image.
336 """
337 physical = self.observationInfo.physical_filter
338 assert physical is not None
339 filter_definitions = self.filterDefinitions
340 assert filter_definitions is not None
341 band = filter_definitions.physical_to_band[physical]
342 return lsst.afw.image.FilterLabel(physical=physical, band=band)
344 def readComponent(self, component: str) -> Any:
345 # Docstring inherited.
346 _ = self.checked_parameters # just for checking; no supported parameters.
347 if component == "image":
348 return self.readImage()
349 elif component == "filter":
350 return self.makeFilterLabel()
351 elif component == "visitInfo":
352 return self.makeVisitInfo()
353 elif component == "detector":
354 assert self.observationInfo.detector_num is not None
355 return self.getDetector(self.observationInfo.detector_num)
356 elif component == "wcs":
357 assert self.observationInfo.detector_num is not None
358 detector = self.getDetector(self.observationInfo.detector_num)
359 visitInfo = self.makeVisitInfo()
360 return self.makeWcs(visitInfo, detector)
361 elif component == "metadata":
362 self.stripMetadata()
363 return self.metadata
364 return None
366 def readFull(self) -> Any:
367 # Docstring inherited.
368 assert self.observationInfo.detector_num is not None
369 amplifier, detector, _ = standardizeAmplifierParameters(
370 self.checked_parameters,
371 self.getDetector(self.observationInfo.detector_num),
372 )
373 if amplifier is not None:
374 amplifier_isolator = lsst.afw.cameraGeom.AmplifierIsolator(
375 amplifier,
376 self.reader.readBBox(),
377 detector,
378 )
379 subimage = amplifier_isolator.transform_subimage(
380 self.reader.read(bbox=amplifier_isolator.subimage_bbox)
381 )
382 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(subimage))
383 exposure.setDetector(amplifier_isolator.make_detector())
384 else:
385 exposure = lsst.afw.image.makeExposure(lsst.afw.image.makeMaskedImage(self.readImage()))
386 exposure.setDetector(detector)
387 self.attachComponentsFromMetadata(exposure)
388 return exposure
390 def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
391 raise FormatterNotImplementedError("Raw data cannot be `put`.")
393 @property
394 def observationInfo(self) -> ObservationInfo:
395 """The `~astro_metadata_translator.ObservationInfo` extracted from
396 this file's metadata (`~astro_metadata_translator.ObservationInfo`,
397 read-only).
398 """
399 if self._observationInfo is None:
400 # Use the primary path rather than any local variant that the
401 # formatter might be using.
402 location = self.file_descriptor.location
403 path = location.path if location is not None else None
404 self._observationInfo = ObservationInfo(
405 self.metadata, translator_class=self.translatorClass, filename=path
406 )
407 return self._observationInfo
409 def attachComponentsFromMetadata(self, exposure: lsst.afw.image.Exposure) -> None:
410 """Attach all `lsst.afw.image.Exposure` components derived from
411 metadata (including the stripped metadata itself).
413 Parameters
414 ----------
415 exposure : `lsst.afw.image.Exposure`
416 Exposure to attach components to (modified in place). Must already
417 have a detector attached.
418 """
419 info = exposure.getInfo()
420 info.id = self.observationInfo.detector_exposure_id
421 info.setFilter(self.makeFilterLabel())
422 info.setVisitInfo(self.makeVisitInfo())
423 info.setWcs(self.makeWcs(info.getVisitInfo(), info.getDetector()))
425 self.stripMetadata()
426 exposure.setMetadata(self.metadata)