Coverage for tests / test_packer.py: 23%
99 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-07 08:47 +0000
1# This file is part of obs_lsst.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://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/>.
22import unittest
24from lsst.daf.butler import DataCoordinate, RegistryConfig
25from lsst.daf.butler.registry.sql_registry import SqlRegistry
26from lsst.obs.lsst import (
27 Latiss,
28 LsstCam,
29 LsstCamImSim,
30 LsstCamPhoSim,
31 LsstCamSim,
32 LsstComCam,
33 LsstComCamSim,
34 LsstTS3,
35 LsstTS8,
36 LsstUCDCam,
37 RubinDimensionPacker,
38)
39from lsst.pex.config import Config
40from lsst.pipe.base import Instrument, ObservationDimensionPacker
43class _TestConfig(Config):
44 packer = Instrument.make_dimension_packer_config_field()
47class RubinDimensionPackerTestCase(unittest.TestCase):
48 """Test the custom data ID packer implementation for the main Rubin
49 instruments.
51 This test mostly checks the data ID packer's methods for self-consistency,
52 and that the Instrument-class overrides work as expected. Direct tests of
53 The packing algorithm is tested against hard-coded values in
54 test_translators.py, since some translators now delegate to it.
55 """
57 def setUp(self) -> None:
58 registry_config = RegistryConfig()
59 registry_config["db"] = "sqlite://"
60 self.registry = SqlRegistry.createFromConfig(registry_config)
61 self.rubin_packer_instruments = [LsstCam, LsstComCam, LsstComCamSim,
62 Latiss]
63 self.old_packer_instruments = [
64 LsstCamImSim,
65 LsstCamPhoSim,
66 LsstTS8,
67 LsstTS3,
68 LsstUCDCam,
69 ]
70 for cls in self.rubin_packer_instruments + self.old_packer_instruments:
71 cls().register(self.registry)
73 def tearDown(self) -> None:
74 self.registry.close()
76 def check_rubin_dimension_packer(
77 self,
78 instrument: Instrument,
79 is_exposure: bool,
80 *,
81 exposure_id: int,
82 day_obs: int,
83 seq_num: int,
84 detector: int,
85 controller: str = "O",
86 is_one_to_one_reinterpretation: bool = False,
87 visit_id: int | None = None,
88 use_controllers: bool = False,
89 ) -> None:
90 """Run tests on an instrument that uses the new Rubin dimension packer.
92 Parameters
93 ----------
94 instrument : `lsst.pipe.base.Instrument`
95 Instrument instance to be tested.
96 is_exposure : `bool`
97 `True` to pack ``{detector, exposure}`` data IDs, `False` to pack
98 ``{detector, visit}`` data IDs.
99 exposure_id : `int`
100 Integer data ID.
101 day_obs : `int`
102 Date of observations as a YYYYMMDD decimal integer, consistent with
103 ``exposure_id``.
104 seq_num : `int`
105 Instrument sequence number, consistent with ``exposure_id``.
106 detector : `int`
107 Integer detector data ID value.
108 controller : `str`, optional
109 Controller code consistent with ``exposure_id``.
110 is_one_to_one_reinterpretation : `bool`, optional
111 If `True`, this is a visit ID that represents the alternate
112 interpretation of that exposure (which must be the first snap in a
113 multi-snap sequence) as a standalone visit.
114 visit_id : `int`
115 Integer visit ID. Must be provided only if
116 ``is_one_to_one_reinterpretatation=True``; otherwise this is the
117 same as ``exposure_id``.
118 use_controllers : `bool`, optional
119 Whether to configure the packer to encode and hence round-trip
120 controller values. This tests more of the functionality but is not
121 the default behavior, since we instead want to assume OCS and
122 save bits for data releases.
123 """
124 if visit_id is None:
125 assert (
126 not is_one_to_one_reinterpretation
127 ), "Test should not infer visit_id in this case."
128 visit_id = exposure_id
129 instrument_data_id = self.registry.expandDataId(instrument=instrument.getName())
130 config = _TestConfig()
131 if use_controllers:
132 config.packer["rubin"].use_controllers()
133 packer = config.packer.apply(instrument_data_id, is_exposure=is_exposure)
134 self.assertIsInstance(packer, RubinDimensionPacker)
135 self.assertEqual(packer.maxBits, 41 if use_controllers else 38)
136 full_data_id = DataCoordinate.standardize(
137 instrument_data_id, exposure=exposure_id, visit=visit_id, detector=detector
138 )
139 packed1 = RubinDimensionPacker.pack_decomposition(
140 day_obs,
141 seq_num,
142 detector,
143 controller,
144 is_one_to_one_reinterpretation=is_one_to_one_reinterpretation,
145 config=packer.config,
146 )
147 packed2 = RubinDimensionPacker.pack_id_pair(
148 exposure_id,
149 detector,
150 is_one_to_one_reinterpretation=is_one_to_one_reinterpretation,
151 config=packer.config,
152 )
153 packed3 = packer.pack(full_data_id)
154 self.assertEqual(packed1, packed2)
155 self.assertEqual(packed1, packed3)
156 (
157 u_day_obs,
158 u_seq_num,
159 u_detector,
160 u_controller,
161 u_is_one_to_one_reinterpretation,
162 ) = RubinDimensionPacker.unpack_decomposition(packed1, config=packer.config)
163 self.assertEqual(u_day_obs, day_obs)
164 self.assertEqual(u_seq_num, seq_num)
165 self.assertEqual(u_detector, detector)
166 self.assertEqual(u_controller, controller)
167 self.assertEqual(u_is_one_to_one_reinterpretation, is_one_to_one_reinterpretation)
168 (
169 u_exposure_id,
170 u_detector,
171 u_is_one_to_one_reinterpretation,
172 ) = RubinDimensionPacker.unpack_id_pair(packed1, config=packer.config)
173 self.assertEqual(u_exposure_id, exposure_id)
174 self.assertEqual(u_detector, detector)
175 self.assertEqual(u_is_one_to_one_reinterpretation, is_one_to_one_reinterpretation)
176 u_data_id = packer.unpack(packed1)
177 self.assertEqual(u_data_id, full_data_id.subset(packer.dimensions))
179 def check_old_dimension_packer(
180 self,
181 instrument: Instrument,
182 is_exposure: bool,
183 ) -> None:
184 """Test that an Instrument's default dimension packer is still
185 `lsst.pipe.base.ObservationDimensionPacker`.
186 """
187 instrument_data_id = self.registry.expandDataId(instrument=instrument.getName())
188 config = _TestConfig()
189 packer = config.packer.apply(instrument_data_id, is_exposure=is_exposure)
190 # This instrument still uses the pipe_base default dimension packer,
191 # which is tested there. Nothing more to do here.
192 self.assertIsInstance(packer, ObservationDimensionPacker)
194 def test_latiss(self):
195 instrument = Latiss()
196 instrument.register(self.registry)
197 # Input values obtained from:
198 # $ butler query-dimension-records /repo/main exposure --where \
199 # "instrument='LATISS'" --limit 1
200 self.check_rubin_dimension_packer(
201 instrument,
202 is_exposure=True,
203 exposure_id=2022062800004,
204 day_obs=20220628,
205 seq_num=4,
206 detector=0,
207 )
208 # Input values obtained from:
209 # $ butler query-dimension-records /repo/main visit --where \
210 # "instrument='LATISS'" --limit 1
211 self.check_rubin_dimension_packer(
212 instrument,
213 is_exposure=False,
214 exposure_id=2021090800749,
215 day_obs=20210908,
216 seq_num=749,
217 detector=0,
218 )
219 # Input data obtained from:
220 # $ butler query-dimension-records /repo/embargo visit --where \
221 # "instrument='LATISS' AND visit_system=0 AND exposure != visit" \
222 # --limit 1
223 self.check_rubin_dimension_packer(
224 instrument,
225 is_exposure=False,
226 exposure_id=2022101101105,
227 day_obs=20221011,
228 seq_num=1105,
229 detector=0,
230 visit_id=92022101101105,
231 is_one_to_one_reinterpretation=True,
232 )
234 def test_lsstCam(self):
235 instrument = LsstCam()
236 instrument.register(self.registry)
237 # Input values obtained from:
238 # $ butler query-dimension-records /repo/main exposure --where \
239 # "instrument='LSSTCam'" --limit 1
240 self.check_rubin_dimension_packer(
241 instrument,
242 is_exposure=True,
243 exposure_id=3021121400075,
244 day_obs=20211214,
245 seq_num=75,
246 detector=150,
247 controller="C",
248 use_controllers=True,
249 )
251 def test_comCam(self):
252 instrument = LsstComCam()
253 instrument.register(self.registry)
254 # Input values obtained from:
255 # $ butler query-dimension-records /repo/main exposure --where \
256 # "instrument='LSSTComCam'" --limit 1
257 self.check_rubin_dimension_packer(
258 instrument,
259 is_exposure=True,
260 exposure_id=2020091100004,
261 day_obs=20200911,
262 seq_num=4,
263 detector=5,
264 )
266 def test_comCamSim(self):
267 instrument = LsstComCamSim()
268 instrument.register(self.registry)
269 # Input values obtained from:
270 # $ butler query-dimension-records data/input/comCamSim exposure \
271 # --where "instrument='LSSTComCamSim'" --limit 1
272 self.check_rubin_dimension_packer(
273 instrument,
274 is_exposure=False,
275 exposure_id=7024032100720,
276 day_obs=20240321,
277 seq_num=720,
278 detector=4,
279 controller="S",
280 use_controllers=True,
281 )
283 def test_lsstCamSim(self):
284 instrument = LsstCamSim()
285 instrument.register(self.registry)
286 # Input values obtained from:
287 # $ butler query-dimension-records data/input/lsstCamSim exposure \
288 # --where "instrument='LSSTCamSim'" --limit 1
289 self.check_rubin_dimension_packer(
290 instrument,
291 is_exposure=True,
292 exposure_id=7024032100720,
293 day_obs=20240321,
294 seq_num=720,
295 detector=94,
296 controller="S",
297 use_controllers=True,
298 )
300 def test_imsim(self):
301 instrument = LsstCamImSim()
302 instrument.register(self.registry)
303 self.check_old_dimension_packer(instrument, is_exposure=True)
304 self.check_old_dimension_packer(instrument, is_exposure=False)
306 def test_phosim(self):
307 instrument = LsstCamPhoSim()
308 instrument.register(self.registry)
309 self.check_old_dimension_packer(instrument, is_exposure=True)
310 self.check_old_dimension_packer(instrument, is_exposure=False)
312 def test_ts3(self):
313 instrument = LsstTS3()
314 instrument.register(self.registry)
315 self.check_old_dimension_packer(instrument, is_exposure=True)
317 def test_ts8(self):
318 instrument = LsstTS8()
319 instrument.register(self.registry)
320 self.check_old_dimension_packer(instrument, is_exposure=True)
322 def test_ucdcam(self):
323 instrument = LsstUCDCam()
324 instrument.register(self.registry)
325 self.check_old_dimension_packer(instrument, is_exposure=True)
328if __name__ == "__main__": 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true
329 unittest.main()