Coverage for python/lsst/images/_observation_summary_stats.py: 94%

121 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-16 15:24 -0700

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. 

11from __future__ import annotations 

12 

13__all__ = ("ObservationSummaryStats",) 

14 

15import dataclasses 

16import math 

17from typing import TYPE_CHECKING, Any, Self, get_origin 

18 

19import pydantic 

20 

21if TYPE_CHECKING: 

22 try: 

23 from lsst.afw.image import ExposureSummaryStats as LegacyExposureSummaryStats 

24 except ImportError: 

25 type LegacyExposureSummaryStats = Any # type: ignore[no-redef] 

26 

27 

28def _default_corners() -> tuple[float, float, float, float]: 

29 return (math.nan, math.nan, math.nan, math.nan) 

30 

31 

32def _is_empty(value: Any) -> bool: 

33 """Return whether a summary-statistic value is unset. 

34 

35 A value counts as unset if it is NaN, or an empty sequence, or a sequence 

36 whose entries are all unset. Such fields carry no information and can be 

37 dropped when converting to or from the legacy representation, allowing the 

38 two representations to define different sets of fields as long as the 

39 fields they do not share are empty. 

40 """ 

41 if isinstance(value, (list, tuple)): 

42 return all(_is_empty(item) for item in value) 

43 return isinstance(value, float) and math.isnan(value) 

44 

45 

46class ObservationSummaryStats(pydantic.BaseModel, ser_json_inf_nan="constants"): 

47 version: int = pydantic.Field(0, description="Version of the model.") 

48 

49 psfSigma: float = pydantic.Field(math.nan, description="PSF determinant radius (pixels).") 

50 

51 psfArea: float = pydantic.Field(math.nan, description="PSF effective area (pixels**2).") 

52 

53 psfIxx: float = pydantic.Field(math.nan, description="PSF shape Ixx (pixels**2).") 

54 

55 psfIyy: float = pydantic.Field(math.nan, description="PSF shape Iyy (pixels**2).") 

56 

57 psfIxy: float = pydantic.Field(math.nan, description="PSF shape Ixy (pixels**2).") 

58 

59 ra: float = pydantic.Field(math.nan, description="Bounding box center Right Ascension (degrees).") 

60 

61 dec: float = pydantic.Field(math.nan, description="Bounding box center Declination (degrees).") 

62 

63 pixelScale: float = pydantic.Field(math.nan, description="Measured detector pixel scale (arcsec/pixel).") 

64 

65 zenithDistance: float = pydantic.Field( 

66 math.nan, description="Bounding box center zenith distance (degrees)." 

67 ) 

68 

69 expTime: float = pydantic.Field(math.nan, description="Exposure time of the exposure (seconds).") 

70 

71 zeroPoint: float = pydantic.Field(math.nan, description="Mean zeropoint in detector (mag).") 

72 

73 skyBg: float = pydantic.Field(math.nan, description="Average sky background (ADU).") 

74 

75 skyNoise: float = pydantic.Field(math.nan, description="Average sky noise (ADU).") 

76 

77 meanVar: float = pydantic.Field(math.nan, description="Mean variance of the weight plane (ADU**2).") 

78 

79 raCorners: tuple[float, float, float, float] = pydantic.Field( 

80 default_factory=_default_corners, description="Right Ascension of bounding box corners (degrees)." 

81 ) 

82 

83 decCorners: tuple[float, float, float, float] = pydantic.Field( 

84 default_factory=_default_corners, description="Declination of bounding box corners (degrees)." 

85 ) 

86 

87 psfAdaptiveThresholdValue: float = pydantic.Field( 

88 math.nan, 

89 description="Threshold value used in the adaptive threshold detection pass for PSF modelling.", 

90 ) 

91 

92 psfAdaptiveIncludeThresholdMultiplier: float = pydantic.Field( 

93 math.nan, 

94 description="Threshold multiplier used in the adaptive threshold detection pass for PSF modelling.", 

95 ) 

96 

97 nShapeletsStar: int = pydantic.Field( 

98 0, 

99 description="Number of sources used in the shapelet decomposition.", 

100 ) 

101 

102 shapeletsOnlyIqScore: float = pydantic.Field( 

103 math.nan, 

104 description=( 

105 "The dimensionless image quality score as determined from the shapelets decomposition " 

106 "that includes power only from the non-atmospheric decomposition coefficients. The " 

107 "score spans the range [0.0, 1.0] with lower values indicating better image quality." 

108 ), 

109 ) 

110 

111 shapeletsIqScore: float = pydantic.Field( 

112 math.nan, 

113 description=( 

114 "The dimensionless image quality score as determined from the shapelets decomposition " 

115 "that includes power from the median centroid offset between those used in the decomposition " 

116 "and those of the centroid slot in addition to non-atmospheric decomposition coefficients. " 

117 "The score spans the range [0.0, 1.0] with lower values indicating better image quality." 

118 ), 

119 ) 

120 

121 shapeletsCoeffs: tuple[float, ...] = pydantic.Field( 

122 default_factory=tuple, 

123 description="Coefficients from the PSF star shapelet decomposition.", 

124 ) 

125 

126 centroidDiffShapeletsVsSlotMedian: float = pydantic.Field( 

127 math.nan, 

128 description=( 

129 "Median centroid difference (sqrt((slot_x - shapelet_x)**2 + (slot_y - shapelet_y)**2)) for " 

130 "sources used in the shapelet decomposition (pixels)." 

131 ), 

132 ) 

133 

134 shapeletsStarEMedian: float = pydantic.Field( 

135 math.nan, 

136 description=( 

137 "Median ellipticity (sqrt(starE1**2.0 + starE2**2.0)) of the sources used in the " 

138 "shapelet decomposition." 

139 ), 

140 ) 

141 

142 shapeletsStarUnNormalizedEMedian: float = pydantic.Field( 

143 math.nan, 

144 description=( 

145 "Median un-normalized ellipticity (sqrt((starXX - starYY)**2.0 + (2.0*starXY)**2.0)) " 

146 "of the sources used in the shapelet decomposition (pixel**2)." 

147 ), 

148 ) 

149 

150 refCatSourceDensity: float = pydantic.Field( 

151 math.nan, 

152 description=( 

153 "Source density for the detector region as computed from the loaded reference catalog " 

154 "(number per degrees**2)." 

155 ), 

156 ) 

157 

158 astromOffsetMean: float = pydantic.Field(math.nan, description="Astrometry match offset mean.") 

159 

160 astromOffsetStd: float = pydantic.Field(math.nan, description="Astrometry match offset stddev.") 

161 

162 nPsfStar: int = pydantic.Field(0, description="Number of stars used for psf model.") 

163 

164 psfStarDeltaE1Median: float = pydantic.Field( 

165 math.nan, description="Psf stars median E1 residual (starE1 - psfE1)." 

166 ) 

167 

168 psfStarDeltaE2Median: float = pydantic.Field( 

169 math.nan, description="Psf stars median E2 residual (starE2 - psfE2)." 

170 ) 

171 

172 psfStarDeltaE1Scatter: float = pydantic.Field( 

173 math.nan, description="Psf stars MAD E1 scatter (starE1 - psfE1)." 

174 ) 

175 

176 psfStarDeltaE2Scatter: float = pydantic.Field( 

177 math.nan, description="Psf stars MAD E2 scatter (starE2 - psfE2)." 

178 ) 

179 

180 psfStarDeltaSizeMedian: float = pydantic.Field( 

181 math.nan, description="Psf stars median size residual (starSize - psfSize)." 

182 ) 

183 

184 psfStarDeltaSizeScatter: float = pydantic.Field( 

185 math.nan, description="Psf stars MAD size scatter (starSize - psfSize)." 

186 ) 

187 

188 psfStarScaledDeltaSizeScatter: float = pydantic.Field( 

189 math.nan, description="Psf stars MAD size scatter scaled by psfSize**2." 

190 ) 

191 

192 psfTraceRadiusDelta: float = pydantic.Field( 

193 math.nan, 

194 description=( 

195 "Delta (max - min) of the model psf trace radius values evaluated on a grid of " 

196 "unmasked pixels (pixels)." 

197 ), 

198 ) 

199 

200 psfApFluxDelta: float = pydantic.Field( 

201 math.nan, 

202 description=( 

203 "Delta (max - min) of the model psf aperture flux (with aperture radius of max(2, 3*psfSigma)) " 

204 "values evaluated on a grid of unmasked pixels." 

205 ), 

206 ) 

207 

208 psfApCorrSigmaScaledDelta: float = pydantic.Field( 

209 math.nan, 

210 description=( 

211 "Delta (max - min) of the psf flux aperture correction factors scaled (divided) by the " 

212 "psfSigma evaluated on a grid of unmasked pixels." 

213 ), 

214 ) 

215 

216 maxDistToNearestPsf: float = pydantic.Field( 

217 math.nan, 

218 description="Maximum distance of an unmasked pixel to its nearest model psf star (pixels).", 

219 ) 

220 

221 starEMedian: float = pydantic.Field( 

222 math.nan, 

223 description=( 

224 "Median ellipticity (sqrt(starE1**2.0 + starE2**2.0)) of the stars used in the PSF model." 

225 ), 

226 ) 

227 

228 starUnNormalizedEMedian: float = pydantic.Field( 

229 math.nan, 

230 description=( 

231 "Median un-normalized ellipticity (sqrt((starXX - starYY)**2.0 + " 

232 "(2.0*starXY)**2.0)) of the stars used in the PSF model." 

233 ), 

234 ) 

235 

236 starComa1Median: float = pydantic.Field( 

237 math.nan, 

238 description=( 

239 "Coma-like higher-order moment combination: median M30 + M12 of the stars used in the PSF model." 

240 ), 

241 ) 

242 

243 starComa2Median: float = pydantic.Field( 

244 math.nan, 

245 description=( 

246 "Coma-like higher-order moment combination: median M21 + M03 of the stars used in the PSF model." 

247 ), 

248 ) 

249 

250 starTrefoil1Median: float = pydantic.Field( 

251 math.nan, 

252 description=( 

253 "Trefoil-like higher-order moment combination: median M30 - 3*M12 " 

254 "of the stars used in the PSF model." 

255 ), 

256 ) 

257 

258 starTrefoil2Median: float = pydantic.Field( 

259 math.nan, 

260 description=( 

261 "Trefoil-like higher-order moment combination: median 3*M21 - M03 " 

262 "of the stars used in the PSF model." 

263 ), 

264 ) 

265 

266 starKurtosisMedian: float = pydantic.Field( 

267 math.nan, 

268 description=( 

269 "Kurtosis-like higher-order moment combination: median M40 + 2*M22 + M04 " 

270 "of the stars used in the PSF model." 

271 ), 

272 ) 

273 

274 starE41Median: float = pydantic.Field( 

275 math.nan, 

276 description=( 

277 "Fourth-order ellipticity-like higher-order moment combination: median M40 - M04 " 

278 "of the stars used in the PSF model." 

279 ), 

280 ) 

281 

282 starE42Median: float = pydantic.Field( 

283 math.nan, 

284 description=( 

285 "Fourth-order ellipticity-like higher-order moment combination: median 2*(M31 + M13) " 

286 "of the stars used in the PSF model." 

287 ), 

288 ) 

289 

290 effTime: float = pydantic.Field( 

291 math.nan, 

292 description="Effective exposure time calculated from psfSigma, skyBg, and zeroPoint (seconds).", 

293 ) 

294 

295 effTimePsfSigmaScale: float = pydantic.Field( 

296 math.nan, description="PSF scaling of the effective exposure time." 

297 ) 

298 

299 effTimeSkyBgScale: float = pydantic.Field( 

300 math.nan, description="Sky background scaling of the effective exposure time." 

301 ) 

302 

303 effTimeZeroPointScale: float = pydantic.Field( 

304 math.nan, description="Zeropoint scaling of the effective exposure time." 

305 ) 

306 

307 magLim: float = pydantic.Field( 

308 math.nan, 

309 description=( 

310 "Magnitude limit at fixed SNR (default SNR=5) calculated from psfSigma, skyBg," 

311 " zeroPoint, and readNoise." 

312 ), 

313 ) 

314 

315 psfTE1e1: float = pydantic.Field( 

316 math.nan, 

317 description=( 

318 "Per-exposure TE1e1 ~ <de1 de1> of PSF residual ellipticity, averaged over theta " 

319 "[0,1] arcmin via treecorr KK correlation. Dimensionless; used to form the " 

320 "full-survey TE1 metric." 

321 ), 

322 ) 

323 

324 psfTE1e2: float = pydantic.Field( 

325 math.nan, 

326 description=( 

327 "Per-exposure TE1e2 ~ <de2 de2> of PSF residual ellipticity, averaged over theta " 

328 "[0,1] arcmin via treecorr KK correlation. Dimensionless; used to form the " 

329 "full-survey TE1 metric." 

330 ), 

331 ) 

332 

333 psfTE1ex: float = pydantic.Field( 

334 math.nan, 

335 description=( 

336 "Per-exposure TE1ex ~ <de1 de2> of PSF residual ellipticity, averaged over theta " 

337 "[0,1] arcmin via treecorr KK correlation. Dimensionless; used to form the " 

338 "full-survey TE1 metric." 

339 ), 

340 ) 

341 

342 psfTE2e1: float = pydantic.Field( 

343 math.nan, 

344 description=( 

345 "Per-exposure TE2e1 ~ <de1 de1> of PSF residual ellipticity, averaged over theta " 

346 "[5,100] arcmin via treecorr KK correlation. Dimensionless; used to form the " 

347 "full-survey TE2 metric." 

348 ), 

349 ) 

350 

351 psfTE2e2: float = pydantic.Field( 

352 math.nan, 

353 description=( 

354 "Per-exposure TE2e2 ~ <de2 de2> of PSF residual ellipticity, averaged over theta " 

355 "[5,100] arcmin via treecorr KK correlation. Dimensionless; used to form the " 

356 "full-survey TE2 metric." 

357 ), 

358 ) 

359 

360 psfTE2ex: float = pydantic.Field( 

361 math.nan, 

362 description=( 

363 "Per-exposure TE2ex ~ <de1 de2> of PSF residual ellipticity, averaged over theta " 

364 "[5,100] arcmin via treecorr KK correlation. Dimensionless; used to form the " 

365 "full-survey TE2 metric." 

366 ), 

367 ) 

368 

369 psfTE3e1: float = pydantic.Field( 

370 math.nan, 

371 description=( 

372 "Per-exposure median-over-CCDs of TE3e1 ~ <de1 de1> of PSF residual ellipticity, " 

373 "where each CCD uses theta within [0,5] arcmin bins. Dimensionless; downstream " 

374 "pipelines take the 85th percentile over images to evaluate TE3." 

375 ), 

376 ) 

377 

378 psfTE3e2: float = pydantic.Field( 

379 math.nan, 

380 description=( 

381 "Per-exposure median-over-CCDs of TE3e2 ~ <de2 de2> of PSF residual ellipticity, " 

382 "where each CCD uses theta within [0,5] arcmin bins. Dimensionless; downstream " 

383 "pipelines take the 85th percentile over images to evaluate TE3." 

384 ), 

385 ) 

386 

387 psfTE3ex: float = pydantic.Field( 

388 math.nan, 

389 description=( 

390 "Per-exposure median-over-CCDs of TE3ex ~ <de1 de2> of PSF residual ellipticity, " 

391 "where each CCD uses theta within [0,5] arcmin bins. Dimensionless; downstream " 

392 "pipelines take the 85th percentile over images to evaluate TE3." 

393 ), 

394 ) 

395 

396 psfTE4e1: float = pydantic.Field( 

397 math.nan, 

398 description=( 

399 "Per-exposure median-over-CCDs of TE4e1 ~ <de1 de1> of PSF residual ellipticity, " 

400 "where each CCD uses theta within [5,20] arcmin bins. Dimensionless; downstream " 

401 "pipelines take the 85th percentile over images to evaluate TE4." 

402 ), 

403 ) 

404 

405 psfTE4e2: float = pydantic.Field( 

406 math.nan, 

407 description=( 

408 "Per-exposure median-over-CCDs of TE4e2 ~ <de2 de2> of PSF residual ellipticity, " 

409 "where each CCD uses theta within [5,20] arcmin bins. Dimensionless; downstream " 

410 "pipelines take the 85th percentile over images to evaluate TE4." 

411 ), 

412 ) 

413 

414 psfTE4ex: float = pydantic.Field( 

415 math.nan, 

416 description=( 

417 "Per-exposure median-over-CCDs of TE4ex ~ <de1 de2> of PSF residual ellipticity, " 

418 "where each CCD uses theta within [5,20] arcmin bins. Dimensionless; downstream " 

419 "pipelines take the 85th percentile over images to evaluate TE4." 

420 ), 

421 ) 

422 

423 def __eq__(self, other: object) -> bool: 

424 if not isinstance(other, ObservationSummaryStats): 424 ↛ 425line 424 didn't jump to line 425 because the condition on line 424 was never true

425 return NotImplemented 

426 for name in ObservationSummaryStats.model_fields: 

427 a = getattr(self, name) 

428 b = getattr(other, name) 

429 if isinstance(a, tuple) and isinstance(b, tuple): 

430 if len(a) != len(b): 430 ↛ 431line 430 didn't jump to line 431 because the condition on line 430 was never true

431 return False 

432 for ai, bi in zip(a, b): 

433 if ai != bi and not (math.isnan(ai) and math.isnan(bi)): 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true

434 return False 

435 elif a != b and not (math.isnan(a) and math.isnan(b)): 

436 return False 

437 return True 

438 

439 @classmethod 

440 def from_legacy(cls, exposure_summary_stats: LegacyExposureSummaryStats) -> Self: 

441 """Return an `ObservationSummaryStats` from a legacy 

442 `lsst.afw.image.ExposureSummaryStats`. 

443 

444 Parameters 

445 ---------- 

446 exposure_summary_stats 

447 Legacy exposure summary statistics to convert. 

448 

449 Notes 

450 ----- 

451 Legacy fields that are empty (NaN) are dropped, so a legacy struct that 

452 carries fields unknown to this class is accepted as long as those 

453 fields are empty. A legacy field that holds a real value but is 

454 unknown here raises `ValueError`, since dropping it would lose data. 

455 """ 

456 known_fields = set(cls.model_fields) 

457 kwargs: dict[str, Any] = {} 

458 for name, value in dataclasses.asdict(exposure_summary_stats).items(): 

459 if _is_empty(value): 

460 continue 

461 if name not in known_fields: 

462 raise ValueError( 

463 f"Legacy field {name!r} has a value ({value!r}) but is not known to " 

464 f"ObservationSummaryStats." 

465 ) 

466 kwargs[name] = value 

467 return cls.model_validate(kwargs) 

468 

469 def to_legacy(self) -> LegacyExposureSummaryStats: 

470 """Convert to an `lsst.afw.image.ExposureSummaryStats` instance. 

471 

472 Notes 

473 ----- 

474 Empty (NaN) fields are not passed to the legacy struct, so fields 

475 defined here that are unknown to the installed version of 

476 `~lsst.afw.image.ExposureSummaryStats` are dropped when empty. A field 

477 that holds a real value but is unknown to the legacy struct raises 

478 `ValueError`, since dropping it would lose data. 

479 """ 

480 from lsst.afw.image import ExposureSummaryStats as LegacyExposureSummaryStats 

481 

482 legacy_fields = {field.name for field in dataclasses.fields(LegacyExposureSummaryStats)} 

483 kwargs: dict[str, Any] = {} 

484 for name, info in ObservationSummaryStats.model_fields.items(): 

485 value = getattr(self, name) 

486 if _is_empty(value): 

487 continue 

488 if name not in legacy_fields: 

489 raise ValueError( 

490 f"Field {name!r} has a value ({value!r}) but is not supported by this " 

491 f"version of lsst.afw.image.ExposureSummaryStats." 

492 ) 

493 # Doing this in general is hard, so we handle the fields that we 

494 # know about and raise if somebody adds a field with a new type 

495 # without updating this function. 

496 if info.annotation in (float, int): 496 ↛ 498line 496 didn't jump to line 498 because the condition on line 496 was always true

497 kwargs[name] = value 

498 elif get_origin(info.annotation) is tuple: 

499 kwargs[name] = list(value) 

500 else: 

501 raise NotImplementedError(f"Unsupported field type: {info.annotation}.") 

502 return LegacyExposureSummaryStats(**kwargs)