Coverage for tests/test_metadetection_shear.py: 26%

1# This file is part of drp_tasks. 

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 unittest 

24from dataclasses import dataclass 

25 

26import numpy as np 

27import pyarrow as pa 

28from felis.datamodel import Schema 

29 

30import lsst.utils.tests 

31from lsst.drp.tasks.metadetection_shear import MetadetectionShearConfig, MetadetectionShearTask 

32from lsst.pipe.base import InvalidQuantumError 

33from lsst.resources import ResourcePath 

34from lsst.skymap import RingsSkyMapConfig 

35 

36 

37class ShearObjectSchemaConsistencyTestCase(unittest.TestCase): 

38 """Verify that the ShearObject table in lsstcam.yaml is a subset of 

39 the schema produced by MetadetectionShearTask.make_metadetect_schema. 

40 """ 

41 

42 def assertFieldsEqual(self, field1, field2): 

43 """Assert that the two fields are identical. 

44 

45 Specifically, it checks that the following are identical: 

46 - name 

47 - type width (if not string) 

48 """ 

49 self.assertEqual(field1.name, field2.name) 

50 if field1.type != "string": 

51 self.assertEqual(field1.type.byte_width, field2.type.byte_width) 

52 

53 def _read_shearobject_columns_from_yaml(self) -> pa.Schema: 

54 dtypes = { 

55 "char": "U2", 

56 "int": np.int32, 

57 "uint": np.uint32, 

58 "float": np.float32, 

59 "double": np.float64, 

60 } 

61 # Load the lsstcam.yaml schema resource from sdm_schemas 

62 

63 resource = ResourcePath("resource://lsst.sdm.schemas/lsstcam.yaml") 

64 schema = Schema.from_uri(resource, context={"id_generation": True}) 

65 # Find ShearObject table and collect column names 

66 for table in schema.tables: 

67 if table.name == "ShearObject": 

68 cols = pa.schema( 

69 [ 

70 pa.field( 

71 col.name, 

72 pa.from_numpy_dtype(dtypes.get(col.datatype, col.datatype)), 

73 ) 

74 for col in table.columns 

75 ] 

76 ) 

77 return cols 

78 

79 def test_yaml_is_subset_of_generated_schema(self) -> None: 

80 sdm_schema = self._read_shearobject_columns_from_yaml() 

81 # Generate schema from task 

82 config = MetadetectionShearTask.ConfigClass() 

83 # Ensure defaults are applied (e.g., shear_bands) 

84 config.setDefaults() 

85 task_schema = MetadetectionShearTask.make_metadetect_schema(config) 

86 

87 missing_field_names = set(sdm_schema.names).difference(task_schema.names) 

88 self.assertFalse(missing_field_names, f"Missing fields in generated schema: {missing_field_names}") 

89 

90 for field in sdm_schema: 

91 with self.subTest(field_name=field.name): 

92 self.assertFieldsEqual(field, task_schema.field(field.name)) 

93 

94 

95@dataclass(frozen=True) 

96class SkymapConfigs: 

97 """Immutable configuration container for standard skymap settings.""" 

98 

99 lsst_cells_v1: RingsSkyMapConfig 

100 lsst_cells_v2: RingsSkyMapConfig 

101 

102 

103class CountCellsAlongEdgesTestCase(unittest.TestCase): 

104 """Test the count_cells_along_edges static method.""" 

105 

106 @classmethod 

107 def setUpClass(cls) -> None: 

108 # Docstring inherited. 

109 

110 ring_skymap_config = RingsSkyMapConfig() 

111 ring_skymap_config.tractBuilder.name = "cells" 

112 ring_skymap_config.numRings = 120 

113 ring_skymap_config.pixelScale = 0.2 

114 ring_skymap_config.projection = "TAN" 

115 ring_skymap_config.raStart = 0.0 

116 ring_skymap_config.rotation = 0.0 

117 ring_skymap_config.tractOverlap = 0.016666666666666666 

118 

119 cell_config = ring_skymap_config.tractBuilder["cells"] 

120 

121 cell_config.cellInnerDimensions = [150, 150] 

122 cell_config.numCellsInPatchBorder = 1 

123 cell_config.numCellsPerPatchInner = 20 

124 

125 cls.standard_skymap_configs = SkymapConfigs( 

126 lsst_cells_v1=ring_skymap_config.copy(), 

127 lsst_cells_v2=ring_skymap_config.copy(), 

128 ) 

129 cls.standard_skymap_configs.lsst_cells_v1.tractBuilder["cells"].cellBorder = 50 

130 cls.standard_skymap_configs.lsst_cells_v2.tractBuilder["cells"].cellBorder = 0 

131 cls.standard_skymap_configs.lsst_cells_v1.freeze() 

132 cls.standard_skymap_configs.lsst_cells_v2.freeze() 

133 

134 cls.custom_skymap_config = ring_skymap_config 

135 

136 def setUp(self): 

137 # Docstring inherited. 

138 self.custom_skymap_config = self.standard_skymap_configs.lsst_cells_v1.copy() 

139 self.metadetection_shear_config = MetadetectionShearConfig() 

140 

141 def test_count_cells_along_edges(self): 

142 """Test count_cells_along_edges""" 

143 

144 lsst_cells_v1 = self.standard_skymap_configs.lsst_cells_v1 

145 num_cells = MetadetectionShearTask.count_cells_along_edges(lsst_cells_v1) 

146 self.assertEqual(num_cells, 0) 

147 

148 lsst_cells_v2 = self.standard_skymap_configs.lsst_cells_v2 

149 num_cells = MetadetectionShearTask.count_cells_along_edges(lsst_cells_v2) 

150 self.assertEqual(num_cells, 1) 

151 

152 def test_validate_skymap_config_lsst_cells_v1(self): 

153 """Test validation passes with lsst_cells_v1 skymap.""" 

154 self.metadetection_shear_config.border = 0 

155 task = MetadetectionShearTask(config=self.metadetection_shear_config) 

156 

157 # This should not raise an exception 

158 task.validate_skymap_config(self.standard_skymap_configs.lsst_cells_v1) 

159 

160 def test_validate_skymap_config_lsst_cells_v2(self): 

161 """Test validation passes with lsst_cells_v2 skymap.""" 

162 self.metadetection_shear_config.border = 50 

163 task = MetadetectionShearTask(config=self.metadetection_shear_config) 

164 

165 # This should not raise an exception 

166 task.validate_skymap_config(self.standard_skymap_configs.lsst_cells_v2) 

167 

168 def test_validate_skymap_config_invalid_tract_builder(self): 

169 """Test validation fails with non-cells tract builder.""" 

170 self.custom_skymap_config.tractBuilder.name = "legacy" 

171 

172 task = MetadetectionShearTask(config=self.metadetection_shear_config) 

173 

174 with self.assertRaises(InvalidQuantumError, msg="requires a cell-based skymap"): 

175 task.validate_skymap_config(self.custom_skymap_config) 

176 

177 def test_validate_skymap_config_no_border_in_both_configs(self): 

178 """Test validation fails when border is 0 in task config and 

179 cellBorder is 0 in skymap. 

180 """ 

181 self.custom_skymap_config.tractBuilder["cells"].cellBorder = 0 

182 

183 config = MetadetectionShearConfig() 

184 config.border = 0 

185 

186 task = MetadetectionShearTask(config=config) 

187 

188 with self.assertRaises( 

189 InvalidQuantumError, msg="requires a positive border to be set either in the skymap config" 

190 ): 

191 task.validate_skymap_config(self.custom_skymap_config) 

192 

193 def test_validate_skymap_config_border_in_both_configs(self): 

194 """Test validation fails when border is set in both task config and 

195 cellBorder in skymap. 

196 """ 

197 self.custom_skymap_config.tractBuilder["cells"].cellBorder = 2 

198 

199 config = MetadetectionShearConfig() 

200 config.border = 10 

201 

202 task = MetadetectionShearTask(config=config) 

203 

204 with self.assertRaises(InvalidQuantumError) as cm: 

205 task.validate_skymap_config(self.custom_skymap_config) 

206 

207 self.assertIn("requires a positive border to be set either in the skymap config", str(cm.exception)) 

208 

209 def test_validate_skymap_config_border_too_large(self): 

210 """Test validation fails when border value exceeds maximum allowed.""" 

211 self.custom_skymap_config.tractBuilder["cells"].cellBorder = 0 # No cell border 

212 

213 config = MetadetectionShearConfig() 

214 config.border = 5000 # Very large border 

215 

216 task = MetadetectionShearTask(config=config) 

217 

218 with self.assertRaises(InvalidQuantumError) as cm: 

219 task.validate_skymap_config(self.custom_skymap_config) 

220 

221 self.assertIn("border value is too large", str(cm.exception)) 

222 

223 def test_validate_skymap_config_pixel_scale_mismatch(self): 

224 """Test validation fails when pixel scale calculations don't match.""" 

225 self.custom_skymap_config.tractBuilder["cells"].cellBorder = 0 

226 self.custom_skymap_config.pixelScale = 0.1 # Very small pixel scale 

227 self.custom_skymap_config.tractOverlap = 0.001 # Small tract overlap 

228 

229 config = MetadetectionShearConfig() 

230 config.border = 10 

231 

232 task = MetadetectionShearTask(config=config) 

233 

234 with self.assertRaises(InvalidQuantumError) as cm: 

235 task.validate_skymap_config(self.custom_skymap_config) 

236 

237 self.assertIn("tract overlap is insufficient given the borders", str(cm.exception)) 

238 

239 def test_validate_skymap_config_success_with_task_border(self): 

240 """Test validation succeeds when border is set in task config only.""" 

241 self.custom_skymap_config.tractBuilder["cells"].cellBorder = 0 # No cell border 

242 

243 config = MetadetectionShearConfig() 

244 config.border = 10 # Reasonable border 

245 

246 task = MetadetectionShearTask(config=config) 

247 

248 # Should not raise an exception 

249 task.validate_skymap_config(self.custom_skymap_config) 

250 

251 def test_validate_skymap_config_success_with_skymap_border(self): 

252 """Test validation succeeds when border is set in skymap config 

253 only. 

254 """ 

255 self.custom_skymap_config.tractBuilder["cells"].cellBorder = 2 # Has cell border 

256 self.custom_skymap_config.tractBuilder["cells"].numCellsInPatchBorder = 1 

257 

258 config = MetadetectionShearConfig() 

259 config.border = 0 

260 

261 task = MetadetectionShearTask(config=config) 

262 

263 # Should not raise an exception 

264 task.validate_skymap_config(self.custom_skymap_config) 

265 

266 

267def setup_module(module): 

268 lsst.utils.tests.init() 

269 

270 

271class MatchMemoryTestCase(lsst.utils.tests.MemoryTestCase): 

272 pass 

273 

274 

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

276 lsst.utils.tests.init() 

277 unittest.main()