21 """Concrete implementations of `PathElementHandler`.
23 The `PathElementHandler` ABC is defined in ``scanner.py`` instead of here to
24 avoid a circular dependency between modules.
26 from __future__
import annotations
28 __all__ = [
"IgnoreHandler",
"SkipHandler",
"SubdirectoryHandler",
"TargetFileHandler"]
30 from abc
import abstractmethod
42 from lsst.daf.butler
import (
48 from ..translators
import Translator
49 from .parser
import PathElementParser
50 from .scanner
import PathElementHandler, DirectoryScanner
53 from lsst.daf.butler
import FormatterParameter
57 """A `PathElementHandler` that matches via a regular expression, and does
60 An `IgnoreHandler` is used to ignore file or directory patterns that can
61 occur at any level in the directory tree, and have no relation to any
62 Gen2 filename template.
66 pattern : `re.Pattern`
67 A regular expression pattern.
69 Whether this handler should be applied to files (`True`) or
70 directories (`False`).
72 def __init__(self, pattern: re.Pattern, isForFiles: bool):
77 __slots__ = (
"_pattern",
"_isForFiles")
80 return f
"{type(self).__name__}({self._pattern}, isForFiles={self._isForFiles})"
92 datasets: Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]], *,
93 predicate: Callable[[DataCoordinate], bool]) -> bool:
102 """An intermediate base class for `PathElementHandler` classes that utilize
103 a `PathElementParser` to match a Gen2 filename template.
107 parser : `PathElementParser`
108 An object that matches the path element this handler is responsible for
109 and extracts a (partial) Gen2 data ID from it.
115 __slots__ = (
"_parser",)
118 return f
"{type(self).__name__}(parser={self._parser})"
121 datasets: Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]], *,
122 predicate: Callable[[DataCoordinate], bool]) -> bool:
125 if nextDataId2
is None:
127 self.
handle(path, nextDataId2, datasets, predicate=predicate)
136 def handle(self, path: str, nextDataId2: dict,
137 datasets: Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]], *,
138 predicate: Callable[[DataCoordinate], bool]):
139 """Customization hook for ``__call__``.
141 Subclasses must override this method, while external callers (i.e.
142 `DirectoryScanner` should instead invoke `__call__`.
147 Full path of the file or directory.
149 Gen2 data ID (usually partial) extracted from the path so far.
150 datasets : `dict` [`DatasetType`, `list` [`FileDataset`] ]
151 Dictionary that found datasets should be added to.
152 predicate : `~collections.abc.Callable`
153 A callable taking a single `DataCoordinate` argument and returning
154 `bool`, indicating whether that (Gen3) data ID represents one
155 that should be included in the scan.
156 formatterMap : `dict`, optional
157 Map dataset type to specialist formatter.
159 raise NotImplementedError()
163 """A `ParsedPathElementHandler` that does nothing with an entry other
164 optionally logging a warning message.
166 A `SkipHandler` is used for Gen2 datasets that we can recognize but do not
167 want to (or cannot) extract Gen3 datasets from, or other files/directories
168 that alway appears at a fixed level in the diectory tree.
172 parser : `PathElementParser`
173 An object that matches the path element this handler is responsible for
174 and extracts a (partial) Gen2 data ID from it.
176 Whether this handler should be applied to files (`True`) or
177 directories (`False`).
178 message : `str`, optional
179 A message to log at warning level when this handler matches a path
180 entry. If `None`, matched entrie will be silently skipped.
182 def __init__(self, parser: PathElementParser, isForFiles: bool, message: Optional[str]):
187 __slots__ = (
"_message",
"_isForFiles")
193 def handle(self, path: str, nextDataId2: dict,
194 datasets: Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]], *,
195 predicate: Callable[[DataCoordinate], bool]):
202 """A `PathElementHandler` that uses a `DirectoryScanner` to recurse.
206 parser : `PathElementParser`
207 An object that matches the path element this handler is responsible for
208 and extracts a (partial) Gen2 data ID from it.
212 The nested `DirectoryScanner` is default-constructed and should be
213 populated with child handlers after the `SubdirectoryHandler` is created.
220 __slots__ = (
"scanner",)
227 datasets: Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]], *,
228 predicate: Callable[[DataCoordinate], bool]):
236 dataId3, _ = self.
translate(nextDataId2, partial=
True)
237 if dataId3
is not None:
238 scan = predicate(dataId3)
243 handler.lastDataId2 = nextDataId2
244 self.
scanner.scan(path, datasets, predicate=predicate)
246 def translate(self, dataId2: dict, *, partial: bool =
False
247 ) -> Tuple[Optional[DataCoordinate], Optional[str]]:
253 result, calibDate = handler.translate(dataId2, partial=
True)
254 if result
is not None:
255 return result, calibDate
258 scanner: DirectoryScanner
259 """Scanner object that holds handlers for the entries of the subdirectory
260 matched by this handler (`DirectoryScanner`).
265 """A `PathElementHandler` that matches files that correspond to target
266 datasets and outputs `FileDataset` instances for them.
270 parser : `PathElementParser`
271 An object that matches the path element this handler is responsible for
272 and extracts a (partial) Gen2 data ID from it.
273 translator : `Translator`
274 Object that translates data IDs from Gen2 to Gen3.
275 datasetType : `lsst.daf.butler.DatasetType`
276 Gen3 dataset type for the datasets this handler matches.
277 formatter : `lsst.daf.butler.Formatter` or `str`, optional
278 A Gen 3 formatter class or fully-qualified name.
280 def __init__(self, parser: PathElementParser, translator: Translator, datasetType: DatasetType,
281 formatter: FormatterParameter =
None):
287 __slots__ = (
"_translator",
"_datasetType",
"_formatter")
290 return f
"{type(self).__name__}({self._translator}, {self._datasetType})"
297 datasets: Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]], *,
298 predicate: Callable[[DataCoordinate], bool]):
300 dataId3, calibDate = self.
translate(nextDataId2, partial=
False)
301 if predicate(dataId3):
309 def translate(self, dataId2: dict, *, partial: bool =
False
310 ) -> Tuple[Optional[DataCoordinate], Optional[str]]:
312 rawDataId3, calibDate = self.
_translator(dataId2, partial=partial)
315 DataCoordinate.standardize(rawDataId3, universe=self.
_datasetType.dimensions.universe),
320 DataCoordinate.standardize(rawDataId3, graph=self.
_datasetType.dimensions),
326 """Handler for FITS files that store image and metadata in multiple HDUs
327 per file, for example DECam raw and Community Pipeline calibrations.
331 For now, this is only used by DECam, and may need to be made more generic
332 (e.g. making ``metadata['CCDNUM']`` use a configurable field) to be used
333 with other obs packages.
336 datasets: Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]], *,
337 predicate: Callable[[DataCoordinate], bool]):
338 dataId3, calibDate = self.
translate(nextDataId2, partial=
True)
340 def get_detectors(filename):
344 for i
in range(1, fitsData.countHdus()):
346 metadata = fitsData.readMetadata()
347 detectors.append(metadata[
'CCDNUM'])
350 if predicate(dataId3):
351 detectors = get_detectors(path)
353 for detector
in detectors:
354 newDataId3 = DataCoordinate.standardize(dataId3,
360 FileDataset(refs=refs, path=path, formatter=self.
_formatter)
363 def translate(self, dataId2: dict, *, partial: bool =
False
364 ) -> Tuple[Optional[DataCoordinate], Optional[str]]:
365 assert partial
is True,
"We always require partial, to ignore 'ccdnum'"
366 rawDataId3, calibDate = self.
_translator(dataId2, partial=partial)
368 DataCoordinate.standardize(rawDataId3, universe=self.
_datasetType.dimensions.universe),