Coverage for tests/test_psfs.py: 50%

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 os 

15import warnings 

16from typing import Any 

17 

18import numpy as np 

19import pytest 

20 

21from lsst.images import Box 

22from lsst.images.psfs import GaussianPointSpreadFunction, PiffWrapper, PointSpreadFunction, PSFExWrapper 

23from lsst.images.psfs._piff import _ArchivePiffWriter 

24from lsst.images.tests import RoundtripFits, RoundtripJson, RoundtripNdf, compare_psf_to_legacy 

25 

26try: 

27 import h5py # noqa: F401 

28 

29 HAVE_H5PY = True 

30except ImportError: 

31 HAVE_H5PY = False 

32 

33try: 

34 from lsst.afw.detection import Psf as LegacyPsf 

35except ImportError: 

36 type LegacyPsf = Any # type: ignore[no-redef] 

37 

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

39 

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

41 

42 

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

44def legacy_piff_psf_and_bbox() -> tuple[LegacyPsf, Box]: 

45 """Return a legacy-wrapped Piff PSF and its bounding box. 

46 

47 Skips if TESTDATA_IMAGES_DIR is unset, piff is unavailable, or afw is 

48 unavailable. 

49 """ 

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

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

52 try: 

53 import piff # noqa: F401 

54 

55 from lsst.afw.image import ExposureFitsReader 

56 except ImportError: 

57 pytest.skip("'piff' or 'lsst.afw.image' could not be imported.") 

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

59 reader = ExposureFitsReader(filename) 

60 legacy_psf = reader.readPsf() 

61 bounds = Box.from_legacy(reader.readBBox()) 

62 return legacy_psf, bounds 

63 

64 

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

66def legacy_psfex_psf_and_bbox() -> tuple[LegacyPsf, Box]: 

67 """Return a legacy PSFEx PSF and its bounding box 

68 

69 Skips if TESTDATA_IMAGES_DIR is unset, afw is unavailable, or psfex is 

70 unavailable. 

71 """ 

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

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

74 try: 

75 from lsst.afw.image import ExposureFitsReader 

76 from lsst.meas.extensions.psfex import PsfexPsf # noqa: F401 

77 except ImportError: 

78 pytest.skip("'lsst.afw.image' or 'lsst.meas.extensions.psfex' could not be imported.") 

79 filename = os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "preliminary_visit_image.fits") 

80 reader = ExposureFitsReader(filename) 

81 legacy_psf = reader.readPsf() 

82 bounds = Box.from_legacy(reader.readBBox()) 

83 return legacy_psf, bounds 

84 

85 

86def test_gaussian() -> None: 

87 """Test the built-in Gaussian PSF implementation.""" 

88 bounds = Box.factory[-1024:1024, -2048:2048] 

89 psf = GaussianPointSpreadFunction(2.5, bounds=bounds, stamp_size=33) 

90 assert psf.bounds == bounds 

91 

92 kernel = psf.compute_kernel_image(x=5.0, y=3.0) 

93 assert kernel.bbox == psf.kernel_bbox 

94 assert abs(float(kernel.array.sum()) - 1.0) < 1e-6 

95 center = kernel.array.shape[0] // 2 

96 assert np.unravel_index(np.argmax(kernel.array), kernel.array.shape) == (center, center) 

97 

98 stellar = psf.compute_stellar_image(x=5.25, y=3.75) 

99 assert stellar.bbox == psf.compute_stellar_bbox(x=5.25, y=3.75) 

100 assert abs(float(stellar.array.sum()) - 1.0) < 1e-6 

101 assert stellar.array[center - 1, center] > stellar.array[center + 1, center] 

102 assert stellar.array[center, center] > stellar.array[center, center - 1] 

103 assert stellar.array[center, center] > stellar.array[center - 1, center] 

104 

105 with RoundtripFits(psf) as roundtrip: 

106 assert roundtrip.result == psf, f"{roundtrip.result} != {psf}" 

107 

108 with pytest.raises(ValueError): 

109 # Even stamp size. 

110 GaussianPointSpreadFunction(2.5, bounds=bounds, stamp_size=32) 

111 

112 with pytest.raises(ValueError): 

113 # Negative stamp size. 

114 GaussianPointSpreadFunction(2.5, bounds=bounds, stamp_size=-33) 

115 

116 with pytest.raises(ValueError): 

117 # Negative sigma. 

118 GaussianPointSpreadFunction(-2.5, bounds=bounds, stamp_size=33) 

119 

120 

121def test_piff_writer_normalizes_tuple_metadata(): # intentionally untyped 

122 """Test that Piff metadata is normalized to JSON-like values.""" 

123 writer = _ArchivePiffWriter() 

124 writer.write_struct( 

125 "interp", 

126 { 

127 "keys": ("u", "v"), 

128 "scale": np.float64(1.5), 

129 "flags": [np.bool_(True), np.int64(3)], 

130 }, 

131 ) 

132 model = writer.serialize(None) # type: ignore[arg-type] 

133 assert model.structs["interp"]["keys"] == ["u", "v"] 

134 assert model.structs["interp"]["scale"] == 1.5 

135 assert model.structs["interp"]["flags"] == [True, 3] 

136 with warnings.catch_warnings(): 

137 warnings.simplefilter("error", UserWarning) 

138 model.model_dump_json() 

139 

140 

141def test_piff(legacy_piff_psf_and_bbox: tuple[LegacyPsf, Box]) -> None: 

142 """Test round-tripping a legacy Piff PSF through FITS, JSON, and NDF 

143 archives. 

144 """ 

145 from piff import PSF 

146 

147 legacy_psf, bounds = legacy_piff_psf_and_bbox 

148 psf = PointSpreadFunction.from_legacy(legacy_psf, bounds) 

149 assert isinstance(psf, PiffWrapper) 

150 assert psf.bounds == bounds 

151 assert isinstance(psf.piff_psf, PSF) 

152 compare_psf_to_legacy(psf, legacy_psf) 

153 with RoundtripFits(psf) as roundtrip1: 

154 pass 

155 compare_psf_to_legacy(roundtrip1.result, legacy_psf) 

156 with RoundtripJson(psf) as roundtrip2: 

157 pass 

158 compare_psf_to_legacy(roundtrip2.result, legacy_psf) 

159 if not HAVE_H5PY: 

160 pytest.skip("h5py is not available.") 

161 with RoundtripNdf(psf) as roundtrip3: 

162 pass 

163 compare_psf_to_legacy(roundtrip3.result, legacy_psf) 

164 legacy_psf_2 = roundtrip1.result.to_legacy() 

165 compare_psf_to_legacy(psf, legacy_psf_2) 

166 assert legacy_psf.getAveragePosition() == legacy_psf_2.getAveragePosition() 

167 

168 

169def test_psfex(legacy_psfex_psf_and_bbox: tuple[LegacyPsf, Box]) -> None: 

170 """Test wrapping a legacy PSFEx PSF and round-tripping through FITS and 

171 JSON. 

172 """ 

173 from lsst.meas.extensions.psfex import PsfexPsf 

174 

175 legacy_psf, bounds = legacy_psfex_psf_and_bbox 

176 psf = PointSpreadFunction.from_legacy(legacy_psf, bounds) 

177 assert isinstance(psf, PSFExWrapper) 

178 assert psf.bounds == bounds 

179 assert isinstance(psf.legacy_psf, PsfexPsf) 

180 compare_psf_to_legacy(psf, legacy_psf) 

181 with RoundtripFits(psf) as roundtrip1: 

182 pass 

183 compare_psf_to_legacy(roundtrip1.result, legacy_psf) 

184 with RoundtripJson(psf) as roundtrip2: 

185 pass 

186 compare_psf_to_legacy(roundtrip2.result, legacy_psf)