Coverage for tests/test_psfs.py: 19%

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 unittest 

16import warnings 

17 

18import numpy as np 

19 

20from lsst.images import Box 

21from lsst.images.psfs import ( 

22 GaussianPointSpreadFunction, 

23 PiffWrapper, 

24 PointSpreadFunction, 

25 PSFExWrapper, 

26) 

27from lsst.images.psfs._piff import _ArchivePiffWriter 

28from lsst.images.tests import ( 

29 RoundtripFits, 

30 RoundtripJson, 

31 RoundtripNdf, 

32 compare_psf_to_legacy, 

33) 

34 

35try: 

36 import h5py # noqa: F401 

37 

38 HAVE_H5PY = True 

39except ImportError: 

40 HAVE_H5PY = False 

41 

42DATA_DIR = os.environ.get("TESTDATA_IMAGES_DIR", None) 

43 

44 

45class PointSpreadFunctionTestCase(unittest.TestCase): 

46 """Tests for the PointSpreadFunction classes.""" 

47 

48 def test_gaussian(self) -> None: 

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

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

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

52 self.assertEqual(psf.bounds, bounds) 

53 

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

55 self.assertEqual(kernel.bbox, psf.kernel_bbox) 

56 self.assertAlmostEqual(float(kernel.array.sum()), 1.0) 

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

58 self.assertEqual(np.unravel_index(np.argmax(kernel.array), kernel.array.shape), (center, center)) 

59 

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

61 self.assertEqual(stellar.bbox, psf.compute_stellar_bbox(x=5.25, y=3.75)) 

62 self.assertAlmostEqual(float(stellar.array.sum()), 1.0) 

63 self.assertGreater(stellar.array[center - 1, center], stellar.array[center + 1, center]) 

64 self.assertGreater(stellar.array[center, center], stellar.array[center, center - 1]) 

65 self.assertGreater(stellar.array[center, center], stellar.array[center - 1, center]) 

66 

67 with RoundtripFits(self, psf) as roundtrip: 

68 self.assertEqual(roundtrip.result, psf, f"{roundtrip.result} != {psf}") 

69 

70 with self.assertRaises(ValueError): 

71 # Even stamp size. 

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

73 

74 with self.assertRaises(ValueError): 

75 # Negative stamp size. 

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

77 

78 with self.assertRaises(ValueError): 

79 # Negative sigma. 

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

81 

82 def test_piff_writer_normalizes_tuple_metadata(self) -> None: 

83 """Piff metadata should be normalized to JSON-like values.""" 

84 writer = _ArchivePiffWriter() 

85 writer.write_struct( 

86 "interp", 

87 { 

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

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

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

91 }, 

92 ) 

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

94 self.assertEqual(model.structs["interp"]["keys"], ["u", "v"]) 

95 self.assertEqual(model.structs["interp"]["scale"], 1.5) 

96 self.assertEqual(model.structs["interp"]["flags"], [True, 3]) 

97 with warnings.catch_warnings(): 

98 warnings.simplefilter("error", UserWarning) 

99 model.model_dump_json() 

100 

101 @unittest.skipUnless(DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not in the environment.") 

102 def test_piff(self) -> None: 

103 """Test that we can: 

104 

105 - read a legacy Piff PSF with afw; 

106 - convert it to the new `PiffWrapper` class; 

107 - get consistent behavior from the two; 

108 - round-trip the new PSF through a FITS archive; 

109 - still get consistent behavior with the round-tripped PSF. 

110 

111 This test is skipped if legacy modules cannot be imported. 

112 """ 

113 try: 

114 from piff import PSF 

115 

116 from lsst.afw.image import ExposureFitsReader 

117 except ImportError: 

118 raise unittest.SkipTest("'lsst.afw.image' could not be imported.") from None 

119 assert DATA_DIR is not None, "Guaranteed by decorator." 

120 filename = os.path.join(DATA_DIR, "dp2", "legacy", "visit_image.fits") 

121 reader = ExposureFitsReader(filename) 

122 legacy_psf = reader.readPsf() 

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

124 psf = PointSpreadFunction.from_legacy(legacy_psf, bounds) 

125 self.assertIsInstance(psf, PiffWrapper) 

126 self.assertEqual(psf.bounds, bounds) 

127 self.assertIsInstance(psf.piff_psf, PSF) 

128 compare_psf_to_legacy(self, psf, legacy_psf) 

129 with RoundtripFits(self, psf) as roundtrip1: 

130 pass 

131 compare_psf_to_legacy(self, roundtrip1.result, legacy_psf) 

132 with RoundtripJson(self, psf) as roundtrip2: 

133 pass 

134 compare_psf_to_legacy(self, roundtrip2.result, legacy_psf) 

135 with self.subTest("NDF round-trip"): 

136 if not HAVE_H5PY: 

137 raise unittest.SkipTest("h5py is not available.") 

138 with RoundtripNdf(self, psf) as roundtrip3: 

139 pass 

140 compare_psf_to_legacy(self, roundtrip3.result, legacy_psf) 

141 

142 @unittest.skipUnless(DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not in the environment.") 

143 def test_psfex(self) -> None: 

144 """Test that we can: 

145 

146 - read a legacy PSFEX PSF with afw; 

147 - wrap it inthe new `LegacyPointSpreadFunction` class; 

148 - get consistent behavior from the two. 

149 

150 This test is skipped if legacy modules cannot be imported. 

151 """ 

152 try: 

153 from lsst.afw.image import ExposureFitsReader 

154 from lsst.meas.extensions.psfex import PsfexPsf 

155 except ImportError: 

156 raise unittest.SkipTest("'lsst.afw.image' could not be imported.") from None 

157 assert DATA_DIR is not None, "Guaranteed by decorator." 

158 filename = os.path.join(DATA_DIR, "dp2", "legacy", "preliminary_visit_image.fits") 

159 reader = ExposureFitsReader(filename) 

160 legacy_psf = reader.readPsf() 

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

162 psf = PointSpreadFunction.from_legacy(legacy_psf, bounds) 

163 self.assertIsInstance(psf, PSFExWrapper) 

164 self.assertEqual(psf.bounds, bounds) 

165 self.assertIsInstance(psf.legacy_psf, PsfexPsf) 

166 compare_psf_to_legacy(self, psf, legacy_psf) 

167 compare_psf_to_legacy(self, psf, legacy_psf) 

168 with RoundtripFits(self, psf) as roundtrip1: 

169 pass 

170 compare_psf_to_legacy(self, roundtrip1.result, legacy_psf) 

171 with RoundtripJson(self, psf) as roundtrip2: 

172 pass 

173 compare_psf_to_legacy(self, roundtrip2.result, legacy_psf) 

174 

175 

176if __name__ == "__main__": 

177 unittest.main()