Coverage for python/lsst/pipe/tasks/extended_psf/extended_psf_image.py: 93%
99 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-25 09:10 +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, 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 bbox, _ = self._handle_getitem_args(bbox)
177 return self._transfer_metadata(
178 ExtendedPsfImage(
179 self.image[bbox],
180 variance=self.variance[bbox],
181 info=self.info,
182 fit=self.fit,
183 ),
184 bbox=bbox,
185 )
187 def __setitem__(self, bbox: Box | EllipsisType, value: ExtendedPsfImage) -> None:
188 self._image[bbox] = value.image
189 self._variance[bbox] = value.variance
191 def __str__(self) -> str:
192 return f"ExtendedPsfImage({self.image!s}, info={self.info!r}, fit={self.fit!r})"
194 __repr__ = __str__
196 def copy(self) -> ExtendedPsfImage:
197 """Deep-copy the profile image and metadata."""
198 return self._transfer_metadata(
199 ExtendedPsfImage(
200 image=self._image.copy(),
201 variance=self._variance.copy(),
202 info=self._info.model_copy(),
203 fit=self._fit.model_copy(),
204 ),
205 copy=True,
206 )
208 def serialize(self, archive: OutputArchive[Any]) -> ExtendedPsfImageSerializationModel:
209 """Serialize the Extended PSF image to an output archive.
211 Parameters
212 ----------
213 archive
214 Archive to write to.
215 """
216 serialized_image = archive.serialize_direct(
217 "image", functools.partial(self.image.serialize, save_projection=False)
218 )
219 serialized_variance = archive.serialize_direct(
220 "variance", functools.partial(self.variance.serialize, save_projection=False)
221 )
222 serialized_info = self.info
223 serialized_fit = self.fit
224 return ExtendedPsfImageSerializationModel(
225 image=serialized_image,
226 variance=serialized_variance,
227 info=serialized_info,
228 fit=serialized_fit,
229 metadata=self.metadata,
230 )
232 @staticmethod
233 def deserialize(
234 model: ExtendedPsfImageSerializationModel[Any], archive: InputArchive[Any], *, bbox: Box | None = None
235 ) -> ExtendedPsfImage:
236 """Deserialize an image from an input archive.
238 Parameters
239 ----------
240 model
241 A Pydantic model representation of the image, holding references
242 to data stored in the archive.
243 archive
244 Archive to read from.
245 bbox
246 Bounding box of a subimage to read instead.
247 """
248 return model.deserialize(archive, bbox=bbox)
250 @staticmethod
251 def _get_archive_tree_type[P: BaseModel](
252 pointer_type: type[P],
253 ) -> type[ExtendedPsfImageSerializationModel[P]]:
254 """Return the serialization model type for this object for an archive
255 type that uses the given pointer type.
256 """
257 return ExtendedPsfImageSerializationModel[pointer_type] # type: ignore
260class ExtendedPsfImageSerializationModel[P: BaseModel](ArchiveTree):
261 """A Pydantic model used to represent a serialized `ExtendedPsfImage`."""
263 SCHEMA_NAME: ClassVar[str] = "extended_psf_image"
264 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
265 MIN_READ_VERSION: ClassVar[int] = 1
266 PUBLIC_TYPE: ClassVar[type] = ExtendedPsfImage
268 image: ImageSerializationModel[P] = Field(
269 description="The main data image.",
270 )
271 variance: ImageSerializationModel[P] = Field(
272 description="Per-pixel variance estimates for the main image."
273 )
274 info: ExtendedPsfImageInfo = Field(
275 description="Additional information about the extended PSF image.",
276 )
277 fit: ExtendedPsfMoffatFit | ExtendedPsfFit = Field(
278 description="The results of an extended PSF fit to the image.",
279 )
281 @property
282 def bbox(self) -> Box:
283 """The bounding box of the image."""
284 return self.image.bbox
286 def deserialize(self, archive: InputArchive[Any], *, bbox: Box | None = None) -> ExtendedPsfImage:
287 """Deserialize an image from an input archive.
289 Parameters
290 ----------
291 archive
292 Archive to read from.
293 bbox
294 Bounding box of a subimage to read instead.
295 """
296 image = self.image.deserialize(archive, bbox=bbox)
297 variance = self.variance.deserialize(archive, bbox=bbox)
298 return ExtendedPsfImage(
299 image,
300 variance=variance,
301 info=self.info,
302 fit=self.fit,
303 )._finish_deserialize(self)