Coverage for tests/test_aggregatedMetadata.py: 97%

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/>. 

21from __future__ import annotations 

22 

23from unittest import TestCase, main 

24 

25import numpy as np 

26 

27import lsst.utils.tests 

28from lsst.analysis.tools.actions.scalar import CountAction, MaxAction, MedianAction, MinAction 

29from lsst.analysis.tools.atools import AggregatedTaskMetadataMetricTool 

30from lsst.analysis.tools.tasks import AggregatedTaskMetadataAnalysisTask 

31from lsst.analysis.tools.tasks.metadataAnalysis import AggregatedTaskMetadataAnalysisConfig 

32from lsst.pipe.base import UpstreamFailureNoWorkFound 

33 

34_PLOT_INFO = {"tableName": "", "run": "", "plotName": ""} 

35 

36 

37def _make_tool(metrics, aggregation_units=None, subtask_names=None): 

38 """Create and finalize a minimal AggregatedTaskMetadataMetricTool. 

39 

40 Parameters 

41 ---------- 

42 metrics : `dict` [`str`, `str`] 

43 Metric names and their units. 

44 aggregation_units : `dict` [`str`, `str`], optional 

45 Fixed units keyed by aggregation suffix. 

46 subtask_names : `dict` [`str`, `str`], optional 

47 Subtask names keyed by metric name. 

48 

49 Returns 

50 ------- 

51 tool : `AggregatedTaskMetadataMetricTool` 

52 A finalized tool. 

53 """ 

54 tool = AggregatedTaskMetadataMetricTool() 

55 tool.taskName = "test" 

56 tool.metrics = metrics 

57 if subtask_names is not None: 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true

58 tool.subTaskNames = subtask_names 

59 if aggregation_units is not None: 

60 tool.aggregationUnits = aggregation_units 

61 

62 aggregations = {"min": MinAction, "max": MaxAction, "median": MedianAction, "ct": CountAction} 

63 for metric in tool.metrics: 

64 sanitized = metric.replace(" ", "_") 

65 for agg_name, agg_cls in aggregations.items(): 

66 action = agg_cls() 

67 action.vectorKey = metric 

68 setattr(tool.process.calculateActions, f"{sanitized}_{agg_name}", action) 

69 

70 tool.finalize() 

71 return tool 

72 

73 

74def _make_task(metrics, subtask_names=None): 

75 """Create a minimal AggregatedTaskMetadataAnalysisTask. 

76 

77 Parameters 

78 ---------- 

79 metrics : `dict` [`str`, `str`] 

80 Metric names and their units. 

81 subtask_names : `dict` [`str`, `str`], optional 

82 Subtask names keyed by metric name. 

83 

84 Returns 

85 ------- 

86 task : `AggregatedTaskMetadataAnalysisTask` 

87 A configured task. 

88 """ 

89 config = AggregatedTaskMetadataAnalysisConfig() 

90 config.connections.inputName = "test_metadata" 

91 config.connections.outputName = "test_metadata_agg" 

92 config.atools.testTool = AggregatedTaskMetadataMetricTool 

93 config.atools.testTool.taskName = "test" 

94 config.atools.testTool.metrics = metrics 

95 if subtask_names is not None: 

96 config.atools.testTool.subTaskNames = subtask_names 

97 

98 action = MedianAction() 

99 action.vectorKey = next(iter(metrics)) 

100 config.atools.testTool.process.calculateActions.metric_median = action 

101 

102 return AggregatedTaskMetadataAnalysisTask(config=config) 

103 

104 

105class _MockHandle: 

106 """Minimal stand-in for a deferred dataset handle.""" 

107 

108 def __init__(self, metadata_dict): 

109 self._metadata = metadata_dict 

110 

111 def get(self): 

112 metadata = self._metadata 

113 

114 class _MockMetadata: 

115 def to_dict(self): 

116 return metadata 

117 

118 return _MockMetadata() 

119 

120 

121def _make_handles(metadata_dicts): 

122 """Create a list of mock handles from a list of metadata dicts. 

123 

124 Parameters 

125 ---------- 

126 metadata_dicts : `list` [`dict`] 

127 One dict per handle, as returned by ``TaskMetadata.to_dict()``. 

128 

129 Returns 

130 ------- 

131 handles : `list` [`_MockHandle`] 

132 """ 

133 return [_MockHandle(md) for md in metadata_dicts] 

134 

135 

136class TestAggregatedTaskMetadataMetricToolFinalize(TestCase): 

137 """Tests for AggregatedTaskMetadataMetricTool.finalize.""" 

138 

139 def testUnitsInheritedFromSourceMetric(self): 

140 """min/max/median aggregations should inherit the source metric's 

141 unit.""" 

142 tool = _make_tool({"nStars": "ct", "psfSigma": "pixel"}) 

143 units = tool.produce.metric.units 

144 self.assertEqual(units["nStars_min"], "ct") 

145 self.assertEqual(units["nStars_max"], "ct") 

146 self.assertEqual(units["nStars_median"], "ct") 

147 self.assertEqual(units["psfSigma_min"], "pixel") 

148 self.assertEqual(units["psfSigma_max"], "pixel") 

149 self.assertEqual(units["psfSigma_median"], "pixel") 

150 

151 def testAggregationUnitsOverride(self): 

152 """aggregationUnits should override the inherited unit for that 

153 suffix.""" 

154 tool = _make_tool( 

155 {"nStars": "ct", "psfSigma": "pixel"}, 

156 aggregation_units={"ct": "ct"}, 

157 ) 

158 units = tool.produce.metric.units 

159 # Count aggregations should use the override unit regardless of source. 

160 self.assertEqual(units["nStars_ct"], "ct") 

161 self.assertEqual(units["psfSigma_ct"], "ct") 

162 # Other aggregations should still inherit. 

163 self.assertEqual(units["psfSigma_min"], "pixel") 

164 

165 def testSpacesInMetricNames(self): 

166 """Metric names containing spaces must be handled gracefully.""" 

167 tool = _make_tool({"n stars": "ct"}) 

168 units = tool.produce.metric.units 

169 # The action name uses underscores; units should still be populated. 

170 self.assertIn("n_stars_min", units) 

171 self.assertIn("n_stars_max", units) 

172 self.assertIn("n_stars_median", units) 

173 self.assertEqual(units["n_stars_min"], "ct") 

174 

175 def testActionWithUnknownVectorKeyRaisesError(self): 

176 """An action whose vectorKey is not in metrics should raise ValueError. 

177 

178 This guards against misconfiguration where someone assigns a vectorKey 

179 that doesn't correspond to any configured metric. 

180 """ 

181 tool = AggregatedTaskMetadataMetricTool() 

182 tool.taskName = "test" 

183 tool.metrics = {"nStars": "ct"} 

184 

185 known_action = MedianAction() 

186 known_action.vectorKey = "nStars" 

187 tool.process.calculateActions.nStars_median = known_action 

188 

189 unknown_action = MedianAction() 

190 unknown_action.vectorKey = "notAMetric" 

191 tool.process.calculateActions.notAMetric_median = unknown_action 

192 

193 with self.assertRaises(ValueError): 

194 tool.finalize() 

195 

196 

197class TestAggregatedTaskMetadataMetricToolNumerics(TestCase): 

198 """Numerical correctness of aggregations.""" 

199 

200 def setUp(self): 

201 self.values = np.array([10.0, 20.0, 30.0, 40.0, 50.0]) 

202 self.tool = _make_tool({"nStars": "ct"}, aggregation_units={"ct": "ct"}) 

203 

204 def _call_tool(self, data): 

205 return self.tool(data, plotInfo=dict(_PLOT_INFO)) 

206 

207 def testMin(self): 

208 result = self._call_tool({"nStars": self.values}) 

209 self.assertAlmostEqual(result["nStars_min"].quantity.value, 10.0) 

210 

211 def testMax(self): 

212 result = self._call_tool({"nStars": self.values}) 

213 self.assertAlmostEqual(result["nStars_max"].quantity.value, 50.0) 

214 

215 def testMedian(self): 

216 result = self._call_tool({"nStars": self.values}) 

217 self.assertAlmostEqual(result["nStars_median"].quantity.value, 30.0) 

218 

219 def testCount(self): 

220 result = self._call_tool({"nStars": self.values}) 

221 self.assertAlmostEqual(result["nStars_ct"].quantity.value, 5.0) 

222 

223 def testCountExcludesNaN(self): 

224 """Count should exclude NaN values.""" 

225 values_with_nan = np.array([10.0, np.nan, 30.0, np.nan, 50.0]) 

226 result = self._call_tool({"nStars": values_with_nan}) 

227 self.assertAlmostEqual(result["nStars_ct"].quantity.value, 3.0) 

228 

229 

230class TestAggregatedTaskMetadataAnalysisTask(TestCase): 

231 """Tests for AggregatedTaskMetadataAnalysisTask._collectData.""" 

232 

233 def testEmptyMetadataRaisesNoWorkFound(self): 

234 """All inputs returning empty metadata should raise NoWorkFound.""" 

235 task = _make_task({"nStars": "ct"}) 

236 handles = _make_handles([{}, {}, {}]) 

237 with self.assertRaises(UpstreamFailureNoWorkFound): 

238 task._collectData(handles, "test") 

239 

240 def testPartialInputsAccepted(self): 

241 """Inputs missing a metric in some (but not all) detectors are fine.""" 

242 task = _make_task({"nStars": "ct"}) 

243 # One input lacks nStars — its value should be NaN. 

244 handles = _make_handles( 

245 [ 

246 {"test": {"nStars": 10.0}}, 

247 {"test": {}}, 

248 {"test": {"nStars": 30.0}}, 

249 ] 

250 ) 

251 data = task._collectData(handles, "test") 

252 self.assertEqual(len(data["nStars"]), 3) 

253 self.assertEqual(data["nStars"][0], 10.0) 

254 self.assertTrue(np.isnan(data["nStars"][1])) 

255 self.assertEqual(data["nStars"][2], 30.0) 

256 

257 def testSubTaskNamesRespected(self): 

258 """Metrics in subtasks should be found under the correct key.""" 

259 task = _make_task( 

260 {"cosmicRayCount": "ct"}, 

261 subtask_names={"cosmicRayCount": "repair"}, 

262 ) 

263 handles = _make_handles( 

264 [ 

265 {"test:repair": {"cosmicRayCount": 5.0}}, 

266 {"test:repair": {"cosmicRayCount": 8.0}}, 

267 ] 

268 ) 

269 data = task._collectData(handles, "test") 

270 self.assertIn("cosmicRayCount", data) 

271 self.assertEqual(len(data["cosmicRayCount"]), 2) 

272 

273 def testTypoInMetricNameRaisesNoWorkFound(self): 

274 """A metric name that doesn't match any key in the metadata raises 

275 NoWorkFound, covering the case of a configuration typo. 

276 """ 

277 task = _make_task({"nSatrs": "ct"}) # typo: nSatrs instead of nStars 

278 handles = _make_handles( 

279 [ 

280 {"test": {"nStars": 10.0}}, 

281 {"test": {"nStars": 20.0}}, 

282 ] 

283 ) 

284 with self.assertRaises(UpstreamFailureNoWorkFound): 

285 task._collectData(handles, "test") 

286 

287 

288class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase): 

289 pass 

290 

291 

292def setup_module(module): 

293 lsst.utils.tests.init() 

294 

295 

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

297 lsst.utils.tests.init() 

298 main()