Coverage for python/lsst/images/_visit_image.py: 47%
437 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-26 09:47 +0000
1# This file is part of lsst-images.
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# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12from __future__ import annotations
14__all__ = ("VisitImage", "VisitImageSerializationModel")
16import functools
17import logging
18import warnings
19from collections.abc import Callable, Mapping, MutableMapping
20from types import EllipsisType
21from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast
23import astropy.io.fits
24import astropy.units
25import numpy as np
26import pydantic
27from astro_metadata_translator import ObservationInfo, VisitInfoTranslator
29from ._backgrounds import BackgroundMap, BackgroundMapSerializationModel
30from ._concrete_bounds import BoundsSerializationModel
31from ._geom import Bounds, Box
32from ._image import Image, ImageSerializationModel
33from ._mask import Mask, MaskPlane, MaskSchema, MaskSerializationModel, get_legacy_visit_image_mask_planes
34from ._masked_image import MaskedImage, MaskedImageSerializationModel
35from ._observation_summary_stats import ObservationSummaryStats
36from ._polygon import Polygon
37from ._transforms import (
38 DetectorFrame,
39 SkyProjection,
40 SkyProjectionAstropyView,
41 SkyProjectionSerializationModel,
42)
43from .aperture_corrections import (
44 ApertureCorrectionMap,
45 ApertureCorrectionMapSerializationModel,
46 aperture_corrections_from_legacy,
47 aperture_corrections_to_legacy,
48)
49from .cameras import Detector, DetectorSerializationModel
50from .describe import DescribeOptions, FieldRole, Report, ReportField
51from .fields import BaseField, Field, FieldSerializationModel, field_from_legacy_photo_calib
52from .fits import FitsOpaqueMetadata
53from .psfs import (
54 GaussianPointSpreadFunction,
55 GaussianPSFSerializationModel,
56 LegacyPointSpreadFunction,
57 PiffSerializationModel,
58 PiffWrapper,
59 PointSpreadFunction,
60 PSFExSerializationModel,
61 PSFExWrapper,
62)
63from .serialization import ArchiveReadError, InputArchive, InvalidParameterError, MetadataValue, OutputArchive
64from .utils import is_none
66if TYPE_CHECKING:
67 try:
68 from lsst.afw.cameraGeom import Detector as LegacyDetector
69 from lsst.afw.image import Exposure as LegacyExposure
70 from lsst.afw.image import FilterLabel as LegacyFilterLabel
71 from lsst.afw.image import VisitInfo as LegacyVisitInfo
72 except ImportError:
73 type LegacyDetector = Any # type: ignore[no-redef]
74 type LegacyExposure = Any # type: ignore[no-redef]
75 type LegacyFilterLabel = Any # type: ignore[no-redef]
76 type LegacyVisitInfo = Any # type: ignore[no-redef]
78_LOG = logging.getLogger("lsst.images")
81class VisitImage(MaskedImage):
82 """A calibrated single-visit image.
84 Parameters
85 ----------
86 image
87 The main image plane. If this has a `SkyProjection`, it will be used
88 for all planes unless a ``sky_projection`` is passed separately.
89 mask
90 A bitmask image that annotates the main image plane. Must have the
91 same bounding box as ``image`` if provided. Any attached
92 ``sky_projection`` is replaced (possibly by `None`).
93 variance
94 The per-pixel uncertainty of the main image as an image of variance
95 values. Must have the same bounding box as ``image`` if provided, and
96 its units must be the square of ``image.unit`` or `None`.
97 Values default to ``1.0``. Any attached ``sky_projection`` is replaced
98 (possibly by `None`).
99 mask_schema
100 Schema for the mask plane. Must be provided if and only if ``mask`` is
101 not provided.
102 sky_projection
103 Projection that maps the pixel grid to the sky. Can only be `None` if
104 a ``sky_projection`` is already attached to ``image``.
105 bounds
106 The region where this image's pixels and other properties are valid.
107 If not provided, the bounding box of the image is used. Other
108 components (``psf``, ``sky_projection``, ``aperture_corrections``,
109 etc.) are assumed to have their own bounds which may or may not be the
110 same as the image bounds. If ``bounds`` extends beyond the image
111 bounding box, the intersection between ``bounds`` and the image
112 bounding box is used instead.
113 obs_info
114 General information about this visit in standardized form.
115 summary_stats
116 Summary statistics associated with this visit. Initialized to default
117 values if not provided.
118 photometric_scaling
119 Field that can be used to multiply a post-ISR image units to yield
120 calibrated image units. This may be a scaling that was already
121 applied (so dividing by it will recover the post-ISR units) or a
122 scaling that has not been applied, depending on ``image.unit``.
123 psf
124 Point-spread function model for this image, or an exception explaining
125 why it could not be read (to be raised if the PSF is requested later).
126 detector
127 Geometry and electronic information for the detector attached to this
128 image.
129 aperture_corrections : `dict` [`str`, `~fields.BaseField`]
130 Mapping from photometry algorithm name to the aperture correction for
131 that algorithm.
132 backgrounds
133 Background models associated with this image.
134 band
135 Name of the passband the image was observed with (this is a shorter,
136 less specific version of ``obs_info.physical_filter``).
137 metadata
138 Arbitrary flexible metadata to associate with the image.
139 """
141 def __init__(
142 self,
143 image: Image,
144 *,
145 mask: Mask | None = None,
146 variance: Image | None = None,
147 mask_schema: MaskSchema | None = None,
148 sky_projection: SkyProjection[DetectorFrame] | None = None,
149 bounds: Bounds | None = None,
150 obs_info: ObservationInfo | None = None,
151 summary_stats: ObservationSummaryStats | None = None,
152 photometric_scaling: Field | None = None,
153 psf: PointSpreadFunction | ArchiveReadError,
154 detector: Detector,
155 aperture_corrections: ApertureCorrectionMap | None = None,
156 backgrounds: BackgroundMap | None = None,
157 band: str,
158 metadata: dict[str, MetadataValue] | None = None,
159 ) -> None:
160 super().__init__(
161 image,
162 mask=mask,
163 variance=variance,
164 mask_schema=mask_schema,
165 sky_projection=sky_projection,
166 metadata=metadata,
167 )
168 if self.image.unit is None:
169 raise TypeError("The image component of a VisitImage must have units.")
170 if self.image.sky_projection is None:
171 raise TypeError("The sky_projection component of a VisitImage cannot be None.")
172 if obs_info is None:
173 raise TypeError("The observation info component of a VisitImage cannot be None.")
174 if obs_info.physical_filter is None: 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 raise ValueError("The obs_info.physical_filter attribute of a VisitImage cannot be None.")
176 self._obs_info = obs_info
177 if not isinstance(self.image.sky_projection.pixel_frame, DetectorFrame):
178 raise TypeError("The sky_projection's pixel frame must be a DetectorFrame for VisitImage.")
179 if summary_stats is None:
180 summary_stats = ObservationSummaryStats()
181 self._summary_stats = summary_stats
182 if photometric_scaling is not None and photometric_scaling.unit is None: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 raise TypeError("If a photometric_scaling is provided, it must have units.")
184 self._photometric_scaling = photometric_scaling
185 self._psf = psf
186 self._detector = detector
187 self._aperture_corrections = aperture_corrections if aperture_corrections is not None else {}
188 self._bounds = bounds if bounds is not None else self.bbox
189 if not self.bbox.contains(self._bounds.bbox):
190 self._bounds = self._bounds.intersection(self.bbox)
191 self._backgrounds = backgrounds if backgrounds is not None else BackgroundMap()
192 self._band = band
194 @property
195 def unit(self) -> astropy.units.UnitBase:
196 """The units of the image plane (`astropy.units.Unit`)."""
197 return cast(astropy.units.UnitBase, super().unit)
199 @property
200 def sky_projection(self) -> SkyProjection[DetectorFrame]:
201 """The projection that maps the pixel grid to the sky
202 (`SkyProjection` [`DetectorFrame`]).
203 """
204 return cast(SkyProjection[DetectorFrame], super().sky_projection)
206 @property
207 def bounds(self) -> Bounds:
208 """The region where pixels are valid (`Bounds`)."""
209 return self._bounds
211 @property
212 def obs_info(self) -> ObservationInfo:
213 """General information about this observation in standard form.
214 (`~astro_metadata_translator.ObservationInfo`).
215 """
216 return self._obs_info
218 @property
219 def physical_filter(self) -> str:
220 """Full name of the physical bandpass filter (`str`)."""
221 assert self._obs_info.physical_filter is not None, "Guaranteed at construction."
222 return self._obs_info.physical_filter
224 @property
225 def band(self) -> str:
226 """Short name of the bandpass filter (`str`)."""
227 return self._band
229 @property
230 def astropy_wcs(self) -> SkyProjectionAstropyView:
231 """An Astropy WCS for the pixel arrays (`SkyProjectionAstropyView`).
233 Notes
234 -----
235 As expected for Astropy WCS objects, this defines pixel coordinates
236 such that the first row and column in the arrays are ``(0, 0)``, not
237 ``bbox.start``, as is the case for `sky_projection`.
239 This object satisfies the `astropy.wcs.wcsapi.BaseHighLevelWCS` and
240 `astropy.wcs.wcsapi.BaseLowLevelWCS` interfaces, but it is not an
241 `astropy.wcs.WCS` (use `fits_wcs` for that).
242 """
243 return cast(SkyProjectionAstropyView, super().astropy_wcs)
245 @property
246 def summary_stats(self) -> ObservationSummaryStats:
247 """Optional summary statistics for this observation
248 (`ObservationSummaryStats`).
249 """
250 return self._summary_stats
252 @property
253 def photometric_scaling(self) -> Field | None:
254 """Field that multiplies a post-ISR image to yield the calibrated
255 image (~`fields.BaseField`).
256 """
257 return self._photometric_scaling
259 @photometric_scaling.setter
260 def photometric_scaling(self, value: Field) -> None:
261 if value.unit is None: 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 raise TypeError("The photometric_scaling for a VisitImage must have units.")
263 self._photometric_scaling = value
265 @property
266 def psf(self) -> PointSpreadFunction:
267 """The point-spread function model for this image
268 (`.psfs.PointSpreadFunction`).
269 """
270 if isinstance(self._psf, ArchiveReadError):
271 raise self._psf
272 return self._psf
274 @property
275 def detector(self) -> Detector:
276 """Geometry and electronic information about the detector
277 (`.cameras.Detector`).
278 """
279 return self._detector
281 @property
282 def aperture_corrections(self) -> ApertureCorrectionMap:
283 """A mapping from photometry algorithm name to the aperture correction
284 field for that algorithm (`dict` [`str`, `~.fields.BaseField`]).
285 """
286 return self._aperture_corrections
288 @property
289 def backgrounds(self) -> BackgroundMap:
290 """A mapping of backgrounds associated with this image
291 (`BackgroundMap`).
292 """
293 return self._backgrounds
295 def __getitem__(self, bbox: Box | EllipsisType) -> VisitImage:
296 bbox, _ = self._handle_getitem_args(bbox)
297 return self._transfer_metadata(
298 VisitImage(
299 self.image[bbox],
300 mask=self.mask[bbox],
301 variance=self.variance[bbox],
302 sky_projection=self.sky_projection,
303 psf=self._psf,
304 obs_info=self.obs_info,
305 bounds=self._bounds, # don't need to intersect here, because __init__ will do that.
306 summary_stats=self.summary_stats,
307 detector=self._detector,
308 photometric_scaling=self._photometric_scaling,
309 aperture_corrections=self.aperture_corrections,
310 backgrounds=self._backgrounds,
311 band=self._band,
312 ),
313 bbox=bbox,
314 )
316 def _summary_stats_describe_options(self, options: DescribeOptions) -> DescribeOptions:
317 """Return the options the summary-statistics child report is built
318 with.
320 Parameters
321 ----------
322 options : `DescribeOptions`
323 Options this image's own report is being built with.
325 Returns
326 -------
327 options : `DescribeOptions`
328 Options for the child report.
330 Notes
331 -----
332 `ObservationSummaryStats.raCorners` and
333 `~ObservationSummaryStats.decCorners` hold the sky corners of the
334 image the statistics were measured on, which the sky projection
335 tabulates more readably. Taking a cutout does not recompute the
336 statistics, so those corners still describe the original image and
337 are worth keeping wherever they no longer match the pixels on hand.
338 """
339 if self.bbox == self.detector.bbox:
340 return options.for_child("corners")
341 return options.for_child()
343 def _describe_psf(self, options: DescribeOptions) -> Report:
344 """Return the report for this image's PSF, or for the error that
345 stopped it being read.
347 Parameters
348 ----------
349 options : `DescribeOptions`
350 Options the child report is built with.
352 Returns
353 -------
354 report : `Report`
355 Report describing the PSF, or naming it unreadable.
357 Notes
358 -----
359 A PSF that could not be read is a supported state of a visit image,
360 not a failure of it, and it is often the only unreadable part: a PSF
361 model can depend on a package the reader does not have installed.
362 Describing it therefore has to say what is missing and why, rather
363 than raise and take the report of everything else with it.
364 """
365 if isinstance(self._psf, ArchiveReadError):
366 return Report(
367 type_name="PointSpreadFunction",
368 summary=f"unreadable ({self._psf})",
369 inline=True,
370 )
371 return self._psf._describe(options)
373 def _describe(self, options: DescribeOptions = DescribeOptions(), /) -> Report:
374 """Return a `Report` describing this visit image.
376 Parameters
377 ----------
378 options : `DescribeOptions`, optional
379 Rendering options; forwarded to all children. Child construction
380 can be expensive, so `DescribeOptions.brief` skips it.
381 """
382 # The image and mask schema are rendered as children below, and the
383 # image renders as ``Image(bbox, dtype)``, which would restate the
384 # shared bbox. Both are REPR_ONLY so only repr sees them.
385 fields = [
386 ReportField(
387 label="image",
388 value=self.image,
389 repr_value=repr(self.image),
390 positional=True,
391 role=FieldRole.REPR_ONLY,
392 ),
393 ReportField(
394 label="mask_schema",
395 value=self.mask.schema,
396 repr_value=repr(self.mask.schema),
397 role=FieldRole.REPR_ONLY,
398 ),
399 ReportField(label="band", value=self.band, role=FieldRole.DERIVED),
400 ReportField(label="physical_filter", value=self.physical_filter, role=FieldRole.DERIVED),
401 ReportField(label="bbox", value=self.bbox, repr_value=repr(self.bbox), role=FieldRole.DERIVED),
402 ]
403 summary = f"VisitImage({self.image!s}, {list(self.mask.schema.names)})"
404 if options.brief:
405 return Report(type_name="VisitImage", summary=summary, fields=fields)
406 child = options.for_child()
407 plane = options.for_child("sky_projection", "bbox")
408 children: dict[str, Report] = {
409 "image": self.image._describe(plane),
410 "mask": self.mask._describe(plane),
411 "variance": self.variance._describe(plane),
412 "sky_projection": self.sky_projection._describe(child, bbox=self.bbox),
413 "psf": self._describe_psf(child),
414 "detector": self.detector._describe(child),
415 "summary_stats": self.summary_stats._describe(self._summary_stats_describe_options(options)),
416 "backgrounds": self.backgrounds._describe(child),
417 }
418 if self.photometric_scaling is not None:
419 children["photometric_scaling"] = self.photometric_scaling._describe(child)
420 return Report(
421 type_name="VisitImage",
422 summary=summary,
423 fields=fields,
424 children=children,
425 )
427 def copy(self, *, copy_detector: bool = False) -> VisitImage:
428 """Deep-copy the visit image.
430 Parameters
431 ----------
432 copy_detector
433 Whether to deep-copy the `detector` attribute.
434 """
435 return self._transfer_metadata(
436 VisitImage(
437 image=self._image.copy(),
438 mask=self._mask.copy(),
439 variance=self._variance.copy(),
440 psf=self._psf,
441 obs_info=self.obs_info,
442 bounds=self._bounds,
443 summary_stats=self.summary_stats.model_copy(),
444 detector=self._detector.copy() if copy_detector else self._detector,
445 photometric_scaling=self._photometric_scaling,
446 aperture_corrections=self.aperture_corrections.copy(),
447 backgrounds=self._backgrounds.copy(),
448 band=self.band,
449 ),
450 copy=True,
451 )
453 def convert_unit(
454 self,
455 unit: astropy.units.UnitBase = astropy.units.nJy,
456 copy: Literal["as-needed"] | bool = True,
457 copy_detector: bool = False,
458 ) -> VisitImage:
459 """Return an equivalent image with different pixel units.
461 Parameters
462 ----------
463 unit
464 The unit to transform to. This may be any of the following:
466 - any unit directly relatable to the current units via Astropy;
467 - any unit relatable to the product of the current units with the
468 `photometric_scaling` (i.e. if the current image is in
469 instrumental units but we know how to calibrate them)
470 - any unit relatable to the quotient of the current units with the
471 `photometric_scaling` (i.e. if the current image is in
472 calibrated units and we want to revert back to instrumental
473 units).
474 copy
475 Whether to copy the images and other components. If `True`, all
476 components that aren't controlled by some other argument will
477 always be deep-copied. If `False`, the operation will fail if the
478 image is not already in the right units. If ``as-needed``, only
479 the image and variance will be copied, and only if they are not
480 already in the right units.
481 copy_detector
482 Whether to deep-copy the `detector` attribute.
484 Returns
485 -------
486 `VisitImage`
487 An image with the given units.
488 """
489 if copy not in (True, False, "as-needed"): 489 ↛ 490line 489 didn't jump to line 490 because the condition on line 489 was never true
490 raise TypeError(f"Invalid value for 'copy' parameter: {copy!r}.")
491 if (factor := _get_unit_conversion_factor(self.unit, unit)) is not None: 491 ↛ 492line 491 didn't jump to line 492 because the condition on line 491 was never true
492 if factor == 1.0:
493 if copy is True: # not "as-needed"
494 return self.copy()
495 else:
496 return self[...]
497 elif copy is False:
498 raise astropy.units.UnitConversionError(
499 f"Units must be converted ({self.unit} -> {unit}), but copy=False."
500 )
501 image = Image(
502 self._image.array * factor, bbox=self.bbox, sky_projection=self.sky_projection, unit=unit
503 )
504 variance = Image(
505 self._variance.array * factor**2,
506 bbox=self.bbox,
507 unit=unit**2,
508 )
509 elif self._photometric_scaling is None: 509 ↛ 510line 509 didn't jump to line 510 because the condition on line 509 was never true
510 raise astropy.units.UnitConversionError(
511 "VisitImage.photometric_scaling is None, and there "
512 f"is no constant conversion from {self.unit} to {unit}."
513 )
514 else:
515 if copy is False: 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true
516 raise astropy.units.UnitConversionError(
517 f"Photometric scaling must be applied to go from ={self.unit} to {unit}, but copy=False."
518 )
519 scaling = self._photometric_scaling
520 assert scaling.unit is not None, "Checked at construction."
521 if (constant_factor := _get_unit_conversion_factor(self.unit * scaling.unit, unit)) is not None:
522 if constant_factor != 1.0: 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true
523 scaling = scaling * constant_factor
524 scaling_array = scaling.render(self.bbox, dtype=self.image.array.dtype).array
525 elif (constant_factor := _get_unit_conversion_factor(self.unit / scaling.unit, unit)) is not None: 525 ↛ 531line 525 didn't jump to line 531 because the condition on line 525 was always true
526 if constant_factor != 1.0: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 scaling = scaling / constant_factor
528 scaling_array = scaling.render(self.bbox, dtype=self.image.array.dtype).array
529 np.true_divide(1.0, scaling_array, out=scaling_array)
530 else:
531 raise astropy.units.UnitConversionError(
532 f"photometric_scaling with units {scaling.unit} does not "
533 f"provide a path from {self.unit} to {unit}."
534 )
535 # We needed to allocate a new array to evaluate the scaling field,
536 # and then we need to allocate another to hold its square for the
537 # variance scaling. But then we can multiply those arrays in-place
538 # to get the output image and variance to avoid yet more
539 # allocations (note we can't instead multiply the visit image's
540 # image and variance arrays in place because they might have other
541 # references that are still associated with the old units).
542 image = Image(scaling_array, bbox=self.bbox, unit=unit)
543 variance = Image(np.square(scaling_array), bbox=self.bbox, unit=unit**2)
544 image.array *= self._image.array
545 variance.array *= self._variance.array
546 copy_components = copy is True
547 return self._transfer_metadata(
548 VisitImage(
549 image=image,
550 mask=self._mask if not copy_components else self._mask.copy(),
551 variance=variance,
552 sky_projection=self.sky_projection, # never copied; immutable
553 obs_info=self.obs_info if not copy_components else self.obs_info.model_copy(),
554 psf=self._psf, # never copied; immutable
555 bounds=self._bounds, # never copied; immutable
556 summary_stats=self.summary_stats if not copy_components else self.summary_stats.model_copy(),
557 detector=self._detector if not copy_detector else self._detector.copy(),
558 photometric_scaling=self._photometric_scaling, # never copied; immutable
559 aperture_corrections=(
560 self.aperture_corrections if not copy_components else self.aperture_corrections.copy()
561 ),
562 backgrounds=self.backgrounds if not copy_components else self.backgrounds.copy(),
563 band=self.band,
564 )
565 )
567 def serialize(self, archive: OutputArchive[Any]) -> VisitImageSerializationModel[Any]:
568 return self._serialize_impl(VisitImageSerializationModel, archive)
570 # This is slightly bad Liskov substitution - we're demanding M be a
571 # VisitImageSerializationModel, not just a MaskedImageSerializationModel,
572 # but that's because we know only `serialize` will call it.
573 def _serialize_impl[M: VisitImageSerializationModel[Any]]( # type: ignore[override]
574 self, model_type: type[M], archive: OutputArchive[Any]
575 ) -> M:
576 result = super()._serialize_impl(model_type, archive)
577 match self._psf:
578 # MyPy is able to figure things out here with this match statement,
579 # but not a single isinstance check on the three types.
580 case PiffWrapper(): 580 ↛ 581line 580 didn't jump to line 581 because the pattern on line 580 never matched
581 result.psf = archive.serialize_direct("psf", self._psf.serialize)
582 case PSFExWrapper(): 582 ↛ 583line 582 didn't jump to line 583 because the pattern on line 582 never matched
583 result.psf = archive.serialize_direct("psf", self._psf.serialize)
584 case GaussianPointSpreadFunction(): 584 ↛ 586line 584 didn't jump to line 586 because the pattern on line 584 always matched
585 result.psf = archive.serialize_direct("psf", self._psf.serialize)
586 case _:
587 raise TypeError(
588 f"Cannot serialize VisitImage with unrecognized PSF type {type(self._psf).__name__}."
589 )
590 assert result.sky_projection is not None, "VisitImage always has a sky_projection."
591 result.obs_info = self.obs_info
592 result.summary_stats = self.summary_stats
593 result.bounds = self._bounds.serialize() if self._bounds != self.bbox else None
594 result.detector = archive.serialize_direct("detector", self._detector.serialize)
595 result.band = self.band
596 result.photometric_scaling = (
597 # MyPy can't quite follow the type union through the serialize
598 # method return types.
599 archive.serialize_direct(
600 "photometric_scaling",
601 self._photometric_scaling.serialize,
602 ) # type: ignore[assignment]
603 if self._photometric_scaling is not None
604 else None
605 )
606 result.aperture_corrections = archive.serialize_direct(
607 "aperture_corrections",
608 functools.partial(ApertureCorrectionMapSerializationModel.serialize, self.aperture_corrections),
609 )
610 result.backgrounds = archive.serialize_direct("backgrounds", self._backgrounds.serialize)
611 return result
613 @staticmethod
614 def _get_archive_tree_type[P: pydantic.BaseModel](
615 pointer_type: type[P],
616 ) -> type[VisitImageSerializationModel[P]]:
617 """Return the serialization model type for this object for an archive
618 type that uses the given pointer type.
619 """
620 return VisitImageSerializationModel[pointer_type] # type: ignore
622 @staticmethod
623 def from_legacy( # type: ignore[override]
624 legacy: LegacyExposure,
625 *,
626 unit: astropy.units.UnitBase | None = None,
627 plane_map: Mapping[str, MaskPlane] | None = None,
628 instrument: str | None = None,
629 visit: int | None = None,
630 ) -> VisitImage:
631 """Convert from an `lsst.afw.image.Exposure` instance.
633 Parameters
634 ----------
635 legacy
636 An `lsst.afw.image.Exposure` instance that will share image and
637 variance (but not mask) pixel data with the returned object.
638 unit
639 Units of the image. If not provided, the ``BUNIT`` metadata
640 key will be used, if available.
641 plane_map
642 A mapping from legacy mask plane name to the new plane name and
643 description. If `None` (default)
644 `get_legacy_visit_image_mask_planes` is used.
645 instrument
646 Name of the instrument. Extracted from the metadata if not
647 provided.
648 visit
649 ID of the visit. Extracted from the metadata if not provided.
650 """
651 if plane_map is None: 651 ↛ 653line 651 didn't jump to line 653 because the condition on line 651 was always true
652 plane_map = get_legacy_visit_image_mask_planes()
653 md = legacy.getMetadata()
654 obs_info = _obs_info_from_md(md, visit_info=legacy.info.getVisitInfo())
655 instrument = _extract_or_check_header(
656 "LSST BUTLER DATAID INSTRUMENT", instrument, md, obs_info.instrument, str
657 )
658 visit = _extract_or_check_header("LSST BUTLER DATAID VISIT", visit, md, obs_info.exposure_id, int)
659 legacy_wcs = legacy.getWcs()
660 if legacy_wcs is None: 660 ↛ 661line 660 didn't jump to line 661 because the condition on line 660 was never true
661 raise ValueError("Exposure does not have a SkyWcs.")
662 legacy_detector = legacy.getDetector()
663 if legacy_detector is None: 663 ↛ 664line 663 didn't jump to line 664 because the condition on line 663 was never true
664 raise ValueError("Exposure does not have a Detector.")
665 detector_bbox = Box.from_legacy(legacy_detector.getBBox())
667 # Update the ObservationInfo from other components.
668 obs_info = _update_obs_info_from_legacy(obs_info, legacy_detector, legacy.info.getFilter())
670 opaque_fits_metadata = FitsOpaqueMetadata()
671 primary_header = astropy.io.fits.Header()
672 with warnings.catch_warnings():
673 # Silence warnings about long keys becoming HIERARCH.
674 warnings.simplefilter("ignore", category=astropy.io.fits.verify.VerifyWarning)
675 for name in md.getOrderedNames():
676 # Some keys may be set more than once.
677 # Write one card per value in those cases.
678 for value in md.getArray(name):
679 primary_header.append((name, value), end=True)
680 metadata = opaque_fits_metadata.extract_legacy_primary_header(primary_header)
681 instrumental_unit = opaque_fits_metadata.get_instrumental_unit() or astropy.units.electron
682 hdr_unit: astropy.units.UnitBase | None = None
683 if hdr_unit_str := md.get("BUNIT"): 683 ↛ 689line 683 didn't jump to line 689 because the condition on line 683 was always true
684 hdr_unit = astropy.units.Unit(hdr_unit_str, format="FITS")
685 if hdr_unit == astropy.units.adu and instrumental_unit == astropy.units.electron: 685 ↛ 688line 685 didn't jump to line 688 because the condition on line 685 was never true
686 # Fix incorrect BUNIT='adu' in LSST
687 # preliminary_visit_image.
688 hdr_unit = astropy.units.electron
689 if unit is None: 689 ↛ 691line 689 didn't jump to line 691 because the condition on line 689 was always true
690 unit = hdr_unit
691 elif hdr_unit is not None and hdr_unit != unit:
692 raise ValueError(f"BUNIT value {hdr_unit} disagrees with given unit {unit}.")
693 sky_projection = SkyProjection.from_legacy(
694 legacy_wcs,
695 DetectorFrame(
696 instrument=instrument,
697 visit=visit,
698 detector=legacy_detector.getId(),
699 bbox=detector_bbox,
700 ),
701 )
702 legacy_psf = legacy.getPsf()
703 if legacy_psf is None: 703 ↛ 704line 703 didn't jump to line 704 because the condition on line 703 was never true
704 raise ValueError("Exposure file does not have a Psf.")
705 psf = PointSpreadFunction.from_legacy(legacy_psf, bounds=detector_bbox)
706 masked_image = MaskedImage.from_legacy(legacy.getMaskedImage(), unit=unit, plane_map=plane_map)
707 legacy_summary_stats = legacy.info.getSummaryStats()
708 legacy_ap_corr_map = legacy.info.getApCorrMap()
709 legacy_polygon = legacy.info.getValidPolygon()
710 legacy_photo_calib = legacy.info.getPhotoCalib()
711 detector = Detector.from_legacy(
712 legacy_detector, instrument=instrument, visit=visit, is_raw_assembled=True
713 )
714 _reconcile_detector_serial(obs_info, detector)
715 result = VisitImage(
716 image=masked_image.image.view(unit=unit),
717 mask=masked_image.mask,
718 variance=masked_image.variance,
719 sky_projection=sky_projection,
720 psf=psf,
721 obs_info=obs_info,
722 summary_stats=(
723 ObservationSummaryStats.from_legacy(legacy_summary_stats)
724 if legacy_summary_stats is not None
725 else None
726 ),
727 detector=detector,
728 aperture_corrections=(
729 aperture_corrections_from_legacy(legacy_ap_corr_map)
730 if legacy_ap_corr_map is not None
731 else None
732 ),
733 bounds=Polygon.from_legacy(legacy_polygon) if legacy_polygon is not None else None,
734 photometric_scaling=(
735 field_from_legacy_photo_calib(
736 legacy_photo_calib, bounds=detector_bbox, instrumental_unit=instrumental_unit
737 )
738 if legacy_photo_calib is not None
739 else None
740 ),
741 band=legacy.info.getFilter().bandLabel,
742 metadata=metadata,
743 )
744 result.metadata["id"] = legacy.info.getId()
745 result._opaque_metadata = opaque_fits_metadata
746 return result
748 def to_legacy(
749 self, *, copy: bool | None = None, plane_map: Mapping[str, MaskPlane] | None = None
750 ) -> LegacyExposure:
751 """Convert to an `lsst.afw.image.Exposure` instance.
753 Parameters
754 ----------
755 copy
756 If `True`, always copy the image and variance pixel data.
757 If `False`, return a view, and raise `TypeError` if the pixel data
758 is read-only (this is not supported by afw). If `None`, only copy
759 if the pixel data is read-only. Mask pixel data is always copied.
760 plane_map
761 A mapping from legacy mask plane name to the new plane name and
762 description. If `None` (default),
763 `get_legacy_visit_image_mask_planes` is used.
764 """
765 from lsst.afw.image import Exposure as LegacyExposure
766 from lsst.afw.image import FilterLabel as LegacyFilterLabel
767 from lsst.obs.base.makeRawVisitInfoViaObsInfo import MakeRawVisitInfoViaObsInfo
769 if plane_map is None: 769 ↛ 771line 769 didn't jump to line 771 because the condition on line 769 was always true
770 plane_map = get_legacy_visit_image_mask_planes()
771 legacy_masked_image = super().to_legacy(copy=copy, plane_map=plane_map)
772 result = LegacyExposure(legacy_masked_image, dtype=self.image.array.dtype)
773 result_info = result.info
774 result_info.setId(self.metadata.get("id"))
775 result_info.setWcs(self.sky_projection.to_legacy())
776 result_info.setDetector(self.detector.to_legacy())
777 result_info.setFilter(LegacyFilterLabel.fromBandPhysical(self.band, self.obs_info.physical_filter))
778 if self._photometric_scaling is not None: 778 ↛ 779line 778 didn't jump to line 779 because the condition on line 778 was never true
779 result_info.setPhotoCalib(self._photometric_scaling.to_legacy_photo_calib(self.unit))
780 else:
781 result_info.setPhotoCalib(BaseField.make_legacy_photo_calib(self.unit))
782 self._fill_legacy_metadata(result_info.getMetadata())
783 if isinstance(self._psf, LegacyPointSpreadFunction): 783 ↛ 785line 783 didn't jump to line 785 because the condition on line 783 was always true
784 result_info.setPsf(self._psf.legacy_psf)
785 elif isinstance(self._psf, PiffWrapper):
786 result_info.setPsf(self._psf.to_legacy())
787 if isinstance(self.bounds, Polygon): 787 ↛ 788line 787 didn't jump to line 788 because the condition on line 787 was never true
788 result_info.setValidPolygon(self.bounds.to_legacy())
789 if self.aperture_corrections: 789 ↛ 790line 789 didn't jump to line 790 because the condition on line 789 was never true
790 result_info.setApCorrMap(aperture_corrections_to_legacy(self.aperture_corrections))
791 result_info.setVisitInfo(MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.obs_info))
792 result_info.setSummaryStats(self.summary_stats.to_legacy())
793 return result
795 @staticmethod
796 def read_legacy( # type: ignore[override]
797 filename: str,
798 *,
799 preserve_quantization: bool = False,
800 plane_map: Mapping[str, MaskPlane] | None = None,
801 instrument: str | None = None,
802 visit: int | None = None,
803 component: Literal[
804 "bbox",
805 "image",
806 "mask",
807 "variance",
808 "sky_projection",
809 "psf",
810 "detector",
811 "photometric_scaling",
812 "obs_info",
813 "summary_stats",
814 "aperture_corrections",
815 ]
816 | None = None,
817 ) -> Any:
818 """Read a FITS file written by `lsst.afw.image.Exposure.writeFits`.
820 Parameters
821 ----------
822 filename
823 Full name of the file.
824 preserve_quantization
825 If `True`, ensure that writing the masked image back out again will
826 exactly preserve quantization-compressed pixel values. This causes
827 the image and variance plane arrays to be marked as read-only and
828 stores the original binary table data for those planes in memory.
829 If the `MaskedImage` is copied, the precompressed pixel values are
830 not transferred to the copy.
831 plane_map
832 A mapping from legacy mask plane name to the new plane name and
833 description. If `None` (default)
834 `get_legacy_visit_image_mask_planes` is used.
835 instrument
836 Name of the instrument. Read from the primary header if not
837 provided.
838 visit
839 ID of the visit. Read from the primary header if not
840 provided.
841 component
842 A component to read instead of the full image.
843 """
844 from lsst.afw.image import ExposureFitsReader
846 reader = ExposureFitsReader(filename)
847 if component == "bbox":
848 return Box.from_legacy(reader.readBBox())
849 legacy_detector = reader.readDetector()
850 if legacy_detector is None:
851 raise ValueError(f"Exposure file {filename!r} does not have a Detector.")
852 detector_bbox = Box.from_legacy(legacy_detector.getBBox())
853 legacy_wcs = None
854 if component in (None, "image", "mask", "variance", "sky_projection"):
855 legacy_wcs = reader.readWcs()
856 if legacy_wcs is None:
857 raise ValueError(f"Exposure file {filename!r} does not have a SkyWcs.")
858 legacy_exposure_info = reader.readExposureInfo()
859 summary_stats = None
860 if component in (None, "summary_stats"):
861 legacy_stats = legacy_exposure_info.getSummaryStats()
862 if legacy_stats is not None:
863 summary_stats = ObservationSummaryStats.from_legacy(legacy_stats)
864 if component == "summary_stats":
865 return summary_stats
866 if component in (None, "psf"):
867 legacy_psf = reader.readPsf()
868 if legacy_psf is None:
869 raise ValueError(f"Exposure file {filename!r} does not have a Psf.")
870 psf = PointSpreadFunction.from_legacy(legacy_psf, bounds=detector_bbox)
871 if component == "psf":
872 return psf
873 aperture_corrections: ApertureCorrectionMap = {}
874 if component in (None, "aperture_corrections"):
875 legacy_ap_corr_map = reader.readApCorrMap()
876 if legacy_ap_corr_map is not None:
877 aperture_corrections = aperture_corrections_from_legacy(legacy_ap_corr_map)
878 if component == "aperture_corrections":
879 return aperture_corrections
880 assert component in (
881 None,
882 "image",
883 "mask",
884 "variance",
885 "sky_projection",
886 "obs_info",
887 "detector",
888 "photometric_scaling",
889 ), component # for MyPy
890 filter_label = reader.readFilter()
891 with astropy.io.fits.open(filename) as hdu_list:
892 primary_header = hdu_list[0].header
893 obs_info = _obs_info_from_md(primary_header)
894 obs_info = _update_obs_info_from_legacy(obs_info, legacy_detector, filter_label)
895 if component == "obs_info":
896 return obs_info
897 instrument = _extract_or_check_header(
898 "LSST BUTLER DATAID INSTRUMENT", instrument, primary_header, obs_info.instrument, str
899 )
900 visit = _extract_or_check_header(
901 "LSST BUTLER DATAID VISIT", visit, primary_header, obs_info.exposure_id, int
902 )
903 opaque_metadata = FitsOpaqueMetadata()
904 # This extraction is destructive, so we need to be sure to pass
905 # this opaque_metadata down to MaskedImage._read_legacy_hdus
906 # so it doesn't try to extract it again.
907 metadata = opaque_metadata.extract_legacy_primary_header(primary_header)
908 if (instrumental_unit := opaque_metadata.get_instrumental_unit()) is None:
909 instrumental_unit = astropy.units.electron
910 photometric_scaling: Field | None = None
911 if component in (None, "photometric_scaling"):
912 legacy_photo_calib = reader.readPhotoCalib()
913 if legacy_photo_calib is not None:
914 photometric_scaling = field_from_legacy_photo_calib(
915 legacy_photo_calib, bounds=detector_bbox, instrumental_unit=instrumental_unit
916 )
917 if component == "photometric_scaling":
918 return photometric_scaling
919 if component in ("detector", None):
920 detector = Detector.from_legacy(
921 legacy_detector, instrument=instrument, visit=visit, is_raw_assembled=True
922 )
923 _reconcile_detector_serial(obs_info, detector)
924 if component == "detector":
925 return detector
926 assert component != "detector", "MyPy can't work this out from the above."
927 sky_projection = SkyProjection.from_legacy(
928 legacy_wcs,
929 DetectorFrame(
930 instrument=instrument,
931 visit=visit,
932 detector=legacy_detector.getId(),
933 bbox=detector_bbox,
934 ),
935 )
936 if component == "sky_projection":
937 return sky_projection
938 if plane_map is None:
939 plane_map = get_legacy_visit_image_mask_planes()
940 from_masked_image = MaskedImage._read_legacy_hdus(
941 hdu_list,
942 filename,
943 opaque_metadata=opaque_metadata,
944 preserve_quantization=preserve_quantization,
945 plane_map=plane_map,
946 component=component,
947 )
948 if component is not None:
949 # This is the image, mask, or variance; attach the sky_projection
950 # and obs_info and return
951 return from_masked_image.view(sky_projection=sky_projection)
952 legacy_polygon = reader.readValidPolygon()
953 result = VisitImage(
954 from_masked_image.image,
955 mask=from_masked_image.mask,
956 variance=from_masked_image.variance,
957 sky_projection=sky_projection,
958 psf=psf,
959 detector=detector,
960 obs_info=obs_info,
961 summary_stats=summary_stats,
962 aperture_corrections=aperture_corrections,
963 bounds=Polygon.from_legacy(legacy_polygon) if legacy_polygon is not None else None,
964 photometric_scaling=photometric_scaling,
965 band=filter_label.bandLabel,
966 metadata=metadata,
967 )
968 result._opaque_metadata = from_masked_image._opaque_metadata
969 result.metadata["id"] = reader.readExposureId()
970 return result
973class VisitImageSerializationModel[P: pydantic.BaseModel](MaskedImageSerializationModel[P]):
974 """A Pydantic model used to represent a serialized `VisitImage`."""
976 SCHEMA_NAME: ClassVar[str] = "visit_image"
977 SCHEMA_VERSION: ClassVar[str] = "1.1.0"
978 MIN_READ_VERSION: ClassVar[int] = 1
979 PUBLIC_TYPE: ClassVar[type] = VisitImage
981 # Inherited attributes are duplicated because that improves the docs
982 # (some limitation in the sphinx/pydantic integration), and these are
983 # important docs.
985 image: ImageSerializationModel[P] = pydantic.Field(description="The main data image.")
986 mask: MaskSerializationModel[P] = pydantic.Field(
987 description="Bitmask that annotates the main image's pixels."
988 )
989 variance: ImageSerializationModel[P] = pydantic.Field(
990 description="Per-pixel variance estimates for the main image."
991 )
992 sky_projection: SkyProjectionSerializationModel[P] = pydantic.Field(
993 description="Projection that maps the pixel grid to the sky.",
994 )
995 psf: PiffSerializationModel | PSFExSerializationModel | GaussianPSFSerializationModel | Any = (
996 pydantic.Field(union_mode="left_to_right", description="PSF model for the image.")
997 )
998 obs_info: ObservationInfo = pydantic.Field(
999 description="Standardized description of visit metadata",
1000 )
1001 photometric_scaling: FieldSerializationModel | None = pydantic.Field(
1002 default=None,
1003 description="Scaling that can be used to multiply a post-ISR image to yield calibrated pixel values.",
1004 )
1005 summary_stats: ObservationSummaryStats = pydantic.Field(
1006 description="Summary statistics for the observation."
1007 )
1008 detector: DetectorSerializationModel = pydantic.Field(
1009 description="Geometry and electronic information for the detector."
1010 )
1011 aperture_corrections: ApertureCorrectionMapSerializationModel = pydantic.Field(
1012 default_factory=ApertureCorrectionMapSerializationModel,
1013 description="Aperture corrections, keyed by flux algorithm.",
1014 )
1015 bounds: BoundsSerializationModel | None = pydantic.Field(
1016 default=None,
1017 description="Pixel validity region, if different from the image bounding box.",
1018 exclude_if=is_none,
1019 )
1020 backgrounds: BackgroundMapSerializationModel = pydantic.Field(
1021 default_factory=BackgroundMapSerializationModel,
1022 description="Background models associated with this image.",
1023 )
1024 band: str = pydantic.Field(description="Short name of the bandpass filter.")
1026 def deserialize(
1027 self, archive: InputArchive[Any], *, bbox: Box | None = None, **kwargs: Any
1028 ) -> VisitImage:
1029 if kwargs: 1029 ↛ 1030line 1029 didn't jump to line 1030 because the condition on line 1029 was never true
1030 raise InvalidParameterError(f"Unrecognized parameters for VisitImage: {set(kwargs.keys())}.")
1031 masked_image = super().deserialize(archive, bbox=bbox)
1032 try:
1033 psf = self.psf.deserialize(archive)
1034 except ArchiveReadError as err:
1035 # Defer this until/unless somebody actually asks for the PSF.
1036 psf = err
1037 detector = self.detector.deserialize(archive)
1038 aperture_corrections = self.aperture_corrections.deserialize(archive)
1039 photometric_scaling = (
1040 self.photometric_scaling.deserialize(archive) if self.photometric_scaling is not None else None
1041 )
1042 return VisitImage(
1043 masked_image.image,
1044 mask=masked_image.mask,
1045 variance=masked_image.variance,
1046 psf=psf,
1047 sky_projection=masked_image.sky_projection,
1048 obs_info=self.obs_info,
1049 summary_stats=self.summary_stats,
1050 detector=detector,
1051 aperture_corrections=aperture_corrections,
1052 photometric_scaling=photometric_scaling,
1053 bounds=self.bounds.deserialize() if self.bounds is not None else None,
1054 backgrounds=self.backgrounds.deserialize(archive),
1055 band=self.band,
1056 )._finish_deserialize(self)
1058 def deserialize_component(self, component: str, archive: InputArchive[Any], **kwargs: Any) -> Any:
1059 if kwargs and component not in ("image", "mask", "variance", "masked_image"): 1059 ↛ 1060line 1059 didn't jump to line 1060 because the condition on line 1059 was never true
1060 raise InvalidParameterError(
1061 f"Unsupported parameters for VisitImage component {component}: {set(kwargs.keys())}."
1062 )
1063 if component == "masked_image":
1064 return super().deserialize(archive, **kwargs)
1065 return super().deserialize_component(component, archive, **kwargs)
1068def _obs_info_from_md(
1069 md: MutableMapping[str, Any], visit_info: LegacyVisitInfo | None = None
1070) -> ObservationInfo:
1071 # Try to get an ObservationInfo from the primary header as if
1072 # it's a raw header. Else fallback.
1073 try:
1074 obs_info = ObservationInfo.from_header(md, quiet=True)
1075 except ValueError:
1076 # Not known translator. Must fall back to visit info. If we have
1077 # an actual VisitInfo, serialize it since we know that it will be
1078 # complete.
1079 if visit_info is not None: 1079 ↛ 1091line 1079 didn't jump to line 1091 because the condition on line 1079 was always true
1080 from lsst.afw.image import setVisitInfoMetadata
1081 from lsst.daf.base import PropertyList
1083 pl = PropertyList()
1084 setVisitInfoMetadata(pl, visit_info)
1085 # Merge so that we still have access to butler provenance.
1086 md.update(pl)
1088 # Try the given header looking for VisitInfo hints.
1089 # We get lots of warnings if nothing can be found. Currently
1090 # no way to disable those without capturing them.
1091 obs_info = ObservationInfo.from_header(md, translator_class=VisitInfoTranslator, quiet=True)
1092 return obs_info
1095def _update_obs_info_from_legacy(
1096 obs_info: ObservationInfo,
1097 detector: LegacyDetector | None = None,
1098 filter_label: LegacyFilterLabel | None = None,
1099) -> ObservationInfo:
1100 extra_md: dict[str, str | int] = {}
1102 if filter_label is not None and filter_label.hasBandLabel(): 1102 ↛ 1109line 1102 didn't jump to line 1109 because the condition on line 1102 was always true
1103 extra_md["physical_filter"] = filter_label.physicalLabel
1105 # Fill in detector metadata, check for consistency.
1106 # ObsInfo detector name and group can not be derived from
1107 # the getName() information without knowing how the components
1108 # are separated.
1109 if detector is not None: 1109 ↛ 1116line 1109 didn't jump to line 1116 because the condition on line 1109 was always true
1110 detector_md = {
1111 "detector_num": detector.getId(),
1112 "detector_unique_name": detector.getName(),
1113 }
1114 extra_md.update(detector_md)
1116 obs_info_updates: dict[str, str | int] = {}
1117 for k, v in extra_md.items():
1118 current = getattr(obs_info, k)
1119 if current is None: 1119 ↛ 1122line 1119 didn't jump to line 1122 because the condition on line 1119 was always true
1120 obs_info_updates[k] = v
1121 continue
1122 if current != v:
1123 raise RuntimeError(
1124 f"ObservationInfo contains value for '{k}' that is inconsistent "
1125 f"with given legacy object: {v} != {current}"
1126 )
1128 if obs_info_updates: 1128 ↛ 1130line 1128 didn't jump to line 1130 because the condition on line 1128 was always true
1129 obs_info = obs_info.model_copy(update=obs_info_updates)
1130 return obs_info
1133def _reconcile_detector_serial(obs_info: ObservationInfo, detector: Detector) -> None:
1134 # Some LSSTCam detector serial numbers are/were incorrect in the camera
1135 # geometry (DM-55080), so if they conflict it's the ObservationInfo (from
1136 # the headers) that's correct.
1137 if obs_info.detector_serial is not None and detector.serial != obs_info.detector_serial: 1137 ↛ 1138line 1137 didn't jump to line 1138 because the condition on line 1137 was never true
1138 _LOG.warning(
1139 "Detector serial from ObservationInfo (%s) for detector %d does not agree "
1140 "with camera geometry %s; assuming the former is correct.",
1141 obs_info.detector_serial,
1142 detector.id,
1143 detector.serial,
1144 )
1145 detector._attributes.serial = obs_info.detector_serial
1148def _extract_or_check_value[T](
1149 key: str,
1150 given_value: T | None,
1151 *sources: tuple[str, T | None],
1152) -> T:
1153 # Compare given value against multiple sources. If given value is not
1154 # supplied return the first non-None value in the reference sources.
1155 if given_value is not None: 1155 ↛ 1167line 1155 didn't jump to line 1167 because the condition on line 1155 was always true
1156 for source_name, source_value in sources:
1157 if source_value is not None and source_value != given_value: 1157 ↛ 1158line 1157 didn't jump to line 1158 because the condition on line 1157 was never true
1158 raise ValueError(
1159 f"Given value {given_value!r} does not match {source_value!r} from {source_name}."
1160 )
1161 if source_value is not None:
1162 # Only check the first non-None source rather than checking
1163 # all supplied values.
1164 break
1165 return given_value
1167 for _, source_value in sources:
1168 if source_value is not None:
1169 return source_value
1171 raise ValueError(f"No value found for {key}.")
1174def _extract_or_check_header[T](
1175 key: str, given_value: T | None, header: Any, obs_info_value: T | None, coerce: Callable[[Any], T]
1176) -> T:
1177 hdr_value: T | None = None
1178 if (hdr_raw_value := header.get(key)) is not None: 1178 ↛ 1179line 1178 didn't jump to line 1179 because the condition on line 1178 was never true
1179 hdr_value = coerce(hdr_raw_value)
1180 return _extract_or_check_value(
1181 key, given_value, ("ObservationInfo", obs_info_value), (f"header key {key}", hdr_value)
1182 )
1185def _get_unit_conversion_factor(
1186 original: astropy.units.UnitBase, new: astropy.units.UnitBase
1187) -> float | None:
1188 try:
1189 return original.to(new)
1190 except astropy.units.UnitConversionError:
1191 return None