Coverage for python/lsst/images/cells/_coadd.py: 46%
241 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-25 08:24 +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__ = ("CellCoadd", "CellCoaddSerializationModel")
16import functools
17from collections.abc import Mapping, Sequence
18from types import EllipsisType
19from typing import TYPE_CHECKING, Any, ClassVar, cast
21import astropy.io.fits
22import astropy.units
23import astropy.wcs
24import pydantic
26from .._backgrounds import BackgroundMap, BackgroundMapSerializationModel
27from .._cell_grid import CellGrid, CellGridBounds, PatchDefinition
28from .._geom import YX, Box
29from .._image import Image, ImageSerializationModel
30from .._mask import Mask, MaskPlane, MaskSchema, MaskSerializationModel, get_legacy_deep_coadd_mask_planes
31from .._masked_image import MaskedImage, MaskedImageSerializationModel
32from .._transforms import SkyProjection, SkyProjectionSerializationModel, TractFrame
33from ..fields import BaseField
34from ..serialization import InputArchive, InvalidParameterError, OutputArchive
35from ._aperture_corrections import CellApertureCorrectionMapSerializationModel, CellField
36from ._provenance import CoaddProvenance, CoaddProvenanceSerializationModel
37from ._psf import CellPointSpreadFunction, CellPointSpreadFunctionSerializationModel
39if TYPE_CHECKING:
40 try:
41 from lsst.afw.image import Exposure as LegacyExposure
42 from lsst.cell_coadds import MultipleCellCoadd as LegacyMultipleCellCoadd
43 from lsst.skymap import TractInfo
44 except ImportError:
45 type LegacyExposure = Any # type: ignore[no-redef]
46 type LegacyMultipleCellCoadd = Any # type: ignore[no-redef]
47 type TractInfo = Any # type: ignore[no-redef]
50class CellCoadd(MaskedImage):
51 """A coadd comprised of cells on a regular grid.
53 Parameters
54 ----------
55 image
56 The main image plane. If this has a `.SkyProjection`, it will be used
57 for all planes unless a ``sky_projection`` is passed separately.
58 mask
59 A bitmask image that annotates the main image plane. Must have the
60 same bounding box as ``image`` if provided. Any attached
61 ``sky_projection`` is replaced (possibly by `None`).
62 variance
63 The per-pixel uncertainty of the main image as an image of variance
64 values. Must have the same bounding box as ``image`` if provided, and
65 its units must be the square of ``image.unit`` or `None`.
66 Values default to ``1.0``. Any attached ``sky_projection`` is replaced
67 (possibly by `None`).
68 mask_fractions
69 A mapping from an input-image mask plane name to an image of the
70 weights sums of that plane.
71 noise_realizations
72 A sequence of images with Monte Carlo realizations of the noise in
73 the coadd.
74 mask_schema
75 Schema for the mask plane. Must be provided if and only if ``mask`` is
76 not provided.
77 sky_projection
78 Projection that maps the pixel grid to the sky. Can only be `None` if
79 a projection is already attached to ``image``.
80 band
81 Name of the band.
82 psf
83 Effective point-spread function for the coadd. The missing cells
84 reported by ``psf.bounds`` are assumed to apply to all image data for
85 that cell as well (i.e. there is a PSF for a cell if and only if
86 there is image data for that cell).
87 aperture_corrections
88 Aperture corrections for different photometry algorithms.
89 patch
90 Identifiers and geometry of the full patch, if the image is confined
91 to a single patch. When present, the cell grid of the PSF and
92 provenance (if provideD) must be the full patch grid, even if its
93 bounds select a subset of that area.
94 provenance
95 Information about the images that went into the coadd.
96 backgrounds
97 Background models associated with this image.
98 """
100 def __init__(
101 self,
102 image: Image,
103 *,
104 mask: Mask | None = None,
105 variance: Image | None = None,
106 mask_fractions: Mapping[str, Image] | None = None,
107 noise_realizations: Sequence[Image] = (),
108 mask_schema: MaskSchema | None = None,
109 sky_projection: SkyProjection[TractFrame] | None = None,
110 band: str | None = None,
111 psf: CellPointSpreadFunction,
112 aperture_corrections: Mapping[str, CellField] | None = None,
113 patch: PatchDefinition | None = None,
114 provenance: CoaddProvenance | None = None,
115 backgrounds: BackgroundMap | None = None,
116 ) -> None:
117 super().__init__(
118 image,
119 mask=mask,
120 variance=variance,
121 mask_schema=mask_schema,
122 sky_projection=sky_projection,
123 )
124 if self.image.unit is None: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 raise TypeError("The image component of a CellCoadd must have units.")
126 if self.image.sky_projection is None: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 raise TypeError("The sky_projection component of a CellCoadd cannot be None.")
128 if not isinstance(self.image.sky_projection.pixel_frame, TractFrame): 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 raise TypeError("The sky_projection's pixel frame must be a TractFrame for CellCoadd.")
130 self._mask_fractions = dict(mask_fractions) if mask_fractions is not None else {}
131 self._noise_realizations = list(noise_realizations)
132 self._band = band
133 if psf.bounds.bbox != self.bbox: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 psf = psf[self.bbox.intersection(psf.bounds.bbox)]
135 self._psf = psf
136 self._aperture_corrections = dict(aperture_corrections) if aperture_corrections is not None else {}
137 for ap_corr_name, ap_corr_field in self._aperture_corrections.items():
138 if ap_corr_field.bounds.grid != self.grid: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 raise ValueError(
140 f"Grids for cell PSF and {ap_corr_name} aperture corrections are not consistent."
141 )
142 self._patch = patch
143 self._provenance = provenance
144 if self._provenance and not self._patch: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 raise TypeError("A CellCoadd cannot carry provenance without a patch definition.")
146 self._backgrounds = backgrounds if backgrounds is not None else BackgroundMap()
148 @property
149 def skymap(self) -> str:
150 """Name of the skymap (`str`)."""
151 return self.sky_projection.pixel_frame.skymap
153 @property
154 def tract(self) -> int:
155 """ID of the tract (`int`)."""
156 return self.sky_projection.pixel_frame.tract
158 @property
159 def patch(self) -> PatchDefinition:
160 """Identifiers and geometry of the full patch, if the image is confined
161 to a single patch (`PatchDefinition`).
162 """
163 if self._patch is None:
164 raise AttributeError("Coadd has no patch information.")
165 return self._patch
167 @property
168 def band(self) -> str | None:
169 """Name of the band (`str` or `None`)."""
170 return self._band
172 @property
173 def mask_fractions(self) -> Mapping[str, Image]:
174 """A mapping from an input-image mask plane name to an image of the
175 weights sums of that plane
176 (`~collections.abc.Mapping` [`str`, `.Image`]).
177 """
178 return self._mask_fractions
180 @property
181 def noise_realizations(self) -> Sequence[Image]:
182 """A sequence of images with Monte Carlo realizations of the noise in
183 the coadd (`~collections.abc.Sequence` [`.Image`]).
184 """
185 return self._noise_realizations
187 @property
188 def unit(self) -> astropy.units.UnitBase:
189 """The units of the image plane (`astropy.units.Unit`)."""
190 return cast(astropy.units.UnitBase, super().unit)
192 @property
193 def sky_projection(self) -> SkyProjection[TractFrame]:
194 """The projection that maps the pixel grid to the sky
195 (`.SkyProjection` [`.TractFrame`]).
196 """
197 return cast(SkyProjection[TractFrame], super().sky_projection)
199 @property
200 def psf(self) -> CellPointSpreadFunction:
201 """Effective point-spread function for the coadd
202 (`CellPointSpreadFunction`).
203 """
204 return self._psf
206 @property
207 def aperture_corrections(self) -> Mapping[str, CellField]:
208 """Aperture corrections for different photometry algorithms
209 (`dict` [`str`, `CellField`]).
210 """
211 return self._aperture_corrections
213 @property
214 def bounds(self) -> CellGridBounds:
215 """The grid of cells that overlap this coadd and a set of missing
216 cells (`CellGridBounds`).
217 """
218 return self._psf.bounds
220 @property
221 def grid(self) -> CellGrid:
222 """The grid of cells that overlap this coadd (`CellGrid`)."""
223 return self._psf.bounds.grid
225 @property
226 def provenance(self) -> CoaddProvenance:
227 """Information about the images that went into the coadd
228 (`CoaddProvenance` or `None`).
229 """
230 if self._provenance is None: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 raise AttributeError("Coadd has no provenance information.")
232 return self._provenance
234 @property
235 def backgrounds(self) -> BackgroundMap:
236 """A mapping of backgrounds associated with this image
237 (`.BackgroundMap`).
238 """
239 return self._backgrounds
241 def __getitem__(self, bbox: Box | EllipsisType) -> CellCoadd:
242 if bbox is ...: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 return self
244 super().__getitem__(bbox)
245 psf = self.psf[bbox]
246 return self._transfer_metadata(
247 CellCoadd(
248 self.image[bbox],
249 mask=self.mask[bbox],
250 variance=self.variance[bbox],
251 sky_projection=self.sky_projection,
252 mask_fractions={k: v[bbox] for k, v in self._mask_fractions.items()},
253 noise_realizations=[v[bbox] for v in self._noise_realizations],
254 band=self.band,
255 psf=psf,
256 patch=self._patch,
257 provenance=(
258 self._provenance.subset(psf.bounds.cell_indices())
259 if self._provenance is not None
260 else None
261 ),
262 backgrounds=self._backgrounds,
263 aperture_corrections=self._aperture_corrections.copy(),
264 ),
265 bbox=bbox,
266 )
268 def __str__(self) -> str:
269 return f"CellCoadd({self.bbox!s}, tract={self.tract})"
271 def __repr__(self) -> str:
272 return str(self)
274 def copy(self) -> CellCoadd:
275 """Deep-copy the coadd."""
276 return self._transfer_metadata(
277 CellCoadd(
278 image=self._image.copy(),
279 mask=self._mask.copy(),
280 variance=self._variance.copy(),
281 sky_projection=self.sky_projection,
282 mask_fractions={k: v.copy() for k, v in self._mask_fractions.items()},
283 noise_realizations=[v.copy() for v in self._noise_realizations],
284 band=self.band,
285 psf=self.psf,
286 patch=self.patch,
287 provenance=self.provenance,
288 backgrounds=self._backgrounds.copy(),
289 aperture_corrections=self._aperture_corrections.copy(),
290 ),
291 copy=True,
292 )
294 def apply_background(self, name: str | None) -> None:
295 """Subtract the background with the given name, modifying the image
296 in place.
298 If ``name`` is `None`, restore the original background.
299 """
300 current_bg = self.backgrounds.subtracted
301 if current_bg is not None:
302 if name == current_bg.name:
303 return
304 self.image.quantity += current_bg.field.render(dtype=self.image.array.dtype).quantity
305 if name is None:
306 self._backgrounds._subtracted = None
307 return
308 new_bg = self.backgrounds[name]
309 self.image.quantity -= new_bg.field.render(dtype=self.image.array.dtype).quantity
310 self._backgrounds._subtracted = name
312 def serialize(self, archive: OutputArchive[Any]) -> CellCoaddSerializationModel:
313 """Serialize the image to an output archive.
315 Parameters
316 ----------
317 archive
318 Archive to write to.
319 """
320 serialized_image = archive.serialize_direct(
321 "image",
322 functools.partial(self.image.serialize, save_projection=False, tile_shape=self.grid.cell_shape),
323 )
324 serialized_mask = archive.serialize_direct(
325 "mask",
326 functools.partial(self.mask.serialize, save_projection=False, tile_shape=self.grid.cell_shape),
327 )
328 serialized_variance = archive.serialize_direct(
329 "variance",
330 functools.partial(
331 self.variance.serialize, save_projection=False, tile_shape=self.grid.cell_shape
332 ),
333 )
334 serialized_projection = archive.serialize_direct("sky_projection", self.sky_projection.serialize)
335 serialized_mask_fractions = {
336 k: archive.serialize_direct(
337 f"mask_fractions/{k}",
338 functools.partial(
339 v.serialize,
340 save_projection=False,
341 tile_shape=self.grid.cell_shape,
342 options_name="mask_fractions",
343 ),
344 )
345 for k, v in self.mask_fractions.items()
346 }
347 serialized_noise_realizations = [
348 archive.serialize_direct(
349 f"noise_realizations/{n}",
350 functools.partial(
351 v.serialize, save_projection=False, tile_shape=self.grid.cell_shape, options_name="image"
352 ),
353 )
354 for n, v in enumerate(self.noise_realizations)
355 ]
356 serialized_psf = archive.serialize_direct("psf", self.psf.serialize)
357 serialized_aperture_corrections = archive.serialize_direct(
358 "aperture_corrections",
359 functools.partial(
360 CellApertureCorrectionMapSerializationModel.serialize, self.aperture_corrections
361 ),
362 )
363 serialized_provenance = (
364 archive.serialize_direct("provenance", self._provenance.serialize)
365 if self._provenance is not None
366 else None
367 )
368 serialized_backgrounds = archive.serialize_direct("background", self._backgrounds.serialize)
369 return CellCoaddSerializationModel(
370 image=serialized_image,
371 mask=serialized_mask,
372 variance=serialized_variance,
373 sky_projection=serialized_projection,
374 mask_fractions=serialized_mask_fractions,
375 noise_realizations=serialized_noise_realizations,
376 band=self._band,
377 psf=serialized_psf,
378 aperture_corrections=serialized_aperture_corrections,
379 patch=self._patch,
380 provenance=serialized_provenance,
381 backgrounds=serialized_backgrounds,
382 metadata=self.metadata,
383 )
385 @staticmethod
386 def _get_archive_tree_type[P: pydantic.BaseModel](
387 pointer_type: type[P],
388 ) -> type[CellCoaddSerializationModel[P]]:
389 """Return the serialization model type for this object for an archive
390 type that uses the given pointer type.
391 """
392 return CellCoaddSerializationModel[pointer_type] # type: ignore
394 @staticmethod
395 def from_legacy( # type: ignore[override]
396 legacy: LegacyMultipleCellCoadd,
397 *,
398 plane_map: Mapping[str, MaskPlane] | None = None,
399 tract_info: TractInfo,
400 bbox: Box | None = None,
401 ) -> CellCoadd:
402 """Convert from a `lsst.cell_coadds.MultipleCellCoadd` instance.
404 Parameters
405 ----------
406 legacy
407 A `lsst.cell_coadds.MultipleCellCoadd` instance to convert.
408 plane_map
409 A mapping from legacy mask plane name to the new plane name and
410 description.
411 tract_info
412 Information about the full tract.
413 bbox
414 Bounding box of the image. The default is to include just the
415 bounding box of the valid cells, which may not cover a full patch.
416 """
417 from lsst.geom import Box2I
419 if plane_map is None:
420 plane_map = get_legacy_deep_coadd_mask_planes()
421 if bbox is None:
422 legacy_bbox = Box2I()
423 for single_cell in legacy.cells.values():
424 legacy_bbox.include(single_cell.inner.bbox)
425 else:
426 legacy_bbox = bbox.to_legacy()
427 legacy_stitched = legacy.stitch(legacy_bbox)
428 unit = astropy.units.Unit(legacy.units.value)
429 tract_bbox = Box.from_legacy(tract_info.getBBox())
430 sky_projection = SkyProjection.from_legacy(
431 legacy.wcs,
432 TractFrame(
433 skymap=legacy.identifiers.skymap,
434 tract=legacy.identifiers.tract,
435 bbox=tract_bbox,
436 ),
437 pixel_bounds=tract_bbox,
438 )
439 band = legacy.identifiers.band
440 image = Image.from_legacy(legacy_stitched.image, unit=unit)
441 mask = Mask.from_legacy(legacy_stitched.mask, plane_map=plane_map)
442 variance = Image.from_legacy(legacy_stitched.variance, unit=unit**2)
443 noise_realizations = [
444 Image.from_legacy(noise_image) for noise_image in legacy_stitched.noise_realizations
445 ]
446 mask_fractions = (
447 {"rejected": Image.from_legacy(legacy_stitched.mask_fractions)}
448 if legacy_stitched.mask_fractions is not None
449 else {}
450 )
451 psf = CellPointSpreadFunction.from_legacy(legacy_stitched.psf, image.bbox)
452 aperture_corrections = {
453 ap_corr_name: CellField.from_legacy_aperture_correction(legacy_ap_corr, psf.bounds)
454 for ap_corr_name, legacy_ap_corr in legacy_stitched.ap_corr_map.items()
455 }
456 patch_info = tract_info[legacy.identifiers.patch]
457 patch = PatchDefinition(
458 id=patch_info.getSequentialIndex(),
459 index=YX(y=legacy.identifiers.patch.y, x=legacy.identifiers.patch.x),
460 inner_bbox=Box.from_legacy(patch_info.getInnerBBox()),
461 cells=CellGrid.from_legacy(legacy.grid),
462 )
463 provenance = CoaddProvenance.from_legacy(legacy)
464 return CellCoadd(
465 image=image,
466 mask=mask,
467 variance=variance,
468 mask_fractions=mask_fractions,
469 noise_realizations=noise_realizations,
470 sky_projection=sky_projection,
471 band=band,
472 psf=psf,
473 aperture_corrections=aperture_corrections,
474 patch=patch,
475 provenance=provenance,
476 )
478 def to_legacy(
479 self, copy: bool | None = None, plane_map: Mapping[str, MaskPlane] | None = None
480 ) -> LegacyMultipleCellCoadd:
481 """Convert to a `lsst.cell_coadds.MultipleCellCoadd` instance.
483 Parameters
484 ----------
485 copy
486 If `True`, always copy the image and variance pixel data.
487 If `False`, return a view, and raise `TypeError` if the pixel data
488 is read-only (this is not supported by afw). If `None`, only copy
489 if the pixel data is read-only. Mask pixel data is always copied.
490 plane_map
491 A mapping from legacy mask plane name to the new plane name and
492 description.
493 """
494 from frozendict import frozendict
496 from lsst.cell_coadds import CellIdentifiers as LegacyCellIdentifiers
497 from lsst.cell_coadds import CoaddUnits as LegacyCoaddUnits
498 from lsst.cell_coadds import CommonComponents as LegacyCommonComponents
499 from lsst.cell_coadds import MultipleCellCoadd as LegacyMultipleCellCoadd
500 from lsst.cell_coadds import OwnedImagePlanes as LegacyOwnedImagePlanes
501 from lsst.cell_coadds import PatchIdentifiers as LegacyPatchIdentifiers
502 from lsst.cell_coadds import SingleCellCoadd as LegacySingleCellCoadd
503 from lsst.skymap import Index2D as LegacyIndex2D
505 if plane_map is None:
506 plane_map = get_legacy_deep_coadd_mask_planes()
507 if self.unit != astropy.units.nJy:
508 raise ValueError("CellCoadd.to_legacy requires nJy pixel units.")
509 legacy_grid = self.grid.to_legacy()
510 visit_polygons = self.provenance.to_legacy_polygon_map()
511 legacy_common = LegacyCommonComponents(
512 units=LegacyCoaddUnits.nJy,
513 wcs=self.sky_projection.to_legacy(),
514 band=self.band,
515 identifiers=LegacyPatchIdentifiers(
516 self.skymap,
517 self.tract,
518 LegacyIndex2D(x=self.patch.index.x, y=self.patch.index.y),
519 band=self.band,
520 ),
521 visit_polygons=visit_polygons,
522 )
523 legacy_inputs = self.provenance.to_legacy_cell_coadd_inputs(visit_polygons.keys())
524 cells: list[LegacySingleCellCoadd] = []
525 for cell_index in self.bounds.cell_indices():
526 cell_bbox = self.grid.bbox_of(cell_index)
527 # Legacy type only has room for one mask_fractions plane.
528 legacy_mask_fractions = (
529 next(iter(self.mask_fractions.values()))[cell_bbox].to_legacy(copy=copy)
530 if self.mask_fractions
531 else None
532 )
533 legacy_planes = LegacyOwnedImagePlanes(
534 image=self.image[cell_bbox].to_legacy(copy=copy),
535 mask=self.mask[cell_bbox].to_legacy(plane_map),
536 variance=self.variance[cell_bbox].to_legacy(copy=copy),
537 mask_fractions=legacy_mask_fractions,
538 noise_realizations=[n[cell_bbox].to_legacy(copy=copy) for n in self.noise_realizations],
539 )
540 legacy_aperture_correction_map = frozendict(
541 {name: field.value_in_cell(cell_index) for name, field in self.aperture_corrections.items()}
542 )
543 cells.append(
544 LegacySingleCellCoadd(
545 legacy_planes,
546 psf=self.psf[cell_index].to_legacy(copy=copy),
547 inner_bbox=cell_bbox.to_legacy(),
548 common=legacy_common,
549 inputs=legacy_inputs[cell_index.to_legacy()],
550 identifiers=LegacyCellIdentifiers(
551 self.skymap,
552 self.tract,
553 legacy_common.identifiers.patch,
554 band=self.band,
555 cell=cell_index.to_legacy(),
556 ),
557 aperture_correction_map=legacy_aperture_correction_map,
558 )
559 )
560 return LegacyMultipleCellCoadd(
561 cells,
562 legacy_grid,
563 outer_cell_size=self.grid.cell_shape.to_legacy_extent(),
564 psf_image_size=self.psf.kernel_bbox.shape.to_legacy_extent(),
565 common=legacy_common,
566 inner_bbox=self.bbox.to_legacy(),
567 )
569 def to_legacy_exposure(
570 self, copy: bool | None = None, plane_map: Mapping[str, MaskPlane] | None = None
571 ) -> LegacyExposure:
572 """Convert to a `lsst.afw.image.Exposure` instance.
574 Parameters
575 ----------
576 copy
577 If `True`, always copy the image and variance pixel data.
578 If `False`, return a view, and raise `TypeError` if the pixel data
579 is read-only (this is not supported by afw). If `None`, only copy
580 if the pixel data is read-only. Mask pixel data is always copied.
581 plane_map
582 A mapping from legacy mask plane name to the new plane name and
583 description.
585 Returns
586 -------
587 `lsst.afw.image.Exposure`
588 A legacy representation of the coadd. This will have its ``wcs``,
589 ``psf``, ``filter``, ``photoCalib``, and ``metadata`` components
590 set. The ``apCorrMap`` component is not set, because there is no
591 true `lsst.afw.math.BoundedField` representation for cell-coadd
592 aperture corrections, and the ``coaddInputs`` component is not set
593 because that data structure cannot fully capture cell-coadd
594 provenance.
596 Notes
597 -----
598 This method requires the `provenance` attribute to have been populated
599 at construction.
600 """
601 from lsst.afw.image import Exposure as LegacyExposure
602 from lsst.afw.image import FilterLabel as LegacyFilterLabel
604 if plane_map is None:
605 plane_map = get_legacy_deep_coadd_mask_planes()
606 legacy_masked_image = super().to_legacy(copy=copy, plane_map=plane_map)
607 result = LegacyExposure(legacy_masked_image, dtype=self.image.array.dtype)
608 result_info = result.info
609 result_info.setWcs(self.sky_projection.to_legacy())
610 result_info.setPsf(self.psf.to_legacy())
611 result_info.setFilter(LegacyFilterLabel.fromBand(self.band))
612 result_info.setPhotoCalib(BaseField.make_legacy_photo_calib(self.unit))
613 # We don't do setCoaddInputs because that data structure can't really
614 # represent cell-coadd provenance accurately, and it's not clear
615 # anything would use it.
616 self._fill_legacy_metadata(result_info.getMetadata())
617 # We can't do setApCorrMap because the legacy
618 # StitchedApertureCorrection is not a real C++ BoundedField, just a
619 # Python duck-alike.
620 return result
623class CellCoaddSerializationModel[P: pydantic.BaseModel](MaskedImageSerializationModel[P]):
624 """A Pydantic model used to represent a serialized `CellCoadd`."""
626 SCHEMA_NAME: ClassVar[str] = "cell_coadd"
627 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
628 MIN_READ_VERSION: ClassVar[int] = 1
629 PUBLIC_TYPE: ClassVar[type] = CellCoadd
631 # Inherited attributes are duplicated because that improves the docs
632 # (some limitation in the sphinx/pydantic integration), and these are
633 # important docs.
635 image: ImageSerializationModel[P] = pydantic.Field(description="The main data image.")
636 mask: MaskSerializationModel[P] = pydantic.Field(
637 description="Bitmask that annotates the main image's pixels."
638 )
639 variance: ImageSerializationModel[P] = pydantic.Field(
640 description="Per-pixel variance estimates for the main image."
641 )
642 sky_projection: SkyProjectionSerializationModel[P] = pydantic.Field(
643 description="Projection that maps the pixel grid to the sky.",
644 )
645 mask_fractions: dict[str, ImageSerializationModel[P]] = pydantic.Field(
646 description=(
647 "A mapping from an input-image mask plane name to an image of the weights sums of that plane."
648 )
649 )
650 noise_realizations: list[ImageSerializationModel[P]] = pydantic.Field(
651 description=(
652 "A mapping from an input-image mask plane name to an image of the weights sums of that plane."
653 )
654 )
655 band: str | None = pydantic.Field(description="Name of the band.")
656 psf: CellPointSpreadFunctionSerializationModel = pydantic.Field(
657 description="Effective point-spread function model for the coadd."
658 )
659 aperture_corrections: CellApertureCorrectionMapSerializationModel | None = pydantic.Field(
660 None, description="Coadded aperture corrections for different photometry algorithms."
661 )
662 patch: PatchDefinition | None = pydantic.Field(description="Identifiers and geometry for the patch.")
663 provenance: CoaddProvenanceSerializationModel | None = pydantic.Field(
664 description="Information about the images that went into the coadd."
665 )
666 backgrounds: BackgroundMapSerializationModel = pydantic.Field(
667 default_factory=BackgroundMapSerializationModel,
668 description="Background models associated with this image.",
669 )
671 def deserialize(
672 self,
673 archive: InputArchive[Any],
674 *,
675 bbox: Box | None = None,
676 provenance: bool = True,
677 **kwargs: Any,
678 ) -> CellCoadd:
679 """Deserialize an image from an input archive.
681 Parameters
682 ----------
683 archive
684 Archive to read from.
685 bbox
686 Bounding box of a subimage to read instead.
687 provenance
688 Whether to read and attach provenance information.
689 **kwargs
690 Unsupported keyword arguments are accepted only to provide better
691 error messages (raising `.serialization.InvalidParameterError`).
692 """
693 if kwargs: 693 ↛ 694line 693 didn't jump to line 694 because the condition on line 693 was never true
694 raise InvalidParameterError(f"Unrecognized parameters for CellCoadd: {set(kwargs.keys())}.")
695 masked_image = super().deserialize(archive, bbox=bbox)
696 mask_fractions = {
697 k.removeprefix("mask_fractions/"): v.deserialize(archive, bbox=bbox)
698 for k, v in self.mask_fractions.items()
699 }
700 noise_realizations = [v.deserialize(archive, bbox=bbox) for v in self.noise_realizations]
701 sky_projection = self.sky_projection.deserialize(archive)
702 psf = self.psf.deserialize(archive, bbox=bbox)
703 aperture_corrections = (
704 self.aperture_corrections.deserialize(archive) if self.aperture_corrections is not None else {}
705 )
706 coadd_provenance: CoaddProvenance | None = None
707 if self.provenance is not None and provenance: 707 ↛ 711line 707 didn't jump to line 711 because the condition on line 707 was always true
708 coadd_provenance = self.provenance.deserialize(archive)
709 if bbox is not None: 709 ↛ 710line 709 didn't jump to line 710 because the condition on line 709 was never true
710 coadd_provenance = coadd_provenance.subset(psf.bounds.cell_indices())
711 backgrounds = self.backgrounds.deserialize(archive)
712 return CellCoadd(
713 masked_image.image,
714 mask=masked_image.mask,
715 variance=masked_image.variance,
716 mask_fractions=mask_fractions,
717 noise_realizations=noise_realizations,
718 sky_projection=sky_projection,
719 band=self.band,
720 psf=psf,
721 aperture_corrections=aperture_corrections,
722 patch=self.patch,
723 provenance=coadd_provenance,
724 backgrounds=backgrounds,
725 )._finish_deserialize(self)
727 def deserialize_component(self, component: str, archive: InputArchive[Any], **kwargs: Any) -> Any:
728 match component:
729 case "mask_fractions":
730 return {
731 name: image_model.deserialize(archive, **kwargs)
732 for name, image_model in self.mask_fractions.items()
733 }
734 case "noise_realizations":
735 return [image_model.deserialize(archive, **kwargs) for image_model in self.noise_realizations]
736 case "aperture_corrections" if self.aperture_corrections is None:
737 # super() delegation handles the not-None case.
738 return {}
739 case "masked_image":
740 return super().deserialize(archive, **kwargs)
741 return super().deserialize_component(component, archive, **kwargs)