Coverage for python/lsst/pipe/tasks/extended_psf/extended_psf_image.py: 92%
101 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-07 01:49 -0700
1# This file is part of pipe_tasks.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <https://www.gnu.org/licenses/>.
22from __future__ import annotations
24__all__ = (
25 "ExtendedPsfImageInfo",
26 "ExtendedPsfImageSerializationModel",
27 "ExtendedPsfImage",
28)
30import functools
31from types import EllipsisType
32from typing import Any, ClassVar
34import numpy as np
35from astropy.units import UnitBase
36from pydantic import BaseModel, Field
38from lsst.images import Box, GeneralizedImage, Image, ImageSerializationModel
39from lsst.images.serialization import ArchiveTree, InputArchive, MetadataValue, OutputArchive
41from .extended_psf_fit import ExtendedPsfFit, ExtendedPsfMoffatFit
44class ExtendedPsfImageInfo(BaseModel):
45 """Additional information about an `ExtendedPsfImage`.
47 Attributes
48 ----------
49 n_stars : `int`, optional
50 Number of stars used to construct the extended PSF image.
51 """
53 n_stars: int | None = None
55 def __str__(self) -> str:
56 attrs = ", ".join(f"{k}={v!r}" for k, v in self.__dict__.items())
57 return f"ExtendedPsfImageInfo({attrs})"
59 __repr__ = __str__
62class ExtendedPsfImage(GeneralizedImage):
63 """A multi-plane image with data (image) and variance planes, and the
64 results of a profile fit to the image.
66 Parameters
67 ----------
68 image : `~lsst.images.Image`
69 The main image plane.
70 variance : `~lsst.images.Image`, optional
71 The per-pixel uncertainty of the main image as an image of variance
72 values. Must have the same bounding box as ``image`` if provided, and
73 its units must be the square of ``image.unit`` or `None`.
74 Values default to ``1.0``. Any attached ``sky_projection`` is replaced
75 (possibly by `None`).
76 info : `ExtendedPsfImageInfo`, optional
77 Additional information about how the extended PSF image was
78 constructed.
79 fit : `ExtendedPsfFit`, optional
80 The results of a profile fit to the image.
81 metadata : `dict` [`str`, `MetadataValue`], optional
82 Arbitrary flexible metadata to associate with the image.
84 Attributes
85 ----------
86 image : `~lsst.images.Image`
87 The main image plane.
88 variance : `~lsst.images.Image`
89 The per-pixel uncertainty of the main image as an image of variance
90 values.
91 info : `ExtendedPsfImageInfo`
92 Additional information about how the extended PSF image was
93 constructed.
94 fit : `ExtendedPsfFit`
95 The results of a profile fit to the image.
96 """
98 def __init__(
99 self,
100 image: Image,
101 *,
102 variance: Image | None = None,
103 info: ExtendedPsfImageInfo | None = None,
104 fit: ExtendedPsfFit | None = None,
105 metadata: dict[str, MetadataValue] | None = None,
106 ):
107 super().__init__(metadata)
108 if variance is None:
109 variance = Image(
110 1.0,
111 dtype=np.float32,
112 bbox=image.bbox,
113 unit=None if image.unit is None else image.unit**2,
114 )
115 else:
116 if image.bbox != variance.bbox:
117 raise ValueError(f"Image ({image.bbox}) and variance ({variance.bbox}) bboxes do not agree.")
118 if image.unit is None:
119 if variance.unit is not None:
120 raise ValueError(f"Image has no units but variance does ({variance.unit}).")
121 elif variance.unit is None: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 variance = variance.view(unit=image.unit**2)
123 elif variance.unit != image.unit**2:
124 raise ValueError(
125 f"Variance unit ({variance.unit}) should be the square of the image unit ({image.unit})."
126 )
127 if info is None:
128 info = ExtendedPsfImageInfo()
129 if fit is None:
130 fit = ExtendedPsfFit(chi2=np.nan, reduced_chi2=np.nan)
131 self._image = image
132 self._variance = variance
133 self._info = info
134 self._fit = fit
136 @property
137 def image(self) -> Image:
138 """The main image plane (`Image`)."""
139 return self._image
141 @property
142 def variance(self) -> Image:
143 """The variance plane (`Image`)."""
144 return self._variance
146 @property
147 def bbox(self) -> Box:
148 """The bounding box shared by both image planes (`Box`)."""
149 return self._image.bbox
151 @property
152 def unit(self) -> UnitBase | None:
153 """The units of the image plane (`astropy.units.Unit` | `None`)."""
154 return self._image.unit
156 @property
157 def sky_projection(self) -> None:
158 """The projection that maps the pixel grid to the sky.
160 ExtendedPsfImage does not support attached projections,
161 so this always returns `None`.
162 """
163 return None
165 @property
166 def info(self) -> ExtendedPsfImageInfo:
167 """Additional information about the image (`ExtendedPsfImageInfo`)."""
168 return self._info
170 @property
171 def fit(self) -> ExtendedPsfFit:
172 """The results of a profile fit to the image."""
173 return self._fit
175 def __getitem__(self, bbox: Box | EllipsisType) -> ExtendedPsfImage:
176 super().__getitem__(bbox)
177 if bbox is ...: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true
178 return self
179 return self._transfer_metadata(
180 ExtendedPsfImage(
181 self.image[bbox],
182 variance=self.variance[bbox],
183 info=self.info,
184 fit=self.fit,
185 ),
186 bbox=bbox,
187 )
189 def __setitem__(self, bbox: Box | EllipsisType, value: ExtendedPsfImage) -> None:
190 self._image[bbox] = value.image
191 self._variance[bbox] = value.variance
193 def __str__(self) -> str:
194 return f"ExtendedPsfImage({self.image!s}, info={self.info!r}, fit={self.fit!r})"
196 __repr__ = __str__
198 def copy(self) -> ExtendedPsfImage:
199 """Deep-copy the profile image and metadata."""
200 return self._transfer_metadata(
201 ExtendedPsfImage(
202 image=self._image.copy(),
203 variance=self._variance.copy(),
204 info=self._info.model_copy(),
205 fit=self._fit.model_copy(),
206 ),
207 copy=True,
208 )
210 def serialize(self, archive: OutputArchive[Any]) -> ExtendedPsfImageSerializationModel:
211 """Serialize the Extended PSF image to an output archive.
213 Parameters
214 ----------
215 archive
216 Archive to write to.
217 """
218 serialized_image = archive.serialize_direct(
219 "image", functools.partial(self.image.serialize, save_projection=False)
220 )
221 serialized_variance = archive.serialize_direct(
222 "variance", functools.partial(self.variance.serialize, save_projection=False)
223 )
224 serialized_info = self.info
225 serialized_fit = self.fit
226 return ExtendedPsfImageSerializationModel(
227 image=serialized_image,
228 variance=serialized_variance,
229 info=serialized_info,
230 fit=serialized_fit,
231 metadata=self.metadata,
232 )
234 @staticmethod
235 def deserialize(
236 model: ExtendedPsfImageSerializationModel[Any], archive: InputArchive[Any], *, bbox: Box | None = None
237 ) -> ExtendedPsfImage:
238 """Deserialize an image from an input archive.
240 Parameters
241 ----------
242 model
243 A Pydantic model representation of the image, holding references
244 to data stored in the archive.
245 archive
246 Archive to read from.
247 bbox
248 Bounding box of a subimage to read instead.
249 """
250 return model.deserialize(archive, bbox=bbox)
252 @staticmethod
253 def _get_archive_tree_type[P: BaseModel](
254 pointer_type: type[P],
255 ) -> type[ExtendedPsfImageSerializationModel[P]]:
256 """Return the serialization model type for this object for an archive
257 type that uses the given pointer type.
258 """
259 return ExtendedPsfImageSerializationModel[pointer_type] # type: ignore
262class ExtendedPsfImageSerializationModel[P: BaseModel](ArchiveTree):
263 """A Pydantic model used to represent a serialized `ExtendedPsfImage`."""
265 SCHEMA_NAME: ClassVar[str] = "extended_psf_image"
266 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
267 MIN_READ_VERSION: ClassVar[int] = 1
268 PUBLIC_TYPE: ClassVar[type] = ExtendedPsfImage
270 image: ImageSerializationModel[P] = Field(
271 description="The main data image.",
272 )
273 variance: ImageSerializationModel[P] = Field(
274 description="Per-pixel variance estimates for the main image."
275 )
276 info: ExtendedPsfImageInfo = Field(
277 description="Additional information about the extended PSF image.",
278 )
279 fit: ExtendedPsfMoffatFit | ExtendedPsfFit = Field(
280 description="The results of an extended PSF fit to the image.",
281 )
283 @property
284 def bbox(self) -> Box:
285 """The bounding box of the image."""
286 return self.image.bbox
288 def deserialize(self, archive: InputArchive[Any], *, bbox: Box | None = None) -> ExtendedPsfImage:
289 """Deserialize an image from an input archive.
291 Parameters
292 ----------
293 archive
294 Archive to read from.
295 bbox
296 Bounding box of a subimage to read instead.
297 """
298 image = self.image.deserialize(archive, bbox=bbox)
299 variance = self.variance.deserialize(archive, bbox=bbox)
300 return ExtendedPsfImage(
301 image,
302 variance=variance,
303 info=self.info,
304 fit=self.fit,
305 )._finish_deserialize(self)