Coverage for tests/test_mask.py: 95%

379 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 

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_non_cell_coadd_mask_planes, 

29 get_legacy_visit_image_mask_planes, 

30) 

31from lsst.images._mask import _guess_legacy_plane_map 

32from lsst.images.describe import Report 

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

34 RoundtripFits, 

35 assert_masks_equal, 

36 compare_mask_to_legacy, 

37 reset_afw_mask_planes, 

38) 

39 

40try: 

41 from lsst.afw.image import MaskedImageReader as LegacyMaskedImageReader 

42 

43except ImportError: 

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

45 

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

47 

48 

49@dataclasses.dataclass 

50class _LegacyTestData: 

51 mask: Mask 

52 reader: LegacyMaskedImageReader 

53 plane_map: dict[str, MaskPlane] 

54 

55 

56@pytest.fixture 

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

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

59 reader for that image. 

60 

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

62 """ 

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

64 from lsst.afw.image import MaskedImageFitsReader 

65 

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

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

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

69 plane_map = get_legacy_visit_image_mask_planes() 

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

71 reader = MaskedImageFitsReader(filename) 

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

73 

74 

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

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

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

78 for i in range(n_planes): 

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

80 planes.extend([None] * n_placeholders) 

81 rng.shuffle(planes) 

82 return planes 

83 

84 

85def test_schema() -> None: 

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

87 rng = np.random.default_rng(500) 

88 planes = make_mask_planes(rng, 17, 5) 

89 with pytest.raises(TypeError): 

90 MaskSchema.bits_per_element(np.float32) 

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

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

93 assert list(schema) == planes 

94 assert len(schema) == len(planes) 

95 assert schema[5] == planes[5] 

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

97 report = schema.describe() 

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

99 assert len(plane_table.rows) == 17 

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

101 bit5 = schema.bit("M5") 

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

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

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

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

106 assert schema.mask_size == 3 

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

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

109 bit7 = schema.bit("M7") 

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

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

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

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

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

115 assert not bitmask57.any() 

116 splits = schema.split(np.int16) 

117 assert len(splits) == 2 

118 assert splits[0].mask_size == 1 

119 assert splits[1].mask_size == 1 

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

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

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

123 

124 

125def test_schema_describe() -> None: 

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

127 rng = np.random.default_rng(500) 

128 planes = make_mask_planes(rng, 3, 0) 

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

130 

131 report = schema.describe() 

132 assert isinstance(report, Report) 

133 assert report.type_name == "MaskSchema" 

134 

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

136 assert len(tables) == 1 

137 table = tables[0] 

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

139 assert len(table.rows) == 3 

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

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

142 

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

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

145 assert reconstructed == schema 

146 

147 

148def test_schema_from_fits_header() -> None: 

149 """Verify MaskSchema.from_fits_header inverts update_header.""" 

150 planes = [ 

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

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

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

154 ] 

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

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

157 schema.update_header(header) 

158 result = MaskSchema.from_fits_header(header) 

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

160 assert list(result) == planes 

161 assert result == schema 

162 

163 

164def test_schema_from_fits_header_preserves_gaps() -> None: 

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

166 numbering. 

167 """ 

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

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

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

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

172 

173 

174def test_schema_from_fits_header_requires_cards() -> None: 

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

176 no MSKN cards. 

177 """ 

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

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

180 

181 

182def test_interpret() -> None: 

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

184 schema. 

185 """ 

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

187 # byte 0, C lands in byte 1. 

188 planes: list[MaskPlane | None] = [ 

189 MaskPlane("A", "a"), 

190 MaskPlane("B", "b"), 

191 None, 

192 None, 

193 None, 

194 None, 

195 None, 

196 None, 

197 MaskPlane("C", "c"), 

198 ] 

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

200 assert schema.mask_size == 2 

201 

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

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

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

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

206 

207 

208def test_basics() -> None: 

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

210 conditions. 

211 """ 

212 rng = np.random.default_rng(500) 

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

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

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

216 mask = Mask( 

217 0, 

218 schema=schema, 

219 bbox=bbox, 

220 metadata={"four_and_a_half": 4.5}, 

221 ) 

222 

223 assert mask[...] is not mask 

224 assert mask.__eq__(42) == NotImplemented 

225 assert mask == mask 

226 assert ( 

227 str(mask) 

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

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

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

231 ) 

232 assert repr(mask).startswith( 

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

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

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

236 

237 with pytest.raises(TypeError): 

238 # No bbox, size or array. 

239 Mask(0, schema=schema) 

240 

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

242 # Box mismatch. 

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

244 

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

246 # Shape mismatch. 

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

248 

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

250 # Cannot be 2-D. 

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

252 

253 

254def test_read_write() -> None: 

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

256 rng = np.random.default_rng(500) 

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

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

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

260 mask = Mask( 

261 0, 

262 schema=schema, 

263 bbox=bbox, 

264 metadata={"four_and_a_half": 4.5}, 

265 ) 

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

267 mask.write(tmpFile) 

268 new = Mask.read(tmpFile) 

269 assert new == mask 

270 # __eq__ ignores metadata. 

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

272 assert new.metadata == mask.metadata 

273 

274 

275def test_serialize_multi() -> None: 

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

277 HDUs). 

278 """ 

279 rng = np.random.default_rng(500) 

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

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

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

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

284 shape = bbox.shape 

285 for plane in schema: 

286 if plane is not None: 

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

288 with RoundtripFits(mask) as roundtrip: 

289 fits = roundtrip.inspect() 

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

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

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

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

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

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

296 n = 0 

297 for plane in planes: 

298 if plane is not None: 

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

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

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

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

303 n += 1 

304 assert_masks_equal(mask, roundtrip.result) 

305 

306 

307def test_add_plane_returns_new_mask() -> None: 

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

309 its views. 

310 """ 

311 rng = np.random.default_rng(500) 

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

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

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

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

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

317 mask.set("M0", m0) 

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

319 original_array = mask.array 

320 

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

322 

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

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

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

326 assert mask.array is original_array 

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

328 assert new_mask.array is not original_array 

329 assert "OUTSIDE_STENCIL" in new_mask.schema.names 

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

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

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

333 assert bit.index == 0 

334 assert bit.mask == 1 << 3 

335 assert new_mask.schema.mask_size == 1 

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

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

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

339 

340 

341def test_add_plane_grows_byte() -> None: 

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

343 second byte. 

344 """ 

345 rng = np.random.default_rng(500) 

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

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

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

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

350 set_planes = {} 

351 for plane in planes: 

352 assert plane is not None 

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

354 mask.set(plane.name, boolean_mask) 

355 set_planes[plane.name] = boolean_mask 

356 

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

358 

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

360 assert mask.schema.mask_size == 1 

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

362 assert bit.index == 1 

363 assert bit.mask == 1 << 0 

364 assert new_mask.schema.mask_size == 2 

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

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

367 # Every pre-existing plane keeps its data. 

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

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

370 

371 

372def test_add_planes_multiple() -> None: 

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

374 rng = np.random.default_rng(500) 

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

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

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

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

379 mask.set("M0", m0) 

380 

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

382 

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

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

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

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

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

388 

389 

390def test_add_planes_drop_reassigns_bits() -> None: 

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

392 rng = np.random.default_rng(500) 

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

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

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

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

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

398 mask.set("A", a) 

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

400 mask.set("C", c) 

401 

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

403 

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

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

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

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

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

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

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

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

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

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

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

415 

416 

417def test_add_planes_with_placeholder() -> None: 

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

419 FITS round-trip. 

420 """ 

421 rng = np.random.default_rng(500) 

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

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

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

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

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

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

428 mask.set("A", a) 

429 mask.set("B", b) 

430 

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

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

433 

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

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

436 assert list(new_mask.schema) == [ 

437 MaskPlane("A", "a"), 

438 None, 

439 MaskPlane("B", "b"), 

440 MaskPlane("C", "c"), 

441 None, 

442 MaskPlane("D", "d"), 

443 ] 

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

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

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

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

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

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

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

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

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

453 

454 with RoundtripFits(new_mask) as roundtrip: 

455 assert_masks_equal(new_mask, roundtrip.result) 

456 

457 

458def test_add_planes_drop_unknown_raises() -> None: 

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

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

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

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

463 

464 

465def test_add_plane_duplicate_raises() -> None: 

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

467 rng = np.random.default_rng(500) 

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

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

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

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

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

473 

474 

475def test_add_plane_roundtrip() -> None: 

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

477 rng = np.random.default_rng(500) 

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

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

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

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

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

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

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

485 mask.write(tmpFile) 

486 new = Mask.read(tmpFile) 

487 assert new == mask 

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

489 assert_masks_equal(new, mask) 

490 

491 

492def test_legacy_non_cell_coadd_plane_map() -> None: 

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

494 plane_map = get_legacy_non_cell_coadd_mask_planes() 

495 assert "SENSOR_EDGE" in plane_map 

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

497 

498 

499def test_guess_legacy_plane_map_coadd_discriminator() -> None: 

500 """Verify INEXACT_PSF routes to a coadd map and SENSOR_EDGE discriminates 

501 non-cell from cell. 

502 """ 

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

504 assert "SENSOR_EDGE" in non_cell 

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

506 assert "SENSOR_EDGE" not in cell 

507 

508 

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

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

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

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

513 legacy_mask = legacy_test_data.reader.readMask() 

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

515 compare_mask_to_legacy( 

516 legacy_test_data.mask, 

517 legacy_test_data.mask.to_legacy(legacy_test_data.plane_map), 

518 legacy_test_data.plane_map, 

519 ) 

520 assert_masks_equal( 

521 legacy_test_data.mask, Mask.from_legacy(legacy_mask, plane_map=legacy_test_data.plane_map) 

522 ) 

523 # Write the mask out in the new format, and test that we can read it back. 

524 with RoundtripFits(legacy_test_data.mask, storage_class="MaskV2") as roundtrip: 

525 pass 

526 assert_masks_equal(roundtrip.result, legacy_test_data.mask) 

527 

528 

529def test_legacy_butler_read(legacy_test_data: _LegacyTestData) -> None: 

530 """Test that a round-tripped MaskV2 can be read back as a legacy afw 

531 Mask via Butler. 

532 """ 

533 with RoundtripFits(legacy_test_data.mask, storage_class="MaskV2") as roundtrip: 

534 legacy_mask = roundtrip.get(storageClass="Mask") 

535 assert isinstance(legacy_mask, lsst.afw.image.Mask) 

536 compare_mask_to_legacy(legacy_test_data.mask, legacy_mask) 

537 

538 

539def test_mask_repr_str_pinned() -> None: 

540 """Mask str/repr match their documented forms.""" 

541 rng = np.random.default_rng(7) 

542 schema = MaskSchema(make_mask_planes(rng, 3, 0), dtype=np.uint8) 

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

544 assert str(mask) == f"Mask({mask.bbox!s}, {list(mask.schema.names)})" 

545 assert repr(mask) == f"Mask(..., bbox={mask.bbox!r}, schema={mask.schema!r})" 

546 

547 

548def test_mask_describe_detail_reports_set_pixel_counts() -> None: 

549 """detail=True adds a 'Set pixels' column with per-plane counts.""" 

550 schema = MaskSchema([MaskPlane("BAD", "bad"), MaskPlane("DET", "detected")], dtype=np.uint8) 

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

552 mask.set("BAD", mask.get("BAD") | (np.arange(20).reshape(4, 5) < 3)) 

553 

554 # Cheap path: no counts, no extra column. 

555 plain = mask.describe().children["schema"] 

556 plain_table = next(t for t in plain.tables if t.title == "Mask planes") 

557 assert "Set pixels" not in plain_table.columns 

558 

559 # Detailed path: 'Set pixels' column present with correct counts. 

560 detailed = mask.describe(detail=True).children["schema"] 

561 table = next(t for t in detailed.tables if t.title == "Mask planes") 

562 assert table.columns[-1] == "Set pixels" 

563 counts = {row[3]: row[-1] for row in table.rows} # row[3] is the Name column 

564 assert counts["BAD"] == 3 

565 assert counts["DET"] == 0 

566 

567 

568def test_compare() -> None: 

569 """Test Mask.compare added/removed counts for a shared plane.""" 

570 schema = MaskSchema([MaskPlane("A", "dA"), MaskPlane("B", "dB")], dtype=np.uint8) 

571 m1 = Mask(0, schema=schema, bbox=Box.factory[0:2, 0:2]) 

572 m1.set("A", np.array([[True, False], [False, True]])) 

573 m2 = Mask(0, schema=schema, bbox=Box.factory[0:2, 0:2]) 

574 m2.set("A", np.array([[True, True], [False, False]])) 

575 # A: m1 set at (0,0),(1,1); m2 set at (0,0),(0,1) 

576 assert m1.compare(m2) == {"A": (1, 1)} 

577 

578 

579def test_compare_identical_is_empty() -> None: 

580 """Test that identical masks yield an empty difference.""" 

581 schema = MaskSchema([MaskPlane("A", "dA")], dtype=np.uint8) 

582 m1 = Mask(0, schema=schema, bbox=Box.factory[0:2, 0:1]) 

583 m1.set("A", np.array([[True], [False]])) 

584 m2 = Mask(0, schema=schema, bbox=Box.factory[0:2, 0:1]) 

585 m2.set("A", np.array([[True], [False]])) 

586 assert m1.compare(m2) == {} 

587 

588 

589def test_compare_one_sided_planes() -> None: 

590 """Test that planes present in only one mask are counted fully.""" 

591 schema_a = MaskSchema([MaskPlane("A", "dA")], dtype=np.uint8) 

592 schema_b = MaskSchema([MaskPlane("B", "dB")], dtype=np.uint8) 

593 m1 = Mask(0, schema=schema_a, bbox=Box.factory[0:2, 0:1]) 

594 m1.set("A", np.array([[True], [False]])) 

595 m2 = Mask(0, schema=schema_b, bbox=Box.factory[0:2, 0:1]) 

596 m2.set("B", np.array([[True], [True]])) 

597 assert m1.compare(m2) == {"A": (1, 0), "B": (0, 2)} 

598 

599 

600def test_compare_bbox_mismatch_raises() -> None: 

601 """Test that differing bounding boxes raise ValueError.""" 

602 schema = MaskSchema([MaskPlane("A", "dA")], dtype=np.uint8) 

603 m1 = Mask(0, schema=schema, bbox=Box.factory[0:2, 0:1]) 

604 m2 = Mask(0, schema=schema, bbox=Box.factory[0:3, 0:1]) 

605 with pytest.raises(ValueError, match="same bounding box"): 

606 m1.compare(m2) 

607 

608 

609def test_compare_mask_to_legacy_notes_plane_diff(reset_afw_mask_planes: None) -> None: # noqa: F811 

610 """Assert that a legacy-mask mismatch adds a per-plane +/- note.""" 

611 schema = MaskSchema([MaskPlane("A", "dA")], dtype=np.uint8) 

612 mask = Mask(0, schema=schema, bbox=Box.factory[0:2, 0:1]) 

613 mask.set("A", np.array([[True], [False]])) 

614 legacy = mask.to_legacy() 

615 mask2 = Mask(0, schema=schema, bbox=Box.factory[0:2, 0:1]) 

616 mask2.set("A", np.array([[True], [True]])) 

617 with pytest.raises(AssertionError) as excinfo: 

618 compare_mask_to_legacy(mask2, legacy) 

619 notes = excinfo.value.__notes__ or [] 

620 assert any("mask[A]: +1 -0" in n for n in notes)