Coverage for tests / test_ingest.py: 19%
253 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-13 01:33 -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 json
23import os
24import pickle
25import shutil
26import tempfile
27import unittest
29import lsst.daf.butler.tests as butlerTests
30from lsst.daf.butler import Butler, Config, DataCoordinate, Registry
31from lsst.daf.butler.registry import ConflictingDefinitionError
32from lsst.obs.base import RawIngestTask
33from lsst.obs.base.ingest_tests import IngestTestBase
34from lsst.obs.base.instrument_tests import DummyCam
35from lsst.utils.introspection import get_full_type_name
37TESTDIR = os.path.abspath(os.path.dirname(__file__))
38INGESTDIR = os.path.join(TESTDIR, "data", "ingest")
41class RawIngestTestCase(IngestTestBase, unittest.TestCase):
42 """Test ingest using JSON sidecar files."""
44 ingestDatasetTypeName = "raw_dict"
45 rawIngestTask = get_full_type_name(RawIngestTask)
46 curatedCalibrationDatasetTypes = ("testCalib",)
47 ingestDir = TESTDIR
48 instrumentClassName = "lsst.obs.base.instrument_tests.DummyCam"
49 file = os.path.join(INGESTDIR, "sidecar_data", "dataset_1.yaml")
50 dataIds = [{"instrument": "DummyCam", "exposure": 100, "detector": 0}]
51 seed_config = os.path.join(TESTDIR, "data", "curated", "seed.yaml")
53 @property
54 def visits(self):
55 with Butler.from_config(self.root, collections=[self.outputRun]) as butler:
56 return {
57 DataCoordinate.standardize(instrument="DummyCam", visit=100, universe=butler.dimensions): [
58 DataCoordinate.standardize(
59 instrument="DummyCam", exposure=100, universe=butler.dimensions
60 )
61 ]
62 }
64 def testWriteCuratedCalibrations(self):
65 # Inject the "data package" location.
66 DummyCam.dataPackageDir = os.path.join(TESTDIR, "data", "curated")
67 return super().testWriteCuratedCalibrations()
69 def _check_obscore(self, registry: Registry, has_visits: bool) -> None:
70 # Docstring inherited from base class.
71 assert registry.obsCoreTableManager is not None
72 with registry.obsCoreTableManager.query(lsst_run=self.outputRun) as result:
73 rows = list(result)
74 self.assertEqual(len(rows), 1)
75 row = rows[0]
77 # No spatial information until visits are defined
78 if not has_visits:
79 self.assertIsNone(row.s_ra)
80 self.assertIsNone(row.s_dec)
81 self.assertIsNone(row.s_fov)
82 self.assertIsNone(row.s_region)
83 else:
84 self.assertIsNotNone(row.s_ra)
85 self.assertIsNotNone(row.s_dec)
86 self.assertIsNotNone(row.s_fov)
87 self.assertRegex(row.s_region, "POLYGON ICRS .*")
90class RawIngestImpliedIndexTestCase(RawIngestTestCase):
91 """Test ingest using JSON index files."""
93 file = os.path.join(INGESTDIR, "indexed_data", "dataset_1.yaml")
96class RawIngestEdgeCaseTestCase(unittest.TestCase):
97 """Test ingest using non-standard approaches including failures.
99 Must create a new butler for each test because dimension records are
100 globals.
101 """
103 def setUp(self):
104 butlerConfig = """
105datastore:
106 # Want to ingest real files so can't use in-memory datastore
107 cls: lsst.daf.butler.datastores.fileDatastore.FileDatastore
108"""
109 self.root = tempfile.mkdtemp(dir=TESTDIR)
110 self.creatorButler = butlerTests.makeTestRepo(self.root, {}, config=Config.fromYaml(butlerConfig))
111 self.enterContext(self.creatorButler)
112 DummyCam().register(self.creatorButler.registry)
114 self.butler = butlerTests.makeTestCollection(self.creatorButler, uniqueId=self.id())
115 self.enterContext(self.butler)
116 self.outputRun = self.butler.run
118 config = RawIngestTask.ConfigClass()
119 self.task = RawIngestTask(config=config, butler=self.butler)
121 # Different test files.
122 self.bad_metadata_file = os.path.join(TESTDIR, "data", "small.fits")
123 self.good_file = os.path.join(INGESTDIR, "sidecar_data", "dataset_2.yaml")
124 self.bad_instrument_file = os.path.join(TESTDIR, "data", "calexp.fits")
126 def tearDown(self):
127 if self.root is not None:
128 shutil.rmtree(self.root, ignore_errors=True)
130 def testSimpleIngest(self):
131 # Use the default per-instrument run for this.
132 self.task.run([self.good_file])
133 datasets = list(self.butler.registry.queryDatasets("raw_dict", collections="DummyCam/raw/all"))
134 self.assertEqual(len(datasets), 1)
136 # Now parallelized.
137 files = [self.good_file, os.path.join(INGESTDIR, "sidecar_data", "dataset_1.yaml")]
138 self.task.run(files, processes=2, run=self.outputRun)
139 datasets = list(self.butler.registry.queryDatasets("raw_dict", collections=self.outputRun))
140 self.assertEqual(len(datasets), 2)
142 def testTimeStampWarning(self):
143 # Now ingest a dataset which should generate a warning because of
144 # the end time being before the begin time.
145 return
146 files = [os.path.join(INGESTDIR, "sidecar_data", "dataset_end.yaml")]
147 with self.assertLogs("lsst.obs.base._instrument", level="WARNING") as cm:
148 self.task.run(files, run=self.outputRun)
150 self.assertIn("has end time before begin time", cm.output[0])
151 records = list(
152 self.butler.registry.queryDimensionRecords(
153 "exposure",
154 where="exposure = exp AND instrument = inst",
155 bind={"exp": 3000, "inst": "DummyCam"},
156 )
157 )
158 record = records[0]
159 timespan = record.timespan
160 self.assertEqual(timespan.begin.isot, timespan.end.isot)
162 def testExplicitIndex(self):
163 files = [os.path.join(INGESTDIR, "indexed_data", "_index.json")]
164 self.task.run(files, run=self.outputRun)
166 datasets = list(self.butler.registry.queryDatasets("raw_dict", collections=self.outputRun))
167 self.assertEqual(len(datasets), 2)
169 # Try again with an explicit index and a file that is in that index.
170 files.append(os.path.join(INGESTDIR, "indexed_data", "dataset_2.yaml"))
171 new_run = self.outputRun + "b"
172 self.task.run(files, run=new_run)
174 datasets = list(self.butler.registry.queryDatasets("raw_dict", collections=self.outputRun))
175 self.assertEqual(len(datasets), 2)
177 # Now with two index files that point to the same files.
178 # Look for the warning from duplication.
179 files = [
180 os.path.join(INGESTDIR, "indexed_data", "_index.json"),
181 os.path.join(INGESTDIR, "indexed_data", "translated_subdir", "_index.json"),
182 ]
183 new_run = self.outputRun + "c"
185 with self.assertLogs(level="WARNING") as cm:
186 self.task.run(files, run=new_run)
187 self.assertIn("already specified in an index file, ignoring content", cm.output[0])
189 datasets = list(self.butler.registry.queryDatasets("raw_dict", collections=self.outputRun))
190 self.assertEqual(len(datasets), 2)
192 # Again with an index file of metadata and one of translated.
193 # Translated should win.
194 # Put the metadata one first to test that order is preserved.
195 files = [
196 os.path.join(INGESTDIR, "indexed_data", "metadata_subdir", "_index.json"),
197 os.path.join(INGESTDIR, "indexed_data", "_index.json"),
198 ]
199 new_run = self.outputRun + "d"
200 with self.assertLogs(level="WARNING") as cm:
201 self.task.run(files, run=new_run)
202 self.assertIn("already specified in an index file but overriding", cm.output[0])
204 # Reversing the order should change the warning.
205 # Again with an index file of metadata and one of translated.
206 # Translated should win.
207 # Put the metadata one first to test that order is preserved.
208 files = [
209 os.path.join(INGESTDIR, "indexed_data", "_index.json"),
210 os.path.join(INGESTDIR, "indexed_data", "metadata_subdir", "_index.json"),
211 ]
213 new_run = self.outputRun + "e"
214 with self.assertLogs(level="WARNING") as cm:
215 self.task.run(files, run=new_run)
216 self.assertIn("already specified in an index file, ignoring", cm.output[0])
218 # Bad index file.
219 files = [os.path.join(INGESTDIR, "indexed_data", "bad_index", "_index.json")]
220 with self.assertRaises(RuntimeError):
221 self.task.run(files, run=self.outputRun)
223 # Bad index file due to bad instrument.
224 files = [os.path.join(INGESTDIR, "indexed_data", "bad_instrument", "_index.json")]
225 with self.assertLogs(level="WARNING") as cm:
226 with self.assertRaises(RuntimeError):
227 self.task.run(files, run=self.outputRun)
228 self.assertIn("Instrument HSC for file", cm.output[0])
230 def testBadExposure(self):
231 """Test that bad exposures trigger the correct failure modes.
233 This is the only test that uses the bad definition of dataset 4
234 because exposure definitions are defined globally in a butler registry.
235 """
236 # Ingest 3 files. 2 of them will implicitly find an index and one
237 # will use a sidecar.
238 files = [os.path.join(INGESTDIR, "indexed_data", f"dataset_{n}.yaml") for n in (1, 2, 3)]
239 new_run = self.outputRun
240 self.task.run(files, run=new_run)
242 datasets = list(self.butler.registry.queryDatasets("raw_dict", collections=new_run))
243 self.assertEqual(len(datasets), 3)
245 # Test fail fast.
246 self.task.config.failFast = True
248 # Ingest files with conflicting exposure definitions.
249 # Ingest 3 files. One of them will implicitly find an index and one
250 # will use a sidecar. The 3rd will fail due to exposure conflict.
251 files = [os.path.join(INGESTDIR, "indexed_data", f"dataset_{n}.yaml") for n in (1, 3, 4)]
252 new_run = self.outputRun + "_bad_exposure"
253 with self.assertRaises(ConflictingDefinitionError):
254 self.task.run(files, run=new_run)
256 def testBadFile(self):
257 """Try to ingest a bad file."""
258 files = [self.bad_metadata_file]
260 with self.assertRaises(RuntimeError) as cm:
261 # Default is to raise an error at the end.
262 self.task.run(files, run=self.outputRun)
263 self.assertIn("Some failures", str(cm.exception))
265 # Including a good file will result in ingest working but still
266 # raises (we might want to move this to solely happen in the
267 # command line invocation).
268 files.append(self.good_file)
270 # Also include a file with unknown instrument.
271 files.append(self.bad_instrument_file)
273 with self.assertRaises(RuntimeError):
274 self.task.run(files, run=self.outputRun)
275 datasets = list(self.butler.registry.queryDatasets("raw_dict", collections=self.outputRun))
276 self.assertEqual(len(datasets), 1)
278 # Fail fast will trigger a run time error with different text.
279 # Use a different output run to be sure we are not failing because
280 # of the attempt to ingest twice.
281 self.task.config.failFast = True
282 new_run = self.outputRun + "b"
283 with self.assertRaises(RuntimeError) as cm:
284 self.task.run([self.bad_metadata_file, self.good_file], run=new_run)
285 self.assertIn("Problem extracting metadata", str(cm.exception))
287 # Attempt to ingest good file again -- this will fail for a different
288 # reason than failed metadata extraction.
289 with self.assertRaises(ConflictingDefinitionError):
290 self.task.run([self.good_file], run=self.outputRun)
292 # Ingest a file with good metadata but unknown instrument.
293 with self.assertRaises(RuntimeError) as cm:
294 self.task.run([self.bad_instrument_file], run=self.outputRun)
295 self.assertIn("Instrument HSC", str(cm.exception))
297 # Ingest of a metadata index file that will fail translation.
298 with self.assertRaises(RuntimeError) as cm:
299 self.task.run([os.path.join(INGESTDIR, "indexed_data", "metadata_subdir", "_index.json")])
300 self.assertIn("Problem extracting metadata", str(cm.exception))
302 # Ingest of a bad index file.
303 with self.assertRaises(RuntimeError) as cm:
304 self.task.run([os.path.join(INGESTDIR, "indexed_data", "bad_index", "_index.json")])
305 self.assertIn("Problem reading index file", str(cm.exception))
307 # Ingest of an implied bad index file.
308 with self.assertRaises(RuntimeError) as cm:
309 self.task.run([os.path.join(INGESTDIR, "indexed_data", "bad_implied", "dataset_2.yaml")])
311 def testCallbacks(self):
312 """Test the callbacks for failures."""
313 # Define the callbacks.
314 metadata_failures = []
315 successes = []
316 ingest_failures = []
318 def on_metadata_failure(filename, exc):
319 metadata_failures.append(filename)
321 def on_success(datasets):
322 successes.append(datasets)
324 def on_ingest_failure(exposure, exc):
325 ingest_failures.append(exposure)
327 # Need our own task instance
328 config = RawIngestTask.ConfigClass()
329 self.task = RawIngestTask(
330 config=config,
331 butler=self.butler,
332 on_metadata_failure=on_metadata_failure,
333 on_success=on_success,
334 on_ingest_failure=on_ingest_failure,
335 )
337 files = [self.good_file, self.bad_metadata_file, self.bad_instrument_file]
339 with self.assertRaises(RuntimeError):
340 self.task.run(files, run=self.outputRun)
342 self.assertEqual(len(successes), 1)
343 self.assertEqual(len(metadata_failures), 2)
344 self.assertEqual(len(ingest_failures), 0)
346 # Try the good one a second time.
347 with self.assertRaises(RuntimeError):
348 self.task.run([self.good_file], run=self.outputRun)
350 self.assertEqual(len(successes), 1)
351 self.assertEqual(len(ingest_failures), 1)
353 # An index file with metadata that won't translate.
354 metadata_failures[:] = []
355 files = [os.path.join(INGESTDIR, "indexed_data", "metadata_subdir", "_index.json")]
356 with self.assertRaises(RuntimeError):
357 self.task.run(files, run=self.outputRun)
358 self.assertEqual(len(metadata_failures), 2)
360 # Bad index file.
361 metadata_failures[:] = []
362 files = [os.path.join(INGESTDIR, "indexed_data", "bad_index", "_index.json")]
363 with self.assertRaises(RuntimeError):
364 self.task.run(files, run=self.outputRun)
365 self.assertEqual(len(metadata_failures), 1)
367 # Ingest two files that have conflicting exposure metadata.
368 ingest_failures[:] = []
369 successes[:] = []
370 # Ingest 4 files. 2 of them will implicitly find an index and one
371 # will use a sidecar. The 4th will fail due to exposure conflict.
372 files = [os.path.join(INGESTDIR, "indexed_data", f"dataset_{n}.yaml") for n in (1, 2, 3, 4)]
373 new_run = self.outputRun + "_fail"
374 with self.assertRaises(RuntimeError):
375 self.task.run(files, run=new_run)
376 self.assertEqual(len(ingest_failures), 1)
377 self.assertEqual(len(successes), 3)
379 def testSkipExistingExposures(self):
380 """Test that skip_existing_exposures=True avoids exceptions from trying
381 to ingest the same file twice.
383 Notes
384 -----
385 This option also prevents not-ingested-yet raws from being ingested
386 when exposure already exists, but that's (A) hard to test given the
387 test data we have now and (B) not really ideal behavior, just behavior
388 we can live with in order to have a way to avoid keep duplicate ingests
389 from being an error.
390 """
391 # Ingest the first time.
392 self.task.run([self.good_file], run=self.outputRun)
393 # Attempt to ingest a second time with skip_existing_exposures=False
394 # (default). This should fail.
395 with self.assertRaises(RuntimeError):
396 self.task.run([self.good_file], run=self.outputRun)
397 # Try again with `skip_existing_exposures=True.
398 self.task.run([self.good_file], run=self.outputRun, skip_existing_exposures=True)
400 def testUpdateExposureRecords(self):
401 """Test that update_exposure_records=True allows metadata to be
402 modified.
403 """
404 config = RawIngestTask.ConfigClass(failFast=True)
405 task = RawIngestTask(config=config, butler=self.butler)
406 with open(os.path.join(INGESTDIR, "sidecar_data", "dataset_1.json")) as file:
407 metadata = json.load(file)
408 # Modify unique identifiers to avoid clashes with ingests from
409 # other test methods in this test case, because those share a a
410 # data repository.
411 metadata["observation_id"] = "DummyDataset_testUpdateExposureRecords"
412 metadata["observation_counter"] = 10
413 metadata["exposure_id"] = 500
414 metadata["exposure_group"] = "50"
415 metadata["visit_id"] = 500
416 base_filename = "dataset"
417 try:
418 # Copy the original file to be ingested (.yaml) to a temporary
419 # directory, and write the new metadata next to it.
420 tmp_dir = tempfile.mkdtemp(dir=TESTDIR)
421 raw_filename = os.path.join(tmp_dir, f"{base_filename}.yaml")
422 sidecar_filename = os.path.join(tmp_dir, f"{base_filename}.json")
423 shutil.copy(self.good_file, raw_filename)
424 with open(sidecar_filename, "w") as sidecar_file:
425 json.dump(metadata, sidecar_file)
426 task.run([raw_filename], run=self.outputRun)
427 (record1,) = set(
428 self.butler.registry.queryDimensionRecords("exposure", instrument="DummyCam", exposure=500)
429 )
430 self.assertEqual(record1.exposure_time, metadata["exposure_time"])
431 # Modify some metadata and repeat the process to update the
432 # exposure.
433 metadata["exposure_time"] *= 2.0
434 metadata["exposure_time_requested"] *= 2.0
435 with open(sidecar_filename, "w") as sidecar_file:
436 json.dump(metadata, sidecar_file)
437 task.run(
438 [raw_filename], run=self.outputRun, skip_existing_exposures=True, update_exposure_records=True
439 )
440 (record2,) = set(
441 self.butler.registry.queryDimensionRecords("exposure", instrument="DummyCam", exposure=500)
442 )
443 self.assertEqual(record2.exposure_time, record1.exposure_time * 2)
444 finally:
445 shutil.rmtree(tmp_dir, ignore_errors=True)
448class TestRawIngestTaskPickle(unittest.TestCase):
449 """Test that pickling of the RawIngestTask works properly."""
451 @classmethod
452 def setUpClass(cls):
453 cls.root = tempfile.mkdtemp(dir=TESTDIR)
454 cls.creatorButler = butlerTests.makeTestRepo(cls.root, {})
456 @classmethod
457 def tearDownClass(cls):
458 if cls.root is not None:
459 shutil.rmtree(cls.root, ignore_errors=True)
461 def setUp(self):
462 self.butler = butlerTests.makeTestCollection(self.creatorButler, uniqueId=self.id())
464 self.config = RawIngestTask.ConfigClass()
465 self.config.transfer = "copy" # safe non-default value
466 self.task = RawIngestTask(config=self.config, butler=self.butler)
468 def testPickleTask(self):
469 stream = pickle.dumps(self.task)
470 copy = pickle.loads(stream)
471 self.enterContext(copy.butler)
472 self.assertEqual(self.task.getFullName(), copy.getFullName())
473 self.assertEqual(self.task.log.name, copy.log.name)
474 self.assertEqual(self.task.config, copy.config)
475 self.assertEqual(self.task.butler._config, copy.butler._config)
476 self.assertEqual(list(self.task.butler.collections.defaults), list(copy.butler.collections.defaults))
477 self.assertEqual(self.task.butler.run, copy.butler.run)
478 self.assertEqual(self.task.universe, copy.universe)
479 self.assertEqual(self.task.datasetType, copy.datasetType)
482if __name__ == "__main__": 482 ↛ 483line 482 didn't jump to line 483 because the condition on line 482 was never true
483 unittest.main()