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

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 @property 

277 def detector(self) -> Detector: 

278 """Geometry and electronic information about the detector 

279 (`.cameras.Detector`). 

280 """ 

281 return self._detector 

282 

283 @property 

284 def aperture_corrections(self) -> ApertureCorrectionMap: 

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

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

287 """ 

288 return self._aperture_corrections 

289 

290 @property 

291 def backgrounds(self) -> BackgroundMap: 

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

293 (`BackgroundMap`). 

294 """ 

295 return self._backgrounds 

296 

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

298 bbox, _ = self._handle_getitem_args(bbox) 

299 return self._transfer_metadata( 

300 VisitImage( 

301 self.image[bbox], 

302 mask=self.mask[bbox], 

303 variance=self.variance[bbox], 

304 sky_projection=self.sky_projection, 

305 psf=self._psf, 

306 obs_info=self.obs_info, 

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

308 summary_stats=self.summary_stats, 

309 detector=self._detector, 

310 photometric_scaling=self._photometric_scaling, 

311 aperture_corrections=self.aperture_corrections, 

312 backgrounds=self._backgrounds, 

313 band=self._band, 

314 ), 

315 bbox=bbox, 

316 ) 

317 

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

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

320 with. 

321 

322 Parameters 

323 ---------- 

324 options : `DescribeOptions` 

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

326 

327 Returns 

328 ------- 

329 options : `DescribeOptions` 

330 Options for the child report. 

331 

332 Notes 

333 ----- 

334 `ObservationSummaryStats.raCorners` and 

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

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

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

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

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

340 """ 

341 if self.bbox == self.detector.bbox: 

342 return options.for_child("corners") 

343 return options.for_child() 

344 

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

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

347 stopped it being read. 

348 

349 Parameters 

350 ---------- 

351 options : `DescribeOptions` 

352 Options the child report is built with. 

353 

354 Returns 

355 ------- 

356 report : `Report` 

357 Report describing the PSF, or naming it unreadable. 

358 

359 Notes 

360 ----- 

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

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

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

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

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

366 """ 

367 if isinstance(self._psf, ArchiveReadError): 

368 return Report( 

369 type_name="PointSpreadFunction", 

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

371 inline=True, 

372 ) 

373 return self._psf._describe(options) 

374 

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

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

377 

378 Parameters 

379 ---------- 

380 options : `DescribeOptions`, optional 

381 Rendering options; forwarded to all children. Child construction 

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

383 """ 

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

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

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

387 fields = [ 

388 ReportField( 

389 label="image", 

390 value=self.image, 

391 repr_value=repr(self.image), 

392 positional=True, 

393 role=FieldRole.REPR_ONLY, 

394 ), 

395 ReportField( 

396 label="mask_schema", 

397 value=self.mask.schema, 

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

399 role=FieldRole.REPR_ONLY, 

400 ), 

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

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

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

404 ] 

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

406 if options.brief: 

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

408 child = options.for_child() 

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

410 children: dict[str, Report] = { 

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

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

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

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

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

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

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

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

419 } 

420 if self.photometric_scaling is not None: 

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

422 return Report( 

423 type_name="VisitImage", 

424 summary=summary, 

425 fields=fields, 

426 children=children, 

427 ) 

428 

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

430 """Deep-copy the visit image. 

431 

432 Parameters 

433 ---------- 

434 copy_detector 

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

436 """ 

437 return self._transfer_metadata( 

438 VisitImage( 

439 image=self._image.copy(), 

440 mask=self._mask.copy(), 

441 variance=self._variance.copy(), 

442 psf=self._psf, 

443 obs_info=self.obs_info, 

444 bounds=self._bounds, 

445 summary_stats=self.summary_stats.model_copy(), 

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

447 photometric_scaling=self._photometric_scaling, 

448 aperture_corrections=self.aperture_corrections.copy(), 

449 backgrounds=self._backgrounds.copy(), 

450 band=self.band, 

451 ), 

452 copy=True, 

453 ) 

454 

455 def convert_unit( 

456 self, 

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

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

459 copy_detector: bool = False, 

460 ) -> VisitImage: 

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

462 

463 Parameters 

464 ---------- 

465 unit 

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

467 

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

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

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

471 instrumental units but we know how to calibrate them) 

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

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

474 calibrated units and we want to revert back to instrumental 

475 units). 

476 copy 

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

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

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

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

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

482 already in the right units. 

483 copy_detector 

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

485 

486 Returns 

487 ------- 

488 `VisitImage` 

489 An image with the given units. 

490 """ 

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

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

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

494 if factor == 1.0: 

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

496 return self.copy() 

497 else: 

498 return self[...] 

499 elif copy is False: 

500 raise astropy.units.UnitConversionError( 

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

502 ) 

503 image = Image( 

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

505 ) 

506 variance = Image( 

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

508 bbox=self.bbox, 

509 unit=unit**2, 

510 ) 

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

512 raise astropy.units.UnitConversionError( 

513 "VisitImage.photometric_scaling is None, and there " 

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

515 ) 

516 else: 

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

518 raise astropy.units.UnitConversionError( 

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

520 ) 

521 scaling = self._photometric_scaling 

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

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

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

525 scaling = scaling * constant_factor 

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

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

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 np.true_divide(1.0, scaling_array, out=scaling_array) 

532 else: 

533 raise astropy.units.UnitConversionError( 

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

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

536 ) 

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

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

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

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

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

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

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

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

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

546 image.array *= self._image.array 

547 variance.array *= self._variance.array 

548 copy_components = copy is True 

549 return self._transfer_metadata( 

550 VisitImage( 

551 image=image, 

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

553 variance=variance, 

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

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

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

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

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

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

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

561 aperture_corrections=( 

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

563 ), 

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

565 band=self.band, 

566 ) 

567 ) 

568 

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

570 return self._serialize_impl(VisitImageSerializationModel, archive) 

571 

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

573 # VisitImageSerializationModel, not just a MaskedImageSerializationModel, 

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

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

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

577 ) -> M: 

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

579 match self._psf: 

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

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

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

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

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

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

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

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

588 case _: 

589 raise TypeError( 

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

591 ) 

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

593 result.obs_info = self.obs_info 

594 result.summary_stats = self.summary_stats 

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

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

597 result.band = self.band 

598 result.photometric_scaling = ( 

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

600 # method return types. 

601 archive.serialize_direct( 

602 "photometric_scaling", 

603 self._photometric_scaling.serialize, 

604 ) # type: ignore[assignment] 

605 if self._photometric_scaling is not None 

606 else None 

607 ) 

608 result.aperture_corrections = archive.serialize_direct( 

609 "aperture_corrections", 

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

611 ) 

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

613 return result 

614 

615 @staticmethod 

616 def _get_archive_tree_type[P: pydantic.BaseModel]( 

617 pointer_type: type[P], 

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

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

620 type that uses the given pointer type. 

621 """ 

622 return VisitImageSerializationModel[pointer_type] # type: ignore 

623 

624 @staticmethod 

625 def from_legacy( # type: ignore[override] 

626 legacy: LegacyExposure, 

627 *, 

628 exposure_record: DimensionRecord | None, 

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

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

631 ) -> VisitImage: 

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

633 

634 Parameters 

635 ---------- 

636 legacy 

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

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

639 exposure_record 

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

641 source of additional required metadata. Pass `None` explicitly to 

642 convert without some `~astro_metadata_translator.ObservationInfo` 

643 fields (with a warning). 

644 unit 

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

646 key will be used, if available. 

647 plane_map 

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

649 description. If `None` (default) 

650 `get_legacy_visit_image_mask_planes` is used. 

651 """ 

652 from lsst.afw.image import setVisitInfoMetadata 

653 from lsst.daf.base import PropertyList 

654 

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

656 plane_map = get_legacy_visit_image_mask_planes() 

657 legacy_wcs = legacy.getWcs() 

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

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

660 legacy_detector = legacy.getDetector() 

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

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

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

664 visit_info = legacy.info.getVisitInfo() 

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

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

667 md = legacy.getMetadata() 

668 primary_header = md.to_astropy_header() 

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

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

671 # attach. 

672 pl = PropertyList() 

673 setVisitInfoMetadata(pl, visit_info) 

674 primary_header.update(pl.to_astropy_header()) 

675 telescope = primary_header.get("TELESCOP") 

676 obs_info = obs_info_from_legacy( 

677 visit_info, 

678 exposure_record, 

679 legacy_detector, 

680 detector_exposure_id=legacy.info.getId(), 

681 telescope=telescope, 

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

683 ) 

684 if exposure_record is not None: 684 ↛ 688line 684 didn't jump to line 688 because the condition on line 684 was always true

685 instrument = exposure_record.instrument 

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

687 else: 

688 instrument = visit_info.getInstrumentLabel() 

689 visit = visit_info.getId() 

690 opaque_fits_metadata = FitsOpaqueMetadata() 

691 metadata = opaque_fits_metadata.extract_legacy_primary_header(primary_header) 

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

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

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

695 hdr_unit = parse_legacy_bunit(hdr_unit_str, instrumental_unit=instrumental_unit) 

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

697 unit = hdr_unit 

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

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

700 sky_projection = SkyProjection.from_legacy( 

701 legacy_wcs, 

702 DetectorFrame( 

703 instrument=instrument, 

704 visit=visit, 

705 detector=legacy_detector.getId(), 

706 bbox=detector_bbox, 

707 ), 

708 ) 

709 legacy_psf = legacy.getPsf() 

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

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

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

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

714 legacy_summary_stats = legacy.info.getSummaryStats() 

715 legacy_ap_corr_map = legacy.info.getApCorrMap() 

716 legacy_polygon = legacy.info.getValidPolygon() 

717 legacy_photo_calib = legacy.info.getPhotoCalib() 

718 detector = Detector.from_legacy( 

719 legacy_detector, instrument=instrument, visit=visit, is_raw_assembled=True 

720 ) 

721 if instrument == "LSSTCam": 

722 obs_info = _fix_lsstcam_detector_serial(obs_info, detector) 

723 result = VisitImage( 

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

725 mask=masked_image.mask, 

726 variance=masked_image.variance, 

727 sky_projection=sky_projection, 

728 psf=psf, 

729 obs_info=obs_info, 

730 summary_stats=( 

731 ObservationSummaryStats.from_legacy(legacy_summary_stats) 

732 if legacy_summary_stats is not None 

733 else None 

734 ), 

735 detector=detector, 

736 aperture_corrections=( 

737 aperture_corrections_from_legacy(legacy_ap_corr_map) 

738 if legacy_ap_corr_map is not None 

739 else None 

740 ), 

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

742 photometric_scaling=( 

743 field_from_legacy_photo_calib( 

744 legacy_photo_calib, bounds=detector_bbox, instrumental_unit=instrumental_unit 

745 ) 

746 if legacy_photo_calib is not None 

747 else None 

748 ), 

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

750 metadata=metadata, 

751 ) 

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

753 result._opaque_metadata = opaque_fits_metadata 

754 return result 

755 

756 def to_legacy( 

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

758 ) -> LegacyExposure: 

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

760 

761 Parameters 

762 ---------- 

763 copy 

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

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

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

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

768 plane_map 

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

770 description. If `None` (default), 

771 `get_legacy_visit_image_mask_planes` is used. 

772 """ 

773 from lsst.afw.image import Exposure as LegacyExposure 

774 from lsst.afw.image import FilterLabel as LegacyFilterLabel 

775 from lsst.obs.base.makeRawVisitInfoViaObsInfo import MakeRawVisitInfoViaObsInfo 

776 

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

778 plane_map = get_legacy_visit_image_mask_planes() 

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

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

781 result_info = result.info 

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

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

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

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

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

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

788 else: 

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

790 self._fill_legacy_metadata(result_info.getMetadata()) 

791 if isinstance(self._psf, LegacyPointSpreadFunction): 791 ↛ 793line 791 didn't jump to line 793 because the condition on line 791 was always true

792 result_info.setPsf(self._psf.legacy_psf) 

793 elif isinstance(self._psf, PiffWrapper): 

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

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

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

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

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

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

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

801 return result 

802 

803 @staticmethod 

804 def read_legacy( # type: ignore[override] 

805 filename: str, 

806 *, 

807 exposure_record: DimensionRecord | None, 

808 preserve_quantization: bool = False, 

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

810 component: Literal[ 

811 "bbox", 

812 "image", 

813 "mask", 

814 "variance",