Coverage for tests / test_instrument.py: 23%
158 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-01 08:20 +0000
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/>.
28"""Tests of the Instrument class."""
30import datetime
31import math
32import unittest
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
42class BaseDummyInstrument(Instrument):
43 """Test instrument base class."""
45 @classmethod
46 def getName(cls):
47 return "DummyInstrument"
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)
63 def getRawFormatter(self, dataId):
64 return JsonFormatter
67class DummyInstrument(BaseDummyInstrument):
68 """Test instrument."""
71class NotInstrument:
72 """Not an instrument class at all."""
74 def __init__(self, collection_prefix: str = ""):
75 self.collection_prefix = collection_prefix
78class BadInstrument(DummyInstrument):
79 """Instrument with wrong class name."""
81 raw_definition = ("raw2", ("instrument", "detector", "exposure"), "StructuredDataDict")
83 @classmethod
84 def getName(cls):
85 return "BadInstrument"
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)
97class UnimportableInstrument(DummyInstrument):
98 """Instrument with class name that does not exist."""
100 @classmethod
101 def getName(cls):
102 return "NoImportInstr"
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)
114class DimensionPackerTestConfig(Config):
115 """Configuration for the dimension packer."""
117 packer = Instrument.make_dimension_packer_config_field()
120class InstrumentTestCase(unittest.TestCase):
121 """Test for Instrument."""
123 def setUp(self):
124 self.instrument = DummyInstrument()
125 self.name = "DummyInstrument"
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")
134 def test_register(self):
135 """Test that register() sets appropriate Dimensions."""
136 butler = create_populated_sqlite_registry()
137 self.enterContext(butler)
138 registry = butler.registry
139 # Check that the registry starts out empty.
140 self.instrument.importAll(registry)
141 self.assertFalse(list(registry.queryDimensionRecords("instrument")))
143 # Register and check again.
144 self.instrument.register(registry)
145 instruments = list(registry.queryDimensionRecords("instrument"))
146 self.assertEqual(len(instruments), 1)
147 self.assertEqual(instruments[0].name, self.name)
148 self.assertEqual(instruments[0].detector_max, 2)
149 self.assertIn("DummyInstrument", instruments[0].class_name)
151 self.instrument.importAll(registry)
152 from_registry = DummyInstrument.fromName("DummyInstrument", registry)
153 self.assertIsInstance(from_registry, Instrument)
154 with self.assertRaises(LookupError):
155 Instrument.fromName("NotThrere", registry)
157 # Register a bad instrument.
158 BadInstrument().register(registry)
159 with self.assertRaises(TypeError):
160 Instrument.fromName("BadInstrument", registry)
162 UnimportableInstrument().register(registry)
163 with self.assertRaises(ImportError):
164 Instrument.fromName("NoImportInstr", registry)
166 # This should work even with the bad class name.
167 self.instrument.importAll(registry)
169 # Check that from_string falls back to fromName.
170 from_string = DummyInstrument.from_string("DummyInstrument", registry)
171 self.assertIsInstance(from_string, type(from_registry))
173 def test_from_string(self):
174 """Test Instrument.from_string works."""
175 with self.assertRaises(RuntimeError):
176 # No registry.
177 DummyInstrument.from_string("DummyInstrument")
179 with self.assertRaises(TypeError):
180 # This will raise because collection_prefix is not understood.
181 DummyInstrument.from_string("lsst.pipe.base.Task")
183 with self.assertRaises(TypeError):
184 # Not an instrument but does have collection_prefix.
185 DummyInstrument.from_string(get_full_type_name(NotInstrument))
187 with self.assertRaises(TypeError):
188 # This will raise because BaseDummyInstrument is not a subclass
189 # of DummyInstrument.
190 DummyInstrument.from_string(get_full_type_name(BaseDummyInstrument))
192 instrument = DummyInstrument.from_string(
193 get_full_type_name(DummyInstrument), collection_prefix="test"
194 )
195 self.assertEqual(instrument.collection_prefix, "test")
197 def test_defaults(self):
198 self.assertEqual(self.instrument.makeDefaultRawIngestRunName(), "DummyInstrument/raw/all")
199 self.assertEqual(
200 self.instrument.makeUnboundedCalibrationRunName("a", "b"), "DummyInstrument/calib/a/b/unbounded"
201 )
202 self.assertEqual(
203 self.instrument.makeCuratedCalibrationRunName("2018-05-04", "a"),
204 "DummyInstrument/calib/a/curated/20180504T000000Z",
205 )
206 self.assertEqual(self.instrument.makeCalibrationCollectionName("c"), "DummyInstrument/calib/c")
207 self.assertEqual(self.instrument.makeRefCatCollectionName(), "refcats")
208 self.assertEqual(self.instrument.makeRefCatCollectionName("a"), "refcats/a")
209 self.assertEqual(self.instrument.makeUmbrellaCollectionName(), "DummyInstrument/defaults")
211 instrument = DummyInstrument(collection_prefix="Different")
212 self.assertEqual(instrument.makeCollectionName("a"), "Different/a")
213 self.assertEqual(self.instrument.makeCollectionName("a"), "DummyInstrument/a")
215 def test_collection_timestamps(self):
216 self.assertEqual(
217 Instrument.formatCollectionTimestamp("2018-05-03"),
218 "20180503T000000Z",
219 )
220 self.assertEqual(
221 Instrument.formatCollectionTimestamp("2018-05-03T14:32:16"),
222 "20180503T143216Z",
223 )
224 self.assertEqual(
225 Instrument.formatCollectionTimestamp("20180503T143216Z"),
226 "20180503T143216Z",
227 )
228 self.assertEqual(
229 Instrument.formatCollectionTimestamp(datetime.datetime(2018, 5, 3, 14, 32, 16)),
230 "20180503T143216Z",
231 )
232 formattedNow = Instrument.makeCollectionTimestamp()
233 self.assertIsInstance(formattedNow, str)
234 datetimeThen1 = datetime.datetime.strptime(formattedNow, "%Y%m%dT%H%M%S%z")
235 self.assertEqual(datetimeThen1.tzinfo, datetime.UTC)
237 with self.assertRaises(TypeError):
238 Instrument.formatCollectionTimestamp(0)
240 def test_dimension_packer_config_defaults(self):
241 """Test the ObservationDimensionPacker class and the Instrument-based
242 systems for constructing it, in the case where the Instrument-defined
243 default is used.
244 """
245 butler = create_populated_sqlite_registry()
246 self.enterContext(butler)
247 registry = butler.registry
248 self.instrument.register(registry)
249 config = DimensionPackerTestConfig()
250 instrument_data_id = registry.expandDataId(instrument=self.name)
251 record = instrument_data_id.records["instrument"]
252 self.check_dimension_packers(
253 registry.dimensions,
254 # Test just one packer-construction signature here as that saves us
255 # from having to insert dimension records for visits, exposures,
256 # and detectors for this test.
257 # Note that we don't need to pass any more than the instrument in
258 # the data ID yet, because we're just constructing packers, not
259 # calling their pack method.
260 visit_packers=[
261 config.packer.apply(instrument_data_id, is_exposure=False),
262 Instrument.make_default_dimension_packer(instrument_data_id, is_exposure=False),
263 ],
264 exposure_packers=[
265 config.packer.apply(instrument_data_id, is_exposure=True),
266 Instrument.make_default_dimension_packer(instrument_data_id, is_exposure=True),
267 ],
268 n_detectors=record.detector_max,
269 n_visits=record.visit_max,
270 n_exposures=record.exposure_max,
271 )
273 def test_dimension_packer_config_override(self):
274 """Test the ObservationDimensionPacker class and the Instrument-based
275 systems for constructing it, in the case where configuration overrides
276 the Instrument's default.
277 """
278 butler = create_populated_sqlite_registry()
279 self.enterContext(butler)
280 registry = butler.registry
281 # Intentionally do not register instrument or insert any other
282 # dimension records to ensure we don't need them in this mode.
283 config = DimensionPackerTestConfig()
284 config.packer["observation"].n_observations = 16
285 config.packer["observation"].n_detectors = 4
286 config.packer.name = "observation"
287 instrument_data_id = DataCoordinate.standardize(instrument=self.name, universe=registry.dimensions)
288 full_data_id = DataCoordinate.standardize(instrument_data_id, detector=1, visit=3, exposure=7)
289 visit_data_id = full_data_id.subset(full_data_id.universe.conform(["visit", "detector"]))
290 exposure_data_id = full_data_id.subset(full_data_id.universe.conform(["exposure", "detector"]))
291 self.check_dimension_packers(
292 registry.dimensions,
293 # Note that we don't need to pass any more than the instrument in
294 # the data ID yet, because we're just constructing packers, not
295 # calling their pack method.
296 visit_packers=[
297 config.packer.apply(visit_data_id),
298 config.packer.apply(full_data_id),
299 config.packer.apply(instrument_data_id, is_exposure=False),
300 ],
301 exposure_packers=[
302 config.packer.apply(instrument_data_id, is_exposure=True),
303 config.packer.apply(exposure_data_id),
304 ],
305 n_detectors=config.packer["observation"].n_detectors,
306 n_visits=config.packer["observation"].n_observations,
307 n_exposures=config.packer["observation"].n_observations,
308 )
310 def check_dimension_packers(
311 self,
312 universe: DimensionUniverse,
313 visit_packers: list[DimensionPacker],
314 exposure_packers: list[DimensionPacker],
315 n_detectors: int,
316 n_visits: int,
317 n_exposures: int,
318 ) -> None:
319 """Test the behavior of one or more dimension packers constructed by
320 an instrument.
322 Parameters
323 ----------
324 universe : `lsst.daf.butler.DimensionUniverse`
325 Data model for butler data IDs.
326 visit_packers : `list` [ `DimensionPacker` ]
327 Packers with ``{visit, detector}`` dimensions to test.
328 exposure_packers : `list` [ `DimensionPacker` ]
329 Packers with ``{exposure, detector}`` dimensions to test.
330 n_detectors : `int`
331 Number of detectors all packers have been configured to reserve
332 space for.
333 n_visits : `int`
334 Number of visits all packers in ``visit_packers`` have been
335 configured to reserve space for.
336 n_exposures : `int`
337 Number of exposures all packers in ``exposure_packers`` have been
338 configured to reserve space for.
339 """
340 full_data_id = DataCoordinate.standardize(
341 instrument=self.name, detector=1, visit=3, exposure=7, universe=universe
342 )
343 visit_data_id = full_data_id.subset(full_data_id.universe.conform(["visit", "detector"]))
344 exposure_data_id = full_data_id.subset(full_data_id.universe.conform(["exposure", "detector"]))
345 for n, packer in enumerate(visit_packers):
346 with self.subTest(n=n):
347 packed_id, max_bits = packer.pack(full_data_id, returnMaxBits=True)
348 self.assertEqual(packed_id, full_data_id["detector"] + full_data_id["visit"] * n_detectors)
349 self.assertEqual(max_bits, math.ceil(math.log2(n_detectors * n_visits)))
350 self.assertEqual(visit_data_id, packer.unpack(packed_id))
351 with self.assertRaisesRegex(ValueError, f"Detector ID {n_detectors} is out of bounds"):
352 packer.pack(instrument=self.name, detector=n_detectors, visit=3)
353 with self.assertRaisesRegex(ValueError, f"Visit ID {n_visits} is out of bounds"):
354 packer.pack(instrument=self.name, detector=1, visit=n_visits)
355 for n, packer in enumerate(exposure_packers):
356 with self.subTest(n=n):
357 packed_id, max_bits = packer.pack(full_data_id, returnMaxBits=True)
358 self.assertEqual(packed_id, full_data_id["detector"] + full_data_id["exposure"] * n_detectors)
359 self.assertEqual(max_bits, math.ceil(math.log2(n_detectors * n_exposures)))
360 self.assertEqual(exposure_data_id, packer.unpack(packed_id))
361 with self.assertRaisesRegex(ValueError, f"Detector ID {n_detectors} is out of bounds"):
362 packer.pack(instrument=self.name, detector=n_detectors, exposure=3)
363 with self.assertRaisesRegex(ValueError, f"Exposure ID {n_exposures} is out of bounds"):
364 packer.pack(instrument=self.name, detector=1, exposure=n_exposures)
367if __name__ == "__main__":
368 unittest.main()