Coverage for tests / test_defineVisits.py: 23%

1# This file is part of obs_base. 

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 <http://www.gnu.org/licenses/>. 

21 

22import os 

23import pickle 

24import shutil 

25import tempfile 

26import unittest 

27import warnings 

28from collections import defaultdict 

29 

30import lsst.daf.butler.tests as butlerTests 

31from lsst.daf.butler import DataCoordinate, DimensionRecord, SerializedDimensionRecord 

32from lsst.daf.butler.registry import ConflictingDefinitionError 

33from lsst.obs.base import DefineVisitsConfig, DefineVisitsTask 

34from lsst.obs.base.instrument_tests import DummyCam 

35from lsst.utils.iteration import ensure_iterable 

36 

37TESTDIR = os.path.dirname(__file__) 

38DATADIR = os.path.join(TESTDIR, "data", "visits") 

39 

40 

41class DefineVisitsBase: 

42 """General set up that can be shared.""" 

43 

44 use_data_ids = True 

45 """Use data IDs when calling defineVisits, else use dimension records.""" 

46 

47 def setUpExposures(self): 

48 """Create a new butler for each test since we are changing dimension 

49 records. 

50 """ 

51 self.root = tempfile.mkdtemp(dir=TESTDIR) 

52 self.creatorButler = butlerTests.makeTestRepo(self.root, {}) 

53 self.enterContext(self.creatorButler) 

54 self.butler = butlerTests.makeTestCollection(self.creatorButler, uniqueId=self.id()) 

55 self.enterContext(self.butler) 

56 

57 self.config = self.get_config() 

58 self.task = DefineVisitsTask(config=self.config, butler=self.butler) 

59 

60 # Need to register the instrument. 

61 DummyCam().register(self.butler.registry) 

62 

63 # Choose serializations based on universe. 

64 universe = self.butler.dimensions 

65 uversion = universe.version 

66 # Not all universe changes result in visible changes. 

67 match uversion: 

68 case uversion if uversion < 2: 

69 raise unittest.SkipTest(f"Universe {uversion} is not compatible with these test files.") 

70 case 2 | 3 | 4 | 5: 

71 # has_simulated, azimuth, seq_start, seq_end. 

72 v = 2 

73 case 6: 

74 # group not group_name, group_id dropped. 

75 v = 6 

76 case 7: 

77 # can_see_sky. 

78 v = 7 

79 case _: 

80 # Might work. 

81 warnings.warn(f"Universe {uversion} has not been validated.") 

82 v = 7 

83 

84 # Read the exposure records. 

85 self.records: dict[int, DimensionRecord] = {} 

86 for i in (347, 348, 349): 

87 with open(os.path.join(DATADIR, f"exp_v{v}_{i}.json")) as fh: 

88 simple = SerializedDimensionRecord.model_validate_json(fh.read()) 

89 self.records[i] = DimensionRecord.from_simple(simple, registry=self.butler.registry) 

90 

91 def define_visits( 

92 self, 

93 exposures: list[DimensionRecord | list[DimensionRecord]], 

94 incremental: bool, 

95 ) -> None: 

96 for records in exposures: 

97 records = list(ensure_iterable(records)) 

98 if "group" in self.butler.dimensions["exposure"].implied: 

99 # This is a group + day_obs universe. 

100 for rec in records: 

101 self.butler.registry.syncDimensionData( 

102 "group", dict(instrument=rec.instrument, name=rec.group) 

103 ) 

104 self.butler.registry.syncDimensionData( 

105 "day_obs", dict(instrument=rec.instrument, id=rec.day_obs) 

106 ) 

107 

108 self.butler.registry.insertDimensionData("exposure", *records) 

109 # Include all records so far in definition. 

110 if self.use_data_ids: 

111 dataIds = list(self.butler.registry.queryDataIds("exposure", instrument="DummyCam")) 

112 else: 

113 dataIds = records 

114 self.task.run(dataIds, incremental=incremental) 

115 

116 

117class DefineVisitsTestCase(unittest.TestCase, DefineVisitsBase): 

118 """Test visit definition.""" 

119 

120 def setUp(self): 

121 self.setUpExposures() 

122 

123 def tearDown(self): 

124 if self.root is not None: 

125 shutil.rmtree(self.root, ignore_errors=True) 

126 

127 def get_config(self) -> DefineVisitsConfig: 

128 return DefineVisitsTask.ConfigClass() 

129 

130 def assertVisits(self): 

131 """Check that the visits were registered as expected.""" 

132 visits = list(self.butler.registry.queryDimensionRecords("visit")) 

133 self.assertEqual(len(visits), 4) 

134 self.assertEqual( 

135 {visit.id for visit in visits}, {2022040500347, 2022040500348, 2022040500349, 92022040500348} 

136 ) 

137 

138 # Ensure that the definitions are correct (ignoring order). 

139 defmap = defaultdict(set) 

140 definitions = list(self.butler.registry.queryDimensionRecords("visit_definition")) 

141 for defn in definitions: 

142 defmap[defn.visit].add(defn.exposure) 

143 

144 self.assertEqual( 

145 dict(defmap), 

146 { 

147 92022040500348: {2022040500348}, 

148 2022040500347: {2022040500347}, 

149 2022040500348: {2022040500348, 2022040500349}, 

150 2022040500349: {2022040500349}, 

151 }, 

152 ) 

153 

154 def test_defineVisits(self): 

155 # Test visit definition with all the records. 

156 self.define_visits([list(self.records.values())], incremental=False) # list inside a list 

157 self.assertVisits() 

158 

159 def test_incremental_cumulative(self): 

160 # Define the visits after each exposure. 

161 self.define_visits(list(self.records.values()), incremental=True) 

162 self.assertVisits() 

163 

164 def test_incremental_cumulative_reverse(self): 

165 # In reverse order we should still eventually end up with the right 

166 # answer. 

167 with self.assertLogs("lsst.defineVisits.groupExposures", level="WARNING") as cm: 

168 self.define_visits(list(reversed(self.records.values())), incremental=True) 

169 self.assertIn("Skipping the multi-snap definition", "\n".join(cm.output)) 

170 self.assertVisits() 

171 

172 def define_visits_incrementally(self, exposure: DimensionRecord) -> None: 

173 if "group" in self.butler.dimensions["exposure"].implied: 

174 self.butler.registry.syncDimensionData( 

175 "group", dict(instrument=exposure.instrument, name=exposure.group) 

176 ) 

177 self.butler.registry.syncDimensionData( 

178 "day_obs", 

179 dict( 

180 instrument=exposure.instrument, 

181 id=exposure.day_obs, 

182 ), 

183 ) 

184 self.butler.registry.insertDimensionData("exposure", exposure) 

185 dataIds = [ 

186 DataCoordinate.standardize( 

187 instrument="DummyCam", exposure=exposure.id, universe=self.butler.dimensions 

188 ) 

189 ] 

190 self.task.run(dataIds, incremental=True) 

191 

192 def test_incremental(self): 

193 for record in self.records.values(): 

194 self.define_visits_incrementally(record) 

195 self.assertVisits() 

196 

197 def test_incremental_reverse(self): 

198 for record in reversed(self.records.values()): 

199 self.define_visits_incrementally(record) 

200 self.assertVisits() 

201 

202 def testPickleTask(self): 

203 stream = pickle.dumps(self.task) 

204 copy = pickle.loads(stream) 

205 self.enterContext(copy.butler) 

206 self.assertEqual(self.task.getFullName(), copy.getFullName()) 

207 self.assertEqual(self.task.log.name, copy.log.name) 

208 self.assertEqual(self.task.config, copy.config) 

209 self.assertEqual(self.task.butler._config, copy.butler._config) 

210 self.assertEqual(list(self.task.butler.collections.defaults), list(copy.butler.collections.defaults)) 

211 self.assertEqual(self.task.butler.run, copy.butler.run) 

212 self.assertEqual(self.task.universe, copy.universe) 

213 

214 

215class DefineVisitsRecordsTestCase(DefineVisitsTestCase): 

216 """Define visits using only dimension records.""" 

217 

218 use_data_ids = False 

219 

220 

221class DefineVisitsGroupingTestCase(unittest.TestCase, DefineVisitsBase): 

222 """Test visit grouping by group metadata.""" 

223 

224 def setUp(self): 

225 self.setUpExposures() 

226 

227 def tearDown(self): 

228 if self.root is not None: 

229 shutil.rmtree(self.root, ignore_errors=True) 

230 

231 def get_config(self) -> DefineVisitsConfig: 

232 config = DefineVisitsTask.ConfigClass() 

233 config.groupExposures.name = "by-group-metadata" 

234 return config 

235 

236 def test_defineVisits(self): 

237 # Test visit definition with all the records. 

238 self.define_visits([list(self.records.values())], incremental=False) # list inside a list 

239 self.assertVisits() 

240 

241 def assertVisits(self): 

242 """Check that the visits were registered as expected.""" 

243 visits = list(self.butler.registry.queryDimensionRecords("visit")) 

244 self.assertEqual(len(visits), 2) 

245 

246 # The visit ID itself depends on which universe we are using. 

247 # It is either calculated or comes from the JSON record. 

248 if "group" in self.butler.dimensions["exposure"].implied: 

249 visit_ids = [20220406025653255, 20220406025807181] 

250 else: 

251 visit_ids = [2291434132550000, 2291434871810000] 

252 self.assertEqual({visit.id for visit in visits}, set(visit_ids)) 

253 

254 # Ensure that the definitions are correct (ignoring order). 

255 defmap = defaultdict(set) 

256 definitions = list(self.butler.registry.queryDimensionRecords("visit_definition")) 

257 for defn in definitions: 

258 defmap[defn.visit].add(defn.exposure) 

259 

260 self.assertEqual( 

261 dict(defmap), 

262 { 

263 visit_ids[0]: {2022040500347}, 

264 visit_ids[1]: {2022040500348, 2022040500349}, 

265 }, 

266 ) 

267 

268 

269class DefineVisitsGroupingRecordsTestCase(DefineVisitsGroupingTestCase): 

270 """Test using dimension records instead of Data IDs.""" 

271 

272 use_data_ids = False 

273 

274 

275class DefineVisitsOneToOneTestCase(unittest.TestCase, DefineVisitsBase): 

276 """Test visit grouping by group metadata.""" 

277 

278 def setUp(self): 

279 self.setUpExposures() 

280 

281 def tearDown(self): 

282 if self.root is not None: 

283 shutil.rmtree(self.root, ignore_errors=True) 

284 

285 def get_config(self) -> DefineVisitsConfig: 

286 config = DefineVisitsTask.ConfigClass() 

287 config.groupExposures.name = "one-to-one" 

288 return config 

289 

290 def test_defineVisits(self): 

291 # Test visit definition with all the records. 

292 self.define_visits([list(self.records.values())], incremental=False) # list inside a list 

293 self.assertVisits() 

294 

295 def assertVisits(self): 

296 """Check that the visits were registered as expected.""" 

297 visits = list(self.butler.registry.queryDimensionRecords("visit")) 

298 self.assertEqual(len(visits), 3) 

299 

300 # For one-to-one the visit ID is the exposure ID. 

301 visit_ids = [rec.id for rec in self.records.values()] 

302 self.assertEqual({visit.id for visit in visits}, set(visit_ids)) 

303 

304 # Ensure that the definitions are correct (ignoring order). 

305 defmap = defaultdict(set) 

306 definitions = list(self.butler.registry.queryDimensionRecords("visit_definition")) 

307 for defn in definitions: 

308 defmap[defn.visit].add(defn.exposure) 

309 

310 self.assertEqual( 

311 dict(defmap), 

312 { 

313 visit_ids[0]: {2022040500347}, 

314 visit_ids[1]: {2022040500348}, 

315 visit_ids[2]: {2022040500349}, 

316 }, 

317 ) 

318 

319 def test_update_records(self): 

320 self.define_visits([list(self.records.values())], incremental=False) # list inside a list 

321 self.assertVisits() 

322 

323 # Modify one of the records. 

324 records = self.records 

325 simple = records[348].to_simple() 

326 simple.record["target_name"] = "new target" 

327 records[348] = DimensionRecord.from_simple(simple, universe=self.butler.dimensions) 

328 self.butler.registry.syncDimensionData("exposure", records[348], update=True) 

329 

330 # Re-run without updates or skipping should fail. 

331 with self.assertRaises(ConflictingDefinitionError): 

332 self.task.run(records.values()) 

333 

334 result = self.task.run(records.values(), skip_conflicting=True) 

335 self.assertEqual(result.n_skipped, 3, str(result)) 

336 

337 # Check that the visit definition did not change. 

338 visit_348 = self.butler.query_dimension_records("visit", where="visit.seq_num = 348")[0] 

339 self.assertEqual(visit_348.target_name, "LATISS_E6A_00000040", visit_348) 

340 

341 # Run with forced update. 

342 result = self.task.run(records.values(), skip_conflicting=True, update_records=True) 

343 

344 # Every record reports it was updated if we are updating, even if 

345 # a record was not really changed. 

346 self.assertEqual(result.n_skipped, 0, str(result)) 

347 self.assertEqual(result.n_fully_updated, 3, str(result)) 

348 visit_348 = self.butler.query_dimension_records("visit", where="visit.seq_num = 348")[0] 

349 self.assertEqual(visit_348.target_name, "new target", visit_348) 

350 

351 

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

353 unittest.main()