Coverage for tests / test_defineVisits.py: 22%

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.obs.base import DefineVisitsConfig, DefineVisitsTask 

33from lsst.obs.base.instrument_tests import DummyCam 

34from lsst.utils.iteration import ensure_iterable 

35 

36TESTDIR = os.path.dirname(__file__) 

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

38 

39 

40class DefineVisitsBase: 

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

42 

43 def setUpExposures(self): 

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

45 records. 

46 """ 

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

48 self.creatorButler = butlerTests.makeTestRepo(self.root, []) 

49 self.enterContext(self.creatorButler) 

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

51 self.enterContext(self.butler) 

52 

53 self.config = self.get_config() 

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

55 

56 # Need to register the instrument. 

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

58 

59 # Choose serializations based on universe. 

60 universe = self.butler.dimensions 

61 uversion = universe.version 

62 # Not all universe changes result in visible changes. 

63 match uversion: 

64 case uversion if uversion < 2: 

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

66 case 2 | 3 | 4 | 5: 

67 # has_simulated, azimuth, seq_start, seq_end. 

68 v = 2 

69 case 6: 

70 # group not group_name, group_id dropped. 

71 v = 6 

72 case 7: 

73 # can_see_sky. 

74 v = 7 

75 case _: 

76 # Might work. 

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

78 v = 7 

79 

80 # Read the exposure records. 

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

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

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

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

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

86 

87 def define_visits( 

88 self, 

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

90 incremental: bool, 

91 ) -> None: 

92 for records in exposures: 

93 records = list(ensure_iterable(records)) 

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

95 # This is a group + day_obs universe. 

96 for rec in records: 

97 self.butler.registry.syncDimensionData( 

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

99 ) 

100 self.butler.registry.syncDimensionData( 

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

102 ) 

103 

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

105 # Include all records so far in definition. 

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

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

108 

109 

110class DefineVisitsTestCase(unittest.TestCase, DefineVisitsBase): 

111 """Test visit definition.""" 

112 

113 def setUp(self): 

114 self.setUpExposures() 

115 

116 def tearDown(self): 

117 if self.root is not None: 

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

119 

120 def get_config(self) -> DefineVisitsConfig: 

121 return DefineVisitsTask.ConfigClass() 

122 

123 def assertVisits(self): 

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

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

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

127 self.assertEqual( 

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

129 ) 

130 

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

132 defmap = defaultdict(set) 

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

134 for defn in definitions: 

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

136 

137 self.assertEqual( 

138 dict(defmap), 

139 { 

140 92022040500348: {2022040500348}, 

141 2022040500347: {2022040500347}, 

142 2022040500348: {2022040500348, 2022040500349}, 

143 2022040500349: {2022040500349}, 

144 }, 

145 ) 

146 

147 def test_defineVisits(self): 

148 # Test visit definition with all the records. 

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

150 self.assertVisits() 

151 

152 def test_incremental_cumulative(self): 

153 # Define the visits after each exposure. 

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

155 self.assertVisits() 

156 

157 def test_incremental_cumulative_reverse(self): 

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

159 # answer. 

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

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

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

163 self.assertVisits() 

164 

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

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

167 self.butler.registry.syncDimensionData( 

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

169 ) 

170 self.butler.registry.syncDimensionData( 

171 "day_obs", 

172 dict( 

173 instrument=exposure.instrument, 

174 id=exposure.day_obs, 

175 ), 

176 ) 

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

178 dataIds = [ 

179 DataCoordinate.standardize( 

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

181 ) 

182 ] 

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

184 

185 def test_incremental(self): 

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

187 self.define_visits_incrementally(record) 

188 self.assertVisits() 

189 

190 def test_incremental_reverse(self): 

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

192 self.define_visits_incrementally(record) 

193 self.assertVisits() 

194 

195 def testPickleTask(self): 

196 stream = pickle.dumps(self.task) 

197 copy = pickle.loads(stream) 

198 self.enterContext(copy.butler) 

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

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

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

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

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

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

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

206 

207 

208class DefineVisitsGroupingTestCase(unittest.TestCase, DefineVisitsBase): 

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

210 

211 def setUp(self): 

212 self.setUpExposures() 

213 

214 def tearDown(self): 

215 if self.root is not None: 

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

217 

218 def get_config(self) -> DefineVisitsConfig: 

219 config = DefineVisitsTask.ConfigClass() 

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

221 return config 

222 

223 def test_defineVisits(self): 

224 # Test visit definition with all the records. 

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

226 self.assertVisits() 

227 

228 def assertVisits(self): 

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

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

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

232 

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

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

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

236 visit_ids = [20220406025653255, 20220406025807181] 

237 else: 

238 visit_ids = [2291434132550000, 2291434871810000] 

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

240 

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

242 defmap = defaultdict(set) 

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

244 for defn in definitions: 

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

246 

247 self.assertEqual( 

248 dict(defmap), 

249 { 

250 visit_ids[0]: {2022040500347}, 

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

252 }, 

253 ) 

254 

255 

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

257 unittest.main()