Coverage for tests/test_completenessPlot.py: 87%

1# This file is part of analysis_tools. 

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 

22 

23import os 

24import shutil 

25import tempfile 

26import unittest 

27 

28import lsst.utils.tests 

29import matplotlib 

30import matplotlib.pyplot as plt 

31import numpy as np 

32from lsst.analysis.tools.actions.plot import CompletenessHist 

33from lsst.analysis.tools.actions.plot.plotUtils import get_and_remove_figure_text 

34from lsst.analysis.tools.math import divide, sqrt 

35 

36# Set to True to debug plot test(s) 

37debug_plot = False 

38if not debug_plot: 38 ↛ 41line 38 didn't jump to line 41 because the condition on line 38 was always true

39 matplotlib.use("Agg") 

40 

41ROOT = os.path.abspath(os.path.dirname(__file__)) 

42filename_texts_ref = os.path.join(ROOT, "data", "test_completenessPlot_texts.txt") 

43path_lines_ref = os.path.join(ROOT, "data", "test_completenessPlot_lines") 

44 

45 

46class CompletenessPlotTestCase(lsst.utils.tests.TestCase): 

47 """CompletenessHist test case.""" 

48 

49 def setUp(self): 

50 self.testDir = tempfile.mkdtemp(dir=ROOT, prefix="test_output_completeness") 

51 

52 plot = CompletenessHist() 

53 

54 plot.action.action.selector_range_ref.vectorKey = "mag_ref" 

55 plot.action.action.selector_range_target.vectorKey = "mag_target" 

56 plot.action.bins.mag_low_max = 25500 

57 plot.action.bins.mag_interval = 100 

58 plot.action.bins.mag_width = 200 

59 

60 band = "r" 

61 

62 mag = 12.5 + 2.5 * np.log10(np.arange(10, 100000)) 

63 zeropoint = mag[-1] + 1 

64 flux = 10 ** (-0.4 * (mag - zeropoint)) 

65 rng = np.random.default_rng(0) 

66 # See usage below if changing this to allow nan/bad extendedness 

67 extendedness = 0.0 + (rng.uniform(size=len(mag)) < 0.99 * (mag - mag[0]) / (mag[-1] - mag[0])) 

68 flux_meas = flux + rng.normal(scale=sqrt(flux * (1 + extendedness))) 

69 flux_pos = flux_meas > 0 

70 flux_err = np.empty_like(flux_meas) 

71 flux_err[flux_pos] = sqrt(flux_meas[flux_pos]) 

72 flux_err[~flux_pos] = 0 

73 detect_sn = divide(flux_meas, flux_err) 

74 good = detect_sn > 3 

75 match_distance = np.full_like(detect_sn, np.nan) 

76 rand_norm_abs = np.abs(rng.normal(size=match_distance.size)) 

77 # make some unmatched as a function of S/N 

78 matched = rand_norm_abs < 0.2 * detect_sn 

79 match_distance[matched] = rand_norm_abs[matched] 

80 

81 flux_meas[~good] = np.nan 

82 

83 # estimated from a DC2 tract 

84 prob_ps = 0.5 * (1 + np.tanh((19.2 - mag) / 2.8)) 

85 is_ps = rng.uniform(size=len(prob_ps)) < prob_ps 

86 is_extended = rng.uniform(size=len(prob_ps)) > prob_ps 

87 is_extended[~good] = np.nan 

88 

89 # add some false detections 

90 false_positive = np.abs(rng.normal(size=match_distance.size)) > 0.5 * detect_sn 

91 flux_meas_extra = flux_meas[false_positive] 

92 nan_extra = np.full_like(flux_meas_extra, np.nan) 

93 action_hist = plot.action.action 

94 action_hist.key_mask_ref = "select_is_ref" 

95 action_hist.key_mask_target = "select_is_target" 

96 is_ps = np.concatenate((is_ps, nan_extra)) 

97 extendedness = np.concatenate((extendedness, extendedness[false_positive])) 

98 is_extended = np.concatenate((is_extended, is_extended[false_positive])) 

99 

100 data = { 

101 "mag_ref": np.concatenate((mag, nan_extra)), 

102 "mag_target": -2.5 * np.log10(np.concatenate((flux_meas, flux_meas_extra))) + zeropoint, 

103 "refcat_is_pointsource": is_ps, 

104 action_hist.key_match_distance: np.concatenate((match_distance, nan_extra)), 

105 action_hist.key_matched_class: is_ps == ~is_extended, 

106 action_hist.key_mask_ref: np.isfinite(is_ps), 

107 action_hist.key_mask_target: np.isfinite(extendedness), 

108 } 

109 

110 results = plot.action(data, band="r") 

111 data.update(results) 

112 self.band = "r" 

113 self.data = data 

114 self.plot = plot 

115 self.plotInfo = {"plotName": "compurity test", "run": "", "tableName": "", "bands": band} 

116 

117 def tearDown(self): 

118 if os.path.exists(self.testDir): 118 ↛ 120line 118 didn't jump to line 120 because the condition on line 118 was always true

119 shutil.rmtree(self.testDir, True) 

120 del self.data 

121 del self.plot 

122 del self.plotInfo 

123 del self.testDir 

124 

125 def test_completenessPlotWithTwoHistsTask(self): 

126 plt.rcParams.update(plt.rcParamsDefault) 

127 result = self.plot( 

128 data=self.data, 

129 band=self.band, 

130 plotInfo=self.plotInfo, 

131 ) 

132 self.assertTrue(isinstance(result, plt.Figure)) 

133 

134 to_compare = True 

135 if to_compare: 135 ↛ 174line 135 didn't jump to line 174 because the condition on line 135 was always true

136 # Set to true to save plots as PNGs 

137 # Use matplotlib.testing.compare.compare_images if needed 

138 save_images = False 

139 if save_images: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 result.savefig(os.path.join(ROOT, "data", "test_completenessPlot.png")) 

141 

142 texts, lines = get_and_remove_figure_text(result) 

143 if save_images: 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true

144 result.savefig(os.path.join(ROOT, "data", "test_completenessPlot_unlabeled.png")) 

145 

146 # Set to true to re-generate reference data 

147 resave = False 

148 

149 # Compare line values 

150 for idx, line in enumerate(lines): 

151 filename = os.path.join(path_lines_ref, f"line_{idx}.txt") 

152 if resave: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 np.savetxt(filename, line) 

154 arr = np.loadtxt(filename) 

155 # Differences of order 1e-12 possible between MacOS and Linux 

156 # Plots are generally not expected to be that precise 

157 # Differences to 1e-3 should not be visible with this test data 

158 # nans are possible here (divide by zero) 

159 self.assertFloatsAlmostEqual(arr, line, atol=1e-3, rtol=1e-4, ignoreNaNs=True) 

160 

161 # Ensure that newlines within labels are replaced by a sentinel 

162 newline = "\n" 

163 newline_replace = "[newline]" 

164 # Compare text labels 

165 if resave: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true

166 with open(filename_texts_ref, "w") as f: 

167 f.writelines(f"{text.strip().replace(newline, newline_replace)}\n" for text in texts) 

168 

169 with open(filename_texts_ref, "r") as f: 

170 texts_ref = set(x.strip() for x in f.readlines()) 

171 texts_set = set(x.strip().replace(newline, newline_replace) for x in texts) 

172 

173 self.assertEqual(texts_ref, texts_set) 

174 if debug_plot: 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true

175 plt.show() 

176 

177 

178class MemoryTester(lsst.utils.tests.MemoryTestCase): 

179 pass 

180 

181 

182def setup_module(module): 

183 lsst.utils.tests.init() 

184 

185 

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

187 lsst.utils.tests.init() 

188 unittest.main()