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

210 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-09-30 11:30 +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 

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 # A caller that passes opaque_metadata has already extracted the 

505 # native metadata from the primary header itself. 

506 native_metadata: dict[str, Any] | None = None 

507 if opaque_metadata is None: 

508 opaque_metadata = fits.FitsOpaqueMetadata() 

509 native_metadata = opaque_metadata.extract_legacy_primary_header(hdu_list[0].header) 

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

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

512 result: Any 

513 with ExitStack() as exit_stack: 

514 if preserve_quantization: 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true

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

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

517 bintable_hdu_list = exit_stack.enter_context( 

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

519 ) 

520 image_bintable_hdu = bintable_hdu_list[1] 

521 variance_bintable_hdu = bintable_hdu_list[3] 

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

523 image = Image._read_legacy_hdu( 

524 hdu_list[1], 

525 opaque_metadata, 

526 preserve_bintable=image_bintable_hdu, 

527 fits_wcs_frame=fits_wcs_frame, 

528 ) 

529 if component == "image": 

530 result = image 

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

532 mask = Mask._read_legacy_hdu( 

533 hdu_list[2], 

534 opaque_metadata, 

535 plane_map=plane_map, 

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

537 ) 

538 if component == "mask": 

539 result = mask 

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

541 variance = Image._read_legacy_hdu( 

542 hdu_list[3], 

543 opaque_metadata, 

544 preserve_bintable=variance_bintable_hdu, 

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

546 ) 

547 if component == "variance": 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true

548 result = variance 

549 if component is None: 

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

551 result._opaque_metadata = opaque_metadata 

552 if native_metadata is not None: 

553 result._metadata = native_metadata 

554 return result 

555 

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

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

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

559 """ 

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

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

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

563 # them (including case preservation). 

564 if self.unit is not None: 

565 try: 

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

567 except ValueError: 

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

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

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

571 # than lying. 

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

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

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

575 # all at once. 

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

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

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

579 # Skip blanks 

580 continue 

581 if card.keyword not in grouped: 

582 grouped[card.keyword] = [] 

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

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

585 legacy_metadata[keyword] = values 

586 for n, (k, v) in enumerate(self._metadata.items()): 

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

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

589 

590 

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

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

593 

594 SCHEMA_NAME: ClassVar[str] = "masked_image" 

595 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

596 MIN_READ_VERSION: ClassVar[int] = 1 

597 PUBLIC_TYPE: ClassVar[type] = MaskedImage 

598 

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

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

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

602 ) 

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

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

605 ) 

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

607 default=None, 

608 exclude_if=is_none, 

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

610 ) 

611 

612 @property 

613 def bbox(self) -> Box: 

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

615 return self.image.bbox 

616 

617 def deserialize( 

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

619 ) -> MaskedImage: 

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

621 

622 Parameters 

623 ---------- 

624 archive 

625 Archive to read from. 

626 bbox 

627 Bounding box of a subimage to read instead. 

628 **kwargs 

629 Unsupported keyword arguments are accepted only to provide better 

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

631 """ 

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

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

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

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

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

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

638 return MaskedImage( 

639 image, mask=mask, variance=variance, sky_projection=sky_projection 

640 )._finish_deserialize(self) 

641 

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

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

644 raise InvalidParameterError( 

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

646 ) 

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