Coverage for python/lsst/pipe/tasks/extended_psf/extended_psf_image.py: 44%
100 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 01:16 -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 ExtendedPsfImageSerializationModel[P: BaseModel](ArchiveTree):
63 """A Pydantic model used to represent a serialized `ExtendedPsfImage`."""
65 SCHEMA_NAME: ClassVar[str] = "extended_psf_image"
66 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
67 MIN_READ_VERSION: ClassVar[int] = 1
69 image: ImageSerializationModel[P] = Field(
70 description="The main data image.",
71 )
72 variance: ImageSerializationModel[P] = Field(
73 description="Per-pixel variance estimates for the main image."
74 )
75 info: ExtendedPsfImageInfo = Field(
76 description="Additional information about the extended PSF image.",
77 )
78 fit: ExtendedPsfMoffatFit | ExtendedPsfFit = Field(
79 description="The results of an extended PSF fit to the image.",
80 )
82 @property
83 def bbox(self) -> Box:
84 """The bounding box of the image."""
85 return self.image.bbox
87 def deserialize(self, archive: InputArchive[Any], *, bbox: Box | None = None) -> ExtendedPsfImage:
88 """Deserialize an image from an input archive.
90 Parameters
91 ----------
92 archive
93 Archive to read from.
94 bbox
95 Bounding box of a subimage to read instead.
96 """
97 image = self.image.deserialize(archive, bbox=bbox)
98 variance = self.variance.deserialize(archive, bbox=bbox)
99 return ExtendedPsfImage(
100 image,
101 variance=variance,
102 info=self.info,
103 fit=self.fit,
104 )._finish_deserialize(self)
107class ExtendedPsfImage(GeneralizedImage):
108 """A multi-plane image with data (image) and variance planes, and the
109 results of a profile fit to the image.
111 Parameters
112 ----------
113 image : `~lsst.images.Image`
114 The main image plane.
115 variance : `~lsst.images.Image`, optional
116 The per-pixel uncertainty of the main image as an image of variance
117 values. Must have the same bounding box as ``image`` if provided, and
118 its units must be the square of ``image.unit`` or `None`.
119 Values default to ``1.0``. Any attached projection is replaced
120 (possibly by `None`).
121 info : `ExtendedPsfImageInfo`, optional
122 Additional information about how the extended PSF image was
123 constructed.
124 fit : `ExtendedPsfFit`, optional
125 The results of a profile fit to the image.
126 metadata : `dict` [`str`, `MetadataValue`], optional
127 Arbitrary flexible metadata to associate with the image.
129 Attributes
130 ----------
131 image : `~lsst.images.Image`
132 The main image plane.
133 variance : `~lsst.images.Image`
134 The per-pixel uncertainty of the main image as an image of variance
135 values.
136 bbox : `~lsst.images.Box`
137 The bounding box shared by both image planes.
138 unit : `astropy.units.Unit` or `None`
139 The units of the image plane, or `None` if the image is dimensionless.
140 projection : `None`
141 The projection that maps the pixel grid to the sky. Always `None` for
142 `ExtendedPsfImage`.
143 info : `ExtendedPsfImageInfo`
144 Additional information about how the extended PSF image was
145 constructed.
146 fit : `ExtendedPsfFit`
147 The results of a profile fit to the image.
148 """
150 def __init__(
151 self,
152 image: Image,
153 *,
154 variance: Image | None = None,
155 info: ExtendedPsfImageInfo | None = None,
156 fit: ExtendedPsfFit | None = None,
157 metadata: dict[str, MetadataValue] | None = None,
158 ):
159 super().__init__(metadata)
160 if variance is None:
161 variance = Image(
162 1.0,
163 dtype=np.float32,
164 bbox=image.bbox,
165 unit=None if image.unit is None else image.unit**2,
166 )
167 else:
168 if image.bbox != variance.bbox:
169 raise ValueError(f"Image ({image.bbox}) and variance ({variance.bbox}) bboxes do not agree.")
170 if image.unit is None:
171 if variance.unit is not None:
172 raise ValueError(f"Image has no units but variance does ({variance.unit}).")
173 elif variance.unit is None:
174 variance = variance.view(unit=image.unit**2)
175 elif variance.unit != image.unit**2:
176 raise ValueError(
177 f"Variance unit ({variance.unit}) should be the square of the image unit ({image.unit})."
178 )
179 if info is None:
180 info = ExtendedPsfImageInfo()
181 if fit is None:
182 fit = ExtendedPsfFit(chi2=np.nan, reduced_chi2=np.nan)
183 self._image = image
184 self._variance = variance
185 self._info = info
186 self._fit = fit
188 @property
189 def image(self) -> Image:
190 """The main image plane (`Image`)."""
191 return self._image
193 @property
194 def variance(self) -> Image:
195 """The variance plane (`Image`)."""
196 return self._variance
198 @property
199 def bbox(self) -> Box:
200 """The bounding box shared by both image planes (`Box`)."""
201 return self._image.bbox
203 @property
204 def unit(self) -> UnitBase | None:
205 """The units of the image plane (`astropy.units.Unit` | `None`)."""
206 return self._image.unit
208 @property
209 def projection(self) -> None:
210 """The projection that maps the pixel grid to the sky.
212 ExtendedPsfImage does not support attached projections,
213 so this always returns `None`.
214 """
215 return None
217 @property
218 def info(self) -> ExtendedPsfImageInfo:
219 """Additional information about the image (`ExtendedPsfImageInfo`)."""
220 return self._info
222 @property
223 def fit(self) -> ExtendedPsfFit:
224 """The results of a profile fit to the image."""
225 return self._fit
227 def __getitem__(self, bbox: Box | EllipsisType) -> ExtendedPsfImage:
228 super().__getitem__(bbox)
229 if bbox is ...:
230 return self
231 return self._transfer_metadata(
232 ExtendedPsfImage(
233 self.image[bbox],
234 variance=self.variance[bbox],
235 info=self.info,
236 fit=self.fit,
237 ),
238 bbox=bbox,
239 )
241 def __setitem__(self, bbox: Box | EllipsisType, value: ExtendedPsfImage) -> None:
242 self._image[bbox] = value.image
243 self._variance[bbox] = value.variance
245 def __str__(self) -> str:
246 return f"ExtendedPsfImage({self.image!s}, info={self.info!r}, fit={self.fit!r})"
248 __repr__ = __str__
250 def copy(self) -> ExtendedPsfImage:
251 """Deep-copy the profile image and metadata."""
252 return self._transfer_metadata(
253 ExtendedPsfImage(
254 image=self._image.copy(),
255 variance=self._variance.copy(),
256 info=self._info.model_copy(),
257 fit=self._fit.model_copy(),
258 ),
259 copy=True,
260 )
262 def serialize(self, archive: OutputArchive[Any]) -> ExtendedPsfImageSerializationModel:
263 """Serialize the Extended PSF image to an output archive.
265 Parameters
266 ----------
267 archive
268 Archive to write to.
269 """
270 serialized_image = archive.serialize_direct(
271 "image", functools.partial(self.image.serialize, save_projection=False)
272 )
273 serialized_variance = archive.serialize_direct(
274 "variance", functools.partial(self.variance.serialize, save_projection=False)
275 )
276 serialized_info = self.info
277 serialized_fit = self.fit
278 return ExtendedPsfImageSerializationModel(
279 image=serialized_image,
280 variance=serialized_variance,
281 info=serialized_info,
282 fit=serialized_fit,
283 metadata=self.metadata,
284 )
286 @staticmethod
287 def deserialize(
288 model: ExtendedPsfImageSerializationModel[Any], archive: InputArchive[Any], *, bbox: Box | None = None
289 ) -> ExtendedPsfImage:
290 """Deserialize an image from an input archive.
292 Parameters
293 ----------
294 model
295 A Pydantic model representation of the image, holding references
296 to data stored in the archive.
297 archive
298 Archive to read from.
299 bbox
300 Bounding box of a subimage to read instead.
301 """
302 return model.deserialize(archive, bbox=bbox)
304 @staticmethod
305 def _get_archive_tree_type[P: BaseModel](
306 pointer_type: type[P],
307 ) -> type[ExtendedPsfImageSerializationModel[P]]:
308 """Return the serialization model type for this object for an archive
309 type that uses the given pointer type.
310 """
311 return ExtendedPsfImageSerializationModel[pointer_type] # type: ignore