Coverage for python/lsst/afw/image/_exposure/_multiband.py: 19%

105 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-05-29 01:21 -0700

1# This file is part of afw. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://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 <http://www.gnu.org/licenses/>. 

21 

22__all__ = ["MultibandExposure", "computePsfImage", "IncompleteDataError"] 

23 

24import numpy as np 

25 

26from lsst.geom import Point2D, Point2I, Box2I 

27from lsst.pex.exceptions import InvalidParameterError 

28from . import Exposure, ExposureF 

29from ..utils import projectImage 

30from .._image._multiband import MultibandTripleBase, MultibandPixel, MultibandImage 

31from .._image._multiband import tripleFromSingles, tripleFromArrays, makeTripleFromKwargs 

32from .._maskedImage import MaskedImage 

33 

34 

35class IncompleteDataError(Exception): 

36 """The PSF could not be computed due to incomplete data 

37 

38 Attributes 

39 ---------- 

40 missingBands: `list[str]` 

41 The bands for which the PSF could not be calculated. 

42 position: `Point2D` 

43 The point at which the PSF could not be calcualted in the 

44 missing bands. 

45 partialPsf: `MultibandImage` 

46 The image of the PSF using only the bands that successfully 

47 computed a PSF image. 

48 

49 Parameters 

50 ---------- 

51 bands : `list` of `str` 

52 The full list of bands in the `MultibandExposure` generating 

53 the PSF. 

54 """ 

55 def __init__(self, bands, position, partialPsf=None): 

56 if partialPsf is None: 

57 missingBands = bands 

58 else: 

59 missingBands = [band for band in bands if band not in partialPsf.bands] 

60 

61 self.missingBands = missingBands 

62 self.position = position 

63 self.partialPsf = partialPsf 

64 message = f"Failed to compute PSF at {position} in {missingBands}" 

65 super().__init__(message) 

66 

67 

68def computePsfImage(psfModels, position, useKernelImage=True): 

69 """Get a multiband PSF image 

70 

71 The PSF Image or PSF Kernel Image is computed for each band 

72 and combined into a (band, y, x) array. 

73 

74 Parameters 

75 ---------- 

76 psfModels : `dict[str, lsst.afw.detection.Psf]` 

77 The list of PSFs in each band. 

78 position : `Point2D` or `tuple` 

79 Coordinates to evaluate the PSF. 

80 useKernelImage: `bool` 

81 Execute ``Psf.computeKernelImage`` when ``True`, 

82 ``PSF/computeImage`` when ``False``. 

83 

84 Returns 

85 ------- 

86 psfs: `lsst.afw.image.MultibandImage` 

87 The multiband PSF image. 

88 """ 

89 psfs = {} 

90 # Make the coordinates into a Point2D (if necessary) 

91 if not isinstance(position, Point2D): 

92 position = Point2D(position[0], position[1]) 

93 

94 incomplete = False 

95 

96 for band, psfModel in psfModels.items(): 

97 try: 

98 if useKernelImage: 

99 psf = psfModel.computeKernelImage(position) 

100 else: 

101 psf = psfModel.computeImage(position) 

102 psfs[band] = psf 

103 except InvalidParameterError: 

104 incomplete = True 

105 

106 if len(psfs) == 0: 

107 raise IncompleteDataError(list(psfModels.keys()), position, None) 

108 

109 left = np.min([psf.getBBox().getMinX() for psf in psfs.values()]) 

110 bottom = np.min([psf.getBBox().getMinY() for psf in psfs.values()]) 

111 right = np.max([psf.getBBox().getMaxX() for psf in psfs.values()]) 

112 top = np.max([psf.getBBox().getMaxY() for psf in psfs.values()]) 

113 bbox = Box2I(Point2I(left, bottom), Point2I(right, top)) 

114 

115 psf_images = [projectImage(psf, bbox) for psf in psfs.values()] 

116 

117 mPsf = MultibandImage.fromImages(list(psfs.keys()), psf_images) 

118 

119 if incomplete: 

120 raise IncompleteDataError(list(psfModels.keys()), position, mPsf) 

121 

122 return mPsf 

123 

124 

125class MultibandExposure(MultibandTripleBase): 

126 """MultibandExposure class 

127 

128 This class acts as a container for multiple `afw.Exposure` objects. 

129 All exposures must have the same bounding box, and the associated 

130 images must all have the same data type. 

131 

132 See `MultibandTripleBase` for parameter definitions. 

133 """ 

134 def __init__(self, bands, image, mask, variance, psfs=None): 

135 super().__init__(bands, image, mask, variance) 

136 if psfs is not None: 

137 for psf, exposure in zip(psfs, self.singles): 

138 exposure.setPsf(psf) 

139 

140 @staticmethod 

141 def fromExposures(bands, singles): 

142 """Construct a MultibandImage from a collection of single band images 

143 

144 see `tripleFromExposures` for a description of parameters 

145 """ 

146 psfs = [s.getPsf() for s in singles] 

147 return tripleFromSingles(MultibandExposure, bands, singles, psfs=psfs) 

148 

149 @staticmethod 

150 def fromArrays(bands, image, mask, variance, bbox=None): 

151 """Construct a MultibandExposure from a collection of arrays 

152 

153 see `tripleFromArrays` for a description of parameters 

154 """ 

155 return tripleFromArrays(MultibandExposure, bands, image, mask, variance, bbox) 

156 

157 @staticmethod 

158 def fromKwargs(bands, bandKwargs, singleType=ExposureF, **kwargs): 

159 """Build a MultibandImage from a set of keyword arguments 

160 

161 see `makeTripleFromKwargs` for a description of parameters 

162 """ 

163 return makeTripleFromKwargs(MultibandExposure, bands, bandKwargs, singleType, **kwargs) 

164 

165 def _buildSingles(self, image=None, mask=None, variance=None): 

166 """Make a new list of single band objects 

167 

168 Parameters 

169 ---------- 

170 image: `list` 

171 List of `Image` objects that represent the image in each band. 

172 mask: `list` 

173 List of `Mask` objects that represent the mask in each band. 

174 variance: `list` 

175 List of `Image` objects that represent the variance in each band. 

176 

177 Returns 

178 ------- 

179 singles: tuple 

180 Tuple of `MaskedImage` objects for each band, 

181 where the `image`, `mask`, and `variance` of each `single` 

182 point to the multiband objects. 

183 """ 

184 singles = [] 

185 if image is None: 

186 image = self.image 

187 if mask is None: 

188 mask = self.mask 

189 if variance is None: 

190 variance = self.variance 

191 

192 dtype = image.array.dtype 

193 for f in self.bands: 

194 maskedImage = MaskedImage(image=image[f], mask=mask[f], variance=variance[f], dtype=dtype) 

195 single = Exposure(maskedImage, dtype=dtype) 

196 singles.append(single) 

197 return tuple(singles) 

198 

199 @staticmethod 

200 def fromButler(butler, bands, *args, **kwargs): 

201 """Load a multiband exposure from a butler 

202 

203 Because each band is stored in a separate exposure file, 

204 this method can be used to load all of the exposures for 

205 a given set of bands 

206 

207 Parameters 

208 ---------- 

209 butler: `lsst.daf.butler.Butler` 

210 Butler connection to use to load the single band 

211 calibrated images 

212 bands: `list` or `str` 

213 List of names for each band 

214 args: `list` 

215 Arguments to the Butler. 

216 kwargs: `dict` 

217 Keyword arguments to pass to the Butler 

218 that are the same in all bands. 

219 

220 Returns 

221 ------- 

222 result: `MultibandExposure` 

223 The new `MultibandExposure` created by combining all of the 

224 single band exposures. 

225 """ 

226 # Load the Exposure in each band 

227 exposures = [] 

228 for band in bands: 

229 exposures.append(butler.get(*args, band=band, **kwargs)) 

230 return MultibandExposure.fromExposures(bands, exposures) 

231 

232 def computePsfKernelImage(self, position): 

233 """Get a multiband PSF image 

234 

235 The PSF Kernel Image is computed for each band 

236 and combined into a (band, y, x) array and stored 

237 as `self._psfImage`. 

238 The result is not cached, so if the same PSF is expected 

239 to be used multiple times it is a good idea to store the 

240 result in another variable. 

241 

242 Parameters 

243 ---------- 

244 position: `Point2D` or `tuple` 

245 Coordinates to evaluate the PSF. 

246 

247 Returns 

248 ------- 

249 self._psfImage: array 

250 The multiband PSF image. 

251 """ 

252 return computePsfImage( 

253 psfModels=self.getPsfs(), 

254 position=position, 

255 useKernelImage=True, 

256 ) 

257 

258 def computePsfImage(self, position=None): 

259 """Get a multiband PSF image 

260 

261 The PSF Kernel Image is computed for each band 

262 and combined into a (band, y, x) array and stored 

263 as `self._psfImage`. 

264 The result is not cached, so if the same PSF is expected 

265 to be used multiple times it is a good idea to store the 

266 result in another variable. 

267 

268 Parameters 

269 ---------- 

270 position: `Point2D` or `tuple` 

271 Coordinates to evaluate the PSF. If `position` is `None` 

272 then `Psf.getAveragePosition()` is used. 

273 

274 Returns 

275 ------- 

276 self._psfImage: array 

277 The multiband PSF image. 

278 """ 

279 return computePsfImage( 

280 psfModels=self.getPsfs(), 

281 position=position, 

282 useKernelImage=True, 

283 ) 

284 

285 def getPsfs(self): 

286 """Extract the PSF model in each band 

287 

288 Returns 

289 ------- 

290 psfs : `dict` of `lsst.afw.detection.Psf` 

291 The PSF in each band 

292 """ 

293 return {band: self[band].getPsf() for band in self.bands} 

294 

295 def _slice(self, bands, bandIndex, indices): 

296 """Slice the current object and return the result 

297 

298 See `Multiband._slice` for a list of the parameters. 

299 This overwrites the base method to attach the PSF to 

300 each individual exposure. 

301 """ 

302 image = self.image._slice(bands, bandIndex, indices) 

303 if self.mask is not None: 

304 mask = self._mask._slice(bands, bandIndex, indices) 

305 else: 

306 mask = None 

307 if self.variance is not None: 

308 variance = self._variance._slice(bands, bandIndex, indices) 

309 else: 

310 variance = None 

311 

312 # If only a single pixel is selected, return the tuple of MultibandPixels 

313 if isinstance(image, MultibandPixel): 

314 if mask is not None: 

315 assert isinstance(mask, MultibandPixel) 

316 if variance is not None: 

317 assert isinstance(variance, MultibandPixel) 

318 return (image, mask, variance) 

319 

320 _psfs = self.getPsfs() 

321 psfs = [_psfs[band] for band in bands] 

322 

323 result = MultibandExposure( 

324 bands=bands, 

325 image=image, 

326 mask=mask, 

327 variance=variance, 

328 psfs=psfs, 

329 ) 

330 

331 assert all([r.getBBox() == result._bbox for r in [result._mask, result._variance]]) 

332 return result