Coverage for python / lsst / images / _mask.py: 24%

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__ = ( 

15 "Mask", 

16 "MaskPlane", 

17 "MaskPlaneBit", 

18 "MaskSchema", 

19 "MaskSerializationModel", 

20 "get_legacy_visit_image_mask_planes", 

21) 

22 

23import dataclasses 

24import math 

25from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence, Set 

26from types import EllipsisType 

27from typing import Any, ClassVar, cast 

28 

29import astropy.io.fits 

30import astropy.wcs 

31import numpy as np 

32import numpy.typing as npt 

33import pydantic 

34from astro_metadata_translator import ObservationInfo 

35 

36from lsst.resources import ResourcePath, ResourcePathExpression 

37 

38from . import fits 

39from ._generalized_image import GeneralizedImage 

40from ._geom import YX, Box 

41from ._transforms import Frame, Projection, ProjectionSerializationModel 

42from .serialization import ( 

43 ArchiveReadError, 

44 ArchiveTree, 

45 ArrayReferenceModel, 

46 InputArchive, 

47 IntegerType, 

48 MetadataValue, 

49 NumberType, 

50 OutputArchive, 

51 is_integer, 

52 no_header_updates, 

53) 

54from .utils import is_none 

55 

56 

57@dataclasses.dataclass(frozen=True) 

58class MaskPlane: 

59 """Name and description of a single plane in a mask array.""" 

60 

61 name: str 

62 """Unique name for the mask plane (`str`).""" 

63 

64 description: str 

65 """Human-readable documentation for the mask plane (`str`).""" 

66 

67 @classmethod 

68 def read_legacy(cls, header: astropy.io.fits.Header) -> dict[str, int]: 

69 """Read mask plane descriptions written by 

70 `lsst.afw.image.Mask.writeFits`. 

71 

72 Parameters 

73 ---------- 

74 header 

75 FITS header. 

76 

77 Returns 

78 ------- 

79 `dict` [`str`, `int`] 

80 A dictionary mapping mask plane name to integer bit index. 

81 """ 

82 result: dict[str, int] = {} 

83 for card in list(header.cards): 

84 if card.keyword.startswith("MP_"): 

85 result[card.keyword.removeprefix("MP_")] = card.value 

86 del header[card.keyword] 

87 return result 

88 

89 

90@dataclasses.dataclass(frozen=True) 

91class MaskPlaneBit: 

92 """The nested array index and mask value associated with a single mask 

93 plane. 

94 """ 

95 

96 index: int 

97 """Index into the last dimension of the mask array where this plane's bit 

98 is stored. 

99 """ 

100 

101 mask: np.integer 

102 """Bitmask that selects just this plane's bit from a mask array value 

103 (`numpy.integer`). 

104 """ 

105 

106 @classmethod 

107 def compute(cls, overall_index: int, stride: int, mask_type: type[np.integer]) -> MaskPlaneBit: 

108 """Construct a `MaskPlaneBit` from the overall index of a plane in a 

109 `MaskSchema` and the stride (number of bits per mask array element). 

110 """ 

111 index, bit = divmod(overall_index, stride) 

112 return cls(index, mask_type(1 << bit)) 

113 

114 

115class MaskSchema: 

116 """A schema for a bit-packed mask array. 

117 

118 Parameters 

119 ---------- 

120 planes 

121 Iterable of `MaskPlane` instances that define the schema. `None` 

122 values may be included to reserve bits for future use. 

123 dtype 

124 The numpy data type of the mask arrays that use this schema. 

125 

126 Notes 

127 ----- 

128 A `MaskSchema` is a collection of mask planes, which each correspond to a 

129 single bit in a mask array. Mask schemas are immutable and associated with 

130 a particular array data type, allowing them to safely precompute the index 

131 and bitmask for each plane. 

132 

133 `MaskSchema` indexing is by integer (the overall index of a plane in the 

134 schema). The `descriptions` attribute may be indexed by plane name to get 

135 the description for that plane, and the `bitmask` method can be used to 

136 obtain an array that can be used to select one or more planes by name in 

137 a mask array that uses this schema. 

138 

139 If no mask planes are provided, a `None` placeholder is automatically 

140 added. 

141 """ 

142 

143 def __init__(self, planes: Iterable[MaskPlane | None], dtype: npt.DTypeLike = np.uint8): 

144 self._planes: tuple[MaskPlane | None, ...] = tuple(planes) or (None,) 

145 self._dtype = cast(np.dtype[np.integer], np.dtype(dtype)) 

146 stride = self.bits_per_element(self._dtype) 

147 self._descriptions = {plane.name: plane.description for plane in self._planes if plane is not None} 

148 self._mask_size = math.ceil(len(self._planes) / stride) 

149 self._bits: dict[str, MaskPlaneBit] = { 

150 plane.name: MaskPlaneBit.compute(n, stride, self._dtype.type) 

151 for n, plane in enumerate(self._planes) 

152 if plane is not None 

153 } 

154 

155 @staticmethod 

156 def bits_per_element(dtype: npt.DTypeLike) -> int: 

157 """Return the number of mask bits per array element for the given 

158 data type. 

159 """ 

160 dtype = np.dtype(dtype) 

161 match dtype.kind: 

162 case "u": 

163 return dtype.itemsize * 8 

164 case "i": 

165 return dtype.itemsize * 8 - 1 

166 case _: 

167 raise TypeError(f"dtype for masks must be an integer; got {dtype} with kind={dtype.kind}.") 

168 

169 def __iter__(self) -> Iterator[MaskPlane | None]: 

170 return iter(self._planes) 

171 

172 def __len__(self) -> int: 

173 return len(self._planes) 

174 

175 def __getitem__(self, i: int) -> MaskPlane | None: 

176 return self._planes[i] 

177 

178 def __repr__(self) -> str: 

179 return f"MaskSchema({list(self._planes)}, dtype={self._dtype!r})" 

180 

181 def __str__(self) -> str: 

182 return "\n".join( 

183 [ 

184 f"{name} [{bit.index}@{hex(bit.mask)}]: {self._descriptions[name]}" 

185 for name, bit in self._bits.items() 

186 ] 

187 ) 

188 

189 def __eq__(self, other: object) -> bool: 

190 if isinstance(other, MaskSchema): 

191 return self._planes == other._planes and self._dtype == other._dtype 

192 return False 

193 

194 @property 

195 def dtype(self) -> np.dtype: 

196 """The numpy data type of the mask arrays that use this schema.""" 

197 return self._dtype 

198 

199 @property 

200 def mask_size(self) -> int: 

201 """The number of elements in the last dimension of any mask array that 

202 uses this schema. 

203 """ 

204 return self._mask_size 

205 

206 @property 

207 def names(self) -> Set[str]: 

208 """The names of the mask planes, in bit order.""" 

209 return self._bits.keys() 

210 

211 @property 

212 def descriptions(self) -> Mapping[str, str]: 

213 """A mapping from plane name to description.""" 

214 return self._descriptions 

215 

216 def bit(self, plane: str) -> MaskPlaneBit: 

217 """Return the last array index and mask for the given mask plane.""" 

218 return self._bits[plane] 

219 

220 def bitmask(self, *planes: str) -> np.ndarray: 

221 """Return a 1-d mask array that represents the union (i.e. bitwise OR) 

222 of the planes with the given names. 

223 

224 Parameters 

225 ---------- 

226 *planes 

227 Mask plane names. 

228 

229 Returns 

230 ------- 

231 numpy.ndarray 

232 A 1-d array with shape ``(mask_size,)``. 

233 """ 

234 result = np.zeros(self.mask_size, dtype=self._dtype) 

235 for plane in planes: 

236 bit = self._bits[plane] 

237 result[bit.index] |= bit.mask 

238 return result 

239 

240 def split(self, dtype: npt.DTypeLike) -> list[MaskSchema]: 

241 """Split the schema into an equivalent series of schemas that each 

242 have a `mask_size` of ``1``, dropping all `None` placeholders. 

243 

244 Parameters 

245 ---------- 

246 dtype 

247 Data type of the new mask pixels. 

248 

249 Returns 

250 ------- 

251 `list` [`MaskSchema`] 

252 A list of mask schemas that together include all planes in 

253 ``self`` and have `mask_size` equal to ``1``. If there are no 

254 mask planes (only `None` placeholders) in ``self``, a single mask 

255 schema with a `None` placeholder is returned; otherwise `None` 

256 placeholders are returned. 

257 """ 

258 dtype = np.dtype(dtype) 

259 planes: list[MaskPlane] = [] 

260 schemas: list[MaskSchema] = [] 

261 n_planes_per_schema = self.bits_per_element(dtype) 

262 for plane in self._planes: 

263 if plane is not None: 

264 planes.append(plane) 

265 if len(planes) == n_planes_per_schema: 

266 schemas.append(MaskSchema(planes, dtype=dtype)) 

267 planes.clear() 

268 if planes: 

269 schemas.append(MaskSchema(planes, dtype=dtype)) 

270 if not schemas: 

271 schemas.append(MaskSchema([None], dtype=dtype)) 

272 return schemas 

273 

274 def update_header(self, header: astropy.io.fits.Header) -> None: 

275 """Add a description of this mask schema to a FITS header.""" 

276 for n, plane in enumerate(self): 

277 if plane is not None: 

278 bit = self.bit(plane.name) 

279 if bit.index != 0: 

280 raise TypeError("Only mask schemas with mask_size==1 can be described in FITS.") 

281 header.set(f"MSKN{n + 1:04d}", plane.name, f"Name for mask plane {n + 1}.") 

282 header.set(f"MSKM{n + 1:04d}", bit.mask, f"Bitmask for plane n={n + 1}; always 1<<(n-1).") 

283 # We don't add a comment to the description card, because it's 

284 # likely to overrun a single card and get the CONTINUE 

285 # treatment . That will cause Astropy to warn about the comment 

286 # being truncated and that's worse than just leaving it 

287 # unexplained; it's pretty obvious from context what it is. 

288 header.set(f"MSKD{n + 1:04d}", plane.description) 

289 

290 def strip_header(self, header: astropy.io.fits.Header) -> None: 

291 """Remove all header cards added by `update_header`.""" 

292 for n, plane in enumerate(self): 

293 if plane is not None: 

294 header.remove(f"MSKN{n + 1:04d}", ignore_missing=True) 

295 header.remove(f"MSKM{n + 1:04d}", ignore_missing=True) 

296 header.remove(f"MSKD{n + 1:04d}", ignore_missing=True) 

297 

298 

299class Mask(GeneralizedImage): 

300 """A 2-d bitmask image backed by a 3-d byte array. 

301 

302 Parameters 

303 ---------- 

304 array_or_fill 

305 Array or fill value for the mask. If a fill value, ``bbox`` or 

306 ``shape`` must be provided. 

307 schema 

308 Schema that defines the planes and their bit assignments. 

309 bbox 

310 Bounding box for the mask. This sets the shape of the first two 

311 dimensions of the array. 

312 start 

313 Logical coordinates of the first pixel in the array, ordered ``y``, 

314 ``x`` (unless an `XY` instance is passed). Ignored if 

315 ``bbox`` is provided. Defaults to zeros. 

316 shape 

317 Leading dimensions of the array, ordered ``y``, ``x`` (unless an `XY` 

318 instance is passed). Only needed if ``array_or_fill`` is not an 

319 array and ``bbox`` is not provided. Like the bbox, this does not 

320 include the last dimension of the array. 

321 projection 

322 Projection that maps the pixel grid to the sky. 

323 obs_info 

324 General information about the associated observation in standardized 

325 form. 

326 metadata 

327 Arbitrary flexible metadata to associate with the mask. 

328 

329 Notes 

330 ----- 

331 Indexing the `array` attribute of a `Mask` does not take into account its 

332 ``start`` offset, but accessing a subimage mask by indexing a `Mask` with 

333 a `Box` does, and the `bbox` of the subimage is set to match its location 

334 within the original mask. 

335 

336 A mask's ``bbox`` corresponds to the leading dimensions of its backing 

337 `numpy.ndarray`, while the last dimension's size is always equal to the 

338 `~MaskSchema.mask_size` of its schema, since a schema can in general 

339 require multiple array elements to represent all of its planes. 

340 """ 

341 

342 def __init__( 

343 self, 

344 array_or_fill: np.ndarray | int = 0, 

345 /, 

346 *, 

347 schema: MaskSchema, 

348 bbox: Box | None = None, 

349 start: Sequence[int] | None = None, 

350 shape: Sequence[int] | None = None, 

351 projection: Projection | None = None, 

352 obs_info: ObservationInfo | None = None, 

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

354 ): 

355 super().__init__(metadata) 

356 if shape is not None: 

357 shape = tuple(shape) 

358 if start is not None: 

359 start = tuple(start) 

360 if isinstance(array_or_fill, np.ndarray): 

361 array = np.array(array_or_fill, dtype=schema.dtype) 

362 if array.ndim != 3: 

363 raise ValueError("Mask array must be 3-d.") 

364 if bbox is None: 

365 bbox = Box.from_shape(array.shape[:-1], start=start) 

366 elif bbox.shape + (schema.mask_size,) != array.shape: 

367 raise ValueError( 

368 f"Explicit bbox shape {bbox.shape} and schema of size {schema.mask_size} do not " 

369 f"match array with shape {array.shape}." 

370 ) 

371 if shape is not None and shape + (schema.mask_size,) != array.shape: 

372 raise ValueError( 

373 f"Explicit shape {shape} and schema of size {schema.mask_size} do " 

374 f"not match array with shape {array.shape}." 

375 ) 

376 

377 else: 

378 if bbox is None: 

379 if shape is None: 

380 raise TypeError("No bbox, size, or array provided.") 

381 bbox = Box.from_shape(shape, start=start) 

382 array = np.full(bbox.shape + (schema.mask_size,), array_or_fill, dtype=schema.dtype) 

383 self._array = array 

384 self._bbox: Box = bbox 

385 self._schema: MaskSchema = schema 

386 self._projection = projection 

387 self._obs_info = obs_info 

388 

389 @property 

390 def array(self) -> np.ndarray: 

391 """The low-level array (`numpy.ndarray`). 

392 

393 Assigning to this attribute modifies the existing array in place; the 

394 bounding box and underlying data pointer are never changed. 

395 """ 

396 return self._array 

397 

398 @array.setter 

399 def array(self, value: np.ndarray | int) -> None: 

400 self._array[:, :] = value 

401 

402 @property 

403 def schema(self) -> MaskSchema: 

404 """Schema that defines the planes and their bit assignments 

405 (`MaskSchema`). 

406 """ 

407 return self._schema 

408 

409 @property 

410 def bbox(self) -> Box: 

411 """2-d bounding box of the mask (`Box`). 

412 

413 This sets the shape of the first two dimensions of the array. 

414 """ 

415 return self._bbox 

416 

417 @property 

418 def projection(self) -> Projection[Any] | None: 

419 """The projection that maps this mask's pixel grid to the sky 

420 (`Projection` | `None`). 

421 

422 Notes 

423 ----- 

424 The pixel coordinates used by this projection account for the bounding 

425 box ``start``; they are not just array indices. 

426 """ 

427 return self._projection 

428 

429 @property 

430 def obs_info(self) -> ObservationInfo | None: 

431 """General information about the associated observation in standard 

432 form. (`~astro_metadata_translator.ObservationInfo` | `None`). 

433 """ 

434 return self._obs_info 

435 

436 def __getitem__(self, bbox: Box | EllipsisType) -> Mask: 

437 if bbox is ...: 

438 return self 

439 super().__getitem__(bbox) 

440 return self._transfer_metadata( 

441 Mask( 

442 self.array[bbox.y.slice_within(self._bbox.y), bbox.x.slice_within(self._bbox.x), :], 

443 bbox=bbox, 

444 schema=self.schema, 

445 ), 

446 bbox=bbox, 

447 ) 

448 

449 def __setitem__(self, bbox: Box | EllipsisType, value: Mask) -> None: 

450 subview = self[bbox] 

451 subview.clear() 

452 subview.update(value) 

453 

454 def __str__(self) -> str: 

455 return f"Mask({self.bbox!s}, {list(self.schema.names)})" 

456 

457 def __repr__(self) -> str: 

458 return f"Mask(..., bbox={self.bbox!r}, schema={self.schema!r})" 

459 

460 def __eq__(self, other: object) -> bool: 

461 if not isinstance(other, Mask): 

462 return NotImplemented 

463 return ( 

464 self._bbox == other._bbox 

465 and self._schema == other._schema 

466 and np.array_equal(self._array, other._array, equal_nan=True) 

467 ) 

468 

469 def copy(self) -> Mask: 

470 """Deep-copy the mask and metadata.""" 

471 return self._transfer_metadata( 

472 Mask( 

473 self._array.copy(), 

474 bbox=self._bbox, 

475 schema=self._schema, 

476 projection=self._projection, 

477 obs_info=self._obs_info, 

478 ), 

479 copy=True, 

480 ) 

481 

482 def view( 

483 self, 

484 *, 

485 schema: MaskSchema | EllipsisType = ..., 

486 projection: Projection | None | EllipsisType = ..., 

487 start: Sequence[int] | EllipsisType = ..., 

488 obs_info: ObservationInfo | None | EllipsisType = ..., 

489 ) -> Mask: 

490 """Make a view of the mask, with optional updates. 

491 

492 Notes 

493 ----- 

494 This can only be used to make changes to schema descriptions; plane 

495 names must remain the same (in the same order). 

496 """ 

497 if schema is ...: 

498 schema = self._schema 

499 else: 

500 if list(schema.names) != list(self.schema.names): 

501 raise ValueError("Cannot create a mask view with a schema with different names.") 

502 if projection is ...: 

503 projection = self._projection 

504 if start is ...: 

505 start = self._bbox.start 

506 if obs_info is ...: 

507 obs_info = self._obs_info 

508 return self._transfer_metadata( 

509 Mask(self._array, start=start, schema=schema, projection=projection, obs_info=obs_info) 

510 ) 

511 

512 def update(self, other: Mask) -> None: 

513 """Update ``self`` to include all common mask values set in ``other``. 

514 

515 Notes 

516 ----- 

517 This only operates on the intersection of the two mask bounding boxes 

518 and the mask planes that are present in both. Mask bits are only set, 

519 not cleared (i.e. this uses ``|=`` updates, not ``=`` assignments). 

520 """ 

521 lhs = self 

522 rhs = other 

523 if other.bbox != self.bbox: 

524 if (bbox := self.bbox.intersection(other.bbox)) is None: 

525 return 

526 lhs = self[bbox] 

527 rhs = other[bbox] 

528 for name in self.schema.names & other.schema.names: 

529 lhs.set(name, rhs.get(name)) 

530 

531 def get(self, plane: str) -> np.ndarray: 

532 """Return a 2-d boolean array for the given mask plane. 

533 

534 Parameters 

535 ---------- 

536 plane 

537 Name of the mask plane. 

538 

539 Returns 

540 ------- 

541 numpy.ndarray 

542 A 2-d boolean array with the same shape as `bbox` that is `True` 

543 where the bit for ``plane`` is set and `False` elsewhere. 

544 """ 

545 bit = self.schema.bit(plane) 

546 return (self._array[..., bit.index] & bit.mask).astype(bool) 

547 

548 def set(self, plane: str, boolean_mask: np.ndarray | EllipsisType = ...) -> None: 

549 """Set a mask plane. 

550 

551 Parameters 

552 ---------- 

553 plane 

554 Name of the mask plane to set 

555 boolean_mask 

556 A 2-d boolean array with the same shape as `bbox` that is `True` 

557 where the bit for ``plane`` should be set and `False` where it 

558 should be left unchanged (*not* set to zero). May be ``...`` to 

559 set the bit everywhere. 

560 """ 

561 bit = self.schema.bit(plane) 

562 if boolean_mask is not ...: 

563 boolean_mask = boolean_mask.astype(bool) 

564 self._array[boolean_mask, bit.index] |= bit.mask 

565 

566 def clear(self, plane: str | None = None, boolean_mask: np.ndarray | EllipsisType = ...) -> None: 

567 """Clear one or more mask planes. 

568 

569 Parameters 

570 ---------- 

571 plane 

572 Name of the mask plane to set. If `None` all mask planes are 

573 cleared. 

574 boolean_mask 

575 A 2-d boolean array with the same shape as `bbox` that is `True` 

576 where the bit for ``plane`` should be cleared and `False` where it 

577 should be left unchanged. May be ``...`` to clear the bit 

578 everywhere. 

579 """ 

580 if boolean_mask is not ...: 

581 boolean_mask = boolean_mask.astype(bool) 

582 if plane is None: 

583 self._array[boolean_mask, :] = 0 

584 else: 

585 bit = self.schema.bit(plane) 

586 self._array[boolean_mask, bit.index] &= ~bit.mask 

587 

588 def serialize[P: pydantic.BaseModel]( 

589 self, 

590 archive: OutputArchive[P], 

591 *, 

592 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

593 save_projection: bool = True, 

594 add_offset_wcs: str | None = "A", 

595 ) -> MaskSerializationModel[P]: 

596 """Serialize the mask to an output archive. 

597 

598 Parameters 

599 ---------- 

600 archive 

601 Archive to write to. 

602 update_header 

603 A callback that will be given the FITS header for the HDU 

604 containing this mask in order to add keys to it. This callback 

605 may be provided but will not be called if the output format is not 

606 FITS. As multiple HDUs may be added, this function may be called 

607 multiple times. 

608 save_projection 

609 If `True`, save the `Projection` attached to the mask, if there 

610 is one. 

611 add_offset_wcs 

612 A FITS WCS single-character suffix to use when adding a linear 

613 WCS that maps the FITS array to the logical pixel coordinates 

614 defined by ``bbox.start``. Set to `None` to not write this WCS. 

615 """ 

616 data: list[ArrayReferenceModel] = [] 

617 for schema_2d in self.schema.split(np.int32): 

618 mask_2d = Mask( 

619 0, bbox=self.bbox, schema=schema_2d, projection=self._projection, obs_info=self._obs_info 

620 ) 

621 mask_2d.update(self) 

622 data.append(mask_2d._serialize_2d(archive, update_header=update_header)) 

623 serialized_projection: ProjectionSerializationModel[P] | None = None 

624 if save_projection and self.projection is not None: 

625 serialized_projection = archive.serialize_direct("projection", self.projection.serialize) 

626 serialized_dtype = NumberType.from_numpy(self.schema.dtype) 

627 assert is_integer(serialized_dtype), "Mask dtypes should always be integers." 

628 return MaskSerializationModel.model_construct( 

629 data=data, 

630 start=list(self.bbox.start), 

631 planes=list(self.schema), 

632 dtype=serialized_dtype, 

633 projection=serialized_projection, 

634 obs_info=self._obs_info, 

635 metadata=self.metadata, 

636 ) 

637 

638 def _serialize_2d[P: pydantic.BaseModel]( 

639 self, 

640 archive: OutputArchive[P], 

641 *, 

642 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

643 save_projection: bool = True, 

644 add_offset_wcs: str | None = "A", 

645 ) -> ArrayReferenceModel: 

646 def _update_header(header: astropy.io.fits.Header) -> None: 

647 update_header(header) 

648 self.schema.update_header(header) 

649 if self.projection is not None: 

650 if self.fits_wcs: 

651 header.update(self.fits_wcs.to_header(relax=True)) 

652 if add_offset_wcs is not None: 

653 fits.add_offset_wcs(header, x=self.bbox.x.start, y=self.bbox.y.start, key=add_offset_wcs) 

654 

655 assert self.array.shape[2] == 1, "Mask should be split before calling this method." 

656 return archive.add_array(self._array[:, :, 0], update_header=_update_header) 

657 

658 @classmethod 

659 def deserialize( 

660 cls, 

661 model: MaskSerializationModel[Any], 

662 archive: InputArchive[Any], 

663 *, 

664 bbox: Box | None = None, 

665 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

666 ) -> Mask: 

667 """Deserialize a mask from an input archive. 

668 

669 Parameters 

670 ---------- 

671 model 

672 A Pydantic model representation of the mask, holding references 

673 to data stored in the archive. 

674 archive 

675 Archive to read from. 

676 bbox 

677 Bounding box of a subimage to read instead. 

678 strip_header 

679 A callable that strips out any FITS header cards added by the 

680 ``update_header`` argument in the corresponding call to 

681 `serialize`. 

682 """ 

683 slices: tuple[slice, ...] | EllipsisType = ... 

684 if bbox is not None: 

685 slices = bbox.slice_within(model.bbox) 

686 else: 

687 bbox = model.bbox 

688 if not is_integer(model.dtype): 

689 raise ArchiveReadError(f"Mask array has a non-integer dtype: {model.dtype}.") 

690 schema = MaskSchema(model.planes, dtype=model.dtype.to_numpy()) 

691 projection = ( 

692 Projection.deserialize(model.projection, archive) if model.projection is not None else None 

693 ) 

694 result = Mask( 

695 0, 

696 schema=schema, 

697 bbox=bbox, 

698 projection=projection, 

699 obs_info=model.obs_info, 

700 ) 

701 schemas_2d = schema.split(np.int32) 

702 if len(schemas_2d) != len(model.data): 

703 raise ArchiveReadError( 

704 f"Number of mask arrays ({len(model.data)}) does not match expectation ({len(schemas_2d)})." 

705 ) 

706 for ref, schema_2d in zip(model.data, schemas_2d): 

707 mask_2d = cls._deserialize_2d( 

708 ref, schema_2d, bbox.start, archive, strip_header=strip_header, slices=slices 

709 ) 

710 result.update(mask_2d) 

711 return result._finish_deserialize(model) 

712 

713 @classmethod 

714 def _deserialize_2d( 

715 cls, 

716 ref: ArrayReferenceModel, 

717 schema_2d: MaskSchema, 

718 start: Sequence[int], 

719 archive: InputArchive[Any], 

720 *, 

721 slices: tuple[slice, ...] | EllipsisType = ..., 

722 strip_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

723 ) -> Mask: 

724 def _strip_header(header: astropy.io.fits.Header) -> None: 

725 strip_header(header) 

726 schema_2d.strip_header(header) 

727 fits.strip_wcs_cards(header) 

728 

729 array_2d = archive.get_array(ref, strip_header=_strip_header, slices=slices) 

730 return Mask(array_2d[:, :, np.newaxis], schema=schema_2d, start=start) 

731 

732 @staticmethod 

733 def _get_archive_tree_type[P: pydantic.BaseModel]( 

734 pointer_type: type[P], 

735 ) -> type[MaskSerializationModel[P]]: 

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

737 type that uses the given pointer type. 

738 """ 

739 return MaskSerializationModel[pointer_type] # type: ignore 

740 

741 _archive_default_name: ClassVar[str] = "mask" 

742 """The name this object should be serialized with when written as the 

743 top-level object. 

744 """ 

745 

746 def write_fits( 

747 self, 

748 filename: str, 

749 *, 

750 compression: fits.FitsCompressionOptions | None = fits.FitsCompressionOptions.DEFAULT, 

751 ) -> None: 

752 """Write the mask to a FITS file. 

753 

754 Parameters 

755 ---------- 

756 filename 

757 Name of the file to write to. Must be a local file. 

758 compression 

759 Compression options. 

760 """ 

761 compression_options = {} 

762 if compression is not fits.FitsCompressionOptions.DEFAULT: 

763 compression_options[self._archive_default_name] = compression 

764 fits.write(self, filename, compression_options) 

765 

766 @staticmethod 

767 def read_fits(url: ResourcePathExpression, *, bbox: Box | None = None) -> Mask: 

768 """Read an image from a FITS file. 

769 

770 Parameters 

771 ---------- 

772 url 

773 URL of the file to read; may be any type supported by 

774 `lsst.resources.ResourcePath`. 

775 bbox 

776 Bounding box of a subimage to read instead. 

777 """ 

778 return fits.read(Mask, url, bbox=bbox).deserialized 

779 

780 @staticmethod 

781 def from_legacy( 

782 legacy: Any, 

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

784 ) -> Mask: 

785 """Convert from an `lsst.afw.image.Mask` instance. 

786 

787 Parameters 

788 ---------- 

789 legacy 

790 An `lsst.afw.image.Mask` instance. This will not share pixel 

791 data with the new object. 

792 plane_map 

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

794 description. 

795 """ 

796 return Mask._from_legacy_array( 

797 legacy.array, 

798 legacy.getMaskPlaneDict(), 

799 start=YX(y=legacy.getY0(), x=legacy.getX0()), 

800 plane_map=plane_map, 

801 ) 

802 

803 def to_legacy(self, plane_map: Mapping[str, MaskPlane] | None = None) -> Any: 

804 """Convert to an `lsst.afw.image.Mask` instance. 

805 

806 The pixel data will not be shared between the two objects. 

807 

808 Parameters 

809 ---------- 

810 plane_map 

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

812 description. 

813 """ 

814 import lsst.afw.image 

815 import lsst.geom 

816 

817 result = lsst.afw.image.Mask(self.bbox.to_legacy()) 

818 if plane_map is None: 

819 plane_map = {plane.name: plane for plane in self.schema if plane is not None} 

820 for old_name, new_plane in plane_map.items(): 

821 old_bit = result.addMaskPlane(old_name) 

822 old_bitmask = 1 << old_bit 

823 result.array[self.get(new_plane.name)] |= old_bitmask 

824 return result 

825 

826 @staticmethod 

827 def _from_legacy_array( 

828 array2d: np.ndarray, 

829 old_planes: Mapping[str, int],