Coverage for python/lsst/obs/base/ingest_tests.py: 27%
214 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-06 01:38 -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 <https://www.gnu.org/licenses/>.
22"""Base class for writing Gen3 raw data ingest tests."""
24__all__ = ("IngestTestBase",)
26import abc
27import os
28import shutil
29import tempfile
30import unittest
32import lsst.afw.cameraGeom
33import lsst.afw.cameraGeom.testUtils # For assertDetectorsEqual
34import lsst.obs.base
35from lsst.daf.butler import Butler, Registry
36from lsst.daf.butler.cli.butler import cli as butlerCli
37from lsst.daf.butler.cli.utils import LogCliRunner
38from lsst.pipe.base import Instrument
39from lsst.resources import ResourcePath
40from lsst.utils import doImportType
42from . import script
45class IngestTestBase(metaclass=abc.ABCMeta):
46 """Base class for tests of gen3 ingest. Subclass from this, then
47 `unittest.TestCase` to get a working test suite.
48 """
50 ingestDir = ""
51 """Root path to ingest files into. Typically `obs_package/tests/`; the
52 actual directory will be a tempdir under this one.
53 """
55 ingestDatasetTypeName = "raw"
56 """The DatasetType to use for the ingest.
58 If this is not an Exposure dataset type the tests will be more limited.
59 """
61 dataIds = []
62 """list of butler data IDs of files that should have been ingested."""
64 file = ""
65 """Full path to a file to ingest in tests."""
67 filterLabel = None
68 """The lsst.afw.image.FilterLabel that should be returned by the above
69 file."""
71 rawIngestTask = "lsst.obs.base.RawIngestTask"
72 """The task to use in the Ingest test."""
74 curatedCalibrationDatasetTypes = None
75 """List or tuple of Datasets types that should be present after calling
76 writeCuratedCalibrations. If `None` writeCuratedCalibrations will
77 not be called and the test will be skipped."""
79 defineVisitsTask = lsst.obs.base.DefineVisitsTask
80 """The task to use to define visits from groups of exposures.
81 This is ignored if ``visits`` is `None`.
82 """
84 visits = {}
85 """A dictionary mapping visit data IDs the lists of exposure data IDs that
86 are associated with them.
87 If this is empty (but not `None`), visit definition will be run but no
88 visits will be expected (e.g. because no exposures are on-sky
89 observations).
90 """
92 seed_config = None
93 """Location of a seed configuration file to pass to butler create.
95 Useful if additional formatters or storage classes need to be defined.
96 """
98 @property
99 @abc.abstractmethod
100 def instrumentClassName(self):
101 """The fully qualified instrument class name.
103 Returns
104 -------
105 `str`
106 The fully qualified instrument class name.
107 """
108 pass
110 @property
111 def instrumentClass(self):
112 """The instrument class."""
113 return doImportType(self.instrumentClassName)
115 @property
116 def instrumentName(self):
117 """The name of the instrument.
119 Returns
120 -------
121 `str`
122 The name of the instrument.
123 """
124 return self.instrumentClass.getName()
126 @classmethod
127 def setUpClass(cls):
128 # Use a temporary working directory.
129 cls.root = tempfile.mkdtemp(dir=cls.ingestDir)
130 cls._createRepo()
132 # Register the instrument and its static metadata.
133 cls._registerInstrument()
135 # Determine the relevant datastore root to use for testing.
136 butler = Butler(cls.root)
137 roots = butler.get_datastore_roots()
138 assert len(roots) == 1 # Only one datastore.
139 cls.datastore_root = list(roots.values())[0]
141 def setUp(self):
142 # Want a unique run name per test.
143 self.outputRun = "raw_ingest_" + self.id()
145 @classmethod
146 def tearDownClass(cls):
147 if os.path.exists(cls.root):
148 shutil.rmtree(cls.root, ignore_errors=True)
150 def verifyIngest(self, files=None, cli=False, fullCheck=False):
151 """
152 Test that RawIngestTask ingested the expected files.
154 Parameters
155 ----------
156 files : `list` [`str`], or None
157 List of files to be ingested, or None to use ``self.file``
158 fullCheck : `bool`, optional
159 If `True`, read the full raw dataset and check component
160 consistency. If `False` check that a component can be read
161 but do not read the entire raw exposure.
163 Notes
164 -----
165 Reading all the ingested test data can be expensive. The code paths
166 for reading the second raw are the same as reading the first so
167 we do not gain anything by doing full checks of everything.
168 Only read full pixel data for first dataset from file.
169 Don't even do that if we are requested not to by the caller.
170 This only really affects files that contain multiple datasets.
171 """
172 butler = Butler(self.root, run=self.outputRun)
173 datasets = list(butler.registry.queryDatasets(self.ingestDatasetTypeName, collections=self.outputRun))
174 self.assertEqual(len(datasets), len(self.dataIds))
176 # Can check that the timespan in the day_obs matches the exposure
177 # record.
178 if "day_obs" in butler.dimensions:
179 days = {
180 (rec.instrument, rec.id): rec.timespan
181 for rec in butler.registry.queryDimensionRecords("day_obs")
182 }
184 exp_records = list(butler.registry.queryDimensionRecords("exposure"))
185 for exp in exp_records:
186 day_span = days[exp.instrument, exp.day_obs]
187 if day_span is not None:
188 self.assertTrue(
189 day_span.contains(exp.timespan.begin), f"Timespan mismatch of {exp} and {day_span}"
190 )
192 # Get the URI to the first dataset and check it is inside the
193 # datastore.
194 datasetUri = butler.getURI(datasets[0])
195 self.assertIsNotNone(datasetUri.relative_to(self.datastore_root))
197 # Get the relevant dataset type.
198 datasetType = butler.get_dataset_type(self.ingestDatasetTypeName)
200 for dataId in self.dataIds:
201 # For testing we only read the entire dataset the first time
202 # round if this is an Exposure. If it's not an Exposure
203 # we always read it completely but we don't read components
204 # because for an arbitrary dataset type we can't easily tell
205 # what component to test.
207 if not datasetType.storageClass.name.startswith("Exposure"):
208 exposure = butler.get(self.ingestDatasetTypeName, dataId)
209 # Could be anything so nothing to test by default
210 continue
212 # Check that we can read metadata from a raw.
213 metadata = butler.get(f"{self.ingestDatasetTypeName}.metadata", dataId)
214 if not fullCheck:
215 continue
216 fullCheck = False
217 exposure = butler.get(self.ingestDatasetTypeName, dataId)
219 # Comparing headers will not work directly because of header
220 # fix up provenance.
221 metadata_headers = metadata.toDict()
222 exposure_headers = exposure.getMetadata().toDict()
223 metadata_headers.pop("HIERARCH ASTRO METADATA FIX DATE", None)
224 exposure_headers.pop("HIERARCH ASTRO METADATA FIX DATE", None)
225 self.assertEqual(metadata_headers, exposure_headers)
227 # Since components follow a different code path we check that
228 # WCS match and also we check that at least the shape
229 # of the image is the same (rather than doing per-pixel equality)
230 wcs = butler.get(f"{self.ingestDatasetTypeName}.wcs", dataId)
231 self.assertEqual(wcs, exposure.getWcs())
233 rawImage = butler.get(f"{self.ingestDatasetTypeName}.image", dataId)
234 self.assertEqual(rawImage.getBBox(), exposure.getBBox())
236 # Check that the filter label got the correct band.
237 filterLabel = butler.get(f"{self.ingestDatasetTypeName}.filter", dataId)
238 self.assertEqual(filterLabel, self.filterLabel)
240 # Check that the exposure's Detector is the same as the component
241 # we would read (this is tricky for LSST, which modifies its
242 # detector at read time; for most other cameras it should be
243 # trivially satisfied.
244 detector = butler.get(f"{self.ingestDatasetTypeName}.detector", dataId)
245 self.assertDetectorsEqual(detector, exposure.getDetector(), compareTransforms=False)
247 self.checkRepo(files=files)
249 def checkRepo(self, files=None):
250 """Check the state of the repository after ingest.
252 This is an optional hook provided for subclasses; by default it does
253 nothing.
255 Parameters
256 ----------
257 files : `list` [`str`], or None
258 List of files to be ingested, or None to use ``self.file``
259 """
260 return
262 @classmethod
263 def _createRepo(cls):
264 """Use the Click `testing` module to call the butler command line api
265 to create a repository.
266 """
267 runner = LogCliRunner()
268 args = []
269 if cls.seed_config:
270 args.extend(["--seed-config", cls.seed_config])
271 result = runner.invoke(butlerCli, ["create", cls.root, *args])
272 # Classmethod so assertEqual does not work.
273 assert result.exit_code == 0, f"output: {result.output} exception: {result.exception}"
275 def _ingestRaws(self, transfer, file=None):
276 """Use the Click `testing` module to call the butler command line api
277 to ingest raws.
279 Parameters
280 ----------
281 transfer : `str`
282 The external data transfer type.
283 file : `str`
284 Path to a file to ingest instead of the default associated with
285 the object.
286 """
287 if file is None:
288 file = self.file
289 runner = LogCliRunner()
290 result = runner.invoke(
291 butlerCli,
292 [
293 "ingest-raws",
294 self.root,
295 file,
296 "--output-run",
297 self.outputRun,
298 "--transfer",
299 transfer,
300 "--ingest-task",
301 self.rawIngestTask,
302 ],
303 )
304 self.assertEqual(result.exit_code, 0, f"output: {result.output} exception: {result.exception}")
306 @classmethod
307 def _registerInstrument(cls):
308 """Use the Click `testing` module to call the butler command line api
309 to register the instrument.
310 """
311 runner = LogCliRunner()
312 result = runner.invoke(butlerCli, ["register-instrument", cls.root, cls.instrumentClassName])
313 # Classmethod so assertEqual does not work.
314 assert result.exit_code == 0, f"output: {result.output} exception: {result.exception}"
316 def _writeCuratedCalibrations(self):
317 """Use the Click `testing` module to call the butler command line api
318 to write curated calibrations.
319 """
320 runner = LogCliRunner()
321 result = runner.invoke(
322 butlerCli, ["write-curated-calibrations", self.root, self.instrumentName, "test"]
323 )
324 self.assertEqual(result.exit_code, 0, f"output: {result.output} exception: {result.exception}")
326 def testLink(self):
327 self._ingestRaws(transfer="link")
328 self.verifyIngest()
330 def testSymLink(self):
331 self._ingestRaws(transfer="symlink")
332 self.verifyIngest()
334 def testDirect(self):
335 self._ingestRaws(transfer="direct")
337 # Check that it really did have a URI outside of datastore.
338 srcUri = ResourcePath(self.file, forceAbsolute=True)
339 butler = Butler(self.root, run=self.outputRun)
340 datasets = list(butler.registry.queryDatasets(self.ingestDatasetTypeName, collections=self.outputRun))
341 datastoreUri = butler.getURI(datasets[0])
342 self.assertEqual(datastoreUri, srcUri)
344 def testCopy(self):
345 self._ingestRaws(transfer="copy")
346 # Only test full read of raws for the copy test. No need to do it
347 # in the other tests since the formatter will be the same in all
348 # cases.
349 self.verifyIngest(fullCheck=True)
351 def testHardLink(self):
352 try:
353 self._ingestRaws(transfer="hardlink")
354 # Running ingest through the Click testing infrastructure causes
355 # the original exception indicating that we can't hard-link
356 # on this filesystem to be turned into a nonzero exit code, which
357 # then trips the test assertion.
358 except (AssertionError, PermissionError) as err:
359 raise unittest.SkipTest(
360 "Skipping hard-link test because input data is on a different filesystem."
361 ) from err
362 self.verifyIngest()
364 def testInPlace(self):
365 """Test that files already in the directory can be added to the
366 registry in-place.
367 """
368 butler = Butler(self.root, run=self.outputRun)
370 # If the test uses an index file the index file needs to also
371 # appear in the datastore root along with the file to be ingested.
372 # In that scenario the file name being used for ingest can not
373 # be modified and must have the same name as found in the index
374 # file itself.
375 source_file_uri = ResourcePath(self.file)
376 index_file = source_file_uri.dirname().join("_index.json")
377 pathInStore = source_file_uri.basename()
378 if index_file.exists():
379 os.symlink(index_file.ospath, self.datastore_root.join("_index.json").ospath)
380 else:
381 # No index file so we are free to pick any name.
382 pathInStore = "prefix-" + pathInStore
384 # Create a symlink to the original file so that it looks like it
385 # is now inside the datastore.
386 newPath = self.datastore_root.join(pathInStore)
387 os.symlink(os.path.abspath(self.file), newPath.ospath)
389 # If there is a sidecar file it needs to be linked in as well
390 # since ingest code does not follow symlinks.
391 sidecar_uri = ResourcePath(source_file_uri).updatedExtension(".json")
392 if sidecar_uri.exists():
393 newSidecar = ResourcePath(newPath).updatedExtension(".json")
394 os.symlink(sidecar_uri.ospath, newSidecar.ospath)
396 # Run ingest with auto mode since that should automatically determine
397 # that an in-place ingest is happening.
398 self._ingestRaws(transfer="auto", file=newPath.ospath)
399 self.verifyIngest()
401 # Recreate a butler post-ingest (the earlier one won't see the
402 # ingested files).
403 butler = Butler(self.root, run=self.outputRun)
405 # Check that the URI associated with this path is the right one.
406 uri = butler.getURI(self.ingestDatasetTypeName, self.dataIds[0])
407 self.assertEqual(uri.relative_to(self.datastore_root), pathInStore)
409 def testFailOnConflict(self):
410 """Re-ingesting the same data into the repository should fail."""
411 self._ingestRaws(transfer="symlink")
412 with self.assertRaises(AssertionError):
413 self._ingestRaws(transfer="symlink")
415 def testWriteCuratedCalibrations(self):
416 """Test that we can ingest the curated calibrations, and read them
417 with `loadCamera` both before and after.
418 """
419 if self.curatedCalibrationDatasetTypes is None:
420 raise unittest.SkipTest("Class requests disabling of writeCuratedCalibrations test")
422 butler = Butler(self.root, writeable=False)
423 collection = self.instrumentClass().makeCalibrationCollectionName("test")
425 # Trying to load a camera with a data ID not known to the registry
426 # is an error, because we can't get any temporal information.
427 with self.assertRaises(LookupError):
428 lsst.obs.base.loadCamera(butler, {"exposure": 0}, collections=collection)
430 # Ingest raws in order to get some exposure records.
431 self._ingestRaws(transfer="auto")
433 # Load camera should returned an unversioned camera because there's
434 # nothing in the repo.
435 camera, isVersioned = lsst.obs.base.loadCamera(butler, self.dataIds[0], collections=collection)
436 self.assertFalse(isVersioned)
437 self.assertIsInstance(camera, lsst.afw.cameraGeom.Camera)
439 self._writeCuratedCalibrations()
441 # Make a new butler instance to make sure we don't have any stale
442 # caches (e.g. of DatasetTypes). Note that we didn't give
443 # _writeCuratedCalibrations the butler instance we had, because it's
444 # trying to test the CLI interface anyway.
445 butler = Butler(self.root, writeable=False)
447 instrumentClass = self.instrumentClass()
448 calibration_names = instrumentClass.getCuratedCalibrationNames()
450 for datasetTypeName in self.curatedCalibrationDatasetTypes:
451 with self.subTest(dtype=datasetTypeName):
452 found = list(
453 butler.registry.queryDatasetAssociations(
454 datasetTypeName,
455 collections=collection,
456 )
457 )
458 self.assertGreater(len(found), 0, f"Checking {datasetTypeName}")
459 self.assertIn(datasetTypeName, calibration_names)
461 # Load camera should returned the versioned camera from the repo.
462 camera, isVersioned = lsst.obs.base.loadCamera(butler, self.dataIds[0], collections=collection)
463 self.assertTrue(isVersioned)
464 self.assertIsInstance(camera, lsst.afw.cameraGeom.Camera)
466 def testDefineVisits(self):
467 if self.visits is None:
468 self.skipTest("Expected visits were not defined.")
469 self._ingestRaws(transfer="link")
471 # Check that obscore table (if configured) has correct contents.
472 butler = Butler(self.root, run=self.outputRun)
473 self._check_obscore(butler.registry, has_visits=False)
475 # Calling defineVisits tests the implementation of the butler command
476 # line interface "define-visits" subcommand. Functions in the script
477 # folder are generally considered protected and should not be used
478 # as public api.
479 script.defineVisits(
480 self.root,
481 config_file=None,
482 collections=self.outputRun,
483 instrument=self.instrumentName,
484 )
486 # Test that we got the visits we expected.
487 visits = butler.registry.queryDataIds(["visit"]).expanded().toSet()
488 self.assertCountEqual(visits, self.visits.keys())
489 instr = Instrument.from_string(self.instrumentName, butler.registry)
490 camera = instr.getCamera()
491 for foundVisit, (expectedVisit, expectedExposures) in zip(visits, self.visits.items(), strict=True):
492 # Test that this visit is associated with the expected exposures.
493 foundExposures = (
494 butler.registry.queryDataIds(["exposure"], dataId=expectedVisit).expanded().toSet()
495 )
496 self.assertCountEqual(foundExposures, expectedExposures)
497 # Test that we have a visit region, and that it contains all of the
498 # detector+visit regions.
499 self.assertIsNotNone(foundVisit.region)
500 detectorVisitDataIds = (
501 butler.registry.queryDataIds(["visit", "detector"], dataId=expectedVisit).expanded().toSet()
502 )
503 self.assertEqual(len(detectorVisitDataIds), len(camera))
504 for dataId in detectorVisitDataIds:
505 self.assertTrue(foundVisit.region.contains(dataId.region))
507 # Check obscore table again.
508 self._check_obscore(butler.registry, has_visits=True)
510 def _check_obscore(self, registry: Registry, has_visits: bool) -> None:
511 """Verify contents of obscore table."""
512 return