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