Coverage for tests / test_defineVisits.py: 24%
142 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-27 01:24 -0800
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/>.
22import os
23import pickle
24import shutil
25import tempfile
26import unittest
27import warnings
28from collections import defaultdict
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
36TESTDIR = os.path.dirname(__file__)
37DATADIR = os.path.join(TESTDIR, "data", "visits")
40class DefineVisitsBase:
41 """General set up that can be shared."""
43 use_data_ids = True
44 """Use data IDs when calling defineVisits, else use dimension records."""
46 def setUpExposures(self):
47 """Create a new butler for each test since we are changing dimension
48 records.
49 """
50 self.root = tempfile.mkdtemp(dir=TESTDIR)
51 self.creatorButler = butlerTests.makeTestRepo(self.root, [])
52 self.enterContext(self.creatorButler)
53 self.butler = butlerTests.makeTestCollection(self.creatorButler, uniqueId=self.id())
54 self.enterContext(self.butler)
56 self.config = self.get_config()
57 self.task = DefineVisitsTask(config=self.config, butler=self.butler)
59 # Need to register the instrument.
60 DummyCam().register(self.butler.registry)
62 # Choose serializations based on universe.
63 universe = self.butler.dimensions
64 uversion = universe.version
65 # Not all universe changes result in visible changes.
66 match uversion:
67 case uversion if uversion < 2:
68 raise unittest.SkipTest(f"Universe {uversion} is not compatible with these test files.")
69 case 2 | 3 | 4 | 5:
70 # has_simulated, azimuth, seq_start, seq_end.
71 v = 2
72 case 6:
73 # group not group_name, group_id dropped.
74 v = 6
75 case 7:
76 # can_see_sky.
77 v = 7
78 case _:
79 # Might work.
80 warnings.warn(f"Universe {uversion} has not been validated.")
81 v = 7
83 # Read the exposure records.
84 self.records: dict[int, DimensionRecord] = {}
85 for i in (347, 348, 349):
86 with open(os.path.join(DATADIR, f"exp_v{v}_{i}.json")) as fh:
87 simple = SerializedDimensionRecord.model_validate_json(fh.read())
88 self.records[i] = DimensionRecord.from_simple(simple, registry=self.butler.registry)
90 def define_visits(
91 self,
92 exposures: list[DimensionRecord | list[DimensionRecord]],
93 incremental: bool,
94 ) -> None:
95 for records in exposures:
96 records = list(ensure_iterable(records))
97 if "group" in self.butler.dimensions["exposure"].implied:
98 # This is a group + day_obs universe.
99 for rec in records:
100 self.butler.registry.syncDimensionData(
101 "group", dict(instrument=rec.instrument, name=rec.group)
102 )
103 self.butler.registry.syncDimensionData(
104 "day_obs", dict(instrument=rec.instrument, id=rec.day_obs)
105 )
107 self.butler.registry.insertDimensionData("exposure", *records)
108 # Include all records so far in definition.
109 if self.use_data_ids:
110 dataIds = list(self.butler.registry.queryDataIds("exposure", instrument="DummyCam"))
111 else:
112 dataIds = records
113 self.task.run(dataIds, incremental=incremental)
116class DefineVisitsTestCase(unittest.TestCase, DefineVisitsBase):
117 """Test visit definition."""
119 def setUp(self):
120 self.setUpExposures()
122 def tearDown(self):
123 if self.root is not None:
124 shutil.rmtree(self.root, ignore_errors=True)
126 def get_config(self) -> DefineVisitsConfig:
127 return DefineVisitsTask.ConfigClass()
129 def assertVisits(self):
130 """Check that the visits were registered as expected."""
131 visits = list(self.butler.registry.queryDimensionRecords("visit"))
132 self.assertEqual(len(visits), 4)
133 self.assertEqual(
134 {visit.id for visit in visits}, {2022040500347, 2022040500348, 2022040500349, 92022040500348}
135 )
137 # Ensure that the definitions are correct (ignoring order).
138 defmap = defaultdict(set)
139 definitions = list(self.butler.registry.queryDimensionRecords("visit_definition"))
140 for defn in definitions:
141 defmap[defn.visit].add(defn.exposure)
143 self.assertEqual(
144 dict(defmap),
145 {
146 92022040500348: {2022040500348},
147 2022040500347: {2022040500347},
148 2022040500348: {2022040500348, 2022040500349},
149 2022040500349: {2022040500349},
150 },
151 )
153 def test_defineVisits(self):
154 # Test visit definition with all the records.
155 self.define_visits([list(self.records.values())], incremental=False) # list inside a list
156 self.assertVisits()
158 def test_incremental_cumulative(self):
159 # Define the visits after each exposure.
160 self.define_visits(list(self.records.values()), incremental=True)
161 self.assertVisits()
163 def test_incremental_cumulative_reverse(self):
164 # In reverse order we should still eventually end up with the right
165 # answer.
166 with self.assertLogs("lsst.defineVisits.groupExposures", level="WARNING") as cm:
167 self.define_visits(list(reversed(self.records.values())), incremental=True)
168 self.assertIn("Skipping the multi-snap definition", "\n".join(cm.output))
169 self.assertVisits()
171 def define_visits_incrementally(self, exposure: DimensionRecord) -> None:
172 if "group" in self.butler.dimensions["exposure"].implied:
173 self.butler.registry.syncDimensionData(
174 "group", dict(instrument=exposure.instrument, name=exposure.group)
175 )
176 self.butler.registry.syncDimensionData(
177 "day_obs",
178 dict(
179 instrument=exposure.instrument,
180 id=exposure.day_obs,
181 ),
182 )
183 self.butler.registry.insertDimensionData("exposure", exposure)
184 dataIds = [
185 DataCoordinate.standardize(
186 instrument="DummyCam", exposure=exposure.id, universe=self.butler.dimensions
187 )
188 ]
189 self.task.run(dataIds, incremental=True)
191 def test_incremental(self):
192 for record in self.records.values():
193 self.define_visits_incrementally(record)
194 self.assertVisits()
196 def test_incremental_reverse(self):
197 for record in reversed(self.records.values()):
198 self.define_visits_incrementally(record)
199 self.assertVisits()
201 def testPickleTask(self):
202 stream = pickle.dumps(self.task)
203 copy = pickle.loads(stream)
204 self.enterContext(copy.butler)
205 self.assertEqual(self.task.getFullName(), copy.getFullName())
206 self.assertEqual(self.task.log.name, copy.log.name)
207 self.assertEqual(self.task.config, copy.config)
208 self.assertEqual(self.task.butler._config, copy.butler._config)
209 self.assertEqual(list(self.task.butler.collections.defaults), list(copy.butler.collections.defaults))
210 self.assertEqual(self.task.butler.run, copy.butler.run)
211 self.assertEqual(self.task.universe, copy.universe)
214class DefineVisitsRecordsTestCase(DefineVisitsTestCase):
215 """Define visits using only dimension records."""
217 use_data_ids = False
220class DefineVisitsGroupingTestCase(unittest.TestCase, DefineVisitsBase):
221 """Test visit grouping by group metadata."""
223 def setUp(self):
224 self.setUpExposures()
226 def tearDown(self):
227 if self.root is not None:
228 shutil.rmtree(self.root, ignore_errors=True)
230 def get_config(self) -> DefineVisitsConfig:
231 config = DefineVisitsTask.ConfigClass()
232 config.groupExposures.name = "by-group-metadata"
233 return config
235 def test_defineVisits(self):
236 # Test visit definition with all the records.
237 self.define_visits([list(self.records.values())], incremental=False) # list inside a list
238 self.assertVisits()
240 def assertVisits(self):
241 """Check that the visits were registered as expected."""
242 visits = list(self.butler.registry.queryDimensionRecords("visit"))
243 self.assertEqual(len(visits), 2)
245 # The visit ID itself depends on which universe we are using.
246 # It is either calculated or comes from the JSON record.
247 if "group" in self.butler.dimensions["exposure"].implied:
248 visit_ids = [20220406025653255, 20220406025807181]
249 else:
250 visit_ids = [2291434132550000, 2291434871810000]
251 self.assertEqual({visit.id for visit in visits}, set(visit_ids))
253 # Ensure that the definitions are correct (ignoring order).
254 defmap = defaultdict(set)
255 definitions = list(self.butler.registry.queryDimensionRecords("visit_definition"))
256 for defn in definitions:
257 defmap[defn.visit].add(defn.exposure)
259 self.assertEqual(
260 dict(defmap),
261 {
262 visit_ids[0]: {2022040500347},
263 visit_ids[1]: {2022040500348, 2022040500349},
264 },
265 )
268class DefineVisitsGroupingRecordsTestCase(DefineVisitsGroupingTestCase):
269 """Test using dimension records instead of Data IDs."""
271 use_data_ids = False
274if __name__ == "__main__": 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true
275 unittest.main()