Coverage for tests/test_visit_image.py: 71%

798 statements  

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

14import dataclasses 

15import math 

16import os 

17import warnings 

18from pathlib import Path 

19from typing import Any, Literal 

20 

21import astropy.io.fits 

22import astropy.units as u 

23import astropy.wcs 

24import numpy as np 

25import pytest 

26from astro_metadata_translator import ObservationInfo 

27 

28from lsst.daf.butler import DimensionUniverse 

29from lsst.images import ( 

30 Background, 

31 BackgroundMap, 

32 Box, 

33 DetectorFrame, 

34 DifferenceImage, 

35 Image, 

36 MaskPlane, 

37 MaskSchema, 

38 ObservationSummaryStats, 

39 Polygon, 

40 SkyProjectionAstropyView, 

41 TractFrame, 

42 VisitImage, 

43 get_legacy_difference_image_mask_planes, 

44 get_legacy_visit_image_mask_planes, 

45 json, 

46) 

47from lsst.images import fits as images_fits 

48from lsst.images.aperture_corrections import ApertureCorrectionMap, aperture_corrections_to_legacy 

49from lsst.images.cameras import Detector 

50from lsst.images.describe import DescribableMixin, DescribeOptions, FieldRole, Report 

51from lsst.images.fields import ChebyshevField, SplineField, SumField, field_from_legacy_photo_calib 

52from lsst.images.fits import ExtensionKey, FitsOpaqueMetadata 

53from lsst.images.psfs import GaussianPointSpreadFunction, PointSpreadFunction 

54from lsst.images.serialization import ArchiveReadError, open_archive, read_archive 

55from lsst.images.tests import ( 

56 DP2_VISIT_DETECTOR_DATA_ID, 

57 RoundtripFits, 

58 RoundtripJson, 

59 RoundtripNdf, 

60 TemporaryButler, 

61 assert_masked_images_equal, 

62 assert_sky_projections_equal, 

63 assert_values_equal, 

64 assert_visit_images_equal, 

65 compare_aperture_corrections_to_legacy, 

66 compare_detector_to_legacy, 

67 compare_mask_to_legacy, 

68 compare_photo_calib_to_legacy, 

69 compare_visit_image_to_legacy, 

70 current_fixture_path, 

71 get_dp2_exposure_record, 

72 make_random_sky_projection, 

73 reset_afw_mask_planes, # noqa: F401 

74) 

75 

76try: 

77 import h5py # noqa: F401 

78 

79 HAVE_H5PY = True 

80except ImportError: 

81 HAVE_H5PY = False 

82 

83try: 

84 from lsst.afw.image import Exposure as LegacyExposure 

85 from lsst.afw.image import VisitInfo as LegacyVisitInfo 

86except ImportError: 

87 type LegacyExposure = Any # type: ignore[no-redef] 

88 type LegacyVisitInfo = Any # type: ignore[no-redef] 

89 

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

91LOCAL_DATA_DIR = os.path.join(os.path.dirname(__file__), "data") 

92FIXTURE_DIR = Path(__file__).parent / "data" / "schemas" 

93MINIMAL_VISIT_DATA_ID = {key: DP2_VISIT_DETECTOR_DATA_ID[key] for key in ("instrument", "visit", "detector")} 

94 

95skip_no_h5py = pytest.mark.skipif(not HAVE_H5PY, reason="h5py is not installed") 

96 

97 

98@pytest.fixture(scope="session") 

99def visit_image_components() -> dict[str, Any]: 

100 """Return a dictionary of VisitImage components.""" 

101 rng = np.random.default_rng(500) 

102 det_frame = DetectorFrame(instrument="Inst", visit=1234, detector=1, bbox=Box.factory[1:4096, 1:4096]) 

103 mask_schema = MaskSchema([MaskPlane("M1", "D1")]) 

104 # The IDs must match the DP2 exposure record that legacy round-trip tests 

105 # supply, because conversion validates the VisitInfo ID against it. 

106 obs_info = ObservationInfo( 

107 instrument="LSSTCam", 

108 detector_num=4, 

109 physical_filter="r1", 

110 exposure_id=2025052000177, 

111 observation_id="MC_O_20250520_000177", 

112 ) 

113 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4) 

114 gaussian_psf = GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13]) 

115 aperture_corrections: ApertureCorrectionMap = { 

116 "flux1": ChebyshevField(det_frame.bbox, np.array([0.75])), 

117 "flux2": ChebyshevField(det_frame.bbox, np.array([0.625])), 

118 } 

119 detector = read_archive(os.path.join(LOCAL_DATA_DIR, "detector.json"), Detector) 

120 # Real visit images have float pixels, and some operations (e.g. rendering 

121 # a photometric scaling) are only defined for floating-point images. 

122 image = Image(42.0, shape=(1024, 1024), unit=u.nJy, dtype=np.float32) 

123 variance = Image(5.0, shape=(1024, 1024), unit=u.nJy * u.nJy, dtype=np.float32) 

124 # polygon is the lower triangle of the image. 

125 polygon = Polygon(x_vertices=[-0.5, 1023.5, -0.5], y_vertices=[-0.5, -0.5, 1023.5]) 

126 sky_projection = make_random_sky_projection(rng, det_frame, det_frame.bbox) 

127 return { 

128 "mask_schema": mask_schema, 

129 "obs_info": obs_info, 

130 "summary_stats": summary_stats, 

131 "gaussian_psf": gaussian_psf, 

132 "aperture_corrections": aperture_corrections, 

133 "detector": detector, 

134 "image": image, 

135 "variance": variance, 

136 "polygon": polygon, 

137 "sky_projection": sky_projection, 

138 } 

139 

140 

141def make_visit_image(components: dict[str, Any]) -> VisitImage: 

142 """Construct a new VisitImage with most components populated.""" 

143 det_frame = components["sky_projection"].pixel_frame 

144 opaque = FitsOpaqueMetadata() 

145 hdr = astropy.io.fits.Header() 

146 with warnings.catch_warnings(): 

147 # Silence warnings about long keys becoming HIERARCH. 

148 warnings.simplefilter("ignore", category=astropy.io.fits.verify.VerifyWarning) 

149 hdr.update({"PLATFORM": "lsstcam", "LSST BUTLER ID": "123456789"}) 

150 opaque.extract_legacy_primary_header(hdr) 

151 # API signature suggests sky_projection and obs_info can be None but 

152 # they are required (unless you pass them in via the image plane). 

153 vi = VisitImage( 

154 components["image"], 

155 variance=components["variance"], 

156 psf=GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13]), 

157 mask_schema=components["mask_schema"], 

158 sky_projection=components["sky_projection"], 

159 obs_info=components["obs_info"], 

160 summary_stats=components["summary_stats"], 

161 detector=components["detector"], 

162 bounds=components["polygon"], 

163 aperture_corrections=components["aperture_corrections"], 

164 band="r", 

165 ) 

166 vi.backgrounds.add( 

167 "standard", 

168 ChebyshevField(det_frame.bbox, np.array([[2.0]])), 

169 description="Background subtracted from the image.", 

170 is_subtracted=True, 

171 ) 

172 vi._opaque_metadata = opaque 

173 return vi 

174 

175 

176def make_simplest_visit_image(components: dict[str, Any]) -> VisitImage: 

177 """Construct a VisitImage with the minimal set of components populated.""" 

178 return VisitImage( 

179 components["image"], 

180 psf=GaussianPointSpreadFunction(2.5, stamp_size=33, bounds=Box.factory[-10:10, -12:13]), 

181 mask_schema=components["mask_schema"], 

182 sky_projection=components["sky_projection"], 

183 detector=components["detector"], 

184 obs_info=components["obs_info"], 

185 band="r", 

186 ) 

187 

188 

189def _make_sum_background_visit_image(components: dict[str, Any], visit_image: VisitImage) -> VisitImage: 

190 """Return a VisitImage whose subtracted background is a SumField.""" 

191 rng = np.random.default_rng(42) 

192 bbox = visit_image.image.sky_projection.pixel_frame.bbox 

193 bin_y = bbox.y.linspace(6) 

194 bin_x = bbox.x.linspace(7) 

195 spline_a = SplineField( 

196 bbox, 

197 rng.standard_normal(size=(bin_y.size, bin_x.size)), 

198 y=bin_y, 

199 x=bin_x, 

200 ) 

201 spline_b = SplineField( 

202 bbox, 

203 rng.standard_normal(size=(bin_y.size, bin_x.size)), 

204 y=bin_y, 

205 x=bin_x, 

206 ) 

207 sum_field = SumField([spline_a, spline_b]) 

208 bg_map = BackgroundMap() 

209 bg_map.add( 

210 "stacked", 

211 sum_field, 

212 description="Two-operand SumField subtracted background.", 

213 is_subtracted=True, 

214 ) 

215 return VisitImage( 

216 components["image"], 

217 variance=components["variance"], 

218 psf=components["gaussian_psf"], 

219 mask_schema=components["mask_schema"], 

220 sky_projection=components["sky_projection"], 

221 obs_info=components["obs_info"], 

222 summary_stats=components["summary_stats"], 

223 detector=components["detector"], 

224 band="r", 

225 backgrounds=bg_map, 

226 ) 

227 

228 

229def _check_sum_background_round_trip(result: VisitImage, original: VisitImage) -> None: 

230 """Assert that a round-tripped SumField background matches the original.""" 

231 subtracted = result.backgrounds.subtracted 

232 assert subtracted is not None 

233 assert isinstance(subtracted.field, SumField) 

234 original_subtracted = original.backgrounds.subtracted 

235 assert original_subtracted is not None 

236 original_field = original_subtracted.field 

237 assert isinstance(original_field, SumField) 

238 round_field = subtracted.field 

239 assert isinstance(round_field, SumField) 

240 assert len(round_field.operands) == len(original_field.operands) 

241 for round_op, orig_op in zip(round_field.operands, original_field.operands, strict=True): 

242 assert round_op == orig_op 

243 

244 

245def test_visit_image_repr_str_pinned(visit_image_components: dict[str, Any]) -> None: 

246 """Pin the exact str and repr output of a VisitImage.""" 

247 visit = make_simplest_visit_image(visit_image_components) 

248 assert str(visit) == "VisitImage(Image([y=0:1024, x=0:1024], float32), ['M1'])" 

249 assert repr(visit) == ( 

250 "VisitImage(Image(..., bbox=Box(y=Interval(start=0, stop=1024), x=Interval(start=0, stop=1024))," 

251 " dtype=dtype('float32')), mask_schema=MaskSchema([MaskPlane(name='M1', description='D1')]," 

252 " dtype=dtype('uint8')))" 

253 ) 

254 

255 

256def test_basics(visit_image_components: dict[str, Any]) -> None: 

257 """Verify VisitImage constructor patterns and required-argument checks.""" 

258 c = visit_image_components 

259 # Test default fill of variance. 

260 visit = make_simplest_visit_image(c) 

261 assert visit.variance.array[0, 0] == 1.0 

262 assert visit[...] is not visit 

263 assert str(visit) == "VisitImage(Image([y=0:1024, x=0:1024], float32), ['M1'])" 

264 assert repr(visit) == ( 

265 "VisitImage(Image(..., bbox=Box(y=Interval(start=0, stop=1024), x=Interval(start=0, stop=1024))," 

266 " dtype=dtype('float32')), mask_schema=MaskSchema([MaskPlane(name='M1', description='D1')]," 

267 " dtype=dtype('uint8')))" 

268 ) 

269 

270 astropy_wcs = visit.astropy_wcs 

271 assert isinstance(astropy_wcs, SkyProjectionAstropyView) 

272 approx_wcs = visit.fits_wcs 

273 assert isinstance(approx_wcs, astropy.wcs.WCS) 

274 

275 with pytest.raises(TypeError): 

276 # Requires a PSF. 

277 VisitImage( 

278 c["image"], 

279 mask_schema=c["mask_schema"], 

280 sky_projection=c["sky_projection"], 

281 obs_info=c["obs_info"], 

282 detector=c["detector"], 

283 band="r", 

284 ) 

285 

286 with pytest.raises(TypeError): 

287 # Requires ObservationInfo. 

288 VisitImage( 

289 c["image"], 

290 psf=c["gaussian_psf"], 

291 mask_schema=c["mask_schema"], 

292 sky_projection=c["sky_projection"], 

293 detector=c["detector"], 

294 band="r", 

295 ) 

296 

297 with pytest.raises(TypeError): 

298 # Requires a sky_projection. 

299 VisitImage( 

300 c["image"], 

301 psf=c["gaussian_psf"], 

302 mask_schema=c["mask_schema"], 

303 obs_info=c["obs_info"], 

304 detector=c["detector"], 

305 band="r", 

306 ) 

307 

308 with pytest.raises(TypeError): 

309 # Requires a detector. 

310 VisitImage( 

311 c["image"], 

312 psf=c["gaussian_psf"], 

313 mask_schema=c["mask_schema"], 

314 sky_projection=c["sky_projection"], 

315 obs_info=c["obs_info"], 

316 band="r", 

317 ) 

318 

319 with pytest.raises(TypeError): 

320 # Requires some form of mask. 

321 VisitImage( 

322 c["image"], 

323 psf=c["gaussian_psf"], 

324 sky_projection=c["sky_projection"], 

325 obs_info=c["obs_info"], 

326 detector=c["detector"], 

327 band="r", 

328 ) 

329 

330 with pytest.raises(TypeError): 

331 VisitImage( 

332 Image(42, shape=(5, 5)), 

333 psf=c["gaussian_psf"], 

334 mask_schema=c["mask_schema"], 

335 sky_projection=c["sky_projection"], 

336 obs_info=c["obs_info"], 

337 detector=c["detector"], 

338 band="r", 

339 ) 

340 

341 # Requires a DetectorFrame. 

342 rng = np.random.default_rng(501) 

343 tract_frame = TractFrame(skymap="Skymap", tract=1, bbox=Box.factory[1:10, 1:10]) 

344 tract_proj = make_random_sky_projection(rng, tract_frame, Box.factory[1:4096, 1:4096]) 

345 with pytest.raises(TypeError): 

346 VisitImage( 

347 c["image"], 

348 sky_projection=tract_proj, 

349 psf=c["gaussian_psf"], 

350 mask_schema=c["mask_schema"], 

351 obs_info=c["obs_info"], 

352 detector=c["detector"], 

353 band="r", 

354 ) 

355 

356 # Variance unit mismatch. 

357 with pytest.raises(ValueError, match="should be the square of the image unit"): 

358 VisitImage( 

359 c["image"], 

360 variance=c["image"], 

361 psf=c["gaussian_psf"], 

362 mask_schema=c["mask_schema"], 

363 sky_projection=c["sky_projection"], 

364 obs_info=c["obs_info"], 

365 detector=c["detector"], 

366 band="r", 

367 ) 

368 

369 

370def test_copy_and_slice(visit_image_components: dict[str, Any]) -> None: 

371 """Verify that copy deep-copies arrays and components while slice shares 

372 them. 

373 """ 

374 c = visit_image_components 

375 visit_image = make_visit_image(c) 

376 copy = visit_image.copy() 

377 copy.image.array[0, 0] = 30.0 

378 assert visit_image.image.array[0, 0] == 42.0 

379 assert copy.image.array[0, 0] == 30.0 

380 subvisit = visit_image[Box.factory[0:5, 0:5]] 

381 # Check summary stats. 

382 assert copy.summary_stats == visit_image.summary_stats 

383 assert copy.summary_stats is not visit_image.summary_stats 

384 assert subvisit.summary_stats == visit_image.summary_stats 

385 assert subvisit.summary_stats is visit_image.summary_stats 

386 # Check aperture corrections. 

387 assert copy.aperture_corrections.keys() == visit_image.aperture_corrections.keys() 

388 assert copy.aperture_corrections is not visit_image.aperture_corrections 

389 assert subvisit.aperture_corrections.keys() == visit_image.aperture_corrections.keys() 

390 assert subvisit.aperture_corrections is visit_image.aperture_corrections 

391 # Check backgrounds. 

392 assert copy.backgrounds.keys() == visit_image.backgrounds.keys() 

393 assert copy.backgrounds is not visit_image.backgrounds 

394 assert subvisit.backgrounds.keys() == visit_image.backgrounds.keys() 

395 assert subvisit.backgrounds is visit_image.backgrounds 

396 # Check bounds. 

397 assert copy.bounds is c["polygon"] 

398 assert subvisit.bounds == subvisit.bbox # original polygon wholly encloses subvisit.bbox 

399 

400 

401def test_obs_info(visit_image_components: dict[str, Any]) -> None: 

402 """Verify that ObservationInfo is present and carries the expected 

403 instrument. 

404 """ 

405 visit_image = make_visit_image(visit_image_components) 

406 assert visit_image.obs_info is not None 

407 assert visit_image.obs_info.instrument == "LSSTCam" 

408 

409 

410def test_summary_stats(visit_image_components: dict[str, Any]) -> None: 

411 """Verify ObservationSummaryStats equality and inequality comparisons.""" 

412 summary_stats = visit_image_components["summary_stats"] 

413 assert summary_stats == ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4) 

414 assert summary_stats != ObservationSummaryStats(psfSigma=2.5) 

415 assert summary_stats != ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2)) 

416 

417 

418def test_summary_stats_to_legacy() -> None: 

419 """Verify ObservationSummaryStats round-trips through the legacy 

420 ExposureSummaryStats even when this package defines fields that the 

421 installed afw does not. 

422 """ 

423 try: 

424 from lsst.afw.image import ExposureSummaryStats 

425 except ImportError: 

426 pytest.skip("lsst.afw.image is not available") 

427 

428 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4) 

429 legacy = summary_stats.to_legacy() 

430 assert isinstance(legacy, ExposureSummaryStats) 

431 assert legacy.psfSigma == 2.5 

432 assert legacy.zeroPoint == 31.4 

433 # Empty (NaN) fields unknown to the legacy struct are dropped, so the 

434 # round trip reproduces the original. 

435 assert ObservationSummaryStats.from_legacy(legacy) == summary_stats 

436 

437 # A real value in a field unknown to the legacy struct cannot be 

438 # represented and must raise rather than be silently dropped. 

439 legacy_fields = {field.name for field in dataclasses.fields(ExposureSummaryStats)} 

440 extra_float_fields = [ 

441 name 

442 for name, info in ObservationSummaryStats.model_fields.items() 

443 if name not in legacy_fields and info.annotation is float 

444 ] 

445 if extra_float_fields: 445 ↛ 446line 445 didn't jump to line 446 because the condition on line 445 was never true

446 with pytest.raises(ValueError, match="not supported by this version"): 

447 ObservationSummaryStats(**{extra_float_fields[0]: 1.0}).to_legacy() 

448 

449 

450def test_summary_stats_from_legacy_unknown_field() -> None: 

451 """Verify from_legacy drops empty unknown fields but errors on set ones.""" 

452 

453 @dataclasses.dataclass 

454 class FakeLegacy: 

455 psfSigma: float = 2.5 

456 notARealField: float = math.nan 

457 

458 # An unknown field that is empty is dropped. 

459 stats = ObservationSummaryStats.from_legacy(FakeLegacy()) 

460 assert stats.psfSigma == 2.5 

461 

462 # An unknown field that holds a real value cannot be represented. 

463 with pytest.raises(ValueError, match="is not known to ObservationSummaryStats"): 

464 ObservationSummaryStats.from_legacy(FakeLegacy(notARealField=1.0)) 

465 

466 

467def test_summary_stats_legacy_butler_read(reset_afw_mask_planes: None) -> None: # noqa: F811 

468 """Test that a round-tripped ObservationSummaryStats can be read back as 

469 a legacy afw ExposureSummaryStats via the converter registered on the 

470 ExposureSummaryStats storage class. 

471 """ 

472 from lsst.afw.image import ExposureSummaryStats as LegacyExposureSummaryStats 

473 

474 summary_stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4) 

475 with RoundtripFits(summary_stats, storage_class="ObservationSummaryStats") as roundtrip: 

476 legacy_stats = roundtrip.get(storageClass="ExposureSummaryStats") 

477 assert isinstance(legacy_stats, LegacyExposureSummaryStats) 

478 assert legacy_stats.psfSigma == 2.5 

479 assert legacy_stats.zeroPoint == 31.4 

480 assert ObservationSummaryStats.from_legacy(legacy_stats) == summary_stats 

481 

482 

483def test_detector_legacy_butler_read(reset_afw_mask_planes: None) -> None: # noqa: F811 

484 """Test that a round-tripped Detector (written under the DetectorV2 

485 storage class) can be read back as a legacy lsst.afw.cameraGeom.Detector 

486 via the converter registered on the Detector storage class. 

487 """ 

488 from lsst.afw.cameraGeom import Detector as LegacyDetector 

489 

490 detector = read_archive(os.path.join(LOCAL_DATA_DIR, "detector.json"), Detector) 

491 with RoundtripFits(detector, storage_class="DetectorV2") as roundtrip: 

492 legacy_detector = roundtrip.get(storageClass="Detector") 

493 assert isinstance(legacy_detector, LegacyDetector) 

494 compare_detector_to_legacy(detector, legacy_detector, is_raw_assembled=True) 

495 

496 

497def test_repeated_metadata_keys_legacy_round_trip( 

498 visit_image_components: dict[str, Any], 

499 reset_afw_mask_planes: None, # noqa: F811 

500) -> None: 

501 """Verify that a header key set more than once keeps all of its values 

502 when a VisitImage is converted to a legacy Exposure and back. 

503 """ 

504 from lsst.afw.detection import GaussianPsf 

505 

506 opaque_metadata = FitsOpaqueMetadata() 

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

508 header.append(("PLATFORM", "lsstcam"), end=True) 

509 # SubtractBackgroundTask writes one BGMEAN card per background fit. 

510 header.append(("BGMEAN", 1.5), end=True) 

511 header.append(("BGMEAN", 2.5), end=True) 

512 opaque_metadata.extract_legacy_primary_header(header) 

513 visit_image = VisitImage( 

514 visit_image_components["image"], 

515 variance=visit_image_components["variance"], 

516 # A legacy-backed PSF, so that to_legacy can attach it to the 

517 # Exposure and from_legacy can read it back. 

518 psf=PointSpreadFunction.from_legacy(GaussianPsf(33, 33, 2.5), bounds=Box.factory[0:1024, 0:1024]), 

519 mask_schema=visit_image_components["mask_schema"], 

520 sky_projection=visit_image_components["sky_projection"], 

521 detector=visit_image_components["detector"], 

522 obs_info=visit_image_components["obs_info"], 

523 band="r", 

524 ) 

525 visit_image._opaque_metadata = opaque_metadata 

526 

527 legacy_exposure = visit_image.to_legacy() 

528 legacy_metadata = legacy_exposure.getMetadata() 

529 assert legacy_metadata.getArray("BGMEAN") == [1.5, 2.5] 

530 assert legacy_metadata["PLATFORM"] == "lsstcam" 

531 # Add a random key directly 

532 legacy_metadata.add("DUMMYVAR", 0.5) 

533 legacy_metadata.add("DUMMYVAR", 1.5) 

534 

535 round_tripped = VisitImage.from_legacy( 

536 legacy_exposure, 

537 exposure_record=get_dp2_exposure_record(DimensionUniverse()), 

538 ) 

539 round_tripped_header = round_tripped._opaque_metadata.headers[ExtensionKey()] 

540 assert "DUMMYVAR" in round_tripped_header 

541 assert "BGMEAN" in round_tripped_header 

542 assert round_tripped_header["PLATFORM"] == "lsstcam" 

543 assert [card.value for card in round_tripped_header.cards if card.keyword == "BGMEAN"] == [1.5, 2.5] 

544 assert [card.value for card in round_tripped_header.cards if card.keyword == "DUMMYVAR"] == [0.5, 1.5] 

545 

546 

547def test_external_metadata_legacy_round_trip( 

548 visit_image_components: dict[str, Any], 

549 reset_afw_mask_planes: None, # noqa: F811 

550 tmp_path: Path, 

551) -> None: 

552 """Verify that native metadata is written to a legacy file as 

553 ``LSST IMAGES`` cards and is not visible as external metadata when the 

554 file is read back. 

555 """ 

556 from lsst.afw.detection import GaussianPsf 

557 

558 opaque_metadata = FitsOpaqueMetadata() 

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

560 header.append(("PLATFORM", "lsstcam"), end=True) 

561 header.append(("BGMEAN", 1.5), end=True) 

562 header.append(("BGMEAN", 2.5), end=True) 

563 # An external ID card must neither block nor replace the native "id". 

564 header.append(("ID", 99), end=True) 

565 opaque_metadata.extract_legacy_primary_header(header) 

566 visit_image = VisitImage( 

567 visit_image_components["image"], 

568 variance=visit_image_components["variance"], 

569 psf=PointSpreadFunction.from_legacy(GaussianPsf(33, 33, 2.5), bounds=Box.factory[0:1024, 0:1024]), 

570 mask_schema=visit_image_components["mask_schema"], 

571 sky_projection=visit_image_components["sky_projection"], 

572 detector=visit_image_components["detector"], 

573 obs_info=visit_image_components["obs_info"], 

574 band="r", 

575 metadata={"native_key": 7, "MixedCase": "yes"}, 

576 ) 

577 visit_image._opaque_metadata = opaque_metadata 

578 path = tmp_path / "legacy.fits" 

579 visit_image.to_legacy().writeFits(str(path)) 

580 

581 with astropy.io.fits.open(path) as hdu_list: 

582 primary = hdu_list[0].header 

583 native_cards = {} 

584 n = 1 

585 while f"LSST IMAGES KEY {n}" in primary: 

586 native_cards[primary[f"LSST IMAGES KEY {n}"]] = primary[f"LSST IMAGES VALUE {n}"] 

587 n += 1 

588 assert native_cards == {"native_key": 7, "MixedCase": "yes"} 

589 assert [card.value for card in primary.cards if card.keyword == "BGMEAN"] == [1.5, 2.5] 

590 

591 result = VisitImage.read_legacy( 

592 str(path), 

593 exposure_record=get_dp2_exposure_record(DimensionUniverse()), 

594 ) 

595 assert result.metadata.native["native_key"] == 7 

596 assert result.metadata.native["MixedCase"] == "yes" 

597 assert "id" in result.metadata.native 

598 assert not [key for key in result.metadata.external if key.startswith("LSST IMAGES")] 

599 assert result.metadata.external.get_all("BGMEAN") == (1.5, 2.5) 

600 assert result.metadata["platform"] == "lsstcam" 

601 assert result.metadata.external["ID"] == 99 

602 

603 # Converting the result back must not duplicate the external cards. 

604 legacy_metadata = result.to_legacy().getMetadata() 

605 assert legacy_metadata.getArray("BGMEAN") == [1.5, 2.5] 

606 assert legacy_metadata["LSST IMAGES KEY 1"] == "native_key" 

607 

608 

609def test_gaussian_psf_legacy_round_trip( 

610 visit_image_components: dict[str, Any], 

611 reset_afw_mask_planes: None, # noqa: F811 

612) -> None: 

613 """Verify that a Gaussian PSF set after construction is attached to the 

614 legacy Exposure that `VisitImage.to_legacy` returns. 

615 """ 

616 from lsst.afw.detection import GaussianPsf 

617 

618 visit_image = make_simplest_visit_image(visit_image_components) 

619 psf = GaussianPointSpreadFunction(3.25, stamp_size=27, bounds=Box.factory[0:1024, 0:1024]) 

620 visit_image.psf = psf 

621 assert visit_image.psf is psf 

622 

623 legacy_psf = visit_image.to_legacy().getPsf() 

624 assert isinstance(legacy_psf, GaussianPsf) 

625 assert legacy_psf.getSigma() == psf.sigma 

626 assert legacy_psf.computeBBox(legacy_psf.getAveragePosition()).getWidth() == 27 

627 

628 

629@skip_no_h5py 

630def test_round_trip_ndf(visit_image_components: dict[str, Any]) -> None: 

631 """Verify NDF round-trip produces a VisitImage equal to the original.""" 

632 visit_image = make_visit_image(visit_image_components) 

633 with RoundtripNdf(visit_image, "VisitImage") as roundtrip: 

634 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False) 

635 

636 

637@skip_no_h5py 

638def test_fits_ndf_consistency(visit_image_components: dict[str, Any]) -> None: 

639 """Verify FITS and NDF backends produce equal VisitImages on round-trip.""" 

640 visit_image = make_visit_image(visit_image_components) 

641 with RoundtripFits(visit_image) as fits_rt, RoundtripNdf(visit_image) as ndf_rt: 

642 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False) 

643 assert_visit_images_equal(visit_image, ndf_rt.result, expect_view=False) 

644 assert_visit_images_equal(fits_rt.result, ndf_rt.result, expect_view=False) 

645 

646 

647def test_fits_json_consistency(visit_image_components: dict[str, Any]) -> None: 

648 """Verify FITS and JSON backends produce equal VisitImages.""" 

649 visit_image = make_visit_image(visit_image_components) 

650 with ( 

651 RoundtripFits(visit_image) as fits_rt, 

652 RoundtripJson(visit_image) as json_rt, 

653 ): 

654 assert_visit_images_equal(visit_image, fits_rt.result, expect_view=False) 

655 assert_visit_images_equal(visit_image, json_rt.result, expect_view=False) 

656 assert_visit_images_equal(fits_rt.result, json_rt.result, expect_view=False) 

657 

658 

659def test_serialized_schema_version(visit_image_components: dict[str, Any]) -> None: 

660 """The written tree's schema_version matches its schema_url. 

661 

662 The image, mask, and variance subtrees are stamped with their own 

663 versions too. 

664 """ 

665 visit_image = make_visit_image(visit_image_components) 

666 tree = json.write(visit_image) 

667 tree_type = type(tree) 

668 assert tree.schema_version == tree_type.SCHEMA_VERSION 

669 assert tree.min_read_version == tree_type.MIN_READ_VERSION 

670 assert tree.schema_url.endswith(f"-{tree.schema_version}") 

671 assert tree.model_dump()["schema_version"] == tree_type.SCHEMA_VERSION 

672 for name in ("image", "mask", "variance"): 

673 component = getattr(tree, name) 

674 assert component.schema_version == type(component).SCHEMA_VERSION, name 

675 

676 

677def test_read_write(visit_image_components: dict[str, Any]) -> None: 

678 """Verify a VisitImage round-trips through FITS with correct compression. 

679 

680 Checks compression headers, subimage reads, equality, and opaque 

681 metadata. Contains only butler-free assertions; component reads live 

682 in `test_read_write_components`. 

683 """ 

684 visit_image = make_visit_image(visit_image_components) 

685 with RoundtripFits(visit_image, "VisitImage") as roundtrip: 

686 # Check that we're still using the right compression, and that we 

687 # wrote WCSs. 

688 fits = roundtrip.inspect() 

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

690 assert fits[1].header["CTYPE1"] == "RA---TAN" 

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

692 assert fits[2].header["CTYPE1"] == "RA---TAN" 

693 assert fits[3].header["ZCMPTYPE"] == "GZIP_2" 

694 assert fits[3].header["CTYPE1"] == "RA---TAN" 

695 # Check a subimage read (no component arg — does not trigger a skip). 

696 subbox = Box.factory[8:13, 9:30] 

697 subimage = roundtrip.get(bbox=subbox) 

698 assert_masked_images_equal(subimage, visit_image[subbox], expect_view=False) 

699 

700 assert_visit_images_equal(roundtrip.result, visit_image, expect_view=False) 

701 # Check that the round-tripped headers are the same (up to card order). 

702 assert len(roundtrip.result._opaque_metadata.headers[ExtensionKey()]) == 1 

703 assert dict(visit_image._opaque_metadata.headers[ExtensionKey()]) == dict( 

704 roundtrip.result._opaque_metadata.headers[ExtensionKey()] 

705 ) 

706 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")] 

707 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")] 

708 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")] 

709 # Spot-check the concrete background contents (names, field types, 

710 # subtracted entry) against the known fixture, so the equality check 

711 # above is not vacuously satisfied by empty background maps. 

712 assert isinstance(roundtrip.result.backgrounds, BackgroundMap) 

713 assert roundtrip.result.backgrounds.keys() == {"standard"} 

714 assert isinstance(roundtrip.result.backgrounds["standard"].field, ChebyshevField) 

715 assert roundtrip.result.backgrounds.subtracted.name == "standard" 

716 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image." 

717 

718 

719def test_read_bounds_and_unit_components(visit_image_components: dict[str, Any], tmp_path: Path) -> None: 

720 """Verify that the bounds and unit components match a full read, both 

721 when the bounds are a polygon and when they default to the bbox. 

722 """ 

723 for n, visit_image in enumerate( 

724 [make_visit_image(visit_image_components), make_simplest_visit_image(visit_image_components)] 

725 ): 

726 path = tmp_path / f"visit_image_{n}.fits" 

727 images_fits.write(visit_image, path) 

728 with open_archive(path) as reader: 

729 assert reader.get_component("bounds") == visit_image.bounds 

730 assert reader.get_component("unit") == visit_image.unit 

731 assert isinstance(make_visit_image(visit_image_components).bounds, Polygon) 

732 

733 

734def test_read_write_components(visit_image_components: dict[str, Any]) -> None: 

735 """Verify component reads and storage-class overrides round-trip correctly. 

736 

737 Requires a butler; skips when `lsst.daf.butler` is absent. 

738 Butler-free assertions live in `test_read_write`. 

739 """ 

740 c = visit_image_components 

741 visit_image = make_visit_image(c) 

742 with RoundtripFits(visit_image, "VisitImage") as roundtrip: 

743 subbox = Box.factory[8:13, 9:30] 

744 subimage = roundtrip.get(bbox=subbox) 

745 

746 # Get an explicit masked image to compare with the subimage. 

747 subimage_masked = roundtrip.get("masked_image", bbox=subbox) 

748 assert_masked_images_equal(subimage_masked, subimage, expect_view=False) 

749 

750 # Get the same masked image in a multi-component get and ensure 

751 # it is the same thing. 

752 components = roundtrip.get("components", components=["masked_image", "psf"], bbox=subbox) 

753 assert set(components) == {"masked_image", "psf"} 

754 assert_masked_images_equal(components["masked_image"], subimage_masked, expect_view=False) 

755 

756 assert roundtrip.get("bbox") == visit_image.bbox 

757 assert roundtrip.get("bounds") == visit_image.bounds 

758 assert roundtrip.get("unit") == visit_image.unit 

759 

760 obs_info = roundtrip.get("obs_info") 

761 assert isinstance(obs_info, ObservationInfo) 

762 assert obs_info == visit_image.obs_info 

763 

764 summary_stats = roundtrip.get("summary_stats") 

765 assert isinstance(summary_stats, ObservationSummaryStats) 

766 assert summary_stats == visit_image.summary_stats 

767 

768 psf = roundtrip.get("psf") 

769 assert isinstance(psf, GaussianPointSpreadFunction) 

770 assert psf.kernel_bbox == c["gaussian_psf"].kernel_bbox 

771 

772 backgrounds = roundtrip.get("backgrounds") 

773 assert isinstance(backgrounds, BackgroundMap) 

774 assert backgrounds.keys() == {"standard"} 

775 assert isinstance(backgrounds["standard"].field, ChebyshevField) 

776 assert backgrounds.subtracted.name == "standard" 

777 assert roundtrip.result.backgrounds.subtracted.description == "Background subtracted from the image." 

778 

779 # Test some components get edge cases. 

780 components = roundtrip.get("components", components="image") 

781 assert isinstance(components["image"], Image) 

782 

783 components = roundtrip.get("components") 

784 assert set(components) == { 

785 "image", 

786 "variance", 

787 "psf", 

788 "bbox", 

789 "mask", 

790 "obs_info", 

791 "backgrounds", 

792 "detector", 

793 "aperture_corrections", 

794 "sky_projection", 

795 "summary_stats", 

796 "photometric_scaling", 

797 "bounds", 

798 "unit", 

799 } 

800 

801 # Butler morphs RuntimeError to ValueError. 

802 with pytest.raises(ValueError, match="component nonexistent"): 

803 roundtrip.get("components", components=["image", "nonexistent"]) 

804 

805 with pytest.raises(ValueError, match="should not be specified"): 

806 roundtrip.get("components", components=["image", "components"]) 

807 

808 with pytest.raises(ValueError, match="empty request"): 

809 roundtrip.get("components", components=[]) 

810 

811 with pytest.raises(ValueError, match="not known to any of the requested components"): 

812 # PSF does not know how to use bbox so this fails. 

813 roundtrip.get("components", components="psf", bbox=subbox) 

814 

815 

816def test_sum_background_round_trip_fits(visit_image_components: dict[str, Any]) -> None: 

817 """Verify FITS backend keeps two same-named SumField operands as distinct 

818 EXTVERs. 

819 """ 

820 visit_image = make_visit_image(visit_image_components) 

821 visit = _make_sum_background_visit_image(visit_image_components, visit_image) 

822 with RoundtripFits(visit) as roundtrip: 

823 _check_sum_background_round_trip(roundtrip.result, visit) 

824 

825 

826@skip_no_h5py 

827def test_sum_background_round_trip_ndf(visit_image_components: dict[str, Any]) -> None: 

828 """Verify NDF backend disambiguates the repeated ``data`` leaf, just as 

829 the FITS backend does. 

830 """ 

831 visit_image = make_visit_image(visit_image_components) 

832 visit = _make_sum_background_visit_image(visit_image_components, visit_image) 

833 with RoundtripNdf(visit) as roundtrip: 

834 _check_sum_background_round_trip(roundtrip.result, visit) 

835 

836 

837@pytest.mark.parametrize( 

838 ("scaling_unit", "operation"), 

839 [(u.electron / u.nJy, "multiply"), (u.nJy / u.electron, "divide")], 

840 ids=["multiply", "divide"], 

841) 

842def test_convert_unit_subimage( 

843 visit_image_components: dict[str, Any], 

844 scaling_unit: u.UnitBase, 

845 operation: Literal["multiply", "divide"], 

846) -> None: 

847 """Verify that converting the units of a subimage applies only the portion 

848 of the photometric scaling that overlaps the subimage. 

849 

850 A photometric scaling keeps the bounds it was modeled over when the image 

851 is subset, so both branches of the conversion must render it over the 

852 subimage's bbox rather than over its own bounds. 

853 """ 

854 visit_image = make_visit_image(visit_image_components) 

855 scaling = ChebyshevField( 

856 visit_image.bbox, 

857 np.array([[4.0, 0.5, 0.125], [0.25, 0.0625, 0.0], [0.03125, 0.0, 0.0]]), 

858 unit=scaling_unit, 

859 ) 

860 visit_image.photometric_scaling = scaling 

861 # Trim a different number of pixels from each side so a scaling rendered 

862 # over the wrong box cannot match by chance. 

863 subbox = Box.factory[ 

864 visit_image.bbox.y.start + 7 : visit_image.bbox.y.stop - 13, 

865 visit_image.bbox.x.start + 3 : visit_image.bbox.x.stop - 21, 

866 ] 

867 subimage = visit_image[subbox] 

868 assert subimage.photometric_scaling.bounds.bbox == visit_image.bbox 

869 

870 converted = subimage.convert_unit(u.electron) 

871 

872 assert converted.unit == u.electron 

873 assert converted.image.bbox == subbox 

874 scaling_array = scaling.render(subbox, dtype=subimage.image.array.dtype).array 

875 if operation == "divide": 

876 scaling_array = 1.0 / scaling_array 

877 assert_values_equal(converted.image.array, subimage.image.array * scaling_array, rtol=1e-5) 

878 assert_values_equal(converted.variance.array, subimage.variance.array * scaling_array**2, rtol=1e-5) 

879 

880 

881@dataclasses.dataclass 

882class _LegacyTestData: 

883 filename: str 

884 plane_map: dict[str, MaskPlane] = dataclasses.field(default_factory=get_legacy_visit_image_mask_planes) 

885 unit: u.Unit = u.nJy 

886 storage_class: str = "VisitImage" 

887 read_cls: type[VisitImage] = VisitImage 

888 exposure_record: Any = None 

889 legacy_exposure: LegacyExposure = dataclasses.field(init=False) 

890 

891 @classmethod 

892 def get( 

893 cls, which: Literal["visit_image", "preliminary_visit_image", "difference_image"] 

894 ) -> _LegacyTestData: 

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

896 pytest.skip("TESTDATA_IMAGES is not set up.") 

897 result = cls( 

898 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", f"{which}.fits"), 

899 ) 

900 match which: 

901 case "preliminary_visit_image": 

902 result.unit = u.electron 

903 case "difference_image": 

904 result.storage_class = "DifferenceImage" 

905 result.read_cls = DifferenceImage 

906 result.plane_map = get_legacy_difference_image_mask_planes() 

907 case "visit_image": 

908 pass 

909 try: 

910 from lsst.afw.image import ExposureFitsReader 

911 

912 result.legacy_exposure = ExposureFitsReader(result.filename).read() 

913 except ImportError: 

914 pytest.skip("lsst.afw.image is not available; cannot read legacy exposures") 

915 # All three legacy files are the same DP2 exposure. 

916 result.exposure_record = get_dp2_exposure_record(DimensionUniverse()) 

917 result.visit_image = result.read_cls.read_legacy( 

918 result.filename, 

919 preserve_quantization=True, 

920 plane_map=result.plane_map, 

921 exposure_record=result.exposure_record, 

922 ) 

923 return result 

924 

925 

926@pytest.fixture(params=["visit_image", "preliminary_visit_image", "difference_image"]) 

927def legacy_test_data(request: pytest.FixtureRequest, reset_afw_mask_planes: None) -> _LegacyTestData: # noqa: F811 

928 """Return legacy test data. 

929 

930 Tests that depend on this parameterized fixture run on all of the legacy 

931 test images. 

932 """ 

933 return _LegacyTestData.get(request.param) 

934 

935 

936@pytest.fixture(params=["visit_image", "difference_image"]) 

937def legacy_test_data_calibrated( 

938 request: pytest.FixtureRequest, 

939 reset_afw_mask_planes: None, # noqa: F811 

940) -> _LegacyTestData: 

941 """Return legacy test data for calibrated images only. 

942 

943 Tests that depend on this parameterized fixture do not run on 

944 preliminary_visit_image, since that has 'electron' pixel units 

945 """ 

946 return _LegacyTestData.get(request.param) 

947 

948 

949def _check_legacy_obs_info(obs_info: ObservationInfo | None) -> None: 

950 """Assert obs_info carries expected LSSTCam/DP2 field values.""" 

951 assert isinstance(obs_info, ObservationInfo) 

952 assert obs_info.instrument == "LSSTCam" 

953 assert obs_info.detector_num == 85, obs_info 

954 assert obs_info.detector_unique_name == "R21_S11", obs_info 

955 assert obs_info.physical_filter == "r_57", obs_info 

956 

957 

958def test_legacy_optional_mask_planes(legacy_test_data_calibrated: _LegacyTestData) -> None: 

959 """Verify that an optional source injection plane converts in both 

960 directions. 

961 """ 

962 legacy = legacy_test_data_calibrated.legacy_exposure.clone() 

963 injected = np.zeros(legacy.mask.array.shape, dtype=bool) 

964 injected[4:9, 3:11] = True 

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

966 legacy.mask.addMaskPlane(old_name) 

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

968 plane_map = legacy_test_data_calibrated.plane_map 

969 

970 image = legacy_test_data_calibrated.read_cls.from_legacy( 

971 legacy, plane_map=plane_map, exposure_record=legacy_test_data_calibrated.exposure_record 

972 ) 

973 

974 assert "INJECTED" in image.mask.schema.names 

975 assert_values_equal(image.mask.get("INJECTED"), injected) 

976 assert "INJECTED_CORE" not in image.mask.schema.names 

977 # Every plane the image really uses, mapped and optional alike. 

978 compare_mask_to_legacy(image.mask, legacy.mask, plane_map) 

979 

980 round_tripped = image.to_legacy() 

981 

982 assert "INJECTED" in round_tripped.mask.getMaskPlaneDict() 

983 assert_values_equal( 

984 (round_tripped.mask.array & round_tripped.mask.getPlaneBitMask("INJECTED")).astype(bool), 

985 injected, 

986 ) 

987 compare_mask_to_legacy(image.mask, round_tripped.mask, plane_map) 

988 

989 

990def test_legacy_errors(legacy_test_data: _LegacyTestData) -> None: 

991 """Verify that from_legacy and read_legacy raise on bad arguments.""" 

992 with pytest.raises(TypeError): 

993 # exposure_record is a required argument. 

994 VisitImage.from_legacy(legacy_test_data.legacy_exposure) 

995 with pytest.raises(ValueError, match="BUNIT value .* disagrees with given unit"): 

996 VisitImage.from_legacy( 

997 legacy_test_data.legacy_exposure, 

998 unit=u.mJy, 

999 exposure_record=legacy_test_data.exposure_record, 

1000 ) 

1001 visit = VisitImage.from_legacy( 

1002 legacy_test_data.legacy_exposure, 

1003 unit=legacy_test_data.unit, 

1004 exposure_record=legacy_test_data.exposure_record, 

1005 ) 

1006 assert visit.unit == legacy_test_data.unit 

1007 

1008 with pytest.raises(TypeError): 

1009 legacy_test_data.read_cls.read_legacy(legacy_test_data.filename) 

1010 

1011 

1012def test_component_reads(legacy_test_data: _LegacyTestData) -> None: 

1013 """Verify that individual components can be read from a legacy FITS 

1014 file. 

1015 """ 

1016 visit = VisitImage.read_legacy( 

1017 legacy_test_data.filename, exposure_record=legacy_test_data.exposure_record 

1018 ) 

1019 proj = VisitImage.read_legacy( 

1020 legacy_test_data.filename, 

1021 exposure_record=legacy_test_data.exposure_record, 

1022 component="sky_projection", 

1023 ) 

1024 assert_sky_projections_equal(proj, visit.sky_projection, expect_identity=False) 

1025 image = VisitImage.read_legacy( 

1026 legacy_test_data.filename, 

1027 exposure_record=legacy_test_data.exposure_record, 

1028 component="image", 

1029 ) 

1030 assert image == visit.image 

1031 assert_sky_projections_equal(proj, image.sky_projection, expect_identity=False) 

1032 variance = VisitImage.read_legacy( 

1033 legacy_test_data.filename, 

1034 exposure_record=legacy_test_data.exposure_record, 

1035 component="variance", 

1036 ) 

1037 assert variance == visit.variance 

1038 assert_sky_projections_equal(proj, variance.sky_projection, expect_identity=False) 

1039 mask = VisitImage.read_legacy( 

1040 legacy_test_data.filename, 

1041 exposure_record=legacy_test_data.exposure_record, 

1042 component="mask", 

1043 ) 

1044 assert mask == visit.mask 

1045 assert_sky_projections_equal(proj, mask.sky_projection, expect_identity=False) 

1046 psf = VisitImage.read_legacy( 

1047 legacy_test_data.filename, 

1048 exposure_record=legacy_test_data.exposure_record, 

1049 component="psf", 

1050 ) 

1051 assert isinstance(psf, PointSpreadFunction) 

1052 obs_info = VisitImage.read_legacy( 

1053 legacy_test_data.filename, 

1054 exposure_record=legacy_test_data.exposure_record, 

1055 component="obs_info", 

1056 ) 

1057 _check_legacy_obs_info(obs_info) 

1058 summary_stats = VisitImage.read_legacy( 

1059 legacy_test_data.filename, 

1060 exposure_record=legacy_test_data.exposure_record, 

1061 component="summary_stats", 

1062 ) 

1063 assert isinstance(summary_stats, ObservationSummaryStats) 

1064 assert summary_stats.nPsfStar == legacy_test_data.legacy_exposure.info.getSummaryStats().nPsfStar 

1065 compare_aperture_corrections_to_legacy( 

1066 VisitImage.read_legacy( 

1067 legacy_test_data.filename, 

1068 exposure_record=legacy_test_data.exposure_record, 

1069 component="aperture_corrections", 

1070 ), 

1071 legacy_test_data.legacy_exposure.info.getApCorrMap(), 

1072 visit.bbox, 

1073 ) 

1074 detector = VisitImage.read_legacy( 

1075 legacy_test_data.filename, 

1076 exposure_record=legacy_test_data.exposure_record, 

1077 component="detector", 

1078 ) 

1079 compare_detector_to_legacy( 

1080 detector, legacy_test_data.legacy_exposure.getDetector(), is_raw_assembled=True 

1081 ) 

1082 photometric_scaling = VisitImage.read_legacy( 

1083 legacy_test_data.filename, 

1084 exposure_record=legacy_test_data.exposure_record, 

1085 component="photometric_scaling", 

1086 ) 

1087 compare_photo_calib_to_legacy( 

1088 photometric_scaling, 

1089 legacy_test_data.legacy_exposure.getPhotoCalib(), 

1090 subimage_bbox=visit.bbox, 

1091 ) 

1092 bounds = VisitImage.read_legacy( 

1093 legacy_test_data.filename, 

1094 exposure_record=legacy_test_data.exposure_record, 

1095 component="bounds", 

1096 ) 

1097 assert bounds == visit.bounds 

1098 unit = VisitImage.read_legacy( 

1099 legacy_test_data.filename, 

1100 exposure_record=legacy_test_data.exposure_record, 

1101 component="unit", 

1102 ) 

1103 assert unit == visit.unit 

1104 

1105 

1106def test_legacy_obs_info(legacy_test_data: _LegacyTestData) -> None: 

1107 """Verify that ObservationInfo is constructed correctly from a legacy 

1108 exposure. 

1109 """ 

1110 legacy = VisitImage.from_legacy( 

1111 legacy_test_data.legacy_exposure, 

1112 plane_map=legacy_test_data.plane_map, 

1113 exposure_record=legacy_test_data.exposure_record, 

1114 ) 

1115 assert legacy.obs_info is not None 

1116 assert legacy.obs_info == legacy_test_data.visit_image.obs_info 

1117 assert legacy.obs_info is not None # for mypy 

1118 assert legacy.obs_info.instrument == "LSSTCam" 

1119 assert legacy.obs_info.detector_num == 85, legacy.obs_info 

1120 assert legacy.obs_info.detector_unique_name == "R21_S11", legacy.obs_info 

1121 assert legacy.obs_info.physical_filter == "r_57", legacy.obs_info 

1122 

1123 

1124def test_aperture_corrections_to_legacy(legacy_test_data: _LegacyTestData) -> None: 

1125 """Verify that aperture corrections round-trip through a legacy 

1126 ApCorrMap. 

1127 """ 

1128 ap_corrections = legacy_test_data.visit_image.aperture_corrections 

1129 legacy_ap_corr_map = aperture_corrections_to_legacy(ap_corrections) 

1130 compare_aperture_corrections_to_legacy( 

1131 ap_corrections, 

1132 legacy_ap_corr_map, 

1133 legacy_test_data.visit_image.bbox, 

1134 ) 

1135 

1136 

1137def _check_legacy_headers(visit_image: VisitImage) -> None: 

1138 """Assert that primary and extension headers are stripped correctly.""" 

1139 header = visit_image._opaque_metadata.headers[ExtensionKey()] 

1140 assert "EXPTIME" in header 

1141 assert header["PLATFORM"] == "lsstcam" 

1142 assert "LSST BUTLER ID" not in header 

1143 assert "AR HDU" not in header 

1144 assert "A_ORDER" not in header 

1145 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("IMAGE"), astropy.io.fits.Header()) 

1146 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("MASK"), astropy.io.fits.Header()) 

1147 assert not visit_image._opaque_metadata.headers.get(ExtensionKey("VARIANCE"), astropy.io.fits.Header()) 

1148 

1149 

1150def test_read_legacy_headers(legacy_test_data: _LegacyTestData) -> None: 

1151 """Verify that headers were stripped and interpreted correctly in 

1152 read_legacy. 

1153 """ 

1154 assert legacy_test_data.visit_image.unit == legacy_test_data.unit 

1155 _check_legacy_headers(legacy_test_data.visit_image) 

1156 

1157 

1158def test_from_legacy_headers(legacy_test_data: _LegacyTestData) -> None: 

1159 """Verify that from_legacy handles primary and extension headers 

1160 correctly. 

1161 """ 

1162 legacy = VisitImage.from_legacy( 

1163 legacy_test_data.legacy_exposure, 

1164 plane_map=legacy_test_data.plane_map, 

1165 exposure_record=legacy_test_data.exposure_record, 

1166 ) 

1167 assert legacy.unit == legacy_test_data.unit 

1168 _check_legacy_headers(legacy) 

1169 

1170 

1171def test_rewrite(legacy_test_data: _LegacyTestData) -> None: 

1172 """Verify that a legacy VisitImage can be rewritten and round-trips both 

1173 pixel values and all components. 

1174 """ 

1175 with RoundtripFits(legacy_test_data.visit_image, legacy_test_data.storage_class) as roundtrip: 

1176 fits = roundtrip.inspect() 

1177 assert fits[1].header["ZCMPTYPE"] == "RICE_1" 

1178 assert fits[1].header["CTYPE1"] == "RA---TAN-SIP" 

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

1180 assert fits[2].header["CTYPE1"] == "RA---TAN-SIP" 

1181 assert fits[3].header["ZCMPTYPE"] == "RICE_1" 

1182 assert fits[3].header["CTYPE1"] == "RA---TAN-SIP" 

1183 subbox = Box.factory[8:13, 9:30] 

1184 subimage = roundtrip.get(bbox=subbox) 

1185 assert_masked_images_equal(subimage, legacy_test_data.visit_image[subbox], expect_view=False) 

1186 alternates: dict[str, Any] = {} 

1187 assert roundtrip.get("bbox") == legacy_test_data.visit_image.bbox 

1188 alternates = { 

1189 k: roundtrip.get(k) 

1190 for k in [ 

1191 "sky_projection", 

1192 "image", 

1193 "mask", 

1194 "variance", 

1195 "psf", 

1196 "obs_info", 

1197 "summary_stats", 

1198 "aperture_corrections", 

1199 "detector", 

1200 "photometric_scaling", 

1201 ] 

1202 } 

1203 legacy_exposure = roundtrip.get(storageClass="Exposure") 

1204 assert isinstance(legacy_exposure, LegacyExposure) 

1205 compare_visit_image_to_legacy( 

1206 legacy_test_data.visit_image, 

1207 legacy_exposure, 

1208 expect_view=False, 

1209 plane_map=legacy_test_data.plane_map, 

1210 **MINIMAL_VISIT_DATA_ID, 

1211 ) 

1212 if legacy_test_data.visit_image.unit == u.nJy: 

1213 assert legacy_exposure.getPhotoCalib()._isConstant 

1214 assert legacy_exposure.getPhotoCalib().getCalibrationMean() == 1.0 

1215 else: 

1216 compare_photo_calib_to_legacy( 

1217 legacy_test_data.visit_image.photometric_scaling, 

1218 legacy_exposure.getPhotoCalib(), 

1219 subimage_bbox=subbox, 

1220 ) 

1221 assert legacy_exposure.info.getId() == legacy_test_data.legacy_exposure.info.getId() 

1222 visit_info = roundtrip.get("obs_info", storageClass="VisitInfo") 

1223 assert isinstance(visit_info, LegacyVisitInfo) 

1224 assert visit_info.getInstrumentLabel() == "LSSTCam" 

1225 

1226 assert_visit_images_equal(roundtrip.result, legacy_test_data.visit_image, expect_view=False) 

1227 assert dict(legacy_test_data.visit_image._opaque_metadata.headers[ExtensionKey()]) == dict( 

1228 roundtrip.result._opaque_metadata.headers[ExtensionKey()] 

1229 ) 

1230 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("IMAGE")] 

1231 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("MASK")] 

1232 assert not roundtrip.result._opaque_metadata.headers[ExtensionKey("VARIANCE")] 

1233 assert roundtrip.result._opaque_metadata.headers[ExtensionKey()]["PLATFORM"] == "lsstcam" 

1234 compare_visit_image_to_legacy( 

1235 roundtrip.result, 

1236 legacy_test_data.legacy_exposure, 

1237 expect_view=False, 

1238 plane_map=legacy_test_data.plane_map, 

1239 **MINIMAL_VISIT_DATA_ID, 

1240 alternates=alternates, 

1241 ) 

1242 compare_visit_image_to_legacy( 

1243 legacy_test_data.read_cls.from_legacy( 

1244 legacy_test_data.legacy_exposure, 

1245 plane_map=legacy_test_data.plane_map, 

1246 exposure_record=legacy_test_data.exposure_record, 

1247 ), 

1248 legacy_test_data.legacy_exposure, 

1249 expect_view=True, 

1250 plane_map=legacy_test_data.plane_map, 

1251 **MINIMAL_VISIT_DATA_ID, 

1252 ) 

1253 

1254 

1255def test_butler_converters(legacy_test_data: _LegacyTestData) -> None: 

1256 """Verify that a VisitImage can be read from a Butler dataset written as 

1257 an Exposure. 

1258 """ 

1259 try: 

1260 from lsst.daf.butler import FileDataset 

1261 except ImportError: 

1262 pytest.skip("lsst.daf.butler could not be imported.") 

1263 

1264 with TemporaryButler(legacy="ExposureF") as helper: 

1265 helper.butler.ingest( 

1266 FileDataset(path=legacy_test_data.filename, refs=[helper.legacy]), transfer="symlink" 

1267 ) 

1268 visit_image_ref = helper.legacy.overrideStorageClass(legacy_test_data.storage_class) 

1269 record_params = {"exposure_record": legacy_test_data.exposure_record} 

1270 with warnings.catch_warnings(): 

1271 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning) 

1272 visit_image = helper.butler.get(visit_image_ref, parameters=record_params) 

1273 assert visit_image._opaque_metadata.precompressed.keys() == set() 

1274 visit_image = helper.butler.get( 

1275 visit_image_ref, parameters={"preserve_quantization": True, **record_params} 

1276 ) 

1277 assert visit_image._opaque_metadata.precompressed.keys() == {"IMAGE", "VARIANCE"} 

1278 bbox = helper.butler.get(visit_image_ref.makeComponentRef("bbox"), parameters=record_params) 

1279 assert bbox == visit_image.bbox 

1280 # bbox/psf/detector reads route through VisitImage and require the 

1281 # exposure record; image/mask/variance route to MaskedImage, which 

1282 # carries no ObservationInfo and must not be given it. 

1283 alternates = { 

1284 k: helper.butler.get(visit_image_ref.makeComponentRef(k)) for k in ["image", "mask", "variance"] 

1285 } | { 

1286 k: helper.butler.get(visit_image_ref.makeComponentRef(k), parameters=record_params) 

1287 for k in ["bbox", "psf", "detector"] 

1288 } 

1289 compare_visit_image_to_legacy( 

1290 visit_image, 

1291 legacy_test_data.legacy_exposure, 

1292 expect_view=False, 

1293 plane_map=legacy_test_data.plane_map, 

1294 alternates=alternates, 

1295 **MINIMAL_VISIT_DATA_ID, 

1296 ) 

1297 helper.butler.pruneDatasets([helper.legacy], purge=True, unstore=True, disassociate=True) 

1298 visit_image.metadata["MixedCaseKey"] = 52 

1299 helper.butler.put(visit_image, visit_image_ref) 

1300 with warnings.catch_warnings(): 

1301 warnings.filterwarnings("ignore", message=".*filter label mismatch.*", category=UserWarning) 

1302 legacy_exposure = helper.butler.get(helper.legacy) 

1303 compare_visit_image_to_legacy( 

1304 visit_image, 

1305 legacy_exposure, 

1306 expect_view=False, 

1307 plane_map=legacy_test_data.plane_map, 

1308 alternates=alternates, 

1309 **MINIMAL_VISIT_DATA_ID, 

1310 ) 

1311 visit_image_2 = helper.butler.get(visit_image_ref, parameters=record_params) 

1312 compare_visit_image_to_legacy( 

1313 visit_image_2, 

1314 legacy_exposure, 

1315 expect_view=False, 

1316 plane_map=legacy_test_data.plane_map, 

1317 alternates=alternates, 

1318 **MINIMAL_VISIT_DATA_ID, 

1319 ) 

1320 assert visit_image_2.metadata["MixedCaseKey"] == 52 

1321 

1322 

1323def test_convert_unit(legacy_test_data_calibrated: _LegacyTestData) -> None: 

1324 """Verify convert_unit round-trips between nJy, mJy, and electron via 

1325 photometric_scaling. 

1326 """ 

1327 from lsst.afw.table import ExposureCatalog 

1328 

1329 legacy_test_data = legacy_test_data_calibrated 

1330 original = legacy_test_data.visit_image.copy() 

1331 with pytest.raises(u.UnitConversionError): 

1332 original.convert_unit(u.electron) 

1333 visit_image_nJy = original.convert_unit(u.nJy, copy=False) 

1334 assert np.may_share_memory(visit_image_nJy.image.array, original.image.array) 

1335 assert np.may_share_memory(visit_image_nJy.variance.array, original.variance.array) 

1336 with pytest.raises(u.UnitConversionError): 

1337 original.convert_unit(u.mJy, copy=False) 

1338 visit_image_mJy = original.convert_unit(u.mJy, copy="as-needed") 

1339 assert visit_image_mJy.unit == u.mJy 

1340 assert_values_equal(visit_image_mJy.image.array, original.image.array * 1e-6, rtol=1e-5) 

1341 assert np.may_share_memory(visit_image_nJy.mask.array, original.mask.array) 

1342 assert_values_equal(visit_image_mJy.variance.array, original.variance.array * 1e-12, rtol=1e-5) 

1343 legacy_exposure_mJy = visit_image_mJy.to_legacy() 

1344 assert_values_equal(legacy_exposure_mJy.getPhotoCalib().getCalibrationMean(), 1e6, rtol=1e-14) 

1345 legacy_masked_image_nJy = legacy_exposure_mJy.getPhotoCalib().calibrateImage( 

1346 legacy_exposure_mJy.maskedImage 

1347 ) 

1348 assert_values_equal(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array, rtol=1e-5) 

1349 assert_values_equal(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array, rtol=1e-5) 

1350 assert np.may_share_memory(visit_image_mJy.mask.array, original.mask.array) 

1351 assert visit_image_mJy.sky_projection is original.sky_projection 

1352 assert visit_image_mJy.obs_info is original.obs_info 

1353 assert visit_image_mJy.summary_stats is original.summary_stats 

1354 assert visit_image_mJy.psf is original.psf 

1355 assert visit_image_mJy.detector is original.detector 

1356 assert visit_image_mJy.bounds is original.bounds 

1357 assert visit_image_mJy.aperture_corrections is original.aperture_corrections 

1358 assert visit_image_mJy.photometric_scaling is original.photometric_scaling 

1359 visit_summary = ExposureCatalog.readFits( 

1360 os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_summary.fits") 

1361 ) 

1362 legacy_photo_calib = visit_summary.find(DP2_VISIT_DETECTOR_DATA_ID["detector"]).getPhotoCalib() 

1363 visit_image_nJy.photometric_scaling = field_from_legacy_photo_calib( 

1364 legacy_photo_calib, bounds=original.detector.bbox, instrumental_unit=u.electron 

1365 ) 

1366 compare_photo_calib_to_legacy( 

1367 visit_image_nJy.photometric_scaling, 

1368 legacy_test_data.legacy_exposure.getPhotoCalib(), 

1369 applied_legacy_photo_calib=legacy_photo_calib, 

1370 subimage_bbox=visit_image_nJy.bbox, 

1371 ) 

1372 with pytest.raises(u.UnitConversionError): 

1373 visit_image_nJy.convert_unit(u.mm) 

1374 with pytest.raises(u.UnitConversionError): 

1375 visit_image_nJy.convert_unit(u.electron, copy=False) 

1376 legacy_masked_image_e = legacy_photo_calib.uncalibrateImage(legacy_test_data.legacy_exposure.maskedImage) 

1377 visit_image_e = visit_image_nJy.convert_unit(u.electron) 

1378 assert_values_equal(visit_image_e.image.array, legacy_masked_image_e.image.array, rtol=1e-5) 

1379 assert_values_equal(visit_image_e.variance.array, legacy_masked_image_e.variance.array, rtol=1e-5) 

1380 assert not np.may_share_memory(visit_image_e.mask.array, visit_image_nJy.mask.array) 

1381 visit_image_mJy.photometric_scaling = visit_image_nJy.photometric_scaling 

1382 visit_image_e = visit_image_mJy.convert_unit(u.electron) 

1383 assert_values_equal(visit_image_e.image.array, legacy_masked_image_e.image.array, rtol=1e-5) 

1384 assert_values_equal(visit_image_e.variance.array, legacy_masked_image_e.variance.array, rtol=1e-5) 

1385 visit_image_nJy_2 = visit_image_e.convert_unit(u.nJy) 

1386 assert_values_equal(visit_image_nJy_2.image.array, visit_image_nJy.image.array, rtol=1e-5) 

1387 assert_values_equal(visit_image_nJy_2.variance.array, original.variance.array, rtol=1e-5) 

1388 visit_image_e.photometric_scaling = visit_image_nJy.photometric_scaling * (1e-6 * u.mJy / u.nJy) 

1389 visit_image_nJy_3 = visit_image_e.convert_unit(u.nJy) 

1390 assert_values_equal(visit_image_nJy_3.image.array, visit_image_nJy.image.array, rtol=1e-5) 

1391 assert_values_equal(visit_image_nJy_3.variance.array, original.variance.array, rtol=1e-5) 

1392 legacy_exposure_e = visit_image_e.to_legacy() 

1393 assert_values_equal( 

1394 legacy_exposure_e.getPhotoCalib().getCalibrationMean(), 

1395 legacy_photo_calib.getCalibrationMean(), 

1396 rtol=1e-5, 

1397 ) 

1398 legacy_masked_image_nJy = legacy_exposure_e.getPhotoCalib().calibrateImage(legacy_exposure_e.maskedImage) 

1399 assert_values_equal(visit_image_nJy.image.array, legacy_masked_image_nJy.image.array, rtol=1e-5) 

1400 assert_values_equal(visit_image_nJy.variance.array, legacy_masked_image_nJy.variance.array, rtol=1e-5) 

1401 

1402 

1403def test_background_map_describe() -> None: 

1404 """An empty BackgroundMap._describe reports no backgrounds inline.""" 

1405 bg_map = BackgroundMap() 

1406 assert isinstance(bg_map, DescribableMixin) 

1407 report = bg_map._describe() 

1408 assert isinstance(report, Report) 

1409 assert report.type_name == "BackgroundMap" 

1410 assert report.inline 

1411 assert report.summary == "no backgrounds" 

1412 assert report.children == {} 

1413 

1414 

1415def test_background_map_with_entries_describe() -> None: 

1416 """BackgroundMap._describe recurses into each background model.""" 

1417 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]])) 

1418 bg_map = BackgroundMap( 

1419 [Background("sky", cheby, "Sky model."), Background("fringe", cheby)], 

1420 subtracted="sky", 

1421 ) 

1422 report = bg_map._describe() 

1423 assert report.type_name == "BackgroundMap" 

1424 assert report.inline 

1425 # The inline summary leads with the subtracted background and marks it, 

1426 # since that is all a composite holding this map will show. 

1427 # Only the subtracted background is marked, not the whole line. 

1428 assert report.summary == "**sky [SUBTRACTED]**; also: fringe" 

1429 assert report.emphasis_markup 

1430 # str reads the summary as text, so the delimiters never reach a user. 

1431 assert report.to_str() == "sky [SUBTRACTED]; also: fringe" 

1432 # Standalone, each background is a child carrying its own model's report. 

1433 assert set(report.children) == {"sky", "fringe"} 

1434 sky = report.children["sky"] 

1435 assert sky.type_name == "ChebyshevField" 

1436 # The marker is on the heading, where it is read before the fields. 

1437 assert sky.title == "ChebyshevField **[SUBTRACTED]**" 

1438 assert sky.emphasis_markup 

1439 sky_fields = {f.label: f.value for f in sky.fields} 

1440 assert sky_fields["description"] == "Sky model." 

1441 # The model's own fields survive alongside the background's attributes. 

1442 assert "bounds" in sky_fields 

1443 # Only the subtracted one is marked, and an absent description is omitted. 

1444 fringe = report.children["fringe"] 

1445 assert fringe.title is None 

1446 assert fringe.emphasis_markup is False 

1447 assert "description" not in {f.label for f in fringe.fields} 

1448 

1449 

1450def test_background_map_describe_none_subtracted() -> None: 

1451 """A map with no subtracted background says so rather than being silent.""" 

1452 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]])) 

1453 bg_map = BackgroundMap([Background("sky", cheby), Background("fringe", cheby)]) 

1454 report = bg_map._describe() 

1455 assert report.summary == "sky, fringe (none subtracted)" 

1456 assert report.emphasis_markup is False 

1457 assert not any(child.emphasis_markup for child in report.children.values()) 

1458 

1459 

1460def test_background_map_describe_leads_with_subtracted() -> None: 

1461 """The subtracted background leads the summary wherever it sits in the 

1462 map, and only its child is marked. 

1463 """ 

1464 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]])) 

1465 bg_map = BackgroundMap( 

1466 [Background("sky", cheby), Background("skyCorr", cheby)], 

1467 subtracted="skyCorr", 

1468 ) 

1469 report = bg_map._describe() 

1470 assert report.to_str() == "skyCorr [SUBTRACTED]; also: sky" 

1471 assert report.children["sky"].emphasis_markup is False 

1472 assert report.children["skyCorr"].emphasis_markup 

1473 

1474 

1475def test_background_map_describe_brief_skips_children() -> None: 

1476 """A brief background map report keeps the summary but not the models.""" 

1477 cheby = ChebyshevField(Box.factory[0:100, 0:200], np.array([[1.0]])) 

1478 bg_map = BackgroundMap([Background("sky", cheby)], subtracted="sky") 

1479 report = bg_map._describe(DescribeOptions(brief=True)) 

1480 assert report.to_str() == "sky [SUBTRACTED]" 

1481 assert report.children == {} 

1482 

1483 

1484def test_visit_image_repr_str_with_unreadable_psf() -> None: 

1485 """Repr and str succeed even when the PSF stored an ArchiveReadError. 

1486 

1487 An unreadable component is a supported state; repr and str read only the 

1488 cheap fields and summary, so they do not build the child tree at all. 

1489 """ 

1490 path = current_fixture_path(FIXTURE_DIR, "visit_image") 

1491 visit_image = read_archive(path) 

1492 visit_image._psf = ArchiveReadError("psf unreadable") 

1493 assert repr(visit_image).startswith("VisitImage(") 

1494 assert str(visit_image).startswith("VisitImage(") 

1495 

1496 

1497def test_visit_image_describe_names_an_unreadable_psf() -> None: 

1498 """A full report says the PSF could not be read, and describes the rest. 

1499 

1500 A PSF model can need a package the reader does not have installed, so an 

1501 unreadable PSF must not cost the report of everything else. 

1502 """ 

1503 path = current_fixture_path(FIXTURE_DIR, "visit_image") 

1504 visit_image = read_archive(path) 

1505 visit_image._psf = ArchiveReadError("Failed to import piff.") 

1506 report = visit_image.describe(detail=True) 

1507 psf = report.children["psf"] 

1508 assert psf.inline 

1509 assert psf.to_str() == "unreadable (Failed to import piff.)" 

1510 # Every other component is described as usual. 

1511 assert "sky_projection" in report.children 

1512 assert "detector" in report.children 

1513 # Both renderers run over the report that contains it. 

1514 assert "unreadable" in report._repr_html_() 

1515 report.__rich__() 

1516 

1517 

1518def test_visit_image_slice_preserves_unreadable_psf() -> None: 

1519 """Slicing propagates a deferred PSF read failure without raising it.""" 

1520 path = current_fixture_path(FIXTURE_DIR, "visit_image") 

1521 visit_image = read_archive(path) 

1522 error = ArchiveReadError("psf unreadable") 

1523 visit_image._psf = error 

1524 

1525 sliced = visit_image[...] 

1526 

1527 assert sliced._psf is error 

1528 with pytest.raises(ArchiveReadError, match="psf unreadable"): 

1529 sliced.psf 

1530 

1531 

1532def test_observation_summary_stats_describe() -> None: 

1533 """ObservationSummaryStats._describe reports the statistics that are set. 

1534 

1535 Unset ones are omitted rather than shown as NaN. 

1536 """ 

1537 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4, ra=180.0, dec=-30.0) 

1538 assert isinstance(stats, DescribableMixin) 

1539 report = stats._describe() 

1540 assert isinstance(report, Report) 

1541 assert report.type_name == "ObservationSummaryStats" 

1542 # Scalars are packed into one group, ordered by name. 

1543 (group,) = report.value_groups 

1544 assert group.role is FieldRole.DERIVED 

1545 assert [name for name, _ in group.values] == sorted(name for name, _ in group.values) 

1546 values = dict(group.values) 

1547 assert values["psfSigma"] == 2.5 

1548 assert values["zeroPoint"] == 31.4 

1549 assert values["ra"] == 180.0 

1550 assert values["dec"] == -30.0 

1551 # Unset statistics are omitted rather than reported as NaN, and the 

1552 # serialization plumbing this class inherits never appears at all. 

1553 assert "expTime" not in values 

1554 assert "skyBg" not in values 

1555 assert not {"metadata", "butler_info", "schema_version"} & set(values) 

1556 

1557 

1558def test_observation_summary_stats_describe_omits_empty_sequences() -> None: 

1559 """Sequence statistics appear only when they carry a value.""" 

1560 empty = ObservationSummaryStats(psfSigma=2.5) 

1561 # raCorners defaults to all-NaN, which carries no more information than an 

1562 # empty sequence does. 

1563 assert all(math.isnan(v) for v in empty.raCorners) 

1564 assert not any(f.label == "raCorners" for f in empty._describe().fields) 

1565 

1566 filled = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2)) 

1567 corners = next(f for f in filled._describe().fields if f.label == "raCorners") 

1568 assert corners.value == (5.2, 5.4, 5.4, 5.2) 

1569 assert corners.role is FieldRole.DERIVED 

1570 

1571 

1572def test_observation_summary_stats_describe_brief_counts() -> None: 

1573 """A brief report gives the number set rather than listing them.""" 

1574 stats = ObservationSummaryStats(psfSigma=2.5, raCorners=(5.2, 5.4, 5.4, 5.2)) 

1575 brief = stats._describe(DescribeOptions(brief=True)) 

1576 assert brief.type_name == "ObservationSummaryStats" 

1577 assert brief.value_groups == [] 

1578 (field,) = brief.fields 

1579 assert field.label == "statistics set" 

1580 assert field.role is FieldRole.DERIVED 

1581 

1582 # The count must agree with what the full report actually shows: two 

1583 # scalars set by default plus psfSigma, and raCorners as a sequence. 

1584 full = stats._describe() 

1585 listed = len(full.fields) + sum(len(g.values) for g in full.value_groups) 

1586 assert field.value.startswith(f"{listed} of ") 

1587 assert brief.to_str() == full.to_str() 

1588 assert f"{listed} of " in brief.to_str() 

1589 

1590 

1591def test_observation_summary_stats_pydantic_repr() -> None: 

1592 """ObservationSummaryStats uses pydantic's repr, not the mixin's.""" 

1593 stats = ObservationSummaryStats(psfSigma=2.5) 

1594 r = repr(stats) 

1595 assert r.startswith("ObservationSummaryStats(") 

1596 assert "psfSigma=2.5" in r 

1597 

1598 

1599def test_observation_summary_stats_str_is_the_report_summary() -> None: 

1600 """The str output reports the count, not pydantic's field-by-field dump. 

1601 

1602 The repr keeps the exhaustive form, since that is the one that 

1603 round-trips. 

1604 """ 

1605 stats = ObservationSummaryStats(psfSigma=2.5, zeroPoint=31.4) 

1606 assert str(stats) == stats.describe().to_str() 

1607 # The two set here, plus the integer counters, which default to a genuine 

1608 # zero rather than to NaN. 

1609 assert str(stats) == "ObservationSummaryStats(4 of 66 statistics set)" 

1610 assert stats.nPsfStar == 0 

1611 assert stats.nShapeletsStar == 0 

1612 # The unset statistics reach repr but not str. 

1613 assert "nan" not in str(stats) 

1614 assert "nan" in repr(stats) 

1615 

1616 

1617def test_archive_tree_repr_omits_schema_bookkeeping() -> None: 

1618 """Schema version fields mirror class constants, so repr leaves them out. 

1619 

1620 They are never passed on construction and say nothing the type does not. 

1621 """ 

1622 stats = ObservationSummaryStats(psfSigma=2.5, indirect=[1]) 

1623 for name in ("schema_version", "min_read_version", "indirect"): 

1624 assert name not in repr(stats), name 

1625 # They are still real fields, and hiding them from repr does not hide them 

1626 # from serialization. 

1627 assert stats.schema_version == ObservationSummaryStats.SCHEMA_VERSION 

1628 dumped = stats.model_dump() 

1629 for name in ("schema_version", "min_read_version", "indirect"): 

1630 assert name in dumped, name 

1631 

1632 

1633def _match_detector_to_image(visit_image: VisitImage) -> VisitImage: 

1634 """Return the visit image with its detector shrunk to the image bbox. 

1635 

1636 Parameters 

1637 ---------- 

1638 visit_image 

1639 Image to adjust. 

1640 

1641 Returns 

1642 ------- 

1643 visit_image : `VisitImage` 

1644 Image whose `~VisitImage.bbox` equals its detector's, as a full-frame 

1645 image read from a butler has. 

1646 """ 

1647 detector = visit_image.detector 

1648 visit_image._detector = Detector( 

1649 detector._attributes.model_copy(update={"bbox": visit_image.bbox}), 

1650 detector.amplifiers, 

1651 detector._frames, 

1652 detector.visit, 

1653 ) 

1654 return visit_image 

1655 

1656 

1657def test_visit_image_describe_hides_summary_stat_corners_for_a_full_frame( 

1658 visit_image_components: dict[str, Any], 

1659) -> None: 

1660 """The sky corners appear once, in the projection's readable table.""" 

1661 visit = _match_detector_to_image(make_visit_image(visit_image_components)) 

1662 visit.summary_stats.raCorners = (1.0, 2.0, 3.0, 4.0) 

1663 visit.summary_stats.decCorners = (-1.0, -2.0, -3.0, -4.0) 

1664 report = visit.describe() 

1665 stats = report.children["summary_stats"] 

1666 assert not {"raCorners", "decCorners"} & {field.label for field in stats.fields} 

1667 # The projection is where they are read instead, and it still has them. 

1668 assert any(table.title == "Corners" for table in report.children["sky_projection"].tables) 

1669 

1670 

1671def test_visit_image_cutout_keeps_summary_stat_corners() -> None: 

1672 """A cutout keeps the statistics' corners, which still describe the whole 

1673 image the statistics were measured on. 

1674 """ 

1675 # The DP2 variant is a cutout of a real image, so it carries the 

1676 # statistics measured on the whole of it. 

1677 path = current_fixture_path(FIXTURE_DIR, "visit_image", variant="dp2") 

1678 visit_image = read_archive(path) 

1679 assert visit_image.bbox != visit_image.detector.bbox 

1680 stats = visit_image.describe().children["summary_stats"] 

1681 assert {"raCorners", "decCorners"} <= {field.label for field in stats.fields}