Coverage for tests/test_packer.py: 23%
97 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-05 00:55 -0800
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 check_rubin_dimension_packer(
74 self,
75 instrument: Instrument,
76 is_exposure: bool,
77 *,
78 exposure_id: int,
79 day_obs: int,
80 seq_num: int,
81 detector: int,
82 controller: str = "O",
83 is_one_to_one_reinterpretation: bool = False,
84 visit_id: int | None = None,
85 use_controllers: bool = False,
86 ) -> None:
87 """Run tests on an instrument that uses the new Rubin dimension packer.
89 Parameters
90 ----------
91 instrument : `lsst.pipe.base.Instrument`
92 Instrument instance to be tested.
93 is_exposure : `bool`
94 `True` to pack ``{detector, exposure}`` data IDs, `False` to pack
95 ``{detector, visit}`` data IDs.
96 exposure_id : `int`
97 Integer data ID.
98 day_obs : `int`
99 Date of observations as a YYYYMMDD decimal integer, consistent with
100 ``exposure_id``.
101 seq_num : `int`
102 Instrument sequence number, consistent with ``exposure_id``.
103 detector : `int`
104 Integer detector data ID value.
105 controller : `str`, optional
106 Controller code consistent with ``exposure_id``.
107 is_one_to_one_reinterpretation : `bool`, optional
108 If `True`, this is a visit ID that represents the alternate
109 interpretation of that exposure (which must be the first snap in a
110 multi-snap sequence) as a standalone visit.
111 visit_id : `int`
112 Integer visit ID. Must be provided only if
113 ``is_one_to_one_reinterpretatation=True``; otherwise this is the
114 same as ``exposure_id``.
115 use_controllers : `bool`, optional
116 Whether to configure the packer to encode and hence round-trip
117 controller values. This tests more of the functionality but is not
118 the default behavior, since we instead want to assume OCS and
119 save bits for data releases.
120 """
121 if visit_id is None:
122 assert (
123 not is_one_to_one_reinterpretation
124 ), "Test should not infer visit_id in this case."
125 visit_id = exposure_id
126 instrument_data_id = self.registry.expandDataId(instrument=instrument.getName())
127 config = _TestConfig()
128 if use_controllers:
129 config.packer["rubin"].use_controllers()
130 packer = config.packer.apply(instrument_data_id, is_exposure=is_exposure)
131 self.assertIsInstance(packer, RubinDimensionPacker)
132 self.assertEqual(packer.maxBits, 41 if use_controllers else 38)
133 full_data_id = DataCoordinate.standardize(
134 instrument_data_id, exposure=exposure_id, visit=visit_id, detector=detector
135 )
136 packed1 = RubinDimensionPacker.pack_decomposition(
137 day_obs,
138 seq_num,
139 detector,
140 controller,
141 is_one_to_one_reinterpretation=is_one_to_one_reinterpretation,
142 config=packer.config,
143 )
144 packed2 = RubinDimensionPacker.pack_id_pair(
145 exposure_id,
146 detector,
147 is_one_to_one_reinterpretation=is_one_to_one_reinterpretation,
148 config=packer.config,
149 )
150 packed3 = packer.pack(full_data_id)
151 self.assertEqual(packed1, packed2)
152 self.assertEqual(packed1, packed3)
153 (
154 u_day_obs,
155 u_seq_num,
156 u_detector,
157 u_controller,
158 u_is_one_to_one_reinterpretation,
159 ) = RubinDimensionPacker.unpack_decomposition(packed1, config=packer.config)
160 self.assertEqual(u_day_obs, day_obs)
161 self.assertEqual(u_seq_num, seq_num)
162 self.assertEqual(u_detector, detector)
163 self.assertEqual(u_controller, controller)
164 self.assertEqual(u_is_one_to_one_reinterpretation, is_one_to_one_reinterpretation)
165 (
166 u_exposure_id,
167 u_detector,
168 u_is_one_to_one_reinterpretation,
169 ) = RubinDimensionPacker.unpack_id_pair(packed1, config=packer.config)
170 self.assertEqual(u_exposure_id, exposure_id)
171 self.assertEqual(u_detector, detector)
172 self.assertEqual(u_is_one_to_one_reinterpretation, is_one_to_one_reinterpretation)
173 u_data_id = packer.unpack(packed1)
174 self.assertEqual(u_data_id, full_data_id.subset(packer.dimensions))
176 def check_old_dimension_packer(
177 self,
178 instrument: Instrument,
179 is_exposure: bool,
180 ) -> None:
181 """Test that an Instrument's default dimension packer is still
182 `lsst.pipe.base.ObservationDimensionPacker`.
183 """
184 instrument_data_id = self.registry.expandDataId(instrument=instrument.getName())
185 config = _TestConfig()
186 packer = config.packer.apply(instrument_data_id, is_exposure=is_exposure)
187 # This instrument still uses the pipe_base default dimension packer,
188 # which is tested there. Nothing more to do here.
189 self.assertIsInstance(packer, ObservationDimensionPacker)
191 def test_latiss(self):
192 instrument = Latiss()
193 instrument.register(self.registry)
194 # Input values obtained from:
195 # $ butler query-dimension-records /repo/main exposure --where \
196 # "instrument='LATISS'" --limit 1
197 self.check_rubin_dimension_packer(
198 instrument,
199 is_exposure=True,
200 exposure_id=2022062800004,
201 day_obs=20220628,
202 seq_num=4,
203 detector=0,
204 )
205 # Input values obtained from:
206 # $ butler query-dimension-records /repo/main visit --where \
207 # "instrument='LATISS'" --limit 1
208 self.check_rubin_dimension_packer(
209 instrument,
210 is_exposure=False,
211 exposure_id=2021090800749,
212 day_obs=20210908,
213 seq_num=749,
214 detector=0,
215 )
216 # Input data obtained from:
217 # $ butler query-dimension-records /repo/embargo visit --where \
218 # "instrument='LATISS' AND visit_system=0 AND exposure != visit" \
219 # --limit 1
220 self.check_rubin_dimension_packer(
221 instrument,
222 is_exposure=False,
223 exposure_id=2022101101105,
224 day_obs=20221011,
225 seq_num=1105,
226 detector=0,
227 visit_id=92022101101105,
228 is_one_to_one_reinterpretation=True,
229 )
231 def test_lsstCam(self):
232 instrument = LsstCam()
233 instrument.register(self.registry)
234 # Input values obtained from:
235 # $ butler query-dimension-records /repo/main exposure --where \
236 # "instrument='LSSTCam'" --limit 1
237 self.check_rubin_dimension_packer(
238 instrument,
239 is_exposure=True,
240 exposure_id=3021121400075,
241 day_obs=20211214,
242 seq_num=75,
243 detector=150,
244 controller="C",
245 use_controllers=True,
246 )
248 def test_comCam(self):
249 instrument = LsstComCam()
250 instrument.register(self.registry)
251 # Input values obtained from:
252 # $ butler query-dimension-records /repo/main exposure --where \
253 # "instrument='LSSTComCam'" --limit 1
254 self.check_rubin_dimension_packer(
255 instrument,
256 is_exposure=True,
257 exposure_id=2020091100004,
258 day_obs=20200911,
259 seq_num=4,
260 detector=5,
261 )
263 def test_comCamSim(self):
264 instrument = LsstComCamSim()
265 instrument.register(self.registry)
266 # Input values obtained from:
267 # $ butler query-dimension-records data/input/comCamSim exposure \
268 # --where "instrument='LSSTComCamSim'" --limit 1
269 self.check_rubin_dimension_packer(
270 instrument,
271 is_exposure=False,
272 exposure_id=7024032100720,
273 day_obs=20240321,
274 seq_num=720,
275 detector=4,
276 controller="S",
277 use_controllers=True,
278 )
280 def test_lsstCamSim(self):
281 instrument = LsstCamSim()
282 instrument.register(self.registry)
283 # Input values obtained from:
284 # $ butler query-dimension-records data/input/lsstCamSim exposure \
285 # --where "instrument='LSSTCamSim'" --limit 1
286 self.check_rubin_dimension_packer(
287 instrument,
288 is_exposure=True,
289 exposure_id=7024032100720,
290 day_obs=20240321,
291 seq_num=720,
292 detector=94,
293 controller="S",
294 use_controllers=True,
295 )
297 def test_imsim(self):
298 instrument = LsstCamImSim()
299 instrument.register(self.registry)
300 self.check_old_dimension_packer(instrument, is_exposure=True)
301 self.check_old_dimension_packer(instrument, is_exposure=False)
303 def test_phosim(self):
304 instrument = LsstCamPhoSim()
305 instrument.register(self.registry)
306 self.check_old_dimension_packer(instrument, is_exposure=True)
307 self.check_old_dimension_packer(instrument, is_exposure=False)
309 def test_ts3(self):
310 instrument = LsstTS3()
311 instrument.register(self.registry)
312 self.check_old_dimension_packer(instrument, is_exposure=True)
314 def test_ts8(self):
315 instrument = LsstTS8()
316 instrument.register(self.registry)
317 self.check_old_dimension_packer(instrument, is_exposure=True)
319 def test_ucdcam(self):
320 instrument = LsstUCDCam()
321 instrument.register(self.registry)
322 self.check_old_dimension_packer(instrument, is_exposure=True)
325if __name__ == "__main__": 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true
326 unittest.main()