Coverage for python/lsst/images/_visit_image.py: 61%

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__ = ("VisitImage", "VisitImageSerializationModel") 

15 

16import functools 

17import logging 

18from collections.abc import Mapping 

19from types import EllipsisType 

20from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast 

21 

22import astropy.io.fits 

23import astropy.units 

24import numpy as np 

25import pydantic 

26from astro_metadata_translator import ObservationInfo 

27 

28from ._backgrounds import BackgroundMap, BackgroundMapSerializationModel 

29from ._concrete_bounds import BoundsSerializationModel 

30from ._geom import Bounds, Box 

31from ._image import Image, ImageSerializationModel 

32from ._mask import Mask, MaskPlane, MaskSchema, MaskSerializationModel, get_legacy_visit_image_mask_planes 

33from ._masked_image import MaskedImage, MaskedImageSerializationModel 

34from ._obs_info_from_legacy import obs_info_from_legacy 

35from ._observation_summary_stats import ObservationSummaryStats 

36from ._polygon import Polygon 

37from ._transforms import ( 

38 DetectorFrame, 

39 SkyProjection, 

40 SkyProjectionAstropyView, 

41 SkyProjectionSerializationModel, 

42) 

43from .aperture_corrections import ( 

44 ApertureCorrectionMap, 

45 ApertureCorrectionMapSerializationModel, 

46 aperture_corrections_from_legacy, 

47 aperture_corrections_to_legacy, 

48) 

49from .cameras import Detector, DetectorSerializationModel 

50from .describe import DescribeOptions, FieldRole, Report, ReportField 

51from .fields import BaseField, Field, FieldSerializationModel, field_from_legacy_photo_calib 

52from .fits import FitsOpaqueMetadata, parse_legacy_bunit 

53from .psfs import ( 

54 GaussianPointSpreadFunction, 

55 GaussianPSFSerializationModel, 

56 LegacyPointSpreadFunction, 

57 PiffSerializationModel, 

58 PiffWrapper, 

59 PointSpreadFunction, 

60 PSFExSerializationModel, 

61 PSFExWrapper, 

62) 

63from .serialization import ArchiveReadError, InputArchive, InvalidParameterError, MetadataValue, OutputArchive 

64from .utils import is_none 

65 

66if TYPE_CHECKING: 

67 try: 

68 from lsst.afw.cameraGeom import Detector as LegacyDetector 

69 from lsst.afw.image import Exposure as LegacyExposure 

70 from lsst.afw.image import FilterLabel as LegacyFilterLabel 

71 from lsst.afw.image import VisitInfo as LegacyVisitInfo 

72 from lsst.daf.butler import DimensionRecord 

73 except ImportError: 

74 type LegacyDetector = Any # type: ignore[no-redef] 

75 type LegacyExposure = Any # type: ignore[no-redef] 

76 type LegacyFilterLabel = Any # type: ignore[no-redef] 

77 type LegacyVisitInfo = Any # type: ignore[no-redef] 

78 type DimensionRecord = Any # type: ignore[no-redef] 

79 

80_LOG = logging.getLogger("lsst.images") 

81 

82 

83class VisitImage(MaskedImage): 

84 """A calibrated single-visit image. 

85 

86 Parameters 

87 ---------- 

88 image 

89 The main image plane. If this has a `SkyProjection`, it will be used 

90 for all planes unless a ``sky_projection`` is passed separately. 

91 mask 

92 A bitmask image that annotates the main image plane. Must have the 

93 same bounding box as ``image`` if provided. Any attached 

94 ``sky_projection`` is replaced (possibly by `None`). 

95 variance 

96 The per-pixel uncertainty of the main image as an image of variance 

97 values. Must have the same bounding box as ``image`` if provided, and 

98 its units must be the square of ``image.unit`` or `None`. 

99 Values default to ``1.0``. Any attached ``sky_projection`` is replaced 

100 (possibly by `None`). 

101 mask_schema 

102 Schema for the mask plane. Must be provided if and only if ``mask`` is 

103 not provided. 

104 sky_projection 

105 Projection that maps the pixel grid to the sky. Can only be `None` if 

106 a ``sky_projection`` is already attached to ``image``. 

107 bounds 

108 The region where this image's pixels and other properties are valid. 

109 If not provided, the bounding box of the image is used. Other 

110 components (``psf``, ``sky_projection``, ``aperture_corrections``, 

111 etc.) are assumed to have their own bounds which may or may not be the 

112 same as the image bounds. If ``bounds`` extends beyond the image 

113 bounding box, the intersection between ``bounds`` and the image 

114 bounding box is used instead. 

115 obs_info 

116 General information about this visit in standardized form. 

117 summary_stats 

118 Summary statistics associated with this visit. Initialized to default 

119 values if not provided. 

120 photometric_scaling 

121 Field that can be used to multiply a post-ISR image units to yield 

122 calibrated image units. This may be a scaling that was already 

123 applied (so dividing by it will recover the post-ISR units) or a 

124 scaling that has not been applied, depending on ``image.unit``. 

125 psf 

126 Point-spread function model for this image, or an exception explaining 

127 why it could not be read (to be raised if the PSF is requested later). 

128 detector 

129 Geometry and electronic information for the detector attached to this 

130 image. 

131 aperture_corrections : `dict` [`str`, `~fields.BaseField`] 

132 Mapping from photometry algorithm name to the aperture correction for 

133 that algorithm. 

134 backgrounds 

135 Background models associated with this image. 

136 band 

137 Name of the passband the image was observed with (this is a shorter, 

138 less specific version of ``obs_info.physical_filter``). 

139 metadata 

140 Arbitrary flexible metadata to associate with the image. 

141 """ 

142 

143 def __init__( 

144 self, 

145 image: Image, 

146 *, 

147 mask: Mask | None = None, 

148 variance: Image | None = None, 

149 mask_schema: MaskSchema | None = None, 

150 sky_projection: SkyProjection[DetectorFrame] | None = None, 

151 bounds: Bounds | None = None, 

152 obs_info: ObservationInfo | None = None, 

153 summary_stats: ObservationSummaryStats | None = None, 

154 photometric_scaling: Field | None = None, 

155 psf: PointSpreadFunction | ArchiveReadError, 

156 detector: Detector, 

157 aperture_corrections: ApertureCorrectionMap | None = None, 

158 backgrounds: BackgroundMap | None = None, 

159 band: str, 

160 metadata: dict[str, MetadataValue] | None = None, 

161 ) -> None: 

162 super().__init__( 

163 image, 

164 mask=mask, 

165 variance=variance, 

166 mask_schema=mask_schema, 

167 sky_projection=sky_projection, 

168 metadata=metadata, 

169 ) 

170 if self.image.unit is None: 

171 raise TypeError("The image component of a VisitImage must have units.") 

172 if self.image.sky_projection is None: 

173 raise TypeError("The sky_projection component of a VisitImage cannot be None.") 

174 if obs_info is None: 

175 raise TypeError("The observation info component of a VisitImage cannot be None.") 

176 if obs_info.physical_filter is None: 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 raise ValueError("The obs_info.physical_filter attribute of a VisitImage cannot be None.") 

178 self._obs_info = obs_info 

179 if not isinstance(self.image.sky_projection.pixel_frame, DetectorFrame): 

180 raise TypeError("The sky_projection's pixel frame must be a DetectorFrame for VisitImage.") 

181 if summary_stats is None: 

182 summary_stats = ObservationSummaryStats() 

183 self._summary_stats = summary_stats 

184 if photometric_scaling is not None and photometric_scaling.unit is None: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true

185 raise TypeError("If a photometric_scaling is provided, it must have units.") 

186 self._photometric_scaling = photometric_scaling 

187 self._psf = psf 

188 self._detector = detector 

189 self._aperture_corrections = aperture_corrections if aperture_corrections is not None else {} 

190 self._bounds = bounds if bounds is not None else self.bbox 

191 if not self.bbox.contains(self._bounds.bbox): 

192 self._bounds = self._bounds.intersection(self.bbox) 

193 self._backgrounds = backgrounds if backgrounds is not None else BackgroundMap() 

194 self._band = band 

195 

196 @property 

197 def unit(self) -> astropy.units.UnitBase: 

198 """The units of the image plane (`astropy.units.Unit`).""" 

199 return cast(astropy.units.UnitBase, super().unit) 

200 

201 @property 

202 def sky_projection(self) -> SkyProjection[DetectorFrame]: 

203 """The projection that maps the pixel grid to the sky 

204 (`SkyProjection` [`DetectorFrame`]). 

205 """ 

206 return cast(SkyProjection[DetectorFrame], super().sky_projection) 

207 

208 @property 

209 def bounds(self) -> Bounds: 

210 """The region where pixels are valid (`Bounds`).""" 

211 return self._bounds 

212 

213 @property 

214 def obs_info(self) -> ObservationInfo: 

215 """General information about this observation in standard form. 

216 (`~astro_metadata_translator.ObservationInfo`). 

217 """ 

218 return self._obs_info 

219 

220 @property 

221 def physical_filter(self) -> str: 

222 """Full name of the physical bandpass filter (`str`).""" 

223 assert self._obs_info.physical_filter is not None, "Guaranteed at construction." 

224 return self._obs_info.physical_filter 

225 

226 @property 

227 def band(self) -> str: 

228 """Short name of the bandpass filter (`str`).""" 

229 return self._band 

230 

231 @property 

232 def astropy_wcs(self) -> SkyProjectionAstropyView: 

233 """An Astropy WCS for the pixel arrays (`SkyProjectionAstropyView`). 

234 

235 Notes 

236 ----- 

237 As expected for Astropy WCS objects, this defines pixel coordinates 

238 such that the first row and column in the arrays are ``(0, 0)``, not 

239 ``bbox.start``, as is the case for `sky_projection`. 

240 

241 This object satisfies the `astropy.wcs.wcsapi.BaseHighLevelWCS` and 

242 `astropy.wcs.wcsapi.BaseLowLevelWCS` interfaces, but it is not an 

243 `astropy.wcs.WCS` (use `fits_wcs` for that). 

244 """ 

245 return cast(SkyProjectionAstropyView, super().astropy_wcs) 

246 

247 @property 

248 def summary_stats(self) -> ObservationSummaryStats: 

249 """Optional summary statistics for this observation 

250 (`ObservationSummaryStats`). 

251 """ 

252 return self._summary_stats 

253 

254 @property 

255 def photometric_scaling(self) -> Field | None: 

256 """Field that multiplies a post-ISR image to yield the calibrated 

257 image (~`fields.BaseField`). 

258 """ 

259 return self._photometric_scaling 

260 

261 @photometric_scaling.setter 

262 def photometric_scaling(self, value: Field) -> None: 

263 if value.unit is None: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true

264 raise TypeError("The photometric_scaling for a VisitImage must have units.") 

265 self._photometric_scaling = value 

266 

267 @property 

268 def psf(self) -> PointSpreadFunction: 

269 """The point-spread function model for this image 

270 (`.psfs.PointSpreadFunction`). 

271 """ 

272 if isinstance(self._psf, ArchiveReadError): 

273 raise self._psf 

274 return self._psf 

275 

276 @psf.setter 

277 def psf(self, value: PointSpreadFunction) -> None: 

278 self._psf = value 

279 

280 @property 

281 def detector(self) -> Detector: 

282 """Geometry and electronic information about the detector 

283 (`.cameras.Detector`). 

284 """ 

285 return self._detector 

286 

287 @property 

288 def aperture_corrections(self) -> ApertureCorrectionMap: 

289 """A mapping from photometry algorithm name to the aperture correction 

290 field for that algorithm (`dict` [`str`, `~.fields.BaseField`]). 

291 """ 

292 return self._aperture_corrections 

293 

294 @property 

295 def backgrounds(self) -> BackgroundMap: 

296 """A mapping of backgrounds associated with this image 

297 (`BackgroundMap`). 

298 """ 

299 return self._backgrounds 

300 

301 def __getitem__(self, bbox: Box | EllipsisType) -> VisitImage: 

302 bbox, _ = self._handle_getitem_args(bbox) 

303 return self._transfer_metadata( 

304 VisitImage( 

305 self.image[bbox], 

306 mask=self.mask[bbox], 

307 variance=self.variance[bbox], 

308 sky_projection=self.sky_projection, 

309 psf=self._psf, 

310 obs_info=self.obs_info, 

311 bounds=self._bounds, # don't need to intersect here, because __init__ will do that. 

312 summary_stats=self.summary_stats, 

313 detector=self._detector, 

314 photometric_scaling=self._photometric_scaling, 

315 aperture_corrections=self.aperture_corrections, 

316 backgrounds=self._backgrounds, 

317 band=self._band, 

318 ), 

319 bbox=bbox, 

320 ) 

321 

322 def _summary_stats_describe_options(self, options: DescribeOptions) -> DescribeOptions: 

323 """Return the options the summary-statistics child report is built 

324 with. 

325 

326 Parameters 

327 ---------- 

328 options : `DescribeOptions` 

329 Options this image's own report is being built with. 

330 

331 Returns 

332 ------- 

333 options : `DescribeOptions` 

334 Options for the child report. 

335 

336 Notes 

337 ----- 

338 `ObservationSummaryStats.raCorners` and 

339 `~ObservationSummaryStats.decCorners` hold the sky corners of the 

340 image the statistics were measured on, which the sky projection 

341 tabulates more readably. Taking a cutout does not recompute the 

342 statistics, so those corners still describe the original image and 

343 are worth keeping wherever they no longer match the pixels on hand. 

344 """ 

345 if self.bbox == self.detector.bbox: 

346 return options.for_child("corners") 

347 return options.for_child() 

348 

349 def _describe_psf(self, options: DescribeOptions) -> Report: 

350 """Return the report for this image's PSF, or for the error that 

351 stopped it being read. 

352 

353 Parameters 

354 ---------- 

355 options : `DescribeOptions` 

356 Options the child report is built with. 

357 

358 Returns 

359 ------- 

360 report : `Report` 

361 Report describing the PSF, or naming it unreadable. 

362 

363 Notes 

364 ----- 

365 A PSF that could not be read is a supported state of a visit image, 

366 not a failure of it, and it is often the only unreadable part: a PSF 

367 model can depend on a package the reader does not have installed. 

368 Describing it therefore has to say what is missing and why, rather 

369 than raise and take the report of everything else with it. 

370 """ 

371 if isinstance(self._psf, ArchiveReadError): 

372 return Report( 

373 type_name="PointSpreadFunction", 

374 summary=f"unreadable ({self._psf})", 

375 inline=True, 

376 ) 

377 return self._psf._describe(options) 

378 

379 def _describe(self, options: DescribeOptions = DescribeOptions(), /) -> Report: 

380 """Return a `Report` describing this visit image. 

381 

382 Parameters 

383 ---------- 

384 options : `DescribeOptions`, optional 

385 Rendering options; forwarded to all children. Child construction 

386 can be expensive, so `DescribeOptions.brief` skips it. 

387 """ 

388 # The image and mask schema are rendered as children below, and the 

389 # image renders as ``Image(bbox, dtype)``, which would restate the 

390 # shared bbox. Both are REPR_ONLY so only repr sees them. 

391 fields = [ 

392 ReportField( 

393 label="image", 

394 value=self.image, 

395 repr_value=repr(self.image), 

396 positional=True, 

397 role=FieldRole.REPR_ONLY, 

398 ), 

399 ReportField( 

400 label="mask_schema", 

401 value=self.mask.schema, 

402 repr_value=repr(self.mask.schema), 

403 role=FieldRole.REPR_ONLY, 

404 ), 

405 ReportField(label="band", value=self.band, role=FieldRole.DERIVED), 

406 ReportField(label="physical_filter", value=self.physical_filter, role=FieldRole.DERIVED), 

407 ReportField(label="bbox", value=self.bbox, repr_value=repr(self.bbox), role=FieldRole.DERIVED), 

408 ] 

409 summary = f"VisitImage({self.image!s}, {list(self.mask.schema.names)})" 

410 if options.brief: 

411 return Report(type_name="VisitImage", summary=summary, fields=fields) 

412 child = options.for_child() 

413 plane = options.for_child("sky_projection", "bbox") 

414 children: dict[str, Report] = { 

415 "image": self.image._describe(plane), 

416 "mask": self.mask._describe(plane), 

417 "variance": self.variance._describe(plane), 

418 "sky_projection": self.sky_projection._describe(child, bbox=self.bbox), 

419 "psf": self._describe_psf(child), 

420 "detector": self.detector._describe(child), 

421 "summary_stats": self.summary_stats._describe(self._summary_stats_describe_options(options)), 

422 "backgrounds": self.backgrounds._describe(child), 

423 } 

424 if self.photometric_scaling is not None: 

425 children["photometric_scaling"] = self.photometric_scaling._describe(child) 

426 return Report( 

427 type_name="VisitImage", 

428 summary=summary, 

429 fields=fields, 

430 children=children, 

431 ) 

432 

433 def copy(self, *, copy_detector: bool = False) -> VisitImage: 

434 """Deep-copy the visit image. 

435 

436 Parameters 

437 ---------- 

438 copy_detector 

439 Whether to deep-copy the `detector` attribute. 

440 """ 

441 return self._transfer_metadata( 

442 VisitImage( 

443 image=self._image.copy(), 

444 mask=self._mask.copy(), 

445 variance=self._variance.copy(), 

446 psf=self._psf, 

447 obs_info=self.obs_info, 

448 bounds=self._bounds, 

449 summary_stats=self.summary_stats.model_copy(), 

450 detector=self._detector.copy() if copy_detector else self._detector, 

451 photometric_scaling=self._photometric_scaling, 

452 aperture_corrections=self.aperture_corrections.copy(), 

453 backgrounds=self._backgrounds.copy(), 

454 band=self.band, 

455 ), 

456 copy=True, 

457 ) 

458 

459 def convert_unit( 

460 self, 

461 unit: astropy.units.UnitBase = astropy.units.nJy, 

462 copy: Literal["as-needed"] | bool = True, 

463 copy_detector: bool = False, 

464 ) -> VisitImage: 

465 """Return an equivalent image with different pixel units. 

466 

467 Parameters 

468 ---------- 

469 unit 

470 The unit to transform to. This may be any of the following: 

471 

472 - any unit directly relatable to the current units via Astropy; 

473 - any unit relatable to the product of the current units with the 

474 `photometric_scaling` (i.e. if the current image is in 

475 instrumental units but we know how to calibrate them) 

476 - any unit relatable to the quotient of the current units with the 

477 `photometric_scaling` (i.e. if the current image is in 

478 calibrated units and we want to revert back to instrumental 

479 units). 

480 copy 

481 Whether to copy the images and other components. If `True`, all 

482 components that aren't controlled by some other argument will 

483 always be deep-copied. If `False`, the operation will fail if the 

484 image is not already in the right units. If ``as-needed``, only 

485 the image and variance will be copied, and only if they are not 

486 already in the right units. 

487 copy_detector 

488 Whether to deep-copy the `detector` attribute. 

489 

490 Returns 

491 ------- 

492 `VisitImage` 

493 An image with the given units. 

494 """ 

495 if copy not in (True, False, "as-needed"): 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true

496 raise TypeError(f"Invalid value for 'copy' parameter: {copy!r}.") 

497 if (factor := _get_unit_conversion_factor(self.unit, unit)) is not None: 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true

498 if factor == 1.0: 

499 if copy is True: # not "as-needed" 

500 return self.copy() 

501 else: 

502 return self[...] 

503 elif copy is False: 

504 raise astropy.units.UnitConversionError( 

505 f"Units must be converted ({self.unit} -> {unit}), but copy=False." 

506 ) 

507 image = Image( 

508 self._image.array * factor, bbox=self.bbox, sky_projection=self.sky_projection, unit=unit 

509 ) 

510 variance = Image( 

511 self._variance.array * factor**2, 

512 bbox=self.bbox, 

513 unit=unit**2, 

514 ) 

515 elif self._photometric_scaling is None: 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true

516 raise astropy.units.UnitConversionError( 

517 "VisitImage.photometric_scaling is None, and there " 

518 f"is no constant conversion from {self.unit} to {unit}." 

519 ) 

520 else: 

521 if copy is False: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true

522 raise astropy.units.UnitConversionError( 

523 f"Photometric scaling must be applied to go from ={self.unit} to {unit}, but copy=False." 

524 ) 

525 scaling = self._photometric_scaling 

526 assert scaling.unit is not None, "Checked at construction." 

527 if (constant_factor := _get_unit_conversion_factor(self.unit * scaling.unit, unit)) is not None: 

528 if constant_factor != 1.0: 528 ↛ 529line 528 didn't jump to line 529 because the condition on line 528 was never true

529 scaling = scaling * constant_factor 

530 scaling_array = scaling.render(self.bbox, dtype=self.image.array.dtype).array 

531 elif (constant_factor := _get_unit_conversion_factor(self.unit / scaling.unit, unit)) is not None: 531 ↛ 537line 531 didn't jump to line 537 because the condition on line 531 was always true

532 if constant_factor != 1.0: 532 ↛ 533line 532 didn't jump to line 533 because the condition on line 532 was never true

533 scaling = scaling / constant_factor 

534 scaling_array = scaling.render(self.bbox, dtype=self.image.array.dtype).array 

535 np.true_divide(1.0, scaling_array, out=scaling_array) 

536 else: 

537 raise astropy.units.UnitConversionError( 

538 f"photometric_scaling with units {scaling.unit} does not " 

539 f"provide a path from {self.unit} to {unit}." 

540 ) 

541 # We needed to allocate a new array to evaluate the scaling field, 

542 # and then we need to allocate another to hold its square for the 

543 # variance scaling. But then we can multiply those arrays in-place 

544 # to get the output image and variance to avoid yet more 

545 # allocations (note we can't instead multiply the visit image's 

546 # image and variance arrays in place because they might have other 

547 # references that are still associated with the old units). 

548 image = Image(scaling_array, bbox=self.bbox, unit=unit) 

549 variance = Image(np.square(scaling_array), bbox=self.bbox, unit=unit**2) 

550 image.array *= self._image.array 

551 variance.array *= self._variance.array 

552 copy_components = copy is True 

553 return self._transfer_metadata( 

554 VisitImage( 

555 image=image, 

556 mask=self._mask if not copy_components else self._mask.copy(), 

557 variance=variance, 

558 sky_projection=self.sky_projection, # never copied; immutable 

559 obs_info=self.obs_info if not copy_components else self.obs_info.model_copy(), 

560 psf=self._psf, # never copied; immutable 

561 bounds=self._bounds, # never copied; immutable 

562 summary_stats=self.summary_stats if not copy_components else self.summary_stats.model_copy(), 

563 detector=self._detector if not copy_detector else self._detector.copy(), 

564 photometric_scaling=self._photometric_scaling, # never copied; immutable 

565 aperture_corrections=( 

566 self.aperture_corrections if not copy_components else self.aperture_corrections.copy() 

567 ), 

568 backgrounds=self.backgrounds if not copy_components else self.backgrounds.copy(), 

569 band=self.band, 

570 ) 

571 ) 

572 

573 def serialize(self, archive: OutputArchive[Any]) -> VisitImageSerializationModel[Any]: 

574 return self._serialize_impl(VisitImageSerializationModel, archive) 

575 

576 # This is slightly bad Liskov substitution - we're demanding M be a 

577 # VisitImageSerializationModel, not just a MaskedImageSerializationModel, 

578 # but that's because we know only `serialize` will call it. 

579 def _serialize_impl[M: VisitImageSerializationModel[Any]]( # type: ignore[override] 

580 self, model_type: type[M], archive: OutputArchive[Any] 

581 ) -> M: 

582 result = super()._serialize_impl(model_type, archive) 

583 match self._psf: 

584 # MyPy is able to figure things out here with this match statement, 

585 # but not a single isinstance check on the three types. 

586 case PiffWrapper(): 586 ↛ 587line 586 didn't jump to line 587 because the pattern on line 586 never matched

587 result.psf = archive.serialize_direct("psf", self._psf.serialize) 

588 case PSFExWrapper(): 588 ↛ 589line 588 didn't jump to line 589 because the pattern on line 588 never matched

589 result.psf = archive.serialize_direct("psf", self._psf.serialize) 

590 case GaussianPointSpreadFunction(): 590 ↛ 592line 590 didn't jump to line 592 because the pattern on line 590 always matched

591 result.psf = archive.serialize_direct("psf", self._psf.serialize) 

592 case _: 

593 raise TypeError( 

594 f"Cannot serialize VisitImage with unrecognized PSF type {type(self._psf).__name__}." 

595 ) 

596 assert result.sky_projection is not None, "VisitImage always has a sky_projection." 

597 result.obs_info = self.obs_info 

598 result.summary_stats = self.summary_stats 

599 result.bounds = self._bounds.serialize() if self._bounds != self.bbox else None 

600 result.detector = archive.serialize_direct("detector", self._detector.serialize) 

601 result.band = self.band 

602 result.photometric_scaling = ( 

603 # MyPy can't quite follow the type union through the serialize 

604 # method return types. 

605 archive.serialize_direct( 

606 "photometric_scaling", 

607 self._photometric_scaling.serialize, 

608 ) # type: ignore[assignment] 

609 if self._photometric_scaling is not None 

610 else None 

611 ) 

612 result.aperture_corrections = archive.serialize_direct( 

613 "aperture_corrections", 

614 functools.partial(ApertureCorrectionMapSerializationModel.serialize, self.aperture_corrections), 

615 ) 

616 result.backgrounds = archive.serialize_direct("backgrounds", self._backgrounds.serialize) 

617 return result 

618 

619 @staticmethod 

620 def _get_archive_tree_type[P: pydantic.BaseModel]( 

621 pointer_type: type[P], 

622 ) -> type[VisitImageSerializationModel[P]]: 

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

624 type that uses the given pointer type. 

625 """ 

626 return VisitImageSerializationModel[pointer_type] # type: ignore 

627 

628 @staticmethod 

629 def from_legacy( # type: ignore[override] 

630 legacy: LegacyExposure, 

631 *, 

632 exposure_record: DimensionRecord | None, 

633 unit: astropy.units.UnitBase | None = None, 

634 plane_map: Mapping[str, MaskPlane] | None = None, 

635 ) -> VisitImage: 

636 """Convert from an `lsst.afw.image.Exposure` instance. 

637 

638 Parameters 

639 ---------- 

640 legacy 

641 An `lsst.afw.image.Exposure` instance that will share image and 

642 variance (but not mask) pixel data with the returned object. 

643 exposure_record 

644 The ``exposure`` dimension record for this observation, as a 

645 source of additional required metadata. It is only consulted when 

646 the legacy metadata does not carry an embedded 

647 `~astro_metadata_translator.ObservationInfo` (flat ``OBS_INFO`` 

648 cards); in that case pass `None` explicitly to convert without 

649 some ``ObservationInfo`` fields (with a warning). 

650 unit 

651 Units of the image. If not provided, the ``BUNIT`` metadata 

652 key will be used, if available. 

653 plane_map 

654 A mapping from legacy mask plane name to the new plane name and 

655 description. If `None` (default) 

656 `get_legacy_visit_image_mask_planes` is used. 

657 """ 

658 from lsst.afw.image import setVisitInfoMetadata 

659 from lsst.daf.base import PropertyList 

660 

661 if plane_map is None: 661 ↛ 663line 661 didn't jump to line 663 because the condition on line 661 was always true

662 plane_map = get_legacy_visit_image_mask_planes() 

663 legacy_wcs = legacy.getWcs() 

664 if legacy_wcs is None: 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true

665 raise ValueError("Exposure does not have a SkyWcs.") 

666 legacy_detector = legacy.getDetector() 

667 if legacy_detector is None: 667 ↛ 668line 667 didn't jump to line 668 because the condition on line 667 was never true

668 raise ValueError("Exposure does not have a Detector.") 

669 detector_bbox = Box.from_legacy(legacy_detector.getBBox()) 

670 visit_info = legacy.info.getVisitInfo() 

671 if visit_info is None: 671 ↛ 672line 671 didn't jump to line 672 because the condition on line 671 was never true

672 raise ValueError("Exposure does not have a VisitInfo.") 

673 md = legacy.getMetadata() 

674 primary_header = md.to_astropy_header() 

675 # afw's ExposureReader strips the VisitInfo's header cards from its 

676 # metadata, but we want to include them in the FITS header we 

677 # attach. 

678 pl = PropertyList() 

679 setVisitInfoMetadata(pl, visit_info) 

680 primary_header.update(pl.to_astropy_header()) 

681 telescope = primary_header.get("TELESCOP") 

682 obs_info = obs_info_from_legacy( 

683 visit_info, 

684 exposure_record, 

685 legacy_detector, 

686 detector_exposure_id=legacy.info.getId(), 

687 telescope=telescope, 

688 physical_filter=legacy.info.getFilter().physicalLabel or None, 

689 header=primary_header, 

690 ) 

691 if exposure_record is not None: 

692 instrument = exposure_record.instrument 

693 visit = exposure_record.id # assume visit ID == exposure ID; always true in practice 

694 else: 

695 instrument = visit_info.getInstrumentLabel() 

696 visit = visit_info.getId() 

697 opaque_fits_metadata = FitsOpaqueMetadata() 

698 metadata = opaque_fits_metadata.extract_legacy_primary_header(primary_header) 

699 instrumental_unit = opaque_fits_metadata.get_instrumental_unit() or astropy.units.electron 

700 hdr_unit: astropy.units.UnitBase | None = None 

701 if hdr_unit_str := md.get("BUNIT"): 701 ↛ 703line 701 didn't jump to line 703 because the condition on line 701 was always true

702 hdr_unit = parse_legacy_bunit(hdr_unit_str, instrumental_unit=instrumental_unit) 

703 if unit is None: 703 ↛ 705line 703 didn't jump to line 705 because the condition on line 703 was always true

704 unit = hdr_unit 

705 elif hdr_unit is not None and hdr_unit != unit: 

706 raise ValueError(f"BUNIT value {hdr_unit} disagrees with given unit {unit}.") 

707 sky_projection = SkyProjection.from_legacy( 

708 legacy_wcs, 

709 DetectorFrame( 

710 instrument=instrument, 

711 visit=visit, 

712 detector=legacy_detector.getId(), 

713 bbox=detector_bbox, 

714 ), 

715 ) 

716 legacy_psf = legacy.getPsf() 

717 if legacy_psf is None: 717 ↛ 718line 717 didn't jump to line 718 because the condition on line 717 was never true

718 raise ValueError("Exposure file does not have a Psf.") 

719 psf = PointSpreadFunction.from_legacy(legacy_psf, bounds=detector_bbox) 

720 masked_image = MaskedImage.from_legacy(legacy.getMaskedImage(), unit=unit, plane_map=plane_map) 

721 legacy_summary_stats = legacy.info.getSummaryStats() 

722 legacy_ap_corr_map = legacy.info.getApCorrMap() 

723 legacy_polygon = legacy.info.getValidPolygon() 

724 legacy_photo_calib = legacy.info.getPhotoCalib() 

725 detector = Detector.from_legacy( 

726 legacy_detector, instrument=instrument, visit=visit, is_raw_assembled=True 

727 ) 

728 if instrument == "LSSTCam": 

729 obs_info = _fix_lsstcam_detector_serial(obs_info, detector) 

730 result = VisitImage( 

731 image=masked_image.image.view(unit=unit), 

732 mask=masked_image.mask, 

733 variance=masked_image.variance, 

734 sky_projection=sky_projection, 

735 psf=psf, 

736 obs_info=obs_info, 

737 summary_stats=( 

738 ObservationSummaryStats.from_legacy(legacy_summary_stats) 

739 if legacy_summary_stats is not None 

740 else None 

741 ), 

742 detector=detector, 

743 aperture_corrections=( 

744 aperture_corrections_from_legacy(legacy_ap_corr_map) 

745 if legacy_ap_corr_map is not None 

746 else None 

747 ), 

748 bounds=Polygon.from_legacy(legacy_polygon) if legacy_polygon is not None else None, 

749 photometric_scaling=( 

750 field_from_legacy_photo_calib( 

751 legacy_photo_calib, bounds=detector_bbox, instrumental_unit=instrumental_unit 

752 ) 

753 if legacy_photo_calib is not None 

754 else None 

755 ), 

756 band=legacy.info.getFilter().bandLabel, 

757 metadata=metadata, 

758 ) 

759 result._metadata["id"] = legacy.info.getId() 

760 result._opaque_metadata = opaque_fits_metadata 

761 return result 

762 

763 def to_legacy( 

764 self, *, copy: bool | None = None, plane_map: Mapping[str, MaskPlane] | None = None 

765 ) -> LegacyExposure: 

766 """Convert to an `lsst.afw.image.Exposure` instance. 

767 

768 Parameters 

769 ---------- 

770 copy 

771 If `True`, always copy the image and variance pixel data. 

772 If `False`, return a view, and raise `TypeError` if the pixel data 

773 is read-only (this is not supported by afw). If `None`, only copy 

774 if the pixel data is read-only. Mask pixel data is always copied. 

775 plane_map 

776 A mapping from legacy mask plane name to the new plane name and 

777 description. If `None` (default), 

778 `get_legacy_visit_image_mask_planes` is used. 

779 """ 

780 from lsst.afw.image import Exposure as LegacyExposure 

781 from lsst.afw.image import FilterLabel as LegacyFilterLabel 

782 from lsst.obs.base.makeRawVisitInfoViaObsInfo import MakeRawVisitInfoViaObsInfo 

783 

784 if plane_map is None: 784 ↛ 786line 784 didn't jump to line 786 because the condition on line 784 was always true

785 plane_map = get_legacy_visit_image_mask_planes() 

786 legacy_masked_image = super().to_legacy(copy=copy, plane_map=plane_map) 

787 result = LegacyExposure(legacy_masked_image, dtype=self.image.array.dtype) 

788 result_info = result.info 

789 result_info.setId(self._metadata.get("id")) 

790 result_info.setWcs(self.sky_projection.to_legacy()) 

791 result_info.setDetector(self.detector.to_legacy()) 

792 result_info.setFilter(LegacyFilterLabel.fromBandPhysical(self.band, self.obs_info.physical_filter)) 

793 if self._photometric_scaling is not None: 793 ↛ 794line 793 didn't jump to line 794 because the condition on line 793 was never true

794 result_info.setPhotoCalib(self._photometric_scaling.to_legacy_photo_calib(self.unit)) 

795 else: 

796 result_info.setPhotoCalib(BaseField.make_legacy_photo_calib(self.unit)) 

797 self._fill_legacy_metadata(result_info.getMetadata()) 

798 if isinstance(self._psf, LegacyPointSpreadFunction): 798 ↛ 799line 798 didn't jump to line 799 because the condition on line 798 was never true

799 result_info.setPsf(self._psf.legacy_psf) 

800 elif isinstance(self._psf, PiffWrapper | GaussianPointSpreadFunction): 800 ↛ 802line 800 didn't jump to line 802 because the condition on line 800 was always true

801 result_info.setPsf(self._psf.to_legacy()) 

802 if isinstance(self.bounds, Polygon): 802 ↛ 803line 802 didn't jump to line 803 because the condition on line 802 was never true

803 result_info.setValidPolygon(self.bounds.to_legacy()) 

804 if self.aperture_corrections: 804 ↛ 805line 804 didn't jump to line 805 because the condition on line 804 was never true

805 result_info.setApCorrMap(aperture_corrections_to_legacy(self.aperture_corrections)) 

806 result_info.setVisitInfo(MakeRawVisitInfoViaObsInfo.observationInfo2visitInfo(self.obs_info)) 

807 result_info.setSummaryStats(self.summary_stats.to_legacy()) 

808 return result 

809 

810 @staticmethod 

811 def read_legacy( # type: ignore[override] 

812 filename: str, 

813 *, 

814 exposure_record: DimensionRecord | None, 

815 preserve_quantization: bool = False,