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