Coverage for python / lsst / obs / base / ingest.py: 15%
372 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-18 01:06 -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/>.
23__all__ = ("RawIngestConfig", "RawIngestTask", "makeTransferChoiceField")
25import json
26import re
27from collections import defaultdict
28from collections.abc import Callable, Iterable, Iterator, MutableMapping, Sequence, Sized
29from dataclasses import InitVar, dataclass
30from multiprocessing import Pool
31from typing import Any, ClassVar
33from astro_metadata_translator import MetadataTranslator, ObservationInfo, merge_headers
34from astro_metadata_translator.indexing import process_index_data, process_sidecar_data
36from lsst.afw.fits import readMetadata
37from lsst.daf.butler import (
38 Butler,
39 CollectionType,
40 DataCoordinate,
41 DatasetIdGenEnum,
42 DatasetRef,
43 DatasetType,
44 DimensionRecord,
45 DimensionUniverse,
46 FileDataset,
47 Formatter,
48 Progress,
49 Timespan,
50)
51from lsst.pex.config import ChoiceField, Config, Field
52from lsst.pipe.base import Instrument, Task
53from lsst.resources import ResourcePath, ResourcePathExpression
54from lsst.utils.timer import time_this, timeMethod
56from ._instrument import makeExposureRecordFromObsInfo
58# multiprocessing.Pool is actually a function, not a type, and the real type
59# isn't exposed, so we can't used it annotations, so we'll just punt on it via
60# this alias instead.
61PoolType = Any
64def _do_nothing(*args: Any, **kwargs: Any) -> None:
65 """Do nothing.
67 This is a function that accepts anything and does nothing.
68 For use as a default in callback arguments.
69 """
70 pass
73def _log_msg_counter(noun: int | Sized) -> tuple[int, str]:
74 """Count the iterable and return the count and plural modifier.
76 Parameters
77 ----------
78 noun : `Sized` or `int`
79 Thing to count. If given an integer it is assumed to be the count
80 to use to calculate modifier.
82 Returns
83 -------
84 num : `int`
85 Number of items found in ``noun``.
86 modifier : `str`
87 Character to add to the end of a string referring to these items
88 to indicate whether it was a single item or not. Returns empty
89 string if there is one item or "s" otherwise.
91 Examples
92 --------
93 .. code-block:: python
95 log.warning("Found %d file%s", *_log_msg_counter(nfiles))
96 """
97 if isinstance(noun, int):
98 num = noun
99 else:
100 num = len(noun)
101 return num, "" if num == 1 else "s"
104@dataclass
105class RawFileDatasetInfo:
106 """Information about a single dataset within a raw file."""
108 dataId: DataCoordinate
109 """Data ID for this file (`lsst.daf.butler.DataCoordinate`)."""
111 obsInfo: ObservationInfo
112 """Standardized observation metadata extracted directly from the file
113 headers (`astro_metadata_translator.ObservationInfo`).
114 """
117@dataclass
118class RawFileData:
119 """Information about a single raw file, used during ingest."""
121 datasets: list[RawFileDatasetInfo]
122 """The information describing each dataset within this raw file.
123 (`list` of `RawFileDatasetInfo`)
124 """
126 filename: ResourcePath
127 """URI of the file this information was extracted from (`str`).
129 This is the path prior to ingest, not the path after ingest.
130 """
132 FormatterClass: type[Formatter]
133 """Formatter class that should be used to ingest this file (`type`; as
134 subclass of `~lsst.daf.butler.Formatter`).
135 """
137 instrument: Instrument | None
138 """The `Instrument` instance associated with this file. Can be `None`
139 if ``datasets`` is an empty list."""
142@dataclass
143class RawExposureData:
144 """Information about a complete raw exposure, used during ingest."""
146 dataId: DataCoordinate
147 """Data ID for this exposure (`lsst.daf.butler.DataCoordinate`).
148 """
150 files: list[RawFileData]
151 """List of structures containing file-level information.
152 """
154 universe: InitVar[DimensionUniverse]
155 """Set of all known dimensions.
156 """
158 record: DimensionRecord
159 """The exposure `DimensionRecord` that must be inserted into the
160 `~lsst.daf.butler.Registry` prior to file-level ingest
161 (`~lsst.daf.butler.DimensionRecord`).
162 """
164 dependencyRecords: dict[str, DimensionRecord]
165 """Additional records that must be inserted into the
166 `~lsst.daf.butler.Registry` prior to ingesting the exposure ``record``
167 (e.g., to satisfy foreign key constraints), indexed by the dimension name.
168 """
171def makeTransferChoiceField(
172 doc: str = "How to transfer files (None for no transfer).", default: str = "auto"
173) -> ChoiceField:
174 """Create a Config field with options for transferring data between repos.
176 The allowed options for the field are exactly those supported by
177 `lsst.daf.butler.Datastore.ingest`.
179 Parameters
180 ----------
181 doc : `str`
182 Documentation for the configuration field.
183 default : `str`, optional
184 Default transfer mode for the field.
186 Returns
187 -------
188 field : `lsst.pex.config.ChoiceField`
189 Configuration field.
190 """
191 return ChoiceField(
192 doc=doc,
193 dtype=str,
194 allowed={
195 "move": "move",
196 "copy": "copy",
197 "auto": "choice will depend on datastore",
198 "direct": "use URI to ingested file directly in datastore",
199 "link": "hard link falling back to symbolic link",
200 "hardlink": "hard link",
201 "symlink": "symbolic (soft) link",
202 "relsymlink": "relative symbolic link",
203 },
204 optional=True,
205 default=default,
206 )
209class RawIngestConfig(Config):
210 """Configuration class for RawIngestTask."""
212 transfer = makeTransferChoiceField()
213 failFast: Field[bool] = Field(
214 dtype=bool,
215 default=False,
216 doc="If True, stop ingest as soon as any problem is encountered with any file. "
217 "Otherwise problem files will be skipped and logged and a report issued at completion.",
218 )
221class RawIngestTask(Task):
222 """Driver Task for ingesting raw data into Gen3 Butler repositories.
224 Parameters
225 ----------
226 config : `RawIngestConfig`
227 Configuration for the task.
228 butler : `~lsst.daf.butler.Butler`
229 Writeable butler instance, with ``butler.run`` set to the appropriate
230 `~lsst.daf.butler.CollectionType.RUN` collection for these raw
231 datasets.
232 on_success : `Callable`, optional
233 A callback invoked when all of the raws associated with an exposure
234 are ingested. Will be passed a list of `FileDataset` objects, each
235 containing one or more resolved `DatasetRef` objects. If this callback
236 raises it will interrupt the entire ingest process, even if
237 `RawIngestConfig.failFast` is `False`.
238 on_metadata_failure : `Callable`, optional
239 A callback invoked when a failure occurs trying to translate the
240 metadata for a file. Will be passed the URI and the exception, in
241 that order, as positional arguments. Guaranteed to be called in an
242 ``except`` block, allowing the callback to re-raise or replace (with
243 ``raise ... from``) to override the task's usual error handling (before
244 `RawIngestConfig.failFast` logic occurs).
245 on_ingest_failure : `Callable`, optional
246 A callback invoked when dimension record or dataset insertion into the
247 database fails for an exposure. Will be passed a `RawExposureData`
248 instance and the exception, in that order, as positional arguments.
249 Guaranteed to be called in an ``except`` block, allowing the callback
250 to re-raise or replace (with ``raise ... from``) to override the task's
251 usual error handling (before `RawIngestConfig.failFast` logic occurs).
252 **kwargs
253 Additional keyword arguments are forwarded to the `lsst.pipe.base.Task`
254 constructor.
256 Notes
257 -----
258 Each instance of `RawIngestTask` writes to the same Butler. Each
259 invocation of `RawIngestTask.run` ingests a list of files.
260 """
262 ConfigClass: ClassVar[type[Config]] = RawIngestConfig
264 _DefaultName: ClassVar[str] = "ingest"
266 def getDatasetType(self) -> DatasetType:
267 """Return the default DatasetType of the datasets ingested by this
268 Task.
270 Returns
271 -------
272 datasetType : `DatasetType`
273 The default dataset type to use for the data being ingested. This
274 is only used if the relevant `~lsst.pipe.base.Instrument` does not
275 define an override.
276 """
277 return DatasetType(
278 "raw",
279 ("instrument", "detector", "exposure"),
280 "Exposure",
281 universe=self.butler.dimensions,
282 )
284 # Mypy can not determine that the config passed to super() is this type.
285 config: RawIngestConfig
287 def __init__(
288 self,
289 config: RawIngestConfig,
290 *,
291 butler: Butler,
292 on_success: Callable[[list[FileDataset]], Any] = _do_nothing,
293 on_metadata_failure: Callable[[ResourcePath, Exception], Any] = _do_nothing,
294 on_ingest_failure: Callable[[RawExposureData, Exception], Any] = _do_nothing,
295 **kwargs: Any,
296 ):
297 config.validate() # Not a CmdlineTask nor PipelineTask, so have to validate the config here.
298 super().__init__(config, **kwargs)
299 self.butler = butler
300 self.universe = self.butler.dimensions
301 self.datasetType = self.getDatasetType()
302 self._on_success = on_success
303 self._on_metadata_failure = on_metadata_failure
304 self._on_ingest_failure = on_ingest_failure
305 self.progress = Progress("obs.base.RawIngestTask")
307 # Import all the instrument classes so that we ensure that we
308 # have all the relevant metadata translators loaded.
309 Instrument.importAll(self.butler.registry)
311 # Read all the instrument records into a cache since they will be
312 # needed later to calculate day_obs timespans, if appropriate.
313 self._instrument_records = {
314 rec.name: rec for rec in butler.registry.queryDimensionRecords("instrument")
315 }
317 def _reduce_kwargs(self) -> dict[str, Any]:
318 # Add extra parameters to pickle.
319 return dict(
320 **super()._reduce_kwargs(),
321 butler=self.butler,
322 on_success=self._on_success,
323 on_metadata_failure=self._on_metadata_failure,
324 on_ingest_failure=self._on_ingest_failure,
325 )
327 def _determine_instrument_formatter(
328 self, dataId: DataCoordinate, filename: ResourcePath
329 ) -> tuple[Instrument | None, type[Formatter]]:
330 """Determine the instrument and formatter class.
332 Parameters
333 ----------
334 dataId : `lsst.daf.butler.DataCoordinate`
335 The dataId associated with this dataset.
336 filename : `lsst.resources.ResourcePath`
337 URI of file used for error reporting.
339 Returns
340 -------
341 instrument : `Instrument` or `None`
342 Instance of the `Instrument` associated with this dataset. `None`
343 indicates that the instrument could not be determined.
344 formatterClass : `type`
345 Class to be used as the formatter for this dataset.
346 """
347 # The data model currently assumes that whilst multiple datasets
348 # can be associated with a single file, they must all share the
349 # same formatter.
350 try:
351 instrument = Instrument.fromName(dataId["instrument"], self.butler.registry) # type: ignore
352 except LookupError as e:
353 self._on_metadata_failure(filename, e)
354 self.log.warning(
355 "Instrument %s for file %s not known to registry", dataId["instrument"], filename
356 )
357 if self.config.failFast:
358 raise RuntimeError(
359 f"Instrument {dataId['instrument']} for file {filename} not known to registry"
360 ) from e
361 FormatterClass = Formatter
362 # Indicate that we could not work out the instrument.
363 instrument = None
364 else:
365 assert instrument is not None, "Should be guaranted by fromName succeeding."
366 FormatterClass = instrument.getRawFormatter(dataId)
367 return instrument, FormatterClass
369 def extractMetadata(self, filename: ResourcePath) -> RawFileData:
370 """Extract and process metadata from a single raw file.
372 Parameters
373 ----------
374 filename : `lsst.resources.ResourcePath`
375 URI to the file.
377 Returns
378 -------
379 data : `RawFileData`
380 A structure containing the metadata extracted from the file,
381 as well as the original filename. All fields will be populated,
382 but the `RawFileData.dataId` attribute will be a minimal
383 (unexpanded) `~lsst.daf.butler.DataCoordinate` instance. The
384 ``instrument`` field will be `None` if there is a problem
385 with metadata extraction.
387 Notes
388 -----
389 Assumes that there is a single dataset associated with the given
390 file. Instruments using a single file to store multiple datasets
391 must implement their own version of this method.
393 By default the method will catch all exceptions unless the ``failFast``
394 configuration item is `True`. If an error is encountered the
395 `_on_metadata_failure()` method will be called. If no exceptions
396 result and an error was encountered the returned object will have
397 a null-instrument class and no datasets.
399 This method supports sidecar JSON files which can be used to
400 extract metadata without having to read the data file itself.
401 The sidecar file is always used if found.
402 """
403 sidecar_fail_msg = "" # Requires prepended space when set.
404 try:
405 sidecar_file = filename.updatedExtension(".json")
406 if sidecar_file.exists():
407 content = json.loads(sidecar_file.read())
408 headers = [process_sidecar_data(content)]
409 sidecar_fail_msg = " (via sidecar)"
410 else:
411 # Read the metadata from the data file itself.
413 # For remote files download the entire file to get the
414 # header. This is very inefficient and it would be better
415 # to have some way of knowing where in the file the headers
416 # are and to only download those parts of the file.
417 with filename.as_local() as local_file:
418 # Read the primary. This might be sufficient.
419 header = readMetadata(local_file.ospath, 0)
420 translator_class = None
422 try:
423 # Try to work out a translator class early.
424 translator_class = MetadataTranslator.determine_translator(
425 header, filename=str(filename)
426 )
427 except ValueError:
428 # Primary header was not sufficient (maybe this file
429 # has been compressed or is a MEF with minimal
430 # primary). Read second header and merge with primary.
431 header = merge_headers([header, readMetadata(local_file.ospath, 1)], mode="overwrite")
433 # Try again to work out a translator class, letting this
434 # fail.
435 if translator_class is None:
436 translator_class = MetadataTranslator.determine_translator(
437 header, filename=str(filename)
438 )
440 # Request the headers to use for ingest
441 headers = list(translator_class.determine_translatable_headers(local_file.ospath, header))
443 # Add each header to the dataset list
444 datasets = [self._calculate_dataset_info(h, filename) for h in headers]
446 except Exception as e:
447 self.log.debug("Problem extracting metadata from %s%s: %s", filename, sidecar_fail_msg, e)
448 # Indicate to the caller that we failed to read.
449 datasets = []
450 formatterClass = Formatter
451 instrument = None
452 self._on_metadata_failure(filename, e)
453 if self.config.failFast:
454 raise RuntimeError(
455 f"Problem extracting metadata for file {filename}{sidecar_fail_msg}"
456 ) from e
457 else:
458 self.log.debug("Extracted metadata for file %s%s", filename, sidecar_fail_msg)
459 # The data model currently assumes that whilst multiple datasets
460 # can be associated with a single file, they must all share the
461 # same formatter.
462 instrument, formatterClass = self._determine_instrument_formatter(datasets[0].dataId, filename)
463 if instrument is None:
464 datasets = []
466 return RawFileData(
467 datasets=datasets,
468 filename=filename,
469 # MyPy wants this to be a non-abstract class, which is not true
470 # for the error case where instrument is None and datasets=[].
471 FormatterClass=formatterClass, # type: ignore
472 instrument=instrument,
473 )
475 @classmethod
476 def getObservationInfoSubsets(cls) -> tuple[set, set]:
477 """Return subsets of fields in the `ObservationInfo` that we care
478 about.
480 These fields will be used in constructing an exposure record.
482 Returns
483 -------
484 required : `set`
485 Set of `ObservationInfo` field names that are required.
486 optional : `set`
487 Set of `ObservationInfo` field names we will use if they are
488 available.
489 """
490 # Marking the new properties "group_counter_*" and
491 # "has_simulated_content" as required, assumes that we either
492 # recreate any existing index/sidecar files that include translated
493 # values, or else allow astro_metadata_translator to fill in
494 # defaults.
495 required = {
496 "datetime_begin",
497 "datetime_end",
498 "detector_num",
499 "exposure_group",
500 "exposure_id",
501 "exposure_time_requested",
502 "group_counter_end",
503 "group_counter_start",
504 "has_simulated_content",
505 "instrument",
506 "observation_id",
507 "observation_type",
508 "observing_day",
509 "physical_filter",
510 }
511 optional = {
512 "altaz_begin",
513 "boresight_rotation_coord",
514 "boresight_rotation_angle",
515 "dark_time",
516 "tracking_radec",
517 "object",
518 "observation_counter",
519 "observation_reason",
520 "observing_day_offset",
521 "science_program",
522 "visit_id",
523 "can_see_sky",
524 }
525 return required, optional
527 def _calculate_dataset_info(
528 self, header: MutableMapping[str, Any] | ObservationInfo, filename: ResourcePath
529 ) -> RawFileDatasetInfo:
530 """Calculate a RawFileDatasetInfo from the supplied information.
532 Parameters
533 ----------
534 header : Mapping or `astro_metadata_translator.ObservationInfo`
535 Header from the dataset or previously-translated content.
536 filename : `lsst.resources.ResourcePath`
537 Filename to use for error messages.
539 Returns
540 -------
541 dataset : `RawFileDatasetInfo`
542 The dataId, and observation information associated with this
543 dataset.
544 """
545 required, optional = self.getObservationInfoSubsets()
546 if isinstance(header, ObservationInfo):
547 obsInfo = header
548 missing = []
549 # Need to check the required properties are present.
550 for property in required:
551 # getattr does not need to be protected because it is using
552 # the defined list above containing properties that must exist.
553 value = getattr(obsInfo, property)
554 if value is None:
555 missing.append(property)
556 if missing:
557 raise ValueError(
558 f"Requested required properties are missing from file {filename}: {missing} (via JSON)"
559 )
561 else:
562 obsInfo = ObservationInfo(
563 header,
564 pedantic=False,
565 filename=str(filename),
566 required=required,
567 subset=required | optional,
568 )
570 dataId = DataCoordinate.standardize(
571 instrument=obsInfo.instrument,
572 exposure=obsInfo.exposure_id,
573 detector=obsInfo.detector_num,
574 universe=self.universe,
575 )
576 return RawFileDatasetInfo(obsInfo=obsInfo, dataId=dataId)
578 def locateAndReadIndexFiles(
579 self, files: Iterable[ResourcePath]
580 ) -> tuple[dict[ResourcePath, Any], list[ResourcePath], set[ResourcePath], set[ResourcePath]]:
581 """Given a list of files, look for index files and read them.
583 Index files can either be explicitly in the list of files to
584 ingest, or else located in the same directory as a file to ingest.
585 Index entries are always used if present.
587 Parameters
588 ----------
589 files : iterable over `lsst.resources.ResourcePath`
590 URIs to the files to be ingested.
592 Returns
593 -------
594 index : `dict` [`ResourcePath`, Any]
595 Merged contents of all relevant index files found. These can
596 be explicitly specified index files or ones found in the
597 directory alongside a data file to be ingested.
598 updated_files : `list` of `ResourcePath`
599 Updated list of the input files with entries removed that were
600 found listed in an index file. Order is not guaranteed to
601 match the order of the files given to this routine.
602 good_index_files: `set` [ `ResourcePath` ]
603 Index files that were successfully read.
604 bad_index_files: `set` [ `ResourcePath` ]
605 Files that looked like index files but failed to read properly.
606 """
607 # Convert the paths to absolute for easy comparison with index content.
608 # Do not convert to real paths since we have to assume that index
609 # files are in this location and not the location which it links to.
610 files = tuple(f.abspath() for f in files)
612 # Index files must be named this.
613 index_root_file = "_index.json"
615 # Group the files by directory.
616 files_by_directory = defaultdict(set)
618 for path in files:
619 directory, file_in_dir = path.split()
620 files_by_directory[directory].add(file_in_dir)
622 # All the metadata read from index files with keys of full path.
623 index_entries: dict[ResourcePath, Any] = {}
625 # Index files we failed to read.
626 bad_index_files = set()
628 # Any good index files that were found and used.
629 good_index_files = set()
631 # Look for index files in those directories.
632 for directory, files_in_directory in files_by_directory.items():
633 possible_index_file = directory.join(index_root_file)
634 if possible_index_file.exists():
635 # If we are explicitly requesting an index file the
636 # messages should be different.
637 index_msg = "inferred"
638 is_implied = True
639 if index_root_file in files_in_directory:
640 index_msg = "explicit"
641 is_implied = False
643 # Try to read the index file and catch and report any
644 # problems.
645 try:
646 content = json.loads(possible_index_file.read())
647 index = process_index_data(content, force_dict=True)
648 # mypy should in theory know that this is a mapping
649 # from the overload type annotation of process_index_data.
650 assert isinstance(index, MutableMapping)
651 except Exception as e:
652 # Only trigger the callback if the index file
653 # was asked for explicitly. Triggering on implied file
654 # might be surprising.
655 if not is_implied:
656 self._on_metadata_failure(possible_index_file, e)
657 if self.config.failFast:
658 raise RuntimeError(
659 f"Problem reading index file from {index_msg} location {possible_index_file}"
660 ) from e
661 bad_index_files.add(possible_index_file)
662 continue
664 self.log.debug("Extracted index metadata from %s file %s", index_msg, possible_index_file)
665 good_index_files.add(possible_index_file)
667 # Go through the index adding entries for files.
668 # If we have non-index files in this directory marked for
669 # ingest we should only get index information for those.
670 # If the index file was explicit we use all entries.
671 if is_implied:
672 files_to_ingest = files_in_directory
673 else:
674 files_to_ingest = set(index)
676 # Copy relevant metadata into a single dict for all index
677 # entries.
678 for file_in_dir in files_to_ingest:
679 # Skip an explicitly specified index file.
680 # This should never happen because an explicit index
681 # file will force ingest of all files in the index
682 # and not use the explicit file list. If somehow
683 # this is not true we continue. Raising an exception
684 # seems like the wrong thing to do since this is harmless.
685 if file_in_dir == index_root_file:
686 self.log.info(
687 "Logic error found scanning directory %s. Please file ticket.", directory
688 )
689 continue
690 if file_in_dir in index:
691 file = directory.join(file_in_dir)
692 if file in index_entries:
693 # ObservationInfo overrides raw metadata
694 if isinstance(index[file_in_dir], ObservationInfo) and not isinstance(
695 index_entries[file], ObservationInfo
696 ):
697 self.log.warning(
698 "File %s already specified in an index file but overriding"
699 " with ObservationInfo content from %s",
700 file,
701 possible_index_file,
702 )
703 else:
704 self.log.warning(
705 "File %s already specified in an index file, ignoring content from %s",
706 file,
707 possible_index_file,
708 )
709 # Do nothing in this case
710 continue
712 index_entries[file] = index[file_in_dir]
714 # Remove files from list that have index entries and also
715 # any files that we determined to be explicit index files
716 # or any index files that we failed to read.
717 filtered = set(files) - set(index_entries) - good_index_files - bad_index_files
719 # The filtered list loses the initial order. Retaining the order
720 # is good for testing but does have a cost if there are many
721 # files when copying the good values out. A dict would have faster
722 # lookups (using the files as keys) but use more memory.
723 ordered = [f for f in filtered if f in files]
725 return index_entries, ordered, good_index_files, bad_index_files
727 def processIndexEntries(self, index_entries: dict[ResourcePath, Any]) -> list[RawFileData]:
728 """Convert index entries to RawFileData.
730 Parameters
731 ----------
732 index_entries : `dict` [`ResourcePath`, Any]
733 Dict indexed by name of file to ingest and with keys either
734 raw metadata or translated
735 `~astro_metadata_translator.ObservationInfo`.
737 Returns
738 -------
739 data : `list` [ `RawFileData` ]
740 Structures containing the metadata extracted from the file,
741 as well as the original filename. All fields will be populated,
742 but the `RawFileData.dataId` attributes will be minimal
743 (unexpanded) `~lsst.daf.butler.DataCoordinate` instances.
744 """
745 fileData = []
746 for filename, metadata in index_entries.items():
747 try:
748 datasets = [self._calculate_dataset_info(metadata, filename)]
749 except Exception as e:
750 self.log.debug("Problem extracting metadata for file %s found in index file: %s", filename, e)
751 datasets = []
752 formatterClass = Formatter
753 instrument = None
754 self._on_metadata_failure(filename, e)
755 if self.config.failFast:
756 raise RuntimeError(
757 f"Problem extracting metadata for file {filename} found in index file"
758 ) from e
759 else:
760 instrument, formatterClass = self._determine_instrument_formatter(
761 datasets[0].dataId, filename
762 )
763 if instrument is None:
764 datasets = []
765 fileData.append(
766 RawFileData(
767 datasets=datasets,
768 filename=filename,
769 # MyPy wants this to be a non-abstract class, which is not
770 # true for the error case where instrument is None and
771 # datasets=[].
772 FormatterClass=formatterClass, # type: ignore
773 instrument=instrument,
774 )
775 )
776 return fileData
778 def groupByExposure(self, files: Iterable[RawFileData]) -> list[RawExposureData]:
779 """Group an iterable of `RawFileData` by exposure.
781 Parameters
782 ----------
783 files : iterable of `RawFileData`
784 File-level information to group.
786 Returns
787 -------
788 exposures : `list` of `RawExposureData`
789 A list of structures that group the file-level information by
790 exposure. All fields will be populated. The
791 `RawExposureData.dataId` attributes will be minimal (unexpanded)
792 `~lsst.daf.butler.DataCoordinate` instances.
793 """
794 exposureDimensions = self.universe["exposure"].minimal_group
795 byExposure = defaultdict(list)
796 for f in files:
797 # Assume that the first dataset is representative for the file.
798 byExposure[f.datasets[0].dataId.subset(exposureDimensions)].append(f)
800 return [
801 RawExposureData(
802 dataId=dataId,
803 files=exposureFiles,
804 universe=self.universe,
805 record=self.makeExposureRecord(exposureFiles[0].datasets[0].obsInfo, self.universe),
806 dependencyRecords=self.makeDependencyRecords(
807 exposureFiles[0].datasets[0].obsInfo, self.universe
808 ),
809 )
810 for dataId, exposureFiles in byExposure.items()
811 ]
813 def makeExposureRecord(
814 self, obsInfo: ObservationInfo, universe: DimensionUniverse, **kwargs: Any
815 ) -> DimensionRecord:
816 """Construct a registry record for an exposure.
818 This is a method that subclasses will often want to customize. This can
819 often be done by calling this base class implementation with additional
820 ``kwargs``.
822 Parameters
823 ----------
824 obsInfo : `ObservationInfo`
825 Observation details for (one of the components of) the exposure.
826 universe : `DimensionUniverse`
827 Set of all known dimensions.
828 **kwargs
829 Additional field values for this record.
831 Returns
832 -------
833 record : `DimensionRecord`
834 The exposure record that must be inserted into the
835 `~lsst.daf.butler.Registry` prior to file-level ingest.
836 """
837 return makeExposureRecordFromObsInfo(obsInfo, universe, **kwargs)
839 def makeDependencyRecords(
840 self, obsInfo: ObservationInfo, universe: DimensionUniverse
841 ) -> dict[str, DimensionRecord]:
842 """Construct dependency records.
844 These dependency records will be inserted into the
845 `~lsst.daf.butler.Registry` before the exposure records, because they
846 are dependencies of the exposure. This allows an opportunity to satisfy
847 foreign key constraints that exist because of dimensions related to the
848 exposure.
850 This is a method that subclasses may want to customize, if they've
851 added dimensions that relate to an exposure.
853 Parameters
854 ----------
855 obsInfo : `ObservationInfo`
856 Observation details for (one of the components of) the exposure.
857 universe : `DimensionUniverse`
858 Set of all known dimensions.
860 Returns
861 -------
862 records : `dict` [`str`, `DimensionRecord`]
863 The records to insert, indexed by dimension name.
864 """
865 records: dict[str, DimensionRecord] = {}
866 if "exposure" not in universe:
867 return records
868 exposure = universe["exposure"]
869 if "group" in exposure.implied:
870 records["group"] = universe["group"].RecordClass(
871 name=obsInfo.exposure_group,
872 instrument=obsInfo.instrument,
873 )
874 if "day_obs" in exposure.implied:
875 if (offset := getattr(obsInfo, "observing_day_offset")) is not None:
876 offset_int = round(offset.to_value("s"))
877 timespan = Timespan.from_day_obs(obsInfo.observing_day, offset_int)
878 else:
879 timespan = None
880 records["day_obs"] = universe["day_obs"].RecordClass(
881 instrument=obsInfo.instrument,
882 id=obsInfo.observing_day,
883 timespan=timespan,
884 )
885 return records
887 def expandDataIds(self, data: RawExposureData) -> RawExposureData:
888 """Expand the data IDs associated with a raw exposure.
890 This adds the metadata records.
892 Parameters
893 ----------
894 exposure : `RawExposureData`
895 A structure containing information about the exposure to be
896 ingested. Must have `RawExposureData.record` populated. Should
897 be considered consumed upon return.
899 Returns
900 -------
901 exposure : `RawExposureData`
902 An updated version of the input structure, with
903 `RawExposureData.dataId` and nested `RawFileData.dataId` attributes
904 updated to data IDs for which
905 `~lsst.daf.butler.DataCoordinate.hasRecords` returns `True`.
906 """
907 # We start by expanded the exposure-level data ID; we won't use that
908 # directly in file ingest, but this lets us do some database lookups
909 # once per exposure instead of once per file later.
910 data.dataId = self.butler.registry.expandDataId(
911 data.dataId,
912 # We pass in the records we'll be inserting shortly so they aren't
913 # looked up from the database. We do expect instrument and filter
914 # records to be retrieved from the database here (though the
915 # Registry may cache them so there isn't a lookup every time).
916 records={"exposure": data.record, **data.dependencyRecords},
917 )
918 # Now we expand the per-file (exposure+detector) data IDs. This time
919 # we pass in the records we just retrieved from the exposure data ID
920 # expansion.
921 for file in data.files:
922 for dataset in file.datasets:
923 dataset.dataId = self.butler.registry.expandDataId(
924 dataset.dataId,
925 records={k: data.dataId.records[k] for k in data.dataId.dimensions.elements},
926 )
927 return data
929 def prep(
930 self, files: Iterable[ResourcePath], *, pool: PoolType | None = None
931 ) -> tuple[Iterator[RawExposureData], list[ResourcePath]]:
932 """Perform all non-database-updating ingest preprocessing steps.
934 Parameters
935 ----------
936 files : iterable over `str` or path-like objects
937 Paths to the files to be ingested. Will be made absolute
938 if they are not already.
939 pool : `multiprocessing.Pool`, optional
940 If not `None`, a process pool with which to parallelize some
941 operations.
943 Returns
944 -------
945 exposures : `Iterator` [ `RawExposureData` ]
946 Data structures containing dimension records, filenames, and data
947 IDs to be ingested (one structure for each exposure).
948 bad_files : `list` of `str`
949 List of all the files that could not have metadata extracted.
950 """
951 mapFunc = map if pool is None else pool.imap_unordered
953 def _partition_good_bad(
954 file_data: Iterable[RawFileData],
955 ) -> tuple[list[RawFileData], list[ResourcePath]]:
956 """Filter out bad files and return good with list of bad."""
957 good_files = []
958 bad_files = []
959 for fileDatum in self.progress.wrap(file_data, desc="Reading image metadata"):
960 if not fileDatum.datasets:
961 bad_files.append(fileDatum.filename)
962 else:
963 good_files.append(fileDatum)
964 return good_files, bad_files
966 # Look for index files and read them.
967 # There should be far fewer index files than data files.
968 index_entries, files, good_index_files, bad_index_files = self.locateAndReadIndexFiles(files)
969 if bad_index_files:
970 self.log.info("Failed to read the following explicitly requested index files:")
971 for bad in sorted(bad_index_files):
972 self.log.info("- %s", bad)
974 # Now convert all the index file entries to standard form for ingest.
975 processed_bad_index_files: list[ResourcePath] = []
976 indexFileData = self.processIndexEntries(index_entries)
977 if indexFileData:
978 indexFileData, processed_bad_index_files = _partition_good_bad(indexFileData)
979 self.log.info(
980 "Successfully extracted metadata for %d file%s found in %d index file%s with %d failure%s",
981 *_log_msg_counter(indexFileData),
982 *_log_msg_counter(good_index_files),
983 *_log_msg_counter(processed_bad_index_files),
984 )
986 # Extract metadata and build per-detector regions.
987 # This could run in a subprocess so collect all output
988 # before looking at failures.
989 fileData: Iterator[RawFileData] = mapFunc(self.extractMetadata, files)
991 # Filter out all the failed reads and store them for later
992 # reporting.
993 good_file_data, bad_files = _partition_good_bad(fileData)
994 self.log.info(
995 "Successfully extracted metadata from %d file%s with %d failure%s",
996 *_log_msg_counter(good_file_data),
997 *_log_msg_counter(bad_files),
998 )
1000 # Combine with data from index files.
1001 good_file_data.extend(indexFileData)
1002 bad_files.extend(processed_bad_index_files)
1003 bad_files.extend(bad_index_files)
1005 # Use that metadata to group files (and extracted metadata) by
1006 # exposure. Never parallelized because it's intrinsically a gather
1007 # step.
1008 exposureData: list[RawExposureData] = self.groupByExposure(good_file_data)
1010 # The next operation operates on RawExposureData instances (one at
1011 # a time) in-place and then returns the modified instance. We call it
1012 # as a pass-through instead of relying on the arguments we pass in to
1013 # have been modified because in the parallel case those arguments are
1014 # going to be pickled and unpickled, and I'm not certain
1015 # multiprocessing is careful enough with that for output arguments to
1016 # work.
1018 # Expand the data IDs to include all dimension metadata; we need this
1019 # because we may need to generate path templates that rely on that
1020 # metadata.
1021 # This is the first step that involves actual database calls (but just
1022 # SELECTs), so if there's going to be a problem with connections vs.
1023 # multiple processes, or lock contention (in SQLite) slowing things
1024 # down, it'll happen here.
1025 return mapFunc(self.expandDataIds, exposureData), bad_files
1027 def ingestExposureDatasets(
1028 self,
1029 exposure: RawExposureData,
1030 datasetType: DatasetType,
1031 *,
1032 run: str,
1033 skip_existing_exposures: bool = False,
1034 track_file_attrs: bool = True,
1035 ) -> list[FileDataset]:
1036 """Ingest all raw files in one exposure.
1038 Parameters
1039 ----------
1040 exposure : `RawExposureData`
1041 A structure containing information about the exposure to be
1042 ingested. Must have `RawExposureData.records` populated and all
1043 data ID attributes expanded.
1044 datasetType : `DatasetType`
1045 The dataset type associated with this exposure.
1046 run : `str`
1047 Name of a RUN-type collection to write to.
1048 skip_existing_exposures : `bool`, optional
1049 If `True` (`False` is default), skip raws that have already been
1050 ingested (i.e. raws for which we already have a dataset with the
1051 same data ID in the target collection, even if from another file).
1052 Note that this is much slower than just not passing
1053 already-ingested files as inputs, because we still need to read and
1054 process metadata to identify which exposures to search for.
1055 track_file_attrs : `bool`, optional
1056 Control whether file attributes such as the size or checksum should
1057 be tracked by the datastore. Whether this parameter is honored
1058 depends on the specific datastore implementation.
1060 Returns
1061 -------
1062 datasets : `list` of `lsst.daf.butler.FileDataset`
1063 Per-file structures identifying the files ingested and their
1064 dataset representation in the data repository.
1065 """
1066 # Raw files are preferentially ingested using a UUID derived from
1067 # the collection name and dataId.
1068 if self.butler.registry.supportsIdGenerationMode(DatasetIdGenEnum.DATAID_TYPE_RUN):
1069 mode = DatasetIdGenEnum.DATAID_TYPE_RUN
1070 else:
1071 mode = DatasetIdGenEnum.UNIQUE
1073 datasets = []
1074 for file in exposure.files:
1075 refs = [
1076 DatasetRef(datasetType, d.dataId, run=run, id_generation_mode=mode) for d in file.datasets
1077 ]
1078 if refs:
1079 datasets.append(
1080 FileDataset(path=file.filename.abspath(), refs=refs, formatter=file.FormatterClass)
1081 )
1083 self.butler.ingest(
1084 *datasets,
1085 transfer=self.config.transfer,
1086 record_validation_info=track_file_attrs,
1087 skip_existing=skip_existing_exposures,
1088 )
1089 return datasets
1091 def ingestFiles(
1092 self,
1093 files: Sequence[ResourcePath],
1094 *,
1095 pool: PoolType | None = None,
1096 processes: int = 1,
1097 run: str | None = None,
1098 skip_existing_exposures: bool = False,
1099 update_exposure_records: bool = False,
1100 track_file_attrs: bool = True,
1101 ) -> tuple[list[DatasetRef], list[ResourcePath], int, int, int]:
1102 """Ingest files into a Butler data repository.
1104 This creates any new exposure or visit Dimension entries needed to
1105 identify the ingested files, creates new Dataset entries in the
1106 Registry and finally ingests the files themselves into the Datastore.
1107 Any needed instrument, detector, and physical_filter Dimension entries
1108 must exist in the Registry before `run` is called.
1110 Parameters
1111 ----------
1112 files : iterable over `lsst.resources.ResourcePath`
1113 URIs to the files to be ingested.
1114 pool : `multiprocessing.Pool`, optional
1115 If not `None`, a process pool with which to parallelize some
1116 operations.
1117 processes : `int`, optional
1118 The number of processes to use. Ignored if ``pool`` is not `None`.
1119 run : `str`, optional
1120 Name of a RUN-type collection to write to, overriding
1121 the default derived from the instrument name.
1122 skip_existing_exposures : `bool`, optional
1123 If `True` (`False` is default), skip raws that have already been
1124 ingested (i.e. raws for which we already have a dataset with the
1125 same data ID in the target collection, even if from another file).
1126 Note that this is much slower than just not passing
1127 already-ingested files as inputs, because we still need to read and
1128 process metadata to identify which exposures to search for. It
1129 also will not work reliably if multiple processes are attempting to
1130 ingest raws from the same exposure concurrently, in that different
1131 processes may still attempt to ingest the same raw and conflict,
1132 causing a failure that prevents other raws from the same exposure
1133 from being ingested.
1134 update_exposure_records : `bool`, optional
1135 If `True` (`False` is default), update existing exposure records
1136 that conflict with the new ones instead of rejecting them. THIS IS
1137 AN ADVANCED OPTION THAT SHOULD ONLY BE USED TO FIX METADATA THAT IS
1138 KNOWN TO BE BAD. This should usually be combined with
1139 ``skip_existing_exposures=True``.
1140 track_file_attrs : `bool`, optional
1141 Control whether file attributes such as the size or checksum should
1142 be tracked by the datastore. Whether this parameter is honored
1143 depends on the specific datastore implementation.
1145 Returns
1146 -------
1147 refs : `list` of `lsst.daf.butler.DatasetRef`
1148 Dataset references for ingested raws.
1149 bad_files : `list` of `ResourcePath`
1150 Given paths that could not be ingested.
1151 n_exposures : `int`
1152 Number of exposures successfully ingested.
1153 n_exposures_failed : `int`
1154 Number of exposures that failed when inserting dimension data.
1155 n_ingests_failed : `int`
1156 Number of exposures that failed when ingesting raw datasets.
1157 """
1158 created_pool = False
1159 if pool is None and processes > 1:
1160 pool = Pool(processes)
1161 created_pool = True
1163 try:
1164 with time_this(
1165 self.log,
1166 msg="Reading metadata from %d file%s",
1167 args=(*_log_msg_counter(files),),
1168 ):
1169 exposureData, bad_files = self.prep(files, pool=pool)
1170 finally:
1171 if created_pool and pool:
1172 # The pool is not needed any more so close it if we created
1173 # it to ensure we clean up resources.
1174 pool.close()
1175 pool.join()
1177 # Up to this point, we haven't modified the data repository at all.
1178 # Now we finally do that, with one transaction per exposure. This is
1179 # not parallelized at present because the performance of this step is
1180 # limited by the database server. That may or may not change in the
1181 # future once we increase our usage of bulk inserts and reduce our
1182 # usage of savepoints; we've tried to get everything but the database
1183 # operations done in advance to reduce the time spent inside
1184 # transactions.
1185 refs = []
1186 runs = set()
1187 datasetTypes: dict[str, DatasetType] = {}
1188 n_exposures = 0
1189 n_exposures_failed = 0
1190 n_ingests_failed = 0
1191 for exposure in self.progress.wrap(exposureData, desc="Ingesting raw exposures"):
1192 assert exposure.record is not None, "Should be guaranteed by prep()"
1193 self.log.debug(
1194 "Attempting to ingest %d file%s from exposure %s:%s",
1195 *_log_msg_counter(exposure.files),
1196 exposure.record.instrument,
1197 exposure.record.obs_id,
1198 )
1200 try:
1201 with time_this(
1202 self.log,
1203 msg="Creating dimension records for instrument %s, exposure %s",
1204 args=(
1205 str(exposure.record.instrument),
1206 str(exposure.record.id),
1207 ),
1208 ):
1209 for name, record in exposure.dependencyRecords.items():
1210 self.butler.registry.syncDimensionData(name, record, update=update_exposure_records)
1211 inserted_or_updated = self.butler.registry.syncDimensionData(
1212 "exposure",
1213 exposure.record,
1214 update=update_exposure_records,
1215 )
1216 except Exception as e:
1217 self._on_ingest_failure(exposure, e)
1218 n_exposures_failed += 1
1219 self.log.warning(
1220 "Exposure %s:%s could not be registered: %s",
1221 exposure.record.instrument,
1222 exposure.record.obs_id,
1223 e,
1224 )
1225 if self.config.failFast:
1226 raise e
1227 continue
1229 if isinstance(inserted_or_updated, dict):
1230 # Exposure is in the registry and we updated it, so
1231 # syncDimensionData returned a dict.
1232 self.log.info(
1233 "Exposure %s:%s was already present, but columns %s were updated.",
1234 exposure.record.instrument,
1235 exposure.record.obs_id,
1236 str(list(inserted_or_updated.keys())),
1237 )
1239 # Determine the instrument so we can work out the dataset type.
1240 instrument = exposure.files[0].instrument
1241 assert instrument is not None, (
1242 "file should have been removed from this list by prep if instrument could not be found"
1243 )
1245 if raw_definition := getattr(instrument, "raw_definition", None):
1246 datasetTypeName, dimensions, storageClass = raw_definition
1247 if not (datasetType := datasetTypes.get(datasetTypeName)):
1248 datasetType = DatasetType(
1249 datasetTypeName, dimensions, storageClass, universe=self.butler.dimensions
1250 )
1251 else:
1252 datasetType = self.datasetType
1253 if datasetType.name not in datasetTypes:
1254 self.butler.registry.registerDatasetType(datasetType)
1255 datasetTypes[datasetType.name] = datasetType
1257 # Override default run if nothing specified explicitly.
1258 if run is None:
1259 this_run = instrument.makeDefaultRawIngestRunName()
1260 else:
1261 this_run = run
1262 if this_run not in runs:
1263 self.butler.registry.registerCollection(this_run, type=CollectionType.RUN)
1264 runs.add(this_run)
1265 try:
1266 datasets_for_exposure = self.ingestExposureDatasets(
1267 exposure,
1268 datasetType=datasetType,
1269 run=this_run,
1270 skip_existing_exposures=skip_existing_exposures,
1271 track_file_attrs=track_file_attrs,
1272 )
1273 except Exception as e:
1274 self._on_ingest_failure(exposure, e)
1275 n_ingests_failed += 1
1276 self.log.warning("Failed to ingest the following for reason: %s", e)
1277 for f in exposure.files:
1278 self.log.warning("- %s", f.filename)
1279 if self.config.failFast:
1280 raise e
1281 continue
1282 else:
1283 self._on_success(datasets_for_exposure)
1284 for dataset in datasets_for_exposure:
1285 refs.extend(dataset.refs)
1287 # Success for this exposure.
1288 n_exposures += 1
1289 self.log.info(
1290 "Exposure %s:%s ingested successfully", exposure.record.instrument, exposure.record.obs_id
1291 )
1293 return refs, bad_files, n_exposures, n_exposures_failed, n_ingests_failed
1295 @timeMethod
1296 def run(
1297 self,
1298 files: Sequence[ResourcePathExpression],
1299 *,
1300 pool: PoolType | None = None,
1301 processes: int = 1,
1302 run: str | None = None,
1303 file_filter: str | re.Pattern = r"\.fit[s]?\b",
1304 group_files: bool = True,
1305 skip_existing_exposures: bool = False,
1306 update_exposure_records: bool = False,
1307 track_file_attrs: bool = True,
1308 ) -> list[DatasetRef]:
1309 """Ingest files into a Butler data repository.
1311 This creates any new exposure or visit Dimension entries needed to
1312 identify the ingested files, creates new Dataset entries in the
1313 Registry and finally ingests the files themselves into the Datastore.
1314 Any needed instrument, detector, and physical_filter Dimension entries
1315 must exist in the Registry before `run` is called.
1317 Parameters
1318 ----------
1319 files : iterable `lsst.resources.ResourcePath`, `str` or path-like
1320 Paths to the files to be ingested. Can refer to directories.
1321 Will be made absolute if they are not already.
1322 pool : `multiprocessing.Pool`, optional
1323 If not `None`, a process pool with which to parallelize some
1324 operations.
1325 processes : `int`, optional
1326 The number of processes to use. Ignored if ``pool`` is not `None`.
1327 run : `str`, optional
1328 Name of a RUN-type collection to write to, overriding
1329 the default derived from the instrument name.
1330 file_filter : `str` or `re.Pattern`, optional
1331 Pattern to use to discover files to ingest within directories.
1332 The default is to search for FITS files. The regex applies to
1333 files within the directory.
1334 group_files : `bool`, optional
1335 Group files by directory if they have been discovered in
1336 directories. Will not affect files explicitly provided.
1337 skip_existing_exposures : `bool`, optional
1338 If `True` (`False` is default), skip raws that have already been
1339 ingested (i.e. raws for which we already have a dataset with the
1340 same data ID in the target collection, even if from another file).
1341 Note that this is much slower than just not passing
1342 already-ingested files as inputs, because we still need to read and
1343 process metadata to identify which exposures to search for. It
1344 also will not work reliably if multiple processes are attempting to
1345 ingest raws from the same exposure concurrently, in that different
1346 processes may still attempt to ingest the same raw and conflict,
1347 causing a failure that prevents other raws from the same exposure
1348 from being ingested.
1349 update_exposure_records : `bool`, optional
1350 If `True` (`False` is default), update existing exposure records
1351 that conflict with the new ones instead of rejecting them. THIS IS
1352 AN ADVANCED OPTION THAT SHOULD ONLY BE USED TO FIX METADATA THAT IS
1353 KNOWN TO BE BAD. This should usually be combined with
1354 ``skip_existing_exposures=True``.
1355 track_file_attrs : `bool`, optional
1356 Control whether file attributes such as the size or checksum should
1357 be tracked by the datastore. Whether this parameter is honored
1358 depends on the specific datastore implementation.
1360 Returns
1361 -------
1362 refs : `list` of `lsst.daf.butler.DatasetRef`
1363 Dataset references for ingested raws.
1365 Notes
1366 -----
1367 This method inserts all datasets for an exposure within a transaction,
1368 guaranteeing that partial exposures are never ingested. The exposure
1369 dimension record is inserted with `Registry.syncDimensionData` first
1370 (in its own transaction), which inserts only if a record with the same
1371 primary key does not already exist. This allows different files within
1372 the same exposure to be ingested in different runs.
1373 """
1374 refs = []
1375 bad_files = []
1376 n_exposures = 0
1377 n_exposures_failed = 0
1378 n_ingests_failed = 0
1379 if group_files:
1380 for group in ResourcePath.findFileResources(files, file_filter, group_files):
1381 new_refs, bad, n_exp, n_exp_fail, n_ingest_fail = self.ingestFiles(
1382 tuple(group),
1383 pool=pool,
1384 processes=processes,
1385 run=run,
1386 skip_existing_exposures=skip_existing_exposures,
1387 update_exposure_records=update_exposure_records,
1388 track_file_attrs=track_file_attrs,
1389 )
1390 refs.extend(new_refs)
1391 bad_files.extend(bad)
1392 n_exposures += n_exp
1393 n_exposures_failed += n_exp_fail
1394 n_ingests_failed += n_ingest_fail
1395 else:
1396 refs, bad_files, n_exposures, n_exposures_failed, n_ingests_failed = self.ingestFiles(
1397 tuple(ResourcePath.findFileResources(files, file_filter, group_files)),
1398 pool=pool,
1399 processes=processes,
1400 run=run,
1401 skip_existing_exposures=skip_existing_exposures,
1402 update_exposure_records=update_exposure_records,
1403 )
1405 had_failure = False
1407 if bad_files:
1408 had_failure = True
1409 self.log.warning("Could not extract observation metadata from the following:")
1410 for f in bad_files:
1411 self.log.warning("- %s", f)
1413 self.log.info(
1414 "Successfully processed data from %d exposure%s with %d failure%s from exposure"
1415 " registration and %d failure%s from file ingest.",
1416 *_log_msg_counter(n_exposures),
1417 *_log_msg_counter(n_exposures_failed),
1418 *_log_msg_counter(n_ingests_failed),
1419 )
1420 if n_exposures_failed > 0 or n_ingests_failed > 0:
1421 had_failure = True
1422 self.log.info("Ingested %d distinct Butler dataset%s", *_log_msg_counter(refs))
1424 if had_failure:
1425 raise RuntimeError("Some failures encountered during ingestion")
1427 return refs