lsst.obs.base.gen2to3.repoWalker.scanner.DirectoryScanner Class Reference

Public Member Functions
def	__init__ (self, Optional[Log] log=None)
def	add (self, PathElementHandler handler)
Iterator[PathElementHandler]	__iter__ (self)
def	scan (self, str path, Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]] datasets, *Callable[[DataCoordinate], bool] predicate)

Public Attributes
	log

Detailed Description

An object that uses `PathElementHandler` instances to process the files
and subdirectories in a directory tree.

Parameters
----------
log : `Log`, optional
    Log to use to report warnings and debug information.

Definition at line 162 of file scanner.py.

Constructor & Destructor Documentation

__init__()

def lsst.obs.base.gen2to3.repoWalker.scanner.DirectoryScanner.__init__	(		self,
		Optional[Log]	log = None
	)

Definition at line 171 of file scanner.py.

    def __init__(self, log: Optional[Log] = None):
        self._files = []
        self._subdirectories = []
        if log is None:
            log = Log.getLogger("obs.base.gen2to3.walker")
        self.log = log
 

Member Function Documentation

__iter__()

Iterator[PathElementHandler] lsst.obs.base.gen2to3.repoWalker.scanner.DirectoryScanner.__iter__

(

self

)

Iterate over all handlers.

Definition at line 194 of file scanner.py.

    def __iter__(self) -> Iterator[PathElementHandler]:
        """Iterate over all handlers.
        """
        yield from self._files
        yield from self._subdirectories
 

add()

def lsst.obs.base.gen2to3.repoWalker.scanner.DirectoryScanner.add	(		self,
		PathElementHandler	handler
	)

Add a new handler to the scanner.

Parameters
----------
handler : `PathElementHandler`
    The handler to be added.

Definition at line 180 of file scanner.py.

    def add(self, handler: PathElementHandler):
        """Add a new handler to the scanner.
 
        Parameters
        ----------
        handler : `PathElementHandler`
            The handler to be added.
        """
        handler.log = self.log
        if handler.isForFiles():
            bisect.insort(self._files, handler)
        else:
            bisect.insort(self._subdirectories, handler)
 

scan()

def lsst.obs.base.gen2to3.repoWalker.scanner.DirectoryScanner.scan	(		self,
		str	path,
		Mapping[DatasetType, Mapping[Optional[str], List[FileDataset]]]	datasets,
		*Callable[[DataCoordinate], bool]	predicate
	)

Process a directory.

Parameters
----------
path : `str`
    Full path to the directory to be processed.
datasets : `dict` [`DatasetType`, `list` ]
    Dictionary that found datasets should be added to.  Nested lists
    elements are tuples of `FileDataset` and an optional "CALIBDATE"
    `str` value (for calibration datasets only).
predicate : `~collections.abc.Callable`
    A callable taking a single `DataCoordinate` argument and returning
    `bool`, indicating whether that (Gen3) data ID represents one
    that should be included in the scan.

Definition at line 200 of file scanner.py.

             predicate: Callable[[DataCoordinate], bool]):
        """Process a directory.
 
        Parameters
        ----------
        path : `str`
            Full path to the directory to be processed.
        datasets : `dict` [`DatasetType`, `list` ]
            Dictionary that found datasets should be added to.  Nested lists
            elements are tuples of `FileDataset` and an optional "CALIBDATE"
            `str` value (for calibration datasets only).
        predicate : `~collections.abc.Callable`
            A callable taking a single `DataCoordinate` argument and returning
            `bool`, indicating whether that (Gen3) data ID represents one
            that should be included in the scan.
        """
        unrecognized = []
        for entry in os.scandir(path):
            if entry.is_file():
                handlers = self._files
            elif entry.is_dir():
                handlers = self._subdirectories
            else:
                continue
            for handler in handlers:
                if handler(entry.path, entry.name, datasets, predicate=predicate):
                    break
            else:
                unrecognized.append(entry.name)
        if unrecognized:
            self.log.warn("Skipped unrecognized entries in %s: %s", path, unrecognized)

Member Data Documentation

log

lsst.obs.base.gen2to3.repoWalker.scanner.DirectoryScanner.log

Definition at line 176 of file scanner.py.

---

The documentation for this class was generated from the following file:

• /j/snowflake/release/lsstsw/stack/0.2.1/Linux64/obs_base/21.0.0-27-gbbd0d29+ae871e0f33/python/lsst/obs/base/gen2to3/repoWalker/scanner.py