Coverage for tests/test_mask.py: 96%

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 

14import dataclasses 

15import os 

16from typing import Any 

17 

18import astropy.io.fits 

19import numpy as np 

20import pytest 

21 

22import lsst.utils.tests 

23from lsst.images import ( 

24 Box, 

25 Mask, 

26 MaskPlane, 

27 MaskSchema, 

28 get_legacy_difference_image_mask_planes, 

29 get_legacy_non_cell_coadd_mask_planes, 

30 get_legacy_optional_mask_planes, 

31 get_legacy_template_mask_planes, 

32 get_legacy_visit_image_mask_planes, 

33) 

34from lsst.images._mask import _guess_legacy_plane_map 

35from lsst.images.describe import Report 

36from lsst.images.tests import ( # noqa: F401 

37 RoundtripFits, 

38 assert_masks_equal, 

39 compare_mask_to_legacy, 

40 reset_afw_mask_planes, 

41) 

42 

43try: 

44 from lsst.afw.image import MaskedImageReader as LegacyMaskedImageReader 

45 

46except ImportError: 

47 type LegacyMaskedImageReader = Any # type: ignore[no-redef] 

48 

49EXTERNAL_DATA_DIR = os.environ.get("TESTDATA_IMAGES_DIR", None) 

50 

51 

52@dataclasses.dataclass 

53class _LegacyTestData: 

54 mask: Mask 

55 reader: LegacyMaskedImageReader 

56 plane_map: dict[str, MaskPlane] 

57 

58 

59@pytest.fixture 

60def legacy_test_data(reset_afw_mask_planes: None) -> _LegacyTestData: # noqa: F811 

61 """Return a Mask read directly from the legacy test dataset and a legacy 

62 reader for that image. 

63 

64 Skips if TESTDATA_IMAGES_DIR is unset or lsst.afw.image is unavailable. 

65 """ 

66 # reset_afw_mask_planes will have already skipped if afw is not available. 

67 from lsst.afw.image import MaskedImageFitsReader 

68 

69 if EXTERNAL_DATA_DIR is None: 69 ↛ 71line 69 didn't jump to line 71 because the condition on line 69 was always true

70 pytest.skip("TESTDATA_IMAGES_DIR is not in the environment.") 

71 filename = os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_image.fits") 

72 plane_map = get_legacy_visit_image_mask_planes() 

73 mask = Mask.read_legacy(filename, ext=2, plane_map=plane_map) 

74 reader = MaskedImageFitsReader(filename) 

75 return _LegacyTestData(mask=mask, reader=reader, plane_map=plane_map) 

76 

77 

78def make_mask_planes(rng: np.random.Generator, n_planes: int, n_placeholders: int) -> list[MaskPlane | None]: 

79 """Return a shuffled list of MaskPlane objects with placeholder Nones.""" 

80 planes: list[MaskPlane | None] = [] 

81 for i in range(n_planes): 

82 planes.append(MaskPlane(f"M{i}", f"D{i}")) 

83 planes.extend([None] * n_placeholders) 

84 rng.shuffle(planes) 

85 return planes 

86 

87 

88def test_schema() -> None: 

89 """Test MaskSchema construction, accessors, and basic operations.""" 

90 rng = np.random.default_rng(500) 

91 planes = make_mask_planes(rng, 17, 5) 

92 with pytest.raises(TypeError): 

93 MaskSchema.bits_per_element(np.float32) 

94 assert MaskSchema.bits_per_element(np.uint8) == 8 

95 schema = MaskSchema(planes, dtype=np.uint8) 

96 assert list(schema) == planes 

97 assert len(schema) == len(planes) 

98 assert schema[5] == planes[5] 

99 assert eval(repr(schema), {"dtype": np.dtype, "MaskSchema": MaskSchema, "MaskPlane": MaskPlane}) == schema 

100 report = schema.describe() 

101 plane_table = next(t for t in report.tables if t.title == "Mask planes") 

102 assert len(plane_table.rows) == 17 

103 assert ["M5" == row[3] for row in plane_table.rows].count(True) == 1 

104 bit5 = schema.bit("M5") 

105 assert schema == MaskSchema(planes, np.uint8) 

106 assert schema != MaskSchema(planes, np.int16) 

107 assert schema != MaskSchema(planes[:-1], np.uint8) 

108 assert schema.dtype == np.dtype(np.uint8) 

109 assert schema.mask_size == 3 

110 assert schema.names == {f"M{i}" for i in range(17)} 

111 assert schema.descriptions == {f"M{i}": f"D{i}" for i in range(17)} 

112 bit7 = schema.bit("M7") 

113 bitmask57 = schema.bitmask("M5", "M7") 

114 assert bitmask57[bit5.index] & bit5.mask 

115 assert bitmask57[bit7.index] & bit7.mask 

116 bitmask57[bit5.index] &= ~bit5.mask 

117 bitmask57[bit7.index] &= ~bit7.mask 

118 assert not bitmask57.any() 

119 splits = schema.split(np.int16) 

120 assert len(splits) == 2 

121 assert splits[0].mask_size == 1 

122 assert splits[1].mask_size == 1 

123 assert list(splits[0]) + list(splits[1]) == [p for p in planes if p is not None] 

124 assert len(splits[0]) == 15 

125 assert len(splits[1]) == 2 

126 

127 

128def test_schema_describe() -> None: 

129 """MaskSchema._describe yields a mask-plane table and eval-able repr.""" 

130 rng = np.random.default_rng(500) 

131 planes = make_mask_planes(rng, 3, 0) 

132 schema = MaskSchema(planes, dtype=np.uint8) 

133 

134 report = schema.describe() 

135 assert isinstance(report, Report) 

136 assert report.type_name == "MaskSchema" 

137 

138 tables = [t for t in report.tables if t.title == "Mask planes"] 

139 assert len(tables) == 1 

140 table = tables[0] 

141 assert table.columns == ["Bit", "Index", "Mask", "Name", "Description"] 

142 assert len(table.rows) == 3 

143 # Names appear in the Name column (index 3). 

144 assert {row[3] for row in table.rows} == {"M0", "M1", "M2"} 

145 

146 # repr is still eval-able and round-trips. 

147 reconstructed = eval(repr(schema), {"dtype": np.dtype, "MaskSchema": MaskSchema, "MaskPlane": MaskPlane}) 

148 assert reconstructed == schema 

149 

150 

151def test_schema_from_fits_header() -> None: 

152 """Verify MaskSchema.from_fits_header inverts update_header.""" 

153 planes = [ 

154 MaskPlane("NO_DATA", "No data was available for this pixel."), 

155 MaskPlane("COSMIC_RAY", "A cosmic ray affected this pixel."), 

156 MaskPlane("DETECTED", "Pixel was part of a detected source."), 

157 ] 

158 schema = MaskSchema(planes, dtype=np.uint8) 

159 header = astropy.io.fits.Header() 

160 schema.update_header(header) 

161 result = MaskSchema.from_fits_header(header) 

162 assert result.dtype == np.dtype(np.uint8) 

163 assert list(result) == planes 

164 assert result == schema 

165 

166 

167def test_schema_from_fits_header_preserves_gaps() -> None: 

168 """Verify None placeholders are reconstructed from gaps in MSKN card 

169 numbering. 

170 """ 

171 planes: list[MaskPlane | None] = [MaskPlane("A", "a"), None, MaskPlane("B", "b")] 

172 header = astropy.io.fits.Header() 

173 MaskSchema(planes, dtype=np.uint8).update_header(header) 

174 assert list(MaskSchema.from_fits_header(header)) == planes 

175 

176 

177def test_schema_from_fits_header_requires_cards() -> None: 

178 """Verify MaskSchema.from_fits_header raises ValueError on a header with 

179 no MSKN cards. 

180 """ 

181 with pytest.raises(ValueError, match="no MSKN cards"): 

182 MaskSchema.from_fits_header(astropy.io.fits.Header()) 

183 

184 

185def test_interpret() -> None: 

186 """Verify interpret returns correct plane names across a multi-byte 

187 schema. 

188 """ 

189 # 3 named planes padded with Nones to exceed 8 bits; A and B land in 

190 # byte 0, C lands in byte 1. 

191 planes: list[MaskPlane | None] = [ 

192 MaskPlane("A", "a"), 

193 MaskPlane("B", "b"), 

194 None, 

195 None, 

196 None, 

197 None, 

198 None, 

199 None, 

200 MaskPlane("C", "c"), 

201 ] 

202 schema = MaskSchema(planes, dtype=np.uint8) 

203 assert schema.mask_size == 2 

204 

205 assert set(schema.interpret(schema.bitmask("A", "C"))) == {"A", "C"} 

206 assert set(schema.interpret(schema.bitmask("B"))) == {"B"} 

207 assert set(schema.interpret(np.zeros(schema.mask_size, dtype=schema.dtype))) == set() 

208 assert set(schema.interpret(schema.bitmask("A", "B", "C"))) == {"A", "B", "C"} 

209 

210 

211def test_basics() -> None: 

212 """Test basic Mask construction, string representation, and error 

213 conditions. 

214 """ 

215 rng = np.random.default_rng(500) 

216 planes = make_mask_planes(rng, 35, n_placeholders=5) 

217 schema = MaskSchema(planes, dtype=np.uint8) 

218 bbox = Box.factory[5:50, 6:60] 

219 mask = Mask( 

220 0, 

221 schema=schema, 

222 bbox=bbox, 

223 metadata={"four_and_a_half": 4.5}, 

224 ) 

225 

226 assert mask[...] is not mask 

227 assert mask.__eq__(42) == NotImplemented 

228 assert mask == mask 

229 assert ( 

230 str(mask) 

231 == "Mask([y=5:50, x=6:60], ['M34', 'M15', 'M29', 'M1', 'M20', 'M11', 'M13', 'M7', 'M17', 'M12', " 

232 "'M31', 'M16', 'M2', 'M3', 'M8', 'M26', 'M22', 'M5', 'M18', 'M19', 'M24', 'M21', 'M27', 'M6', " 

233 "'M28', 'M10', 'M4', 'M23', 'M0', 'M25', 'M9', 'M14', 'M33', 'M32', 'M30'])" 

234 ) 

235 assert repr(mask).startswith( 

236 "Mask(..., bbox=Box(y=Interval(start=5, stop=50), x=Interval(start=6, stop=60)), " 

237 "schema=MaskSchema([MaskPlane(name='M34', description='D34')" 

238 ), f"Repr: {mask!r}" 

239 

240 with pytest.raises(TypeError): 

241 # No bbox, size or array. 

242 Mask(0, schema=schema) 

243 

244 with pytest.raises(ValueError, match="Explicit bbox shape"): 

245 # Box mismatch. 

246 Mask(mask.array, schema=schema, bbox=Box.factory[0:20, -5:45]) 

247 

248 with pytest.raises(ValueError, match="Explicit shape"): 

249 # Shape mismatch. 

250 Mask(mask.array, schema=schema, shape=(5, 10, 5)) 

251 

252 with pytest.raises(ValueError, match="Mask array must be 3-d"): 

253 # Cannot be 2-D. 

254 Mask(mask.array.reshape((2430, 5)), schema=schema, bbox=Box.factory[0:20, -5:45]) 

255 

256 

257def test_read_write() -> None: 

258 """Test explicit calls to Mask.read and Mask.write through FITS.""" 

259 rng = np.random.default_rng(500) 

260 planes = make_mask_planes(rng, 35, n_placeholders=5) 

261 schema = MaskSchema(planes, dtype=np.uint8) 

262 bbox = Box.factory[5:50, 6:60] 

263 mask = Mask( 

264 0, 

265 schema=schema, 

266 bbox=bbox, 

267 metadata={"four_and_a_half": 4.5}, 

268 ) 

269 with lsst.utils.tests.getTempFilePath(".fits") as tmpFile: 

270 mask.write(tmpFile) 

271 new = Mask.read(tmpFile) 

272 assert new == mask 

273 # __eq__ ignores metadata. 

274 assert new.metadata["four_and_a_half"] == 4.5 

275 assert new.metadata == mask.metadata 

276 

277 

278def test_serialize_multi() -> None: 

279 """Test serializing a mask with more than 31 mask planes (multiple 

280 HDUs). 

281 """ 

282 rng = np.random.default_rng(500) 

283 planes = make_mask_planes(rng, 35, n_placeholders=5) 

284 schema = MaskSchema(planes, dtype=np.uint8) 

285 bbox = Box.factory[5:50, 6:60] 

286 mask = Mask(0, schema=schema, bbox=bbox, metadata={"four_and_a_half": 4.5}) 

287 shape = bbox.shape 

288 for plane in schema: 

289 if plane is not None: 

290 mask.set(plane.name, rng.random(shape) > 0.5) 

291 with RoundtripFits(mask) as roundtrip: 

292 fits = roundtrip.inspect() 

293 assert fits[1].header["EXTNAME"] == "MASK" 

294 assert fits[1].header.get("EXTVER", 1) == 1 

295 assert fits[1].header["ZCMPTYPE"] == "GZIP_2" 

296 assert fits[2].header["EXTNAME"] == "MASK" 

297 assert fits[2].header["EXTVER"] == 2 

298 assert fits[2].header["ZCMPTYPE"] == "GZIP_2" 

299 n = 0 

300 for plane in planes: 

301 if plane is not None: 

302 hdu = fits[1] if n < 31 else fits[2] 

303 assert hdu.header[f"MSKN{(n % 31):04d}"] == plane.name 

304 assert hdu.header[f"MSKM{(n % 31):04d}"] == 1 << (n % 31) 

305 assert hdu.header[f"MSKD{(n % 31):04d}"] == plane.description 

306 n += 1 

307 assert_masks_equal(mask, roundtrip.result) 

308 

309 

310def test_add_plane_returns_new_mask() -> None: 

311 """Verify add_plane returns a new mask without modifying the original or 

312 its views. 

313 """ 

314 rng = np.random.default_rng(500) 

315 planes = make_mask_planes(rng, 3, n_placeholders=0) 

316 schema = MaskSchema(planes, dtype=np.uint8) 

317 bbox = Box.factory[5:50, 6:60] 

318 mask = Mask(0, schema=schema, bbox=bbox) 

319 m0 = rng.random(bbox.shape) > 0.5 

320 mask.set("M0", m0) 

321 view = mask[bbox] # shares the array and old schema with mask 

322 original_array = mask.array 

323 

324 new_mask = mask.add_plane("OUTSIDE_STENCIL", "Pixel lies outside the stencil.") 

325 

326 # The original mask and any views keep the old schema and array. 

327 assert "OUTSIDE_STENCIL" not in mask.schema.names 

328 assert "OUTSIDE_STENCIL" not in view.schema.names 

329 assert mask.array is original_array 

330 # The new mask reallocated a fresh array and carries the new plane. 

331 assert new_mask.array is not original_array 

332 assert "OUTSIDE_STENCIL" in new_mask.schema.names 

333 assert new_mask.schema.descriptions["OUTSIDE_STENCIL"] == "Pixel lies outside the stencil." 

334 # The new plane is the fourth (overall index 3) so it lives in byte 0. 

335 bit = new_mask.schema.bit("OUTSIDE_STENCIL") 

336 assert bit.index == 0 

337 assert bit.mask == 1 << 3 

338 assert new_mask.schema.mask_size == 1 

339 # Existing plane data is preserved and the new plane starts all-False. 

340 np.testing.assert_array_equal(new_mask.get("M0"), m0) 

341 assert not new_mask.get("OUTSIDE_STENCIL").any() 

342 

343 

344def test_add_plane_grows_byte() -> None: 

345 """Verify adding a ninth plane crosses the 8-plane boundary into a 

346 second byte. 

347 """ 

348 rng = np.random.default_rng(500) 

349 planes = make_mask_planes(rng, 8, n_placeholders=0) 

350 schema = MaskSchema(planes, dtype=np.uint8) 

351 bbox = Box.factory[5:50, 6:60] 

352 mask = Mask(0, schema=schema, bbox=bbox) 

353 set_planes = {} 

354 for plane in planes: 

355 assert plane is not None 

356 boolean_mask = rng.random(bbox.shape) > 0.5 

357 mask.set(plane.name, boolean_mask) 

358 set_planes[plane.name] = boolean_mask 

359 

360 new_mask = mask.add_plane("OUTSIDE_STENCIL", "Pixel lies outside the stencil.") 

361 

362 # The original is unchanged; the new mask spills into a second byte. 

363 assert mask.schema.mask_size == 1 

364 bit = new_mask.schema.bit("OUTSIDE_STENCIL") 

365 assert bit.index == 1 

366 assert bit.mask == 1 << 0 

367 assert new_mask.schema.mask_size == 2 

368 assert new_mask.array.shape == bbox.shape + (2,) 

369 assert not new_mask.get("OUTSIDE_STENCIL").any() 

370 # Every pre-existing plane keeps its data. 

371 for name, boolean_mask in set_planes.items(): 

372 np.testing.assert_array_equal(new_mask.get(name), boolean_mask) 

373 

374 

375def test_add_planes_multiple() -> None: 

376 """Verify add_planes adds several planes in a single call.""" 

377 rng = np.random.default_rng(500) 

378 planes = make_mask_planes(rng, 3, n_placeholders=0) 

379 bbox = Box.factory[0:4, 0:5] 

380 mask = Mask(0, schema=MaskSchema(planes, dtype=np.uint8), bbox=bbox) 

381 m0 = rng.random(bbox.shape) > 0.5 

382 mask.set("M0", m0) 

383 

384 new_mask = mask.add_planes([MaskPlane("A", "plane a"), MaskPlane("B", "plane b")]) 

385 

386 assert set(mask.schema.names) == {"M0", "M1", "M2"} # original unchanged 

387 assert set(new_mask.schema.names) == {"M0", "M1", "M2", "A", "B"} 

388 np.testing.assert_array_equal(new_mask.get("M0"), m0) 

389 assert not new_mask.get("A").any() 

390 assert not new_mask.get("B").any() 

391 

392 

393def test_add_planes_drop_reassigns_bits() -> None: 

394 """Verify dropping a plane compacts the schema and repacks pixel values.""" 

395 rng = np.random.default_rng(500) 

396 bbox = Box.factory[0:4, 0:5] 

397 schema = MaskSchema([MaskPlane("A", "a"), MaskPlane("B", "b"), MaskPlane("C", "c")], dtype=np.uint8) 

398 mask = Mask(0, schema=schema, bbox=bbox) 

399 a = rng.random(bbox.shape) > 0.5 

400 c = rng.random(bbox.shape) > 0.5 

401 mask.set("A", a) 

402 mask.set("B", rng.random(bbox.shape) > 0.5) 

403 mask.set("C", c) 

404 

405 new_mask = mask.add_planes([MaskPlane("D", "d")], drop=["B"]) 

406 

407 # B is gone; D is appended after the retained planes. 

408 assert list(new_mask.schema.names) == ["A", "C", "D"] 

409 assert "B" not in new_mask.schema.names 

410 # C moved down from bit 2 to bit 1; D takes bit 2. 

411 assert new_mask.schema.bit("A").mask == 1 << 0 

412 assert new_mask.schema.bit("C").mask == 1 << 1 

413 assert new_mask.schema.bit("D").mask == 1 << 2 

414 # Retained pixel values follow their planes; the new plane is cleared. 

415 np.testing.assert_array_equal(new_mask.get("A"), a) 

416 np.testing.assert_array_equal(new_mask.get("C"), c) 

417 assert not new_mask.get("D").any() 

418 

419 

420def test_add_planes_with_placeholder() -> None: 

421 """Verify None placeholders reserve bits and survive add_planes and a 

422 FITS round-trip. 

423 """ 

424 rng = np.random.default_rng(500) 

425 bbox = Box.factory[0:4, 0:5] 

426 # Schema with a pre-existing placeholder reserving bit 1. 

427 schema = MaskSchema([MaskPlane("A", "a"), None, MaskPlane("B", "b")], dtype=np.uint8) 

428 mask = Mask(0, schema=schema, bbox=bbox) 

429 a = rng.random(bbox.shape) > 0.5 

430 b = rng.random(bbox.shape) > 0.5 

431 mask.set("A", a) 

432 mask.set("B", b) 

433 

434 # Append a block that itself contains an interior placeholder. 

435 new_mask = mask.add_planes([MaskPlane("C", "c"), None, MaskPlane("D", "d")]) 

436 

437 # The pre-existing placeholder stays at bit 1; the added placeholder 

438 # stays between C and D (bit 4), not at the end. 

439 assert list(new_mask.schema) == [ 

440 MaskPlane("A", "a"), 

441 None, 

442 MaskPlane("B", "b"), 

443 MaskPlane("C", "c"), 

444 None, 

445 MaskPlane("D", "d"), 

446 ] 

447 assert new_mask.schema.bit("A").mask == 1 << 0 

448 assert new_mask.schema.bit("B").mask == 1 << 2 

449 assert new_mask.schema.bit("C").mask == 1 << 3 

450 assert new_mask.schema.bit("D").mask == 1 << 5 

451 # Retained pixel values follow their planes; new planes start cleared. 

452 np.testing.assert_array_equal(new_mask.get("A"), a) 

453 np.testing.assert_array_equal(new_mask.get("B"), b) 

454 assert not new_mask.get("C").any() 

455 assert not new_mask.get("D").any() 

456 

457 with RoundtripFits(new_mask) as roundtrip: 

458 assert_masks_equal(new_mask, roundtrip.result) 

459 

460 

461def test_add_planes_drop_unknown_raises() -> None: 

462 """Verify dropping a non-existent plane raises ValueError.""" 

463 mask = Mask(0, schema=MaskSchema([MaskPlane("A", "a")], dtype=np.uint8), bbox=Box.factory[0:2, 0:2]) 

464 with pytest.raises(ValueError, match="Cannot drop mask planes"): 

465 mask.add_planes([], drop=["NOPE"]) 

466 

467 

468def test_add_plane_duplicate_raises() -> None: 

469 """Verify adding a plane whose name already exists raises ValueError.""" 

470 rng = np.random.default_rng(500) 

471 planes = make_mask_planes(rng, 3, n_placeholders=0) 

472 schema = MaskSchema(planes, dtype=np.uint8) 

473 mask = Mask(0, schema=schema, bbox=Box.factory[0:4, 0:4]) 

474 with pytest.raises(ValueError, match="already exists"): 

475 mask.add_plane("M0", "Duplicate of an existing plane.") 

476 

477 

478def test_add_plane_roundtrip() -> None: 

479 """Verify a runtime-added plane and its data survive a FITS round-trip.""" 

480 rng = np.random.default_rng(500) 

481 planes = make_mask_planes(rng, 8, n_placeholders=0) 

482 schema = MaskSchema(planes, dtype=np.uint8) 

483 bbox = Box.factory[5:50, 6:60] 

484 mask = Mask(0, schema=schema, bbox=bbox) 

485 mask = mask.add_plane("OUTSIDE_STENCIL", "Pixel lies outside the stencil.") 

486 mask.set("OUTSIDE_STENCIL", rng.random(bbox.shape) > 0.5) 

487 with lsst.utils.tests.getTempFilePath(".fits") as tmpFile: 

488 mask.write(tmpFile) 

489 new = Mask.read(tmpFile) 

490 assert new == mask 

491 assert new.schema.descriptions["OUTSIDE_STENCIL"] == "Pixel lies outside the stencil." 

492 assert_masks_equal(new, mask) 

493 

494 

495def test_legacy_non_cell_coadd_plane_map() -> None: 

496 """Verify the non-cell coadd map defines a distinct SENSOR_EDGE plane.""" 

497 plane_map = get_legacy_non_cell_coadd_mask_planes() 

498 assert "SENSOR_EDGE" in plane_map 

499 assert plane_map["SENSOR_EDGE"].name == "SENSOR_EDGE" 

500 

501 

502def test_legacy_optional_plane_map() -> None: 

503 """Verify the source injection planes are optional, and so belong to no 

504 image's plane map. 

505 """ 

506 optional = get_legacy_optional_mask_planes() 

507 assert set(optional) == { 

508 "INJECTED", 

509 "INJECTED_CORE", 

510 "INJECTED_TEMPLATE", 

511 "INJECTED_CORE_TEMPLATE", 

512 } 

513 for old_name, plane in optional.items(): 

514 assert plane.name == old_name 

515 for plane_map in ( 

516 get_legacy_visit_image_mask_planes(), 

517 get_legacy_difference_image_mask_planes(), 

518 get_legacy_non_cell_coadd_mask_planes(), 

519 get_legacy_template_mask_planes(), 

520 ): 

521 assert not (plane_map.keys() & optional.keys()) 

522 

523 

524def test_from_legacy_adds_optional_planes_in_use(reset_afw_mask_planes: None) -> None: # noqa: F811 

525 """Verify an optional plane joins the schema when the legacy mask has 

526 pixels set in it, and costs nothing when it does not. 

527 """ 

528 # reset_afw_mask_planes will have already skipped if afw is not available. 

529 from lsst.afw.image import Mask as LegacyMask 

530 

531 legacy = LegacyMask(5, 4) 

532 injected = np.arange(20).reshape(4, 5) < 3 

533 for old_name in ("INJECTED", "INJECTED_CORE"): 

534 legacy.addMaskPlane(old_name) 

535 legacy.array[injected] |= legacy.getPlaneBitMask("INJECTED") 

536 plane_map = get_legacy_visit_image_mask_planes() 

537 

538 mask = Mask.from_legacy(legacy, plane_map=plane_map) 

539 

540 assert "INJECTED" in mask.schema.names 

541 np.testing.assert_array_equal(mask.get("INJECTED"), injected) 

542 # INJECTED_CORE is defined but unused, so it takes up no bit. 

543 assert "INJECTED_CORE" not in mask.schema.names 

544 assert len(mask.schema) == len(plane_map) + 1 

545 # The plane keeps its legacy name on the way back out. 

546 round_tripped = mask.to_legacy(plane_map) 

547 np.testing.assert_array_equal( 

548 (round_tripped.array & round_tripped.getPlaneBitMask("INJECTED")).astype(bool), injected 

549 ) 

550 

551 

552def test_legacy_template_plane_map() -> None: 

553 """Verify the template map has the coadd planes and HIGH_VARIANCE.""" 

554 plane_map = get_legacy_template_mask_planes() 

555 non_cell_coadd = get_legacy_non_cell_coadd_mask_planes() 

556 assert non_cell_coadd.keys() < plane_map.keys() 

557 assert plane_map["HIGH_VARIANCE"].name == "HIGH_VARIANCE" 

558 

559 

560def test_guess_legacy_plane_map_coadd_discriminator() -> None: 

561 """Verify INEXACT_PSF routes to a coadd map, SENSOR_EDGE discriminates 

562 non-cell from cell, and HIGH_VARIANCE marks a template. 

563 """ 

564 non_cell = _guess_legacy_plane_map({"INEXACT_PSF": 11, "SENSOR_EDGE": 14}) 

565 assert "SENSOR_EDGE" in non_cell 

566 assert "HIGH_VARIANCE" not in non_cell 

567 cell = _guess_legacy_plane_map({"INEXACT_PSF": 11}) 

568 assert "SENSOR_EDGE" not in cell 

569 template = _guess_legacy_plane_map({"INEXACT_PSF": 11, "SENSOR_EDGE": 14, "HIGH_VARIANCE": 15}) 

570 assert "HIGH_VARIANCE" in template 

571 

572 

573def test_to_legacy_keeps_unmapped_planes(reset_afw_mask_planes: None) -> None: # noqa: F811 

574 """Verify a plane the map does not name keeps its pixels and its name. 

575 

576 A template is a warped coadd stored as a difference image, so it has 

577 CLIPPED set while the difference image map has no entry for it. 

578 """ 

579 plane_map = get_legacy_non_cell_coadd_mask_planes() 

580 mask = Mask(0, schema=MaskSchema(list(plane_map.values())), bbox=Box.factory[0:4, 0:5]) 

581 clipped = np.arange(20).reshape(4, 5) < 3 

582 saturated = np.arange(20).reshape(4, 5) > 17 

583 mask.set("CLIPPED", clipped) 

584 mask.set("SATURATED", saturated) 

585 

586 legacy = mask.to_legacy(get_legacy_difference_image_mask_planes()) 

587 

588 # The unmapped coadd plane is written under the name it already has. 

589 assert "CLIPPED" in legacy.getMaskPlaneDict() 

590 np.testing.assert_array_equal((legacy.array & legacy.getPlaneBitMask("CLIPPED")).astype(bool), clipped) 

591 # A mapped plane still comes back under its legacy name. 

592 np.testing.assert_array_equal((legacy.array & legacy.getPlaneBitMask("SAT")).astype(bool), saturated) 

593 assert "SATURATED" not in legacy.getMaskPlaneDict() 

594 assert_masks_equal(Mask.from_legacy(legacy, plane_map=plane_map), mask) 

595 

596 

597def test_legacy_round_trip_with_optional_plane(reset_afw_mask_planes: None) -> None: # noqa: F811 

598 """Verify a legacy mask survives a round trip through `Mask`, with the 

599 mapped planes renamed both ways and an optional plane carried as it is. 

600 """ 

601 # reset_afw_mask_planes will have already skipped if afw is not available. 

602 from lsst.afw.image import Mask as LegacyMask 

603 

604 legacy = LegacyMask(5, 4) 

605 rows = {"SAT": 0, "INTRP": 1, "INJECTED": 2} 

606 pixels = {} 

607 for old_name, row in rows.items(): 

608 legacy.addMaskPlane(old_name) 

609 selection = np.zeros(legacy.array.shape, dtype=bool) 

610 selection[row, :] = True 

611 legacy.array[selection] |= legacy.getPlaneBitMask(old_name) 

612 pixels[old_name] = selection 

613 # Defined by some other image in this process, but unused here. 

614 legacy.addMaskPlane("INJECTED_CORE") 

615 plane_map = get_legacy_visit_image_mask_planes() 

616 

617 mask = Mask.from_legacy(legacy, plane_map=plane_map) 

618 

619 assert {"SATURATED", "INTERPOLATED", "INJECTED"} <= set(mask.schema.names) 

620 assert "INJECTED_CORE" not in mask.schema.names 

621 np.testing.assert_array_equal(mask.get("SATURATED"), pixels["SAT"]) 

622 np.testing.assert_array_equal(mask.get("INJECTED"), pixels["INJECTED"]) 

623 

624 round_tripped = mask.to_legacy(plane_map) 

625 

626 compare_mask_to_legacy(mask, round_tripped, plane_map) 

627 for old_name, selection in pixels.items(): 

628 np.testing.assert_array_equal( 

629 (round_tripped.array & round_tripped.getPlaneBitMask(old_name)).astype(bool), selection 

630 ) 

631 # The mapped planes go back out under their legacy names, and the 

632 # optional plane under the only name it has. 

633 assert "SATURATED" not in round_tripped.getMaskPlaneDict() 

634 assert "INJECTED" in round_tripped.getMaskPlaneDict() 

635 

636 

637def test_read_legacy_adds_optional_planes_in_use(reset_afw_mask_planes: None) -> None: # noqa: F811 

638 """Verify reading a legacy FITS mask resolves the optional planes the 

639 same way converting one in memory does. 

640 """ 

641 # reset_afw_mask_planes will have already skipped if afw is not available. 

642 from lsst.afw.image import MaskedImageF 

643 

644 # `read_legacy` reads the MASK extension of a masked image or exposure. 

645 masked_image = MaskedImageF(5, 4) 

646 legacy = masked_image.mask 

647 injected = np.arange(20).reshape(4, 5) < 3 

648 for old_name in ("INJECTED", "INJECTED_CORE"): 

649 legacy.addMaskPlane(old_name) 

650 legacy.array[injected] |= legacy.getPlaneBitMask("INJECTED") 

651 

652 with lsst.utils.tests.getTempFilePath(".fits") as tmpFile: 

653 masked_image.writeFits(tmpFile) 

654 mask = Mask.read_legacy(tmpFile, ext=2, plane_map=get_legacy_visit_image_mask_planes()) 

655 

656 assert "INJECTED" in mask.schema.names 

657 np.testing.assert_array_equal(mask.get("INJECTED"), injected) 

658 assert "INJECTED_CORE" not in mask.schema.names 

659 

660 

661def test_legacy(legacy_test_data: _LegacyTestData) -> None: 

662 """Test Mask.read_legacy, Mask.to_legacy, and Mask.from_legacy.""" 

663 assert legacy_test_data.mask.schema.names == {p.name for p in legacy_test_data.plane_map.values()} 

664 assert legacy_test_data.mask.bbox == Box.from_legacy(legacy_test_data.reader.readBBox()) 

665 legacy_mask = legacy_test_data.reader.readMask() 

666 compare_mask_to_legacy(legacy_test_data.mask, legacy_mask, legacy_test_data.plane_map) 

667 compare_mask_to_legacy( 

668 legacy_test_data.mask,