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

207 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-09-26 02:37 -0700

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12from __future__ import annotations 

13 

14__all__ = ("MaskedImage", "MaskedImageSerializationModel") 

15 

16import functools 

17from collections.abc import Mapping 

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 try: 

52 from lsst.daf.base import PropertyList 

53 except ImportError: 

54 type PropertyList = Any # type: ignore[no-redef] 

55 

56 

57class MaskedImage(GeneralizedImage): 

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

59 

60 Parameters 

61 ---------- 

62 image 

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

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

65 mask 

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

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

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

69 variance 

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

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

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

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

74 (possibly by `None`). 

75 mask_schema 

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

77 not provided. 

78 sky_projection 

79 Projection that maps the pixel grid to the sky. 

80 metadata 

81 Arbitrary flexible metadata to associate with the image. 

82 """ 

83 

84 def __init__( 

85 self, 

86 image: Image, 

87 *, 

88 mask: Mask | None = None, 

89 variance: Image | None = None, 

90 mask_schema: MaskSchema | None = None, 

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

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

93 ) -> None: 

94 super().__init__(metadata) 

95 if sky_projection is None: 

96 sky_projection = image.sky_projection 

97 else: 

98 image = image.view(sky_projection=sky_projection) 

99 if mask is None: 

100 if mask_schema is None: 

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

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

103 elif mask_schema is not None: 

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

105 else: 

106 if image.bbox != mask.bbox: 

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

108 mask = mask.view(sky_projection=sky_projection) 

109 if variance is None: 

110 variance = Image( 

111 1.0, 

112 dtype=np.float32, 

113 bbox=image.bbox, 

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

115 sky_projection=sky_projection, 

116 ) 

117 else: 

118 if image.bbox != variance.bbox: 

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

120 variance = variance.view(sky_projection=sky_projection) 

121 if image.unit is None: 

122 if variance.unit is not None: 

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

124 elif variance.unit is None: 

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

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

127 raise ValueError( 

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

129 ) 

130 self._image = image 

131 self._mask = mask 

132 self._variance = variance 

133 

134 @property 

135 def image(self) -> Image: 

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

137 return self._image 

138 

139 @property 

140 def mask(self) -> Mask: 

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

142 

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

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

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

146 with the image's. 

147 """ 

148 return self._mask 

149 

150 @mask.setter 

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

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

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

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

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

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

157 # are the same projections. 

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

159 

160 @property 

161 def variance(self) -> Image: 

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

163 return self._variance 

164 

165 @property 

166 def bbox(self) -> Box: 

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

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

169 """ 

170 return self._image.bbox 

171 

172 @property 

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

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

175 return self._image.unit 

176 

177 @property 

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

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

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

181 """ 

182 return self._image.sky_projection 

183 

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

185 bbox, _ = self._handle_getitem_args(bbox) 

186 return self._transfer_metadata( 

187 MaskedImage( 

188 # Projection and obs_info propagate from the image. 

189 self.image[bbox], 

190 mask=self.mask[bbox], 

191 variance=self.variance[bbox], 

192 ), 

193 bbox=bbox, 

194 ) 

195 

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

197 self._image[bbox] = value.image 

198 self._mask[bbox] = value.mask 

199 self._variance[bbox] = value.variance 

200 

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

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

203 

204 Parameters 

205 ---------- 

206 options : `DescribeOptions`, optional 

207 Rendering options; forwarded to all children. 

208 """ 

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

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

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

212 fields = [ 

213 ReportField( 

214 label="image", 

215 value=self.image, 

216 repr_value=repr(self.image), 

217 positional=True, 

218 role=FieldRole.REPR_ONLY, 

219 ), 

220 ReportField( 

221 label="mask_schema", 

222 value=self.mask.schema, 

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

224 role=FieldRole.REPR_ONLY, 

225 ), 

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

227 ] 

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

229 if options.brief: 

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

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

232 children = { 

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

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

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

236 } 

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

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

239 return Report( 

240 type_name="MaskedImage", 

241 summary=summary, 

242 fields=fields, 

243 children=children, 

244 ) 

245 

246 def copy(self) -> MaskedImage: 

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

248 return self._transfer_metadata( 

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

250 copy=True, 

251 ) 

252 

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

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

255 

256 Parameters 

257 ---------- 

258 archive 

259 Archive to write to. 

260 """ 

261 return self._serialize_impl(MaskedImageSerializationModel, archive) 

262 

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

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

265 ) -> M: 

266 serialized_image = archive.serialize_direct( 

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

268 ) 

269 serialized_mask = archive.serialize_direct( 

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

271 ) 

272 serialized_variance = archive.serialize_direct( 

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

274 ) 

275 serialized_projection = ( 

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

277 if self.sky_projection is not None 

278 else None 

279 ) 

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

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

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

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

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

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

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

287 # makes it incorrectly unhappy about the return type. 

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

289 image=serialized_image, 

290 mask=serialized_mask, 

291 variance=serialized_variance, 

292 sky_projection=serialized_projection, 

293 metadata=self.metadata, 

294 ) 

295 

296 @staticmethod 

297 def _get_archive_tree_type[P: pydantic.BaseModel]( 

298 pointer_type: type[P], 

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

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

301 type that uses the given pointer type. 

302 """ 

303 return MaskedImageSerializationModel[pointer_type] # type: ignore 

304 

305 @staticmethod 

306 def from_legacy( 

307 legacy: LegacyMaskedImage, 

308 *, 

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

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

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

312 ) -> MaskedImage: 

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

314 

315 Parameters 

316 ---------- 

317 legacy 

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

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

320 unit 

321 Units of the image. 

322 plane_map 

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

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

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

326 mask actually has set. 

327 sky_projection 

328 Projection from pixels to xky. 

329 """ 

330 return MaskedImage( 

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

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

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

334 ) 

335 

336 def to_legacy( 

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

338 ) -> LegacyMaskedImage: 

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

340 

341 Parameters 

342 ---------- 

343 copy 

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

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

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

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

348 plane_map 

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

350 description. 

351 """ 

352 import lsst.afw.image 

353 

354 return lsst.afw.image.MaskedImage( 

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

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

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

358 dtype=self.image.array.dtype, 

359 ) 

360 

361 @classmethod 

362 def from_hdu_list( 

363 cls, 

364 hdu_list: astropy.io.fits.HDUList, 

365 *, 

366 fits_wcs_frame: Frame | None = DEFAULT_PIXEL_FRAME, 

367 ) -> MaskedImage: 

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

369 ``lsst.images`` FITS HDU list. 

370 

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

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

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

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

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

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

377 

378 Parameters 

379 ---------- 

380 hdu_list 

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

382 a primary HDU. 

383 fits_wcs_frame 

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

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

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

387 projection. 

388 

389 Returns 

390 ------- 

391 `~lsst.images.MaskedImage` 

392 The reconstructed masked image. 

393 

394 Raises 

395 ------ 

396 ValueError 

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

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

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

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

401 here and would otherwise be silently dropped). 

402 

403 Notes 

404 ----- 

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

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

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

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

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

410 `read_legacy`. 

411 

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

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

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

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

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

417 

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

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

420 headers. 

421 """ 

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

423 if n_mask_hdus > 1: 

424 raise ValueError( 

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

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

427 ) 

428 mask_hdu = hdu_list["MASK"] 

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

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

431 opaque_metadata = fits.FitsOpaqueMetadata() 

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

433 image = Image._read_legacy_hdu( 

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

435 ) 

436 mask = Mask._read_legacy_hdu( 

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

438 ) 

439 variance = Image._read_legacy_hdu( 

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

441 ) 

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

443 result._opaque_metadata = opaque_metadata 

444 return result 

445 

446 @staticmethod 

447 def read_legacy( 

448 uri: ResourcePathExpression, 

449 *, 

450 preserve_quantization: bool = False, 

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

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

453 fits_wcs_frame: Frame | None = None, 

454 ) -> Any: 

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

456 

457 Parameters 

458 ---------- 

459 uri 

460 URI or file name. 

461 preserve_quantization 

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

463 exactly preserve quantization-compressed pixel values. This causes 

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

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

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

467 pixel values are not transferred to the copy. 

468 plane_map 

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

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

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

472 mask actually has set. 

473 component 

474 A component to read instead of the full image. 

475 fits_wcs_frame 

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

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

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

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

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

481 """ 

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

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

484 return MaskedImage._read_legacy_hdus( 

485 hdu_list, 

486 uri, 

487 preserve_quantization=preserve_quantization, 

488 plane_map=plane_map, 

489 component=component, 

490 fits_wcs_frame=fits_wcs_frame, 

491 ) 

492 

493 @staticmethod 

494 def _read_legacy_hdus( 

495 hdu_list: astropy.io.fits.HDUList, 

496 uri: ResourcePathExpression, 

497 *, 

498 opaque_metadata: fits.FitsOpaqueMetadata | None = None, 

499 preserve_quantization: bool = False, 

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

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

502 fits_wcs_frame: Frame | None = None, 

503 ) -> Any: 

504 if opaque_metadata is None: 

505 opaque_metadata = fits.FitsOpaqueMetadata() 

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

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

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

509 result: Any 

510 with ExitStack() as exit_stack: 

511 if preserve_quantization: 

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

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

514 bintable_hdu_list = exit_stack.enter_context( 

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

516 ) 

517 image_bintable_hdu = bintable_hdu_list[1] 

518 variance_bintable_hdu = bintable_hdu_list[3] 

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

520 image = Image._read_legacy_hdu( 

521 hdu_list[1], 

522 opaque_metadata, 

523 preserve_bintable=image_bintable_hdu, 

524 fits_wcs_frame=fits_wcs_frame, 

525 ) 

526 if component == "image": 

527 result = image 

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

529 mask = Mask._read_legacy_hdu( 

530 hdu_list[2], 

531 opaque_metadata, 

532 plane_map=plane_map, 

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

534 ) 

535 if component == "mask": 

536 result = mask 

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

538 variance = Image._read_legacy_hdu( 

539 hdu_list[3], 

540 opaque_metadata, 

541 preserve_bintable=variance_bintable_hdu, 

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

543 ) 

544 if component == "variance": 

545 result = variance 

546 if component is None: 

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

548 result._opaque_metadata = opaque_metadata 

549 return result 

550 

551 def _fill_legacy_metadata(self, legacy_metadata: PropertyList) -> None: 

552 """Fill a legacy PropertyList with metadata suitable for an 

553 `lsst.afw.image.Exposure` representation of this object. 

554 """ 

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

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

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

558 # them (including case preservation). 

559 if self.unit is not None: 559 ↛ 568line 559 didn't jump to line 568 because the condition on line 559 was always true

560 try: 

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

562 except ValueError: 

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

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

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

566 # than lying. 

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

568 if isinstance(self._opaque_metadata, fits.FitsOpaqueMetadata): 568 ↛ 581line 568 didn't jump to line 581 because the condition on line 568 was always true

569 # Group all cards with the same keyword into a list, and write them 

570 # all at once. 

571 grouped: dict[str, list[Any]] = {} 

572 for card in self._opaque_metadata.headers[fits.ExtensionKey()].cards: 

573 if not card.keyword: 573 ↛ 575line 573 didn't jump to line 575 because the condition on line 573 was never true

574 # Skip blanks 

575 continue 

576 if card.keyword not in grouped: 

577 grouped[card.keyword] = [] 

578 grouped[card.keyword].append(card.value) 

579 for keyword, values in grouped.items(): 

580 legacy_metadata[keyword] = values 

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

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

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

584 

585 

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

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

588 

589 SCHEMA_NAME: ClassVar[str] = "masked_image" 

590 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

591 MIN_READ_VERSION: ClassVar[int] = 1 

592 PUBLIC_TYPE: ClassVar[type] = MaskedImage 

593 

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

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

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

597 ) 

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

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

600 ) 

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

602 default=None, 

603 exclude_if=is_none, 

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

605 ) 

606 

607 @property 

608 def bbox(self) -> Box: 

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

610 return self.image.bbox 

611 

612 def deserialize( 

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

614 ) -> MaskedImage: 

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

616 

617 Parameters 

618 ---------- 

619 archive 

620 Archive to read from. 

621 bbox 

622 Bounding box of a subimage to read instead. 

623 **kwargs 

624 Unsupported keyword arguments are accepted only to provide better 

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

626 """ 

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

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

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

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

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

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

633 return MaskedImage( 

634 image, mask=mask, variance=variance, sky_projection=sky_projection 

635 )._finish_deserialize(self) 

636 

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

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

639 raise InvalidParameterError( 

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

641 ) 

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