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