Coverage for tests/test_instrument.py: 24%

1# This file is part of pipe_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 software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <http://www.gnu.org/licenses/>. 

27 

28"""Tests of the Instrument class.""" 

29 

30import datetime 

31import math 

32import unittest 

33 

34from lsst.daf.butler import DataCoordinate, DimensionPacker, DimensionUniverse 

35from lsst.daf.butler.formatters.json import JsonFormatter 

36from lsst.daf.butler.tests.utils import create_populated_sqlite_registry 

37from lsst.pex.config import Config 

38from lsst.pipe.base import Instrument 

39from lsst.utils.introspection import get_full_type_name 

40 

41 

42class BaseDummyInstrument(Instrument): 

43 """Test instrument base class.""" 

44 

45 @classmethod 

46 def getName(cls): 

47 return "DummyInstrument" 

48 

49 def register(self, registry, update=False): 

50 detector_max = 2 

51 visit_max = 10 

52 exposure_max = 8 

53 record = { 

54 "instrument": self.getName(), 

55 "class_name": get_full_type_name(DummyInstrument), 

56 "detector_max": detector_max, 

57 "visit_max": visit_max, 

58 "exposure_max": exposure_max, 

59 } 

60 with registry.transaction(): 

61 registry.syncDimensionData("instrument", record, update=update) 

62 

63 def getRawFormatter(self, dataId): 

64 return JsonFormatter 

65 

66 

67class DummyInstrument(BaseDummyInstrument): 

68 """Test instrument.""" 

69 

70 

71class NotInstrument: 

72 """Not an instrument class at all.""" 

73 

74 def __init__(self, collection_prefix: str = ""): 

75 self.collection_prefix = collection_prefix 

76 

77 

78class BadInstrument(DummyInstrument): 

79 """Instrument with wrong class name.""" 

80 

81 raw_definition = ("raw2", ("instrument", "detector", "exposure"), "StructuredDataDict") 

82 

83 @classmethod 

84 def getName(cls): 

85 return "BadInstrument" 

86 

87 def register(self, registry, update=False): 

88 # Register a bad class name 

89 record = { 

90 "instrument": self.getName(), 

91 "class_name": "builtins.str", 

92 "detector_max": 1, 

93 } 

94 registry.syncDimensionData("instrument", record, update=update) 

95 

96 

97class UnimportableInstrument(DummyInstrument): 

98 """Instrument with class name that does not exist.""" 

99 

100 @classmethod 

101 def getName(cls): 

102 return "NoImportInstr" 

103 

104 def register(self, registry, update=False): 

105 # Register a bad class name 

106 record = { 

107 "instrument": self.getName(), 

108 "class_name": "not.importable", 

109 "detector_max": 1, 

110 } 

111 registry.syncDimensionData("instrument", record, update=update) 

112 

113 

114class DimensionPackerTestConfig(Config): 

115 """Configuration for the dimension packer.""" 

116 

117 packer = Instrument.make_dimension_packer_config_field() 

118 

119 

120class InstrumentTestCase(unittest.TestCase): 

121 """Test for Instrument.""" 

122 

123 def setUp(self): 

124 self.instrument = DummyInstrument() 

125 self.name = "DummyInstrument" 

126 

127 def test_basics(self): 

128 self.assertEqual(self.instrument.getName(), self.name) 

129 self.assertEqual(self.instrument.getRawFormatter({}), JsonFormatter) 

130 self.assertIsNone(DummyInstrument.raw_definition) 

131 raw = BadInstrument.raw_definition 

132 self.assertEqual(raw[2], "StructuredDataDict") 

133 

134 def test_register(self): 

135 """Test that register() sets appropriate Dimensions.""" 

136 registry = create_populated_sqlite_registry().registry 

137 # Check that the registry starts out empty. 

138 self.instrument.importAll(registry) 

139 self.assertFalse(list(registry.queryDimensionRecords("instrument"))) 

140 

141 # Register and check again. 

142 self.instrument.register(registry) 

143 instruments = list(registry.queryDimensionRecords("instrument")) 

144 self.assertEqual(len(instruments), 1) 

145 self.assertEqual(instruments[0].name, self.name) 

146 self.assertEqual(instruments[0].detector_max, 2) 

147 self.assertIn("DummyInstrument", instruments[0].class_name) 

148 

149 self.instrument.importAll(registry) 

150 from_registry = DummyInstrument.fromName("DummyInstrument", registry) 

151 self.assertIsInstance(from_registry, Instrument) 

152 with self.assertRaises(LookupError): 

153 Instrument.fromName("NotThrere", registry) 

154 

155 # Register a bad instrument. 

156 BadInstrument().register(registry) 

157 with self.assertRaises(TypeError): 

158 Instrument.fromName("BadInstrument", registry) 

159 

160 UnimportableInstrument().register(registry) 

161 with self.assertRaises(ImportError): 

162 Instrument.fromName("NoImportInstr", registry) 

163 

164 # This should work even with the bad class name. 

165 self.instrument.importAll(registry) 

166 

167 # Check that from_string falls back to fromName. 

168 from_string = DummyInstrument.from_string("DummyInstrument", registry) 

169 self.assertIsInstance(from_string, type(from_registry)) 

170 

171 def test_from_string(self): 

172 """Test Instrument.from_string works.""" 

173 with self.assertRaises(RuntimeError): 

174 # No registry. 

175 DummyInstrument.from_string("DummyInstrument") 

176 

177 with self.assertRaises(TypeError): 

178 # This will raise because collection_prefix is not understood. 

179 DummyInstrument.from_string("lsst.pipe.base.Task") 

180 

181 with self.assertRaises(TypeError): 

182 # Not an instrument but does have collection_prefix. 

183 DummyInstrument.from_string(get_full_type_name(NotInstrument)) 

184 

185 with self.assertRaises(TypeError): 

186 # This will raise because BaseDummyInstrument is not a subclass 

187 # of DummyInstrument. 

188 DummyInstrument.from_string(get_full_type_name(BaseDummyInstrument)) 

189 

190 instrument = DummyInstrument.from_string( 

191 get_full_type_name(DummyInstrument), collection_prefix="test" 

192 ) 

193 self.assertEqual(instrument.collection_prefix, "test") 

194 

195 def test_defaults(self): 

196 self.assertEqual(self.instrument.makeDefaultRawIngestRunName(), "DummyInstrument/raw/all") 

197 self.assertEqual( 

198 self.instrument.makeUnboundedCalibrationRunName("a", "b"), "DummyInstrument/calib/a/b/unbounded" 

199 ) 

200 self.assertEqual( 

201 self.instrument.makeCuratedCalibrationRunName("2018-05-04", "a"), 

202 "DummyInstrument/calib/a/curated/20180504T000000Z", 

203 ) 

204 self.assertEqual(self.instrument.makeCalibrationCollectionName("c"), "DummyInstrument/calib/c") 

205 self.assertEqual(self.instrument.makeRefCatCollectionName(), "refcats") 

206 self.assertEqual(self.instrument.makeRefCatCollectionName("a"), "refcats/a") 

207 self.assertEqual(self.instrument.makeUmbrellaCollectionName(), "DummyInstrument/defaults") 

208 

209 instrument = DummyInstrument(collection_prefix="Different") 

210 self.assertEqual(instrument.makeCollectionName("a"), "Different/a") 

211 self.assertEqual(self.instrument.makeCollectionName("a"), "DummyInstrument/a") 

212 

213 def test_collection_timestamps(self): 

214 self.assertEqual( 

215 Instrument.formatCollectionTimestamp("2018-05-03"), 

216 "20180503T000000Z", 

217 ) 

218 self.assertEqual( 

219 Instrument.formatCollectionTimestamp("2018-05-03T14:32:16"), 

220 "20180503T143216Z", 

221 ) 

222 self.assertEqual( 

223 Instrument.formatCollectionTimestamp("20180503T143216Z"), 

224 "20180503T143216Z", 

225 ) 

226 self.assertEqual( 

227 Instrument.formatCollectionTimestamp(datetime.datetime(2018, 5, 3, 14, 32, 16)), 

228 "20180503T143216Z", 

229 ) 

230 formattedNow = Instrument.makeCollectionTimestamp() 

231 self.assertIsInstance(formattedNow, str) 

232 datetimeThen1 = datetime.datetime.strptime(formattedNow, "%Y%m%dT%H%M%S%z") 

233 self.assertEqual(datetimeThen1.tzinfo, datetime.UTC) 

234 

235 with self.assertRaises(TypeError): 

236 Instrument.formatCollectionTimestamp(0) 

237 

238 def test_dimension_packer_config_defaults(self): 

239 """Test the ObservationDimensionPacker class and the Instrument-based 

240 systems for constructing it, in the case where the Instrument-defined 

241 default is used. 

242 """ 

243 registry = create_populated_sqlite_registry().registry 

244 self.instrument.register(registry) 

245 config = DimensionPackerTestConfig() 

246 instrument_data_id = registry.expandDataId(instrument=self.name) 

247 record = instrument_data_id.records["instrument"] 

248 self.check_dimension_packers( 

249 registry.dimensions, 

250 # Test just one packer-construction signature here as that saves us 

251 # from having to insert dimension records for visits, exposures, 

252 # and detectors for this test. 

253 # Note that we don't need to pass any more than the instrument in 

254 # the data ID yet, because we're just constructing packers, not 

255 # calling their pack method. 

256 visit_packers=[ 

257 config.packer.apply(instrument_data_id, is_exposure=False), 

258 Instrument.make_default_dimension_packer(instrument_data_id, is_exposure=False), 

259 ], 

260 exposure_packers=[ 

261 config.packer.apply(instrument_data_id, is_exposure=True), 

262 Instrument.make_default_dimension_packer(instrument_data_id, is_exposure=True), 

263 ], 

264 n_detectors=record.detector_max, 

265 n_visits=record.visit_max, 

266 n_exposures=record.exposure_max, 

267 ) 

268 

269 def test_dimension_packer_config_override(self): 

270 """Test the ObservationDimensionPacker class and the Instrument-based 

271 systems for constructing it, in the case where configuration overrides 

272 the Instrument's default. 

273 """ 

274 registry = create_populated_sqlite_registry().registry 

275 # Intentionally do not register instrument or insert any other 

276 # dimension records to ensure we don't need them in this mode. 

277 config = DimensionPackerTestConfig() 

278 config.packer["observation"].n_observations = 16 

279 config.packer["observation"].n_detectors = 4 

280 config.packer.name = "observation" 

281 instrument_data_id = DataCoordinate.standardize(instrument=self.name, universe=registry.dimensions) 

282 full_data_id = DataCoordinate.standardize(instrument_data_id, detector=1, visit=3, exposure=7) 

283 visit_data_id = full_data_id.subset(full_data_id.universe.conform(["visit", "detector"])) 

284 exposure_data_id = full_data_id.subset(full_data_id.universe.conform(["exposure", "detector"])) 

285 self.check_dimension_packers( 

286 registry.dimensions, 

287 # Note that we don't need to pass any more than the instrument in 

288 # the data ID yet, because we're just constructing packers, not 

289 # calling their pack method. 

290 visit_packers=[ 

291 config.packer.apply(visit_data_id), 

292 config.packer.apply(full_data_id), 

293 config.packer.apply(instrument_data_id, is_exposure=False), 

294 ], 

295 exposure_packers=[ 

296 config.packer.apply(instrument_data_id, is_exposure=True), 

297 config.packer.apply(exposure_data_id), 

298 ], 

299 n_detectors=config.packer["observation"].n_detectors, 

300 n_visits=config.packer["observation"].n_observations, 

301 n_exposures=config.packer["observation"].n_observations, 

302 ) 

303 

304 def check_dimension_packers( 

305 self, 

306 universe: DimensionUniverse, 

307 visit_packers: list[DimensionPacker], 

308 exposure_packers: list[DimensionPacker], 

309 n_detectors: int, 

310 n_visits: int, 

311 n_exposures: int, 

312 ) -> None: 

313 """Test the behavior of one or more dimension packers constructed by 

314 an instrument. 

315 

316 Parameters 

317 ---------- 

318 universe : `lsst.daf.butler.DimensionUniverse` 

319 Data model for butler data IDs. 

320 visit_packers : `list` [ `DimensionPacker` ] 

321 Packers with ``{visit, detector}`` dimensions to test. 

322 exposure_packers : `list` [ `DimensionPacker` ] 

323 Packers with ``{exposure, detector}`` dimensions to test. 

324 n_detectors : `int` 

325 Number of detectors all packers have been configured to reserve 

326 space for. 

327 n_visits : `int` 

328 Number of visits all packers in ``visit_packers`` have been 

329 configured to reserve space for. 

330 n_exposures : `int` 

331 Number of exposures all packers in ``exposure_packers`` have been 

332 configured to reserve space for. 

333 """ 

334 full_data_id = DataCoordinate.standardize( 

335 instrument=self.name, detector=1, visit=3, exposure=7, universe=universe 

336 ) 

337 visit_data_id = full_data_id.subset(full_data_id.universe.conform(["visit", "detector"])) 

338 exposure_data_id = full_data_id.subset(full_data_id.universe.conform(["exposure", "detector"])) 

339 for n, packer in enumerate(visit_packers): 

340 with self.subTest(n=n): 

341 packed_id, max_bits = packer.pack(full_data_id, returnMaxBits=True) 

342 self.assertEqual(packed_id, full_data_id["detector"] + full_data_id["visit"] * n_detectors) 

343 self.assertEqual(max_bits, math.ceil(math.log2(n_detectors * n_visits))) 

344 self.assertEqual(visit_data_id, packer.unpack(packed_id)) 

345 with self.assertRaisesRegex(ValueError, f"Detector ID {n_detectors} is out of bounds"): 

346 packer.pack(instrument=self.name, detector=n_detectors, visit=3) 

347 with self.assertRaisesRegex(ValueError, f"Visit ID {n_visits} is out of bounds"): 

348 packer.pack(instrument=self.name, detector=1, visit=n_visits) 

349 for n, packer in enumerate(exposure_packers): 

350 with self.subTest(n=n): 

351 packed_id, max_bits = packer.pack(full_data_id, returnMaxBits=True) 

352 self.assertEqual(packed_id, full_data_id["detector"] + full_data_id["exposure"] * n_detectors) 

353 self.assertEqual(max_bits, math.ceil(math.log2(n_detectors * n_exposures))) 

354 self.assertEqual(exposure_data_id, packer.unpack(packed_id)) 

355 with self.assertRaisesRegex(ValueError, f"Detector ID {n_detectors} is out of bounds"): 

356 packer.pack(instrument=self.name, detector=n_detectors, exposure=3) 

357 with self.assertRaisesRegex(ValueError, f"Exposure ID {n_exposures} is out of bounds"): 

358 packer.pack(instrument=self.name, detector=1, exposure=n_exposures) 

359 

360 

361if __name__ == "__main__": 

362 unittest.main()