Coverage for tests/test_inject_engine.py: 97%

1# This file is part of source_injection. 

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# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <https://www.gnu.org/licenses/>. 

21 

22import logging 

23import unittest 

24from types import GeneratorType 

25 

26import galsim 

27import numpy as np 

28from galsim import BoundsI, GSObject 

29 

30import lsst.utils.tests 

31from lsst.geom import Point2D, SpherePoint, degrees 

32from lsst.source.injection.inject_engine import ( 

33 generate_galsim_objects, 

34 get_gain_map, 

35 infer_gain_from_image, 

36 inject_galsim_objects_into_exposure, 

37 make_galsim_object, 

38) 

39from lsst.source.injection.utils.test_utils import make_test_exposure, make_test_injection_catalog 

40from lsst.utils.tests import TestCase 

41 

42 

43class InjectEngineTestCase(TestCase): 

44 """Test the inject_engine.py module.""" 

45 

46 def setUp(self): 

47 """Set up synthetic source injection inputs. 

48 

49 This method sets up a noisy synthetic image with Gaussian PSFs injected 

50 into the frame, an example source injection catalog, and a generator of 

51 GalSim objects intended for injection. 

52 """ 

53 self.exposure = make_test_exposure() 

54 self.injection_catalog = make_test_injection_catalog( 

55 self.exposure.getWcs(), 

56 self.exposure.getBBox(), 

57 ) 

58 self.galsim_objects = generate_galsim_objects( 

59 injection_catalog=self.injection_catalog, 

60 photo_calib=self.exposure.photoCalib, 

61 wcs=self.exposure.wcs, 

62 fits_alignment="wcs", 

63 stamp_prefix="", 

64 ) 

65 

66 def tearDown(self): 

67 del self.exposure 

68 del self.injection_catalog 

69 del self.galsim_objects 

70 

71 def test_make_galsim_object(self): 

72 source_data = self.injection_catalog[0] 

73 sky_coords = SpherePoint(float(source_data["ra"]), float(source_data["dec"]), degrees) 

74 pixel_coords = self.exposure.wcs.skyToPixel(sky_coords) 

75 inst_flux = self.exposure.photoCalib.magnitudeToInstFlux(source_data["mag"], pixel_coords) 

76 object = make_galsim_object( 

77 source_data=source_data, 

78 source_type=source_data["source_type"], 

79 inst_flux=inst_flux, 

80 ) 

81 self.assertIsInstance(object, GSObject) 

82 self.assertIsInstance(object, getattr(galsim, source_data["source_type"])) 

83 

84 def test_generate_galsim_objects(self): 

85 self.assertTrue(isinstance(self.galsim_objects, GeneratorType)) 

86 for galsim_object in self.galsim_objects: 

87 self.assertIsInstance(galsim_object, tuple) 

88 self.assertEqual(len(galsim_object), 4) 

89 self.assertIsInstance(galsim_object[0], SpherePoint) # RA/Dec 

90 self.assertIsInstance(galsim_object[1], Point2D) # x/y 

91 self.assertIsInstance(galsim_object[2], int) # draw size 

92 self.assertIsInstance(galsim_object[3], GSObject) # GSObject 

93 

94 def test_infer_gain_nonexistent_mask(self): 

95 """Test that infer_gain_from_image returns a gain value and logs a 

96 warning when provided with nonexistent mask plane names. 

97 """ 

98 logger = logging.getLogger(__name__) 

99 with self.assertLogs(logger, level="WARNING") as cm: 

100 gain = infer_gain_from_image( 

101 self.exposure, 

102 bad_mask_names=["NONEXISTENT_MASK1", "NONEXISTENT_MASK2"], 

103 logger=logger, 

104 ) 

105 self.assertTrue(any("NONEXISTENT_MASK1" in msg for msg in cm.output)) 

106 self.assertIsInstance(gain, float) 

107 self.assertTrue(np.isfinite(gain)) 

108 gain_no_mask = infer_gain_from_image(self.exposure, bad_mask_names=[]) 

109 self.assertAlmostEqual(gain, gain_no_mask) 

110 

111 def test_infer_gain_no_valid_pixels(self): 

112 """Test that infer_gain_from_image returns NaN (rather than raising) 

113 when a region has no valid pixels to fit. 

114 """ 

115 logger = logging.getLogger(__name__) 

116 # Every pixel flagged with a bad mask plane. 

117 all_bad = make_test_exposure() 

118 all_bad.mask.addMaskPlane("BAD") 

119 all_bad.mask.array[:] = all_bad.mask.getPlaneBitMask("BAD") 

120 with self.assertLogs(logger, level="WARNING"): 

121 gain = infer_gain_from_image(all_bad, bad_mask_names=["BAD"], logger=logger) 

122 self.assertTrue(np.isnan(gain)) 

123 

124 # Every variance pixel non-finite. 

125 all_nan = make_test_exposure() 

126 all_nan.variance.array[:] = np.nan 

127 self.assertTrue(np.isnan(infer_gain_from_image(all_nan, bad_mask_names=[]))) 

128 

129 def test_get_gain_map_no_valid_pixels(self): 

130 """Test that get_gain_map produces a finite, positive map even when no 

131 region can be fit, falling back to unit gain. 

132 """ 

133 all_bad = make_test_exposure() 

134 all_bad.mask.addMaskPlane("BAD") 

135 all_bad.mask.array[:] = all_bad.mask.getPlaneBitMask("BAD") 

136 gain_map = get_gain_map(all_bad, bad_mask_names=["BAD"]) 

137 self.assertTrue(np.all(np.isfinite(gain_map.array))) 

138 self.assertTrue(np.all(gain_map.array > 0)) 

139 

140 def test_inject_galsim_objects_into_exposure(self): 

141 flux0 = np.sum(self.exposure.image.array) 

142 injected_outputs = inject_galsim_objects_into_exposure( 

143 exposure=self.exposure, 

144 objects=self.galsim_objects, 

145 mask_plane_name="INJECTED", 

146 calib_flux_radius=12.0, 

147 draw_size_max=1000, 

148 add_noise=False, 

149 ) 

150 pc = self.exposure.getPhotoCalib() 

151 inst_fluxes = [float(pc.magnitudeToInstFlux(mag)) for mag in self.injection_catalog["mag"]] 

152 self.assertAlmostEqual( 

153 np.sum(self.exposure.image.array) - flux0, 

154 np.sum(inst_fluxes), 

155 delta=0.00015 * np.sum(inst_fluxes), 

156 ) 

157 self.assertEqual(len(injected_outputs[0]), len(self.injection_catalog["ra"])) 

158 self.assertTrue(all(isinstance(injected_output, list) for injected_output in injected_outputs)) 

159 self.assertTrue(all(isinstance(item, int) for item in injected_outputs[0])) # draw sizes 

160 self.assertTrue(all(isinstance(item, BoundsI) for item in injected_outputs[1])) # common bounds 

161 self.assertTrue(all(isinstance(item, bool) for item in injected_outputs[2])) # FFT size errors 

162 self.assertTrue(all(isinstance(item, bool) for item in injected_outputs[3])) # PSF compute errors 

163 

164 

165class MemoryTestCase(lsst.utils.tests.MemoryTestCase): 

166 """Test memory usage of functions in this script.""" 

167 

168 pass 

169 

170 

171def setup_module(module): 

172 """Configure pytest.""" 

173 lsst.utils.tests.init() 

174 

175 

176if __name__ == "__main__": 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true

177 lsst.utils.tests.init() 

178 unittest.main()