Coverage for python/lsst/images/psfs/_legacy.py: 1%

1# This file is part of lsst-images. 

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# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12from __future__ import annotations 

13 

14__all__ = ("LegacyPointSpreadFunction", "PSFExSerializationModel", "PSFExWrapper") 

15 

16from functools import cached_property 

17from typing import Any, ClassVar 

18 

19import numpy as np 

20import pydantic 

21 

22from .. import serialization 

23from .._concrete_bounds import BoundsSerializationModel 

24from .._geom import Bounds, Box 

25from .._image import Image 

26from ._base import PointSpreadFunction 

27 

28 

29class LegacyPointSpreadFunction(PointSpreadFunction): 

30 """A PSF model backed by a legacy `lsst.afw.detection.Psf` object. 

31 

32 Parameters 

33 ---------- 

34 impl 

35 An `lsst.afw.detection.Psf` instance. 

36 bounds 

37 The pixel-coordinate region where the model can safely be evaluated. 

38 

39 Notes 

40 ----- 

41 This wrapper is usable as-is on any `lsst.afw.detection.Psf` instance, 

42 but subclasses (e.g. `PSFExWrapper`) must be used for serialization. 

43 """ 

44 

45 def __init__(self, impl: Any, bounds: Bounds) -> None: 

46 self._impl = impl 

47 self._bounds = bounds 

48 

49 @property 

50 def bounds(self) -> Bounds: 

51 return self._bounds 

52 

53 @cached_property 

54 def kernel_bbox(self) -> Box: 

55 from lsst.geom import Box2I, Point2D 

56 

57 biggest = Box2I() 

58 for y, x in self._bounds.bbox.boundary(): 

59 biggest.include(self._impl.computeKernelBBox(Point2D(x, y))) 

60 return Box.from_legacy(biggest) 

61 

62 def compute_kernel_image(self, *, x: float, y: float) -> Image: 

63 from lsst.geom import Point2D 

64 

65 result = Image.from_legacy(self._impl.computeKernelImage(Point2D(x, y))) 

66 if result.bbox != self.kernel_bbox: 

67 # afw does not guarantee a consistent kernel_bbox, but we do now. 

68 padded = Image(0.0, bbox=self.kernel_bbox, dtype=np.float64) 

69 padded[self.kernel_bbox] = result[self.kernel_bbox] 

70 result = padded 

71 return result 

72 

73 def compute_stellar_image(self, *, x: float, y: float) -> Image: 

74 from lsst.geom import Point2D 

75 

76 return Image.from_legacy(self._impl.computeImage(Point2D(x, y))) 

77 

78 def compute_stellar_bbox(self, *, x: float, y: float) -> Box: 

79 from lsst.geom import Point2D 

80 

81 return Box.from_legacy(self._impl.computeImageBBox(Point2D(x, y))) 

82 

83 @property 

84 def legacy_psf(self) -> Any: 

85 """The backing `lsst.afw.detection.Psf` object.""" 

86 return self._impl 

87 

88 @classmethod 

89 def from_legacy(cls, legacy_psf: Any, bounds: Bounds) -> LegacyPointSpreadFunction: 

90 from lsst.meas.extensions.psfex import PsfexPsf 

91 

92 if isinstance(legacy_psf, PsfexPsf): 

93 return PSFExWrapper(legacy_psf, bounds) 

94 return cls(impl=legacy_psf, bounds=bounds) 

95 

96 def to_legacy(self) -> Any: 

97 return self.legacy_psf 

98 

99 

100class PSFExWrapper(LegacyPointSpreadFunction): 

101 """A specialization of LegacyPointSpreadFunction for the PSFEx backend. 

102 

103 Parameters 

104 ---------- 

105 impl 

106 A `lsst.meas.extensions.psfex.PsfexPsf` instance. 

107 bounds 

108 The pixel-coordinate region where the model can safely be 

109 evaluated. 

110 """ 

111 

112 def __init__(self, impl: Any, bounds: Bounds) -> None: 

113 from lsst.meas.extensions.psfex import PsfexPsf 

114 

115 if not isinstance(impl, PsfexPsf): 

116 raise TypeError(f"{impl!r} is not a PSFEx object.") 

117 super().__init__(impl, bounds) 

118 

119 def serialize(self, archive: serialization.OutputArchive[Any]) -> PSFExSerializationModel: 

120 """Serialize the PSF to an archive. 

121 

122 This method is intended to be usable as the callback function passed to 

123 `.serialization.OutputArchive.serialize_direct` or 

124 `.serialization.OutputArchive.serialize_pointer`. 

125 

126 Parameters 

127 ---------- 

128 archive 

129 Archive to write to. 

130 """ 

131 data = self._impl.getSerializationData() 

132 shape = tuple(reversed(data.size)) 

133 array_ref = archive.add_array(data.comp.reshape(*shape), name="parameters") 

134 return PSFExSerializationModel( 

135 average_x=data.average_x, 

136 average_y=data.average_y, 

137 pixel_step=data.pixel_step, 

138 group=data.group, 

139 degree=data.degree, 

140 basis=data.basis, 

141 coeff=data.coeff, 

142 parameters=array_ref, 

143 context=data.context, 

144 bounds=self.bounds.serialize(), 

145 ) 

146 

147 @staticmethod 

148 def _get_archive_tree_type( 

149 pointer_type: type[pydantic.BaseModel], 

150 ) -> type[PSFExSerializationModel]: 

151 """Return the serialization model type for this object for an archive 

152 type that uses the given pointer type. 

153 """ 

154 return PSFExSerializationModel 

155 

156 

157class PSFExSerializationModel(serialization.ArchiveTree): 

158 """Serialization model for PSFEx PSFs.""" 

159 

160 SCHEMA_NAME: ClassVar[str] = "psfex_psf" 

161 SCHEMA_VERSION: ClassVar[str] = "1.0.0.dev0" 

162 MIN_READ_VERSION: ClassVar[int] = 1 

163 PUBLIC_TYPE: ClassVar[type] = PSFExWrapper 

164 

165 average_x: float = pydantic.Field( 

166 description="Average X position of the stars used to build this PSF model." 

167 ) 

168 

169 average_y: float = pydantic.Field( 

170 description="Average Y position of the stars used to build this PSF model." 

171 ) 

172 

173 pixel_step: float = pydantic.Field( 

174 description="Size of a model pixel, as a fraction or multiple of the native pixel size." 

175 ) 

176 

177 group: list[int] = pydantic.Field( 

178 default_factory=lambda: [0, 0], 

179 exclude_if=lambda v: v == [0, 0], 

180 description="Number of model groups in each dimension.", 

181 ) 

182 

183 degree: list[int] = pydantic.Field(description="Polynomial degree for each model group.") 

184 

185 basis: list[float] = pydantic.Field(description="Basis function values.") 

186 

187 coeff: list[float] = pydantic.Field(description="Polynomial coefficients.") 

188 

189 parameters: serialization.ArrayReferenceModel | serialization.InlineArrayModel = pydantic.Field( 

190 description="Reference to an array with the complete model parameters." 

191 ) 

192 

193 context: serialization.InlineArray = pydantic.Field(description="Internal PSFEx context array.") 

194 

195 bounds: BoundsSerializationModel = pydantic.Field(description="Validity range for this PSF model.") 

196 

197 model_config = pydantic.ConfigDict(ser_json_inf_nan="constants") 

198 

199 def deserialize(self, archive: serialization.InputArchive[Any], **kwargs: Any) -> PSFExWrapper: 

200 """Deserialize the PSF from an archive. 

201 

202 This method is intended to be usable as the callback function passed to 

203 `.serialization.InputArchive.deserialize_pointer`. 

204 

205 Parameters 

206 ---------- 

207 archive 

208 Archive to read from. 

209 **kwargs 

210 Unsupported keyword arguments are accepted only to provide 

211 better error messages (raising 

212 `.serialization.InvalidParameterError`). 

213 """ 

214 if kwargs: 

215 raise serialization.InvalidParameterError( 

216 f"Unrecognized parameters for PsfExWrapper: {set(kwargs.keys())}." 

217 ) 

218 try: 

219 from lsst.meas.extensions.psfex import PsfexPsf, PsfexPsfSerializationData 

220 except ImportError: 

221 raise serialization.ArchiveReadError("Failed to import lsst.meas.extensions.psfex.") from None 

222 

223 parameters = archive.get_array(self.parameters).astype(np.float32) 

224 data = PsfexPsfSerializationData() 

225 data.average_x = self.average_x 

226 data.average_y = self.average_y 

227 data.pixel_step = self.pixel_step 

228 data.group = self.group 

229 data.degree = self.degree 

230 data.basis = self.basis 

231 data.coeff = self.coeff 

232 data.size = list(reversed(parameters.shape)) 

233 data.comp = parameters.flatten() 

234 data.context = self.context 

235 legacy_psf = PsfexPsf.fromSerializationData(data) 

236 return PSFExWrapper(legacy_psf, self.bounds.deserialize())