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