Coverage for tests / test_coaddInputAnalysis.py: 26%

95 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-26 09:36 +0000

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 

22import unittest 

23 

24import astropy.table 

25import numpy as np 

26 

27import lsst.utils.tests 

28from lsst.analysis.tools.tasks.coaddInputAnalysis import ( 

29 CoaddInputAnalysisConfig, 

30 CoaddInputAnalysisTask, 

31) 

32from lsst.geom import SpherePoint, degrees 

33from lsst.sphgeom import ConvexPolygon 

34 

35 

36class MockRawRef: 

37 """Minimal stand-in for a butler raw data reference.""" 

38 

39 def __init__(self, detector): 

40 self.dataId = {"detector": detector} 

41 

42 

43def makePatchPoly(ra_center, dec_center, half_size=0.5): 

44 """Return a square ConvexPolygon centred at (ra_center, dec_center).""" 

45 corners = [ 

46 SpherePoint(ra_center - half_size, dec_center - half_size, degrees).getVector(), 

47 SpherePoint(ra_center + half_size, dec_center - half_size, degrees).getVector(), 

48 SpherePoint(ra_center + half_size, dec_center + half_size, degrees).getVector(), 

49 SpherePoint(ra_center - half_size, dec_center + half_size, degrees).getVector(), 

50 ] 

51 return ConvexPolygon.convexHull(corners) 

52 

53 

54def makeImageCornersRow(visit, detector, ra_center, dec_center, half_size=0.1): 

55 """Return a single-row imageCorners Table for one detector.""" 

56 return astropy.table.Table( 

57 { 

58 "visitId": [visit], 

59 "detector": [detector], 

60 "llcra": [ra_center - half_size], 

61 "llcdec": [dec_center - half_size], 

62 "ulcra": [ra_center - half_size], 

63 "ulcdec": [dec_center + half_size], 

64 "urcra": [ra_center + half_size], 

65 "urcdec": [dec_center + half_size], 

66 "lrcra": [ra_center + half_size], 

67 "lrcdec": [dec_center - half_size], 

68 } 

69 ) 

70 

71 

72class TestMakeData(lsst.utils.tests.TestCase): 

73 """Tests for CoaddInputAnalysisTask.makeData.""" 

74 

75 def setUp(self): 

76 self.task = CoaddInputAnalysisTask(config=CoaddInputAnalysisConfig()) 

77 # 1-degree square patch centred at (180, 0). 

78 self.patchPoly = makePatchPoly(180.0, 0.0, half_size=0.5) 

79 

80 def test_overlap_and_in_coadd(self): 

81 """PVI inside patch and in the coadd is correctly flagged.""" 

82 visit, detector = 100, 5 

83 imageCorners = makeImageCornersRow(visit, detector, 180.0, 0.0) 

84 rawsByVisit = {visit: [MockRawRef(detector)]} 

85 inCoadd = {(visit, detector)} 

86 

87 data = self.task.makeData(inCoadd, self.patchPoly, imageCorners, rawsByVisit) 

88 

89 self.assertEqual(len(data), 1) 

90 self.assertTrue(data["patchOverlap"][0]) 

91 self.assertTrue(data["visitSummaryRecord"][0]) 

92 self.assertTrue(data["inCoadd"][0]) 

93 

94 def test_no_overlap_not_in_coadd(self): 

95 """PVI outside patch and not in the coadd is correctly flagged.""" 

96 visit, detector = 100, 5 

97 imageCorners = makeImageCornersRow(visit, detector, 190.0, 0.0) 

98 rawsByVisit = {visit: [MockRawRef(detector)]} 

99 

100 data = self.task.makeData(set(), self.patchPoly, imageCorners, rawsByVisit) 

101 

102 self.assertEqual(len(data), 1) 

103 self.assertFalse(data["patchOverlap"][0]) 

104 self.assertTrue(data["visitSummaryRecord"][0]) 

105 self.assertFalse(data["inCoadd"][0]) 

106 

107 def test_non_finite_corners_treated_as_no_overlap(self): 

108 """ 

109 A PVI with a non-finite corner coordinate is treated as 

110 not overlapping. 

111 """ 

112 visit, detector = 100, 5 

113 imageCorners = makeImageCornersRow(visit, detector, 180.0, 0.0) 

114 imageCorners["llcra"][0] = np.nan 

115 rawsByVisit = {visit: [MockRawRef(detector)]} 

116 

117 data = self.task.makeData(set(), self.patchPoly, imageCorners, rawsByVisit) 

118 

119 self.assertFalse(data["patchOverlap"][0]) 

120 self.assertTrue(data["visitSummaryRecord"][0]) 

121 

122 def test_visit_absent_from_image_corners(self): 

123 """A visit with no imageCorners entry gets all-False flags.""" 

124 visit, detector = 100, 5 

125 imageCorners = astropy.table.Table( 

126 { 

127 col: [] 

128 for col in [ 

129 "visitId", 

130 "detector", 

131 "llcra", 

132 "llcdec", 

133 "ulcra", 

134 "ulcdec", 

135 "urcra", 

136 "urcdec", 

137 "lrcra", 

138 "lrcdec", 

139 ] 

140 } 

141 ) 

142 rawsByVisit = {visit: [MockRawRef(detector)]} 

143 

144 data = self.task.makeData(set(), self.patchPoly, imageCorners, rawsByVisit) 

145 

146 self.assertEqual(len(data), 1) 

147 self.assertFalse(data["visitSummaryRecord"][0]) 

148 self.assertFalse(data["patchOverlap"][0]) 

149 self.assertFalse(data["inCoadd"][0]) 

150 

151 def test_detector_absent_from_visit_corners(self): 

152 """ 

153 A detector with no row in imageCorners gets False for corner-derived 

154 flags. 

155 """ 

156 visit, detector = 100, 5 

157 other_detector = 6 

158 imageCorners = makeImageCornersRow(visit, other_detector, 180.0, 0.0) 

159 rawsByVisit = {visit: [MockRawRef(detector)]} 

160 

161 data = self.task.makeData(set(), self.patchPoly, imageCorners, rawsByVisit) 

162 

163 self.assertFalse(data["visitSummaryRecord"][0]) 

164 self.assertFalse(data["patchOverlap"][0]) 

165 

166 def test_duplicate_detector_row_raises(self): 

167 """ 

168 Duplicate (visit, detector) rows in imageCorners raise RuntimeError. 

169 """ 

170 visit, detector = 100, 5 

171 row = makeImageCornersRow(visit, detector, 180.0, 0.0) 

172 imageCorners = astropy.table.vstack([row, row]) 

173 rawsByVisit = {visit: [MockRawRef(detector)]} 

174 

175 with self.assertRaises(RuntimeError): 

176 self.task.makeData(set(), self.patchPoly, imageCorners, rawsByVisit) 

177 

178 def test_multiple_visits_and_detectors(self): 

179 """ 

180 Multiple visits and detectors produce the correct number of rows with 

181 correct flags. 

182 """ 

183 v1, v2 = 100, 200 

184 d1, d2, d3 = 5, 6, 7 

185 imageCorners = astropy.table.vstack( 

186 [ 

187 makeImageCornersRow(v1, d1, 180.0, 0.0), # overlaps 

188 makeImageCornersRow(v1, d2, 190.0, 0.0), # no overlap 

189 makeImageCornersRow(v2, d3, 180.0, 0.0), # overlaps 

190 ] 

191 ) 

192 rawsByVisit = { 

193 v1: [MockRawRef(d1), MockRawRef(d2)], 

194 v2: [MockRawRef(d3)], 

195 } 

196 inCoadd = {(v1, d1), (v2, d3)} 

197 

198 data = self.task.makeData(inCoadd, self.patchPoly, imageCorners, rawsByVisit) 

199 data.sort(["visit", "detector"]) 

200 

201 self.assertEqual(len(data), 3) 

202 

203 row_v1d1 = data[(data["visit"] == v1) & (data["detector"] == d1)][0] 

204 self.assertTrue(row_v1d1["patchOverlap"]) 

205 self.assertTrue(row_v1d1["inCoadd"]) 

206 

207 row_v1d2 = data[(data["visit"] == v1) & (data["detector"] == d2)][0] 

208 self.assertFalse(row_v1d2["patchOverlap"]) 

209 self.assertFalse(row_v1d2["inCoadd"]) 

210 

211 row_v2d3 = data[(data["visit"] == v2) & (data["detector"] == d3)][0] 

212 self.assertTrue(row_v2d3["patchOverlap"]) 

213 self.assertTrue(row_v2d3["inCoadd"]) 

214 

215 

216class TestMemory(lsst.utils.tests.MemoryTestCase): 

217 pass 

218 

219 

220def setup_module(module): 

221 lsst.utils.tests.init() 

222 

223 

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

225 lsst.utils.tests.init() 

226 unittest.main()