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 not in {"bool", "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 "boolean": bool, 

61 } 

62 # Load the lsstcam.yaml schema resource from sdm_schemas 

63 

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

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

66 # Find ShearObject table and collect column names 

67 for table in schema.tables: 

68 if table.name == "ShearObject": 

69 cols = pa.schema( 

70 [ 

71 pa.field( 

72 col.name, 

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

74 ) 

75 for col in table.columns 

76 ] 

77 ) 

78 return cols 

79 

80 def test_yaml_is_subset_of_generated_schema(self) -> None: 

81 sdm_schema = self._read_shearobject_columns_from_yaml() 

82 # Generate schema from task 

83 config = MetadetectionShearTask.ConfigClass() 

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

85 config.setDefaults() 

86 task_schema = MetadetectionShearTask.make_metadetect_schema(config) 

87 

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

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

90 

91 for field in sdm_schema: 

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

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

94 

95 

96@dataclass(frozen=True) 

97class SkymapConfigs: 

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

99 

100 lsst_cells_v1: RingsSkyMapConfig 

101 lsst_cells_v2: RingsSkyMapConfig 

102 

103 

104class CountCellsAlongEdgesTestCase(unittest.TestCase): 

105 """Test the count_cells_along_edges static method.""" 

106 

107 @classmethod 

108 def setUpClass(cls) -> None: 

109 # Docstring inherited. 

110 

111 ring_skymap_config = RingsSkyMapConfig() 

112 ring_skymap_config.tractBuilder.name = "cells" 

113 ring_skymap_config.numRings = 120 

114 ring_skymap_config.pixelScale = 0.2 

115 ring_skymap_config.projection = "TAN" 

116 ring_skymap_config.raStart = 0.0 

117 ring_skymap_config.rotation = 0.0 

118 ring_skymap_config.tractOverlap = 0.016666666666666666 

119 

120 cell_config = ring_skymap_config.tractBuilder["cells"] 

121 

122 cell_config.cellInnerDimensions = [150, 150] 

123 cell_config.numCellsInPatchBorder = 1 

124 cell_config.numCellsPerPatchInner = 20 

125 

126 cls.standard_skymap_configs = SkymapConfigs( 

127 lsst_cells_v1=ring_skymap_config.copy(), 

128 lsst_cells_v2=ring_skymap_config.copy(), 

129 ) 

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

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

132 cls.standard_skymap_configs.lsst_cells_v1.freeze() 

133 cls.standard_skymap_configs.lsst_cells_v2.freeze() 

134 

135 cls.custom_skymap_config = ring_skymap_config 

136 

137 def setUp(self): 

138 # Docstring inherited. 

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

140 self.metadetection_shear_config = MetadetectionShearConfig() 

141 

142 def test_count_cells_along_edges(self): 

143 """Test count_cells_along_edges""" 

144 

145 lsst_cells_v1 = self.standard_skymap_configs.lsst_cells_v1 

146 num_cells = MetadetectionShearTask.count_cells_along_edges(lsst_cells_v1) 

147 self.assertEqual(num_cells, 0) 

148 

149 lsst_cells_v2 = self.standard_skymap_configs.lsst_cells_v2 

150 num_cells = MetadetectionShearTask.count_cells_along_edges(lsst_cells_v2) 

151 self.assertEqual(num_cells, 1) 

152 

153 def test_validate_skymap_config_lsst_cells_v1(self): 

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

155 self.metadetection_shear_config.border = 0 

156 task = MetadetectionShearTask(config=self.metadetection_shear_config) 

157 

158 # This should not raise an exception 

159 task.validate_skymap_config(self.standard_skymap_configs.lsst_cells_v1) 

160 

161 def test_validate_skymap_config_lsst_cells_v2(self): 

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

163 self.metadetection_shear_config.border = 50 

164 task = MetadetectionShearTask(config=self.metadetection_shear_config) 

165 

166 # This should not raise an exception 

167 task.validate_skymap_config(self.standard_skymap_configs.lsst_cells_v2) 

168 

169 def test_validate_skymap_config_invalid_tract_builder(self): 

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

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

172 

173 task = MetadetectionShearTask(config=self.metadetection_shear_config) 

174 

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

176 task.validate_skymap_config(self.custom_skymap_config) 

177 

178 def test_validate_skymap_config_no_border_in_both_configs(self): 

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

180 cellBorder is 0 in skymap. 

181 """ 

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

183 

184 config = MetadetectionShearConfig() 

185 config.border = 0 

186 

187 task = MetadetectionShearTask(config=config) 

188 

189 with self.assertRaises( 

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

191 ): 

192 task.validate_skymap_config(self.custom_skymap_config) 

193 

194 def test_validate_skymap_config_border_in_both_configs(self): 

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

196 cellBorder in skymap. 

197 """ 

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

199 

200 config = MetadetectionShearConfig() 

201 config.border = 10 

202 

203 task = MetadetectionShearTask(config=config) 

204 

205 with self.assertRaises(InvalidQuantumError) as cm: 

206 task.validate_skymap_config(self.custom_skymap_config) 

207 

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

209 

210 def test_validate_skymap_config_border_too_large(self): 

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

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

213 

214 config = MetadetectionShearConfig() 

215 config.border = 5000 # Very large border 

216 

217 task = MetadetectionShearTask(config=config) 

218 

219 with self.assertRaises(InvalidQuantumError) as cm: 

220 task.validate_skymap_config(self.custom_skymap_config) 

221 

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

223 

224 def test_validate_skymap_config_pixel_scale_mismatch(self): 

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

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

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

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

229 

230 config = MetadetectionShearConfig() 

231 config.border = 10 

232 

233 task = MetadetectionShearTask(config=config) 

234 

235 with self.assertRaises(InvalidQuantumError) as cm: 

236 task.validate_skymap_config(self.custom_skymap_config) 

237 

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

239 

240 def test_validate_skymap_config_success_with_task_border(self): 

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

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

243 

244 config = MetadetectionShearConfig() 

245 config.border = 10 # Reasonable border 

246 

247 task = MetadetectionShearTask(config=config) 

248 

249 # Should not raise an exception 

250 task.validate_skymap_config(self.custom_skymap_config) 

251 

252 def test_validate_skymap_config_success_with_skymap_border(self): 

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

254 only. 

255 """ 

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

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

258 

259 config = MetadetectionShearConfig() 

260 config.border = 0 

261 

262 task = MetadetectionShearTask(config=config) 

263 

264 # Should not raise an exception 

265 task.validate_skymap_config(self.custom_skymap_config) 

266 

267 

268def setup_module(module): 

269 lsst.utils.tests.init() 

270 

271 

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

273 pass 

274 

275 

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

277 lsst.utils.tests.init() 

278 unittest.main()