Coverage for python/lsst/images/_masked_image.py: 44%

199 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-30 08:38 +0000

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__ = ("MaskedImage", "MaskedImageSerializationModel") 

15 

16import functools 

17from collections.abc import Mapping, MutableMapping 

18from contextlib import ExitStack 

19from types import EllipsisType 

20from typing import TYPE_CHECKING, Any, ClassVar, Literal 

21 

22import astropy.io.fits 

23import astropy.units 

24import astropy.wcs 

25import numpy as np 

26import pydantic 

27 

28from lsst.resources import ResourcePath, ResourcePathExpression 

29 

30from . import fits 

31from ._generalized_image import GeneralizedImage 

32from ._geom import Box 

33from ._image import DEFAULT_PIXEL_FRAME, Image, ImageSerializationModel 

34from ._mask import Mask, MaskPlane, MaskSchema, MaskSerializationModel 

35from ._transforms import Frame, SkyProjection, SkyProjectionSerializationModel 

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

37from .serialization import ( 

38 ArchiveTree, 

39 InputArchive, 

40 InvalidParameterError, 

41 MetadataValue, 

42 OutputArchive, 

43) 

44from .utils import is_none 

45 

46if TYPE_CHECKING: 

47 try: 

48 from lsst.afw.image import MaskedImage as LegacyMaskedImage 

49 except ImportError: 

50 type LegacyMaskedImage = Any # type: ignore[no-redef] 

51 

52 

53class MaskedImage(GeneralizedImage): 

54 """A multi-plane image with data (image), mask, and variance planes. 

55 

56 Parameters 

57 ---------- 

58 image 

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

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

61 mask 

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

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

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

65 variance 

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

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

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

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

70 (possibly by `None`). 

71 mask_schema 

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

73 not provided. 

74 sky_projection 

75 Projection that maps the pixel grid to the sky. 

76 metadata 

77 Arbitrary flexible metadata to associate with the image. 

78 """ 

79 

80 def __init__( 

81 self, 

82 image: Image, 

83 *, 

84 mask: Mask | None = None, 

85 variance: Image | None = None, 

86 mask_schema: MaskSchema | None = None, 

87 sky_projection: SkyProjection[Any] | None = None, 

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

89 ) -> None: 

90 super().__init__(metadata) 

91 if sky_projection is None: 

92 sky_projection = image.sky_projection 

93 else: 

94 image = image.view(sky_projection=sky_projection) 

95 if mask is None: 

96 if mask_schema is None: 

97 raise TypeError("'mask_schema' must be provided if 'mask' is not.") 

98 mask = Mask(schema=mask_schema, bbox=image.bbox, sky_projection=sky_projection) 

99 elif mask_schema is not None: 

100 raise TypeError("'mask_schema' may not be provided if 'mask' is.") 

101 else: 

102 if image.bbox != mask.bbox: 

103 raise ValueError(f"Image ({image.bbox}) and mask ({mask.bbox}) bboxes do not agree.") 

104 mask = mask.view(sky_projection=sky_projection) 

105 if variance is None: 

106 variance = Image( 

107 1.0, 

108 dtype=np.float32, 

109 bbox=image.bbox, 

110 unit=None if image.unit is None else image.unit**2, 

111 sky_projection=sky_projection, 

112 ) 

113 else: 

114 if image.bbox != variance.bbox: 

115 raise ValueError(f"Image ({image.bbox}) and variance ({variance.bbox}) bboxes do not agree.") 

116 variance = variance.view(sky_projection=sky_projection) 

117 if image.unit is None: 

118 if variance.unit is not None: 

119 raise ValueError(f"Image has no units but variance does ({variance.unit}).") 

120 elif variance.unit is None: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 variance = variance.view(unit=image.unit**2) 

122 elif variance.unit != image.unit**2: 

123 raise ValueError( 

124 f"Variance unit ({variance.unit}) should be the square of the image unit ({image.unit})." 

125 ) 

126 self._image = image 

127 self._mask = mask 

128 self._variance = variance 

129 

130 @property 

131 def image(self) -> Image: 

132 """The main image plane (`~lsst.images.Image`).""" 

133 return self._image 

134 

135 @property 

136 def mask(self) -> Mask: 

137 """The mask plane (`~lsst.images.Mask`). 

138 

139 Assigning a new `~lsst.images.Mask` (for example one returned by 

140 `~lsst.images.Mask.add_planes`) replaces the mask plane. The new mask 

141 must share this image's bounding box; its sky projection is replaced 

142 with the image's. 

143 """ 

144 return self._mask 

145 

146 @mask.setter 

147 def mask(self, value: Mask) -> None: 

148 if value.bbox != self._image.bbox: 

149 raise ValueError(f"Image ({self._image.bbox}) and mask ({value.bbox}) bboxes do not agree.") 

150 if self._image.sky_projection != value.sky_projection: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 raise ValueError("Image sky projection and new mask sky projection do not agree") 

152 # Use a view to ensure that the WCS instances across the masked image 

153 # are the same projections. 

154 self._mask = value.view(sky_projection=self._image.sky_projection) 

155 

156 @property 

157 def variance(self) -> Image: 

158 """The variance plane (`~lsst.images.Image`).""" 

159 return self._variance 

160 

161 @property 

162 def bbox(self) -> Box: 

163 """The bounding box shared by all three image planes 

164 (`~lsst.images.Box`). 

165 """ 

166 return self._image.bbox 

167 

168 @property 

169 def unit(self) -> astropy.units.UnitBase | None: 

170 """The units of the image plane (`astropy.units.Unit` | `None`).""" 

171 return self._image.unit 

172 

173 @property 

174 def sky_projection(self) -> SkyProjection[Any] | None: 

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

176 (`~lsst.images.SkyProjection` | `None`). 

177 """ 

178 return self._image.sky_projection 

179 

180 def __getitem__(self, bbox: Box | EllipsisType) -> MaskedImage: 

181 bbox, _ = self._handle_getitem_args(bbox) 

182 return self._transfer_metadata( 

183 MaskedImage( 

184 # Projection and obs_info propagate from the image. 

185 self.image[bbox], 

186 mask=self.mask[bbox], 

187 variance=self.variance[bbox], 

188 ), 

189 bbox=bbox, 

190 ) 

191 

192 def __setitem__(self, bbox: Box | EllipsisType, value: MaskedImage) -> None: 

193 self._image[bbox] = value.image 

194 self._mask[bbox] = value.mask 

195 self._variance[bbox] = value.variance 

196 

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

198 """Return a `Report` describing this masked image. 

199 

200 Parameters 

201 ---------- 

202 options : `DescribeOptions`, optional 

203 Rendering options; forwarded to all children. 

204 """ 

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

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

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

208 fields = [ 

209 ReportField( 

210 label="image", 

211 value=self.image, 

212 repr_value=repr(self.image), 

213 positional=True, 

214 role=FieldRole.REPR_ONLY, 

215 ), 

216 ReportField( 

217 label="mask_schema", 

218 value=self.mask.schema, 

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

220 role=FieldRole.REPR_ONLY, 

221 ), 

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

223 ] 

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

225 if options.brief: 

226 return Report(type_name="MaskedImage", summary=summary, fields=fields) 

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

228 children = { 

229 "image": self._image._describe(plane), 

230 "mask": self._mask._describe(plane), 

231 "variance": self._variance._describe(plane), 

232 } 

233 if self.sky_projection is not None: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true

234 children["sky_projection"] = self.sky_projection._describe(options.for_child(), bbox=self.bbox) 

235 return Report( 

236 type_name="MaskedImage", 

237 summary=summary, 

238 fields=fields, 

239 children=children, 

240 ) 

241 

242 def copy(self) -> MaskedImage: 

243 """Deep-copy the masked image and metadata.""" 

244 return self._transfer_metadata( 

245 MaskedImage(image=self._image.copy(), mask=self._mask.copy(), variance=self._variance.copy()), 

246 copy=True, 

247 ) 

248 

249 def serialize(self, archive: OutputArchive[Any]) -> MaskedImageSerializationModel[Any]: 

250 """Serialize the masked image to an output archive. 

251 

252 Parameters 

253 ---------- 

254 archive 

255 Archive to write to. 

256 """ 

257 return self._serialize_impl(MaskedImageSerializationModel, archive) 

258 

259 def _serialize_impl[M: MaskedImageSerializationModel[Any]]( 

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

261 ) -> M: 

262 serialized_image = archive.serialize_direct( 

263 "image", functools.partial(self.image.serialize, save_projection=False) 

264 ) 

265 serialized_mask = archive.serialize_direct( 

266 "mask", functools.partial(self.mask.serialize, save_projection=False) 

267 ) 

268 serialized_variance = archive.serialize_direct( 

269 "variance", functools.partial(self.variance.serialize, save_projection=False) 

270 ) 

271 serialized_projection = ( 

272 archive.serialize_direct("sky_projection", self.sky_projection.serialize) 

273 if self.sky_projection is not None 

274 else None 

275 ) 

276 # When M is a subclass of MaskedImageSerializationModel, it probably 

277 # has fields that aren't being set here. We're intentionally making use 

278 # of the fact that model_construct doesn't guard against that so we can 

279 # instead set them in a subclass implementation later, after calling 

280 # super() to construct an instance of the right type. MyPy is actually 

281 # fine with this, but only because it incorrectly thinks model_type is 

282 # only ever MaskedImageSerializationModel, not a subclass, and that 

283 # makes it incorrectly unhappy about the return type. 

284 return model_type.model_construct( # type: ignore[return-value] 

285 image=serialized_image, 

286 mask=serialized_mask, 

287 variance=serialized_variance, 

288 sky_projection=serialized_projection, 

289 metadata=self.metadata, 

290 ) 

291 

292 @staticmethod 

293 def _get_archive_tree_type[P: pydantic.BaseModel]( 

294 pointer_type: type[P], 

295 ) -> type[MaskedImageSerializationModel[P]]: 

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

297 type that uses the given pointer type. 

298 """ 

299 return MaskedImageSerializationModel[pointer_type] # type: ignore 

300 

301 @staticmethod 

302 def from_legacy( 

303 legacy: LegacyMaskedImage, 

304 *, 

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

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

307 sky_projection: SkyProjection[Any] | None = None, 

308 ) -> MaskedImage: 

309 """Convert from an `lsst.afw.image.MaskedImage` instance. 

310 

311 Parameters 

312 ---------- 

313 legacy 

314 An `lsst.afw.image.MaskedImage` instance that will share image and 

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

316 unit 

317 Units of the image. 

318 plane_map 

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

320 description. If not provided, the right legacy mask plane will be 

321 guessed, but this can depend on which mask planes the legacy 

322 mask actually has set. 

323 sky_projection 

324 Projection from pixels to xky. 

325 """ 

326 return MaskedImage( 

327 image=Image.from_legacy(legacy.getImage(), unit, sky_projection=sky_projection), 

328 mask=Mask.from_legacy(legacy.getMask(), plane_map, sky_projection=sky_projection), 

329 variance=Image.from_legacy(legacy.getVariance(), sky_projection=sky_projection), 

330 ) 

331 

332 def to_legacy( 

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

334 ) -> LegacyMaskedImage: 

335 """Convert to an `lsst.afw.image.MaskedImage` instance. 

336 

337 Parameters 

338 ---------- 

339 copy 

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

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

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

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

344 plane_map 

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

346 description. 

347 """ 

348 import lsst.afw.image 

349 

350 return lsst.afw.image.MaskedImage( 

351 self.image.to_legacy(copy=copy), 

352 mask=self.mask.to_legacy(plane_map), 

353 variance=self.variance.to_legacy(copy=copy), 

354 dtype=self.image.array.dtype, 

355 ) 

356 

357 @classmethod 

358 def from_hdu_list( 

359 cls, 

360 hdu_list: astropy.io.fits.HDUList, 

361 *, 

362 fits_wcs_frame: Frame | None = DEFAULT_PIXEL_FRAME, 

363 ) -> MaskedImage: 

364 """Reconstruct a `~lsst.images.MaskedImage` from a cut-down 

365 ``lsst.images`` FITS HDU list. 

366 

367 This assumes the ``PRIMARY``, ``IMAGE``, ``MASK``, and ``VARIANCE`` 

368 HDUs written for the masked-image cut-outs produced by 

369 ``dax_images_cutout``: a real ``lsst.images`` file with its JSON-tree, 

370 index, and any nested-archive HDUs dropped. The reconstructed object 

371 can be re-serialized as a normal ``lsst.images`` file (with schema and 

372 index) so it can be read with the full ``lsst.images`` infrastructure. 

373 

374 Parameters 

375 ---------- 

376 hdu_list 

377 HDU list with ``IMAGE``, ``MASK``, and ``VARIANCE`` extensions and 

378 a primary HDU. 

379 fits_wcs_frame 

380 Pixel-grid `~lsst.images.Frame` for the 

381 `~lsst.images.SkyProjection` reconstructed from the FITS WCS. 

382 Defaults to a plain pixel frame; pass `None` to skip attaching a 

383 projection. 

384 

385 Returns 

386 ------- 

387 `~lsst.images.MaskedImage` 

388 The reconstructed masked image. 

389 

390 Raises 

391 ------ 

392 ValueError 

393 Raised if the ``MASK`` HDU has neither ``MSKN`` nor ``MP_`` mask- 

394 plane cards, since the mask schema cannot then be reconstructed, or 

395 if ``hdu_list`` contains more than one ``MASK`` HDU (multiple 

396 ``MASK`` extensions, distinguished by ``EXTVER``, are not handled 

397 here and would otherwise be silently dropped). 

398 

399 Notes 

400 ----- 

401 Both mask-plane conventions are supported: the self-describing 

402 ``MSKN``/``MSKM``/``MSKD`` cards written by ``lsst.images``, and the 

403 legacy `lsst.afw.image` ``MP_*`` cards (as produced by 

404 ``dax_images_cutout`` from afw-written images). Legacy masks are 

405 mapped to a new schema with the same plane-guessing used by 

406 `read_legacy`. 

407 

408 Unlike `read_legacy`, the legacy ``MP_*`` mask-plane cards are kept 

409 (not stripped) for backwards compatibility, since this path 

410 reconstructs a file that may still be read by legacy tooling. They are 

411 re-indexed to the reshuffled schema so each ``MP_`` bit matches the 

412 plane's position in the written ``MSKN`` layout. 

413 

414 The headers of the HDUs in ``hdu_list`` are modified in place: the WCS 

415 and mask-schema cards interpreted here are stripped from the caller's 

416 headers. 

417 """ 

418 n_mask_hdus = sum(1 for hdu in hdu_list if hdu.name == "MASK") 

419 if n_mask_hdus > 1: 

420 raise ValueError( 

421 f"Found {n_mask_hdus} MASK HDUs; from_hdu_list supports only a single MASK " 

422 "extension and would otherwise silently drop mask information from the others." 

423 ) 

424 mask_hdu = hdu_list["MASK"] 

425 if not any(card.keyword.startswith(("MSKN", "MP_")) for card in mask_hdu.header.cards): 

426 raise ValueError("MASK HDU has no MSKN or MP_ cards; cannot reconstruct the mask schema.") 

427 opaque_metadata = fits.FitsOpaqueMetadata() 

428 opaque_metadata.add_cutdown_primary_header(hdu_list[0].header) 

429 image = Image._read_legacy_hdu( 

430 hdu_list["IMAGE"], opaque_metadata, preserve_bintable=None, fits_wcs_frame=fits_wcs_frame 

431 ) 

432 mask = Mask._read_legacy_hdu( 

433 mask_hdu, opaque_metadata, fits_wcs_frame=None, strip_legacy_planes=False 

434 ) 

435 variance = Image._read_legacy_hdu( 

436 hdu_list["VARIANCE"], opaque_metadata, preserve_bintable=None, fits_wcs_frame=None 

437 ) 

438 result = cls(image, mask=mask, variance=variance) 

439 result._opaque_metadata = opaque_metadata 

440 return result 

441 

442 @staticmethod 

443 def read_legacy( 

444 uri: ResourcePathExpression, 

445 *, 

446 preserve_quantization: bool = False, 

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

448 component: Literal["image", "mask", "variance"] | None = None, 

449 fits_wcs_frame: Frame | None = None, 

450 ) -> Any: 

451 """Read a FITS file written by `lsst.afw.image.MaskedImage.writeFits`. 

452 

453 Parameters 

454 ---------- 

455 uri 

456 URI or file name. 

457 preserve_quantization 

458 If `True`, ensure that writing the masked image back out again will 

459 exactly preserve quantization-compressed pixel values. This causes 

460 the image and variance plane arrays to be marked as read-only and 

461 stores the original binary table data for those planes in memory. 

462 If the `~lsst.images.MaskedImage` is copied, the precompressed 

463 pixel values are not transferred to the copy. 

464 plane_map 

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

466 description. If not provided, the right legacy mask plane will be 

467 guessed, but this can depend on which mask planes the legacy 

468 mask actually has set. 

469 component 

470 A component to read instead of the full image. 

471 fits_wcs_frame 

472 If not `None` and the HDU containing the image plane has a FITS 

473 WCS, attach a `~lsst.images.SkyProjection` to the returned masked 

474 image by converting that WCS. When ``component`` is one of 

475 ``"image"``, ``"mask"``, or ``"variance"``, a FITS WCS from the 

476 component HDU is used instead (all three should have the same WCS). 

477 """ 

478 fs, fspath = ResourcePath(uri).to_fsspec() 

479 with fs.open(fspath) as stream, astropy.io.fits.open(stream) as hdu_list: 

480 return MaskedImage._read_legacy_hdus( 

481 hdu_list, 

482 uri, 

483 preserve_quantization=preserve_quantization, 

484 plane_map=plane_map, 

485 component=component, 

486 fits_wcs_frame=fits_wcs_frame, 

487 ) 

488 

489 @staticmethod 

490 def _read_legacy_hdus( 

491 hdu_list: astropy.io.fits.HDUList, 

492 uri: ResourcePathExpression, 

493 *, 

494 opaque_metadata: fits.FitsOpaqueMetadata | None = None, 

495 preserve_quantization: bool = False, 

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

497 component: Literal["image", "mask", "variance"] | None, 

498 fits_wcs_frame: Frame | None = None, 

499 ) -> Any: 

500 if opaque_metadata is None: 

501 opaque_metadata = fits.FitsOpaqueMetadata() 

502 opaque_metadata.extract_legacy_primary_header(hdu_list[0].header) 

503 image_bintable_hdu: astropy.io.fits.BinTableHDU | None = None 

504 variance_bintable_hdu: astropy.io.fits.BinTableHDU | None = None 

505 result: Any 

506 with ExitStack() as exit_stack: 

507 if preserve_quantization: 

508 fs, fspath = ResourcePath(uri).to_fsspec() 

509 bintable_stream = exit_stack.enter_context(fs.open(fspath)) 

510 bintable_hdu_list = exit_stack.enter_context( 

511 astropy.io.fits.open(bintable_stream, disable_image_compression=True) 

512 ) 

513 image_bintable_hdu = bintable_hdu_list[1] 

514 variance_bintable_hdu = bintable_hdu_list[3] 

515 if component is None or component == "image": 

516 image = Image._read_legacy_hdu( 

517 hdu_list[1], 

518 opaque_metadata, 

519 preserve_bintable=image_bintable_hdu, 

520 fits_wcs_frame=fits_wcs_frame, 

521 ) 

522 if component == "image": 

523 result = image 

524 if component is None or component == "mask": 

525 mask = Mask._read_legacy_hdu( 

526 hdu_list[2], 

527 opaque_metadata, 

528 plane_map=plane_map, 

529 fits_wcs_frame=fits_wcs_frame if component is not None else None, 

530 ) 

531 if component == "mask": 

532 result = mask 

533 if component is None or component == "variance": 

534 variance = Image._read_legacy_hdu( 

535 hdu_list[3], 

536 opaque_metadata, 

537 preserve_bintable=variance_bintable_hdu, 

538 fits_wcs_frame=fits_wcs_frame if component is not None else None, 

539 ) 

540 if component == "variance": 

541 result = variance 

542 if component is None: 

543 result = MaskedImage(image, mask=mask, variance=variance) 

544 result._opaque_metadata = opaque_metadata 

545 return result 

546 

547 def _fill_legacy_metadata(self, legacy_metadata: MutableMapping[str, Any]) -> None: 

548 """Fill a legacy mutable mapping (e.g `lsst.daf.base.PropertySet`) 

549 with metadata suitable for an `lsst.afw.image.Exposure` representation 

550 of this object. 

551 """ 

552 # We just dump all of the FITS headers and non-FITS metadata into the 

553 # legacy metadata component, to make sure we have everything. We dump 

554 # the latter into a pair of special cards to be able to full round-trip 

555 # them (including case preservation). 

556 if self.unit is not None: 

557 try: 

558 legacy_metadata["BUNIT"] = self.unit.to_string(format="fits") 

559 except ValueError: 

560 # Write units that astropy doesn't think FITS will accept 

561 # anyway; FITS standard says "SHOULD" about using its 

562 # recommended units, and coloring outside the lines is better 

563 # than lying. 

564 legacy_metadata["BUNIT"] = self.unit.to_string() 

565 if isinstance(self._opaque_metadata, fits.FitsOpaqueMetadata): 

566 legacy_metadata.update(self._opaque_metadata.headers[fits.ExtensionKey()]) 

567 for n, (k, v) in enumerate(self.metadata.items()): 

568 legacy_metadata[f"LSST IMAGES KEY {n + 1}"] = k 

569 legacy_metadata[f"LSST IMAGES VALUE {n + 1}"] = v 

570 

571 

572class MaskedImageSerializationModel[P: pydantic.BaseModel](ArchiveTree): 

573 """A Pydantic model used to represent a serialized `MaskedImage`.""" 

574 

575 SCHEMA_NAME: ClassVar[str] = "masked_image" 

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

577 MIN_READ_VERSION: ClassVar[int] = 1 

578 PUBLIC_TYPE: ClassVar[type] = MaskedImage 

579 

580 image: ImageSerializationModel[P] = pydantic.Field(description="The main data image.") 

581 mask: MaskSerializationModel[P] = pydantic.Field( 

582 description="Bitmask that annotates the main image's pixels." 

583 ) 

584 variance: ImageSerializationModel[P] = pydantic.Field( 

585 description="Per-pixel variance estimates for the main image." 

586 ) 

587 sky_projection: SkyProjectionSerializationModel[P] | None = pydantic.Field( 

588 default=None, 

589 exclude_if=is_none, 

590 description="Projection that maps the pixel grid to the sky.", 

591 ) 

592 

593 @property 

594 def bbox(self) -> Box: 

595 """The bounding box of the image.""" 

596 return self.image.bbox 

597 

598 def deserialize( 

599 self, archive: InputArchive[Any], *, bbox: Box | None = None, **kwargs: Any 

600 ) -> MaskedImage: 

601 """Deserialize an image from an input archive. 

602 

603 Parameters 

604 ---------- 

605 archive 

606 Archive to read from. 

607 bbox 

608 Bounding box of a subimage to read instead. 

609 **kwargs 

610 Unsupported keyword arguments are accepted only to provide better 

611 error messages (raising `serialization.InvalidParameterError`). 

612 """ 

613 if kwargs: 613 ↛ 614line 613 didn't jump to line 614 because the condition on line 613 was never true

614 raise InvalidParameterError(f"Unrecognized parameters for MaskedImage: {set(kwargs.keys())}.") 

615 image = self.image.deserialize(archive, bbox=bbox) 

616 mask = self.mask.deserialize(archive, bbox=bbox) 

617 variance = self.variance.deserialize(archive, bbox=bbox) 

618 sky_projection = self.sky_projection.deserialize(archive) if self.sky_projection is not None else None 

619 return MaskedImage( 

620 image, mask=mask, variance=variance, sky_projection=sky_projection 

621 )._finish_deserialize(self) 

622 

623 def deserialize_component(self, component: str, archive: InputArchive[Any], **kwargs: Any) -> Any: 

624 if component == "bbox" and kwargs: 624 ↛ 625line 624 didn't jump to line 625 because the condition on line 624 was never true

625 raise InvalidParameterError( 

626 f"Unrecognized parameters for MaskedImage.bbox: {set(kwargs.keys())}." 

627 ) 

628 return super().deserialize_component(component, archive, **kwargs)