lsst.pipe.drivers.ingestDriver.PoolIngestTask Class Reference

Inheritance diagram for lsst.pipe.drivers.ingestDriver.PoolIngestTask:

Public Member Functions
def	batchWallTime (cls, time, parsedCmd, numCores)
	Return walltime request for batch job. More...
def	parseAndRun (cls, *args, **kwargs)
def	runFileWrapper (self, struct, args)
def	run (self, args)
def	writeConfig (self, *args, **kwargs)
def	writeMetadata (self, *args, **kwargs)
def	parseAndSubmit (cls, args=None, **kwargs)
def	batchCommand (cls, args)
	Return command to run CmdLineTask. More...
def	logOperation (self, operation, catch=False, trace=True)
	Provide a context manager for logging an operation. More...
def	parseAndRun (cls)
def	prepareTask (cls, root=None, dryrun=False, mode="move", create=False, ignoreIngested=False)
def	ingest (self, infile, outfile, mode="move", dryrun=False)
def	isBadFile (self, filename, badFileList)
def	isBadId (self, info, badIdList)
def	expandFiles (self, fileNameList)
	Expand a set of filenames and globs, returning a list of filenames. More...
def	runFile (self, infile, registry, args, pos)
	Examine and ingest a single file. More...
def	ingestFiles (self, fileList)

Static Public Attributes
	ConfigClass = IngestConfig
	ArgumentParser = IngestArgumentParser

Detailed Description

Parallel version of IngestTask

Definition at line 7 of file ingestDriver.py.

Member Function Documentation

batchCommand()

def lsst.ctrl.pool.parallel.BatchCmdLineTask.batchCommand (   cls,   args  )

inherited

Return command to run CmdLineTask.

    @param cls: Class
    @param args: Parsed batch job arguments (from BatchArgumentParser)

Definition at line 476 of file parallel.py.

    def batchCommand(cls, args):
        """!Return command to run CmdLineTask
 
        @param cls: Class
        @param args: Parsed batch job arguments (from BatchArgumentParser)
        """
        job = args.job if args.job is not None else "job"
        module = cls.__module__
        script = ("import os; os.umask(%#05o); " +
                  "import lsst.base; lsst.base.disableImplicitThreading(); " +
                  "import lsst.ctrl.pool.log; lsst.ctrl.pool.log.jobLog(\"%s\"); ") % (UMASK, job)
 
        if args.batchStats:
            script += ("import lsst.ctrl.pool.parallel; import atexit; " +
                       "atexit.register(lsst.ctrl.pool.parallel.printProcessStats); ")
 
        script += "import %s; %s.%s.parseAndRun();" % (module, module, cls.__name__)
 
        profilePre = "import cProfile; import os; cProfile.run(\"\"\""
        profilePost = "\"\"\", filename=\"profile-" + job + "-%s-%d.dat\" % (os.uname()[1], os.getpid()))"
 
        return ("python -c '" + (profilePre if args.batchProfile else "") + script +
                (profilePost if args.batchProfile else "") + "' " + shCommandFromArgs(args.leftover) +
                " --noExit")
 

batchWallTime()

def lsst.pipe.drivers.ingestDriver.PoolIngestTask.batchWallTime	(		cls,
			time,
			parsedCmd,
			numCores
	)

Return walltime request for batch job.

    Subclasses should override if the walltime should be calculated
    differently (e.g., addition of some serial time).

    @param cls: Class
    @param time: Requested time per iteration
    @param parsedCmd: Results of argument parsing
    @param numCores: Number of cores

Reimplemented from lsst.ctrl.pool.parallel.BatchCmdLineTask.

Definition at line 10 of file ingestDriver.py.

    def batchWallTime(cls, time, parsedCmd, numCores):
        return float(time)*len(parsedCmd.files)/numCores
 

expandFiles()

def lsst.pipe.tasks.ingest.IngestTask.expandFiles (   self,   fileNameList  )

inherited

Expand a set of filenames and globs, returning a list of filenames.

    @param fileNameList A list of files and glob patterns

    N.b. globs obey Posix semantics, so a pattern that matches nothing is returned unchanged

Definition at line 557 of file ingest.py.

    def expandFiles(self, fileNameList):
        """!Expand a set of filenames and globs, returning a list of filenames
 
        @param fileNameList A list of files and glob patterns
 
        N.b. globs obey Posix semantics, so a pattern that matches nothing is returned unchanged
        """
        filenameList = []
        for globPattern in fileNameList:
            files = glob(globPattern)
 
            if not files:               # posix behaviour is to return pattern unchanged
                self.log.warning("%s doesn't match any file", globPattern)
                continue
 
            filenameList.extend(files)
 
        return filenameList
 

ingest()

def lsst.pipe.tasks.ingest.IngestTask.ingest (   self,   infile,   outfile,   mode = "move",   dryrun = False  )

inherited

Ingest a file into the image repository.

@param infile  Name of input file
@param outfile Name of output file (file in repository)
@param mode    Mode of ingest (copy/link/move/skip)
@param dryrun  Only report what would occur?
@param Success boolean

Definition at line 478 of file ingest.py.

    def ingest(self, infile, outfile, mode="move", dryrun=False):
        """Ingest a file into the image repository.
 
        @param infile  Name of input file
        @param outfile Name of output file (file in repository)
        @param mode    Mode of ingest (copy/link/move/skip)
        @param dryrun  Only report what would occur?
        @param Success boolean
        """
        if mode == "skip":
            return True
        if dryrun:
            self.log.info("Would %s from %s to %s", mode, infile, outfile)
            return True
        try:
            outdir = os.path.dirname(outfile)
            if not os.path.isdir(outdir):
                try:
                    os.makedirs(outdir)
                except OSError as exc:
                    # Silently ignore mkdir failures due to race conditions
                    if not os.path.isdir(outdir):
                        raise RuntimeError(f"Failed to create directory {outdir}") from exc
            if os.path.lexists(outfile):
                if self.config.clobber:
                    os.unlink(outfile)
                else:
                    raise RuntimeError("File %s already exists; consider --config clobber=True" % outfile)
 
            if mode == "copy":
                assertCanCopy(infile, outfile)
                shutil.copyfile(infile, outfile)
            elif mode == "link":
                if os.path.exists(outfile):
                    if os.path.samefile(infile, outfile):
                        self.log.debug("Already linked %s to %s: ignoring", infile, outfile)
                    else:
                        self.log.warning("%s already has a file at the target location (%s): ignoring "
                                         "(set clobber=True to overwrite)", infile, outfile)
                    return False
                os.symlink(os.path.abspath(infile), outfile)
            elif mode == "move":
                assertCanCopy(infile, outfile)
                shutil.move(infile, outfile)
            else:
                raise AssertionError("Unknown mode: %s" % mode)
            self.log.info("%s --<%s>--> %s", infile, mode, outfile)
        except Exception as e:
            self.log.warning("Failed to %s %s to %s: %s", mode, infile, outfile, e)
            if not self.config.allowError:
                raise RuntimeError(f"Failed to {mode} {infile} to {outfile}") from e
            return False
        return True
 

ingestFiles()

def lsst.pipe.tasks.ingest.IngestTask.ingestFiles (   self,   fileList  )

inherited

Ingest specified file or list of files and add them to the registry.

This method can only be called if `prepareTask` was used.

Parameters
----------
fileList : `str` or `list` [`str`]
    Pathname or list of pathnames of files to ingest.

Definition at line 632 of file ingest.py.

    def ingestFiles(self, fileList):
        """Ingest specified file or list of files and add them to the registry.
 
        This method can only be called if `prepareTask` was used.
 
        Parameters
        ----------
        fileList : `str` or `list` [`str`]
            Pathname or list of pathnames of files to ingest.
        """
        if not hasattr(self, "_args"):
            raise RuntimeError("Task not created with prepareTask")
        if isinstance(fileList, str):
            fileList = [fileList]
        self._args.files = fileList
        self.run(self._args)
 
 

isBadFile()

def lsst.pipe.tasks.ingest.IngestTask.isBadFile (   self,   filename,   badFileList  )

inherited

Return whether the file qualifies as bad

We match against the list of bad file patterns.

Definition at line 532 of file ingest.py.

    def isBadFile(self, filename, badFileList):
        """Return whether the file qualifies as bad
 
        We match against the list of bad file patterns.
        """
        filename = os.path.basename(filename)
        if not badFileList:
            return False
        for badFile in badFileList:
            if fnmatch(filename, badFile):
                return True
        return False
 

isBadId()

def lsst.pipe.tasks.ingest.IngestTask.isBadId (   self,   info,   badIdList  )

inherited

Return whether the file information qualifies as bad

We match against the list of bad data identifiers.

Definition at line 545 of file ingest.py.

    def isBadId(self, info, badIdList):
        """Return whether the file information qualifies as bad
 
        We match against the list of bad data identifiers.
        """
        if not badIdList:
            return False
        for badId in badIdList:
            if all(info[key] == value for key, value in badId.items()):
                return True
        return False
 

logOperation()

def lsst.ctrl.pool.parallel.BatchCmdLineTask.logOperation (   self,   operation,   catch = False,   trace = True  )

inherited

Provide a context manager for logging an operation.

    @param operation: description of operation (string)
    @param catch: Catch all exceptions?
    @param trace: Log a traceback of caught exception?

    Note that if 'catch' is True, all exceptions are swallowed, but there may
    be other side-effects such as undefined variables.

Definition at line 502 of file parallel.py.

    def logOperation(self, operation, catch=False, trace=True):
        """!Provide a context manager for logging an operation
 
        @param operation: description of operation (string)
        @param catch: Catch all exceptions?
        @param trace: Log a traceback of caught exception?
 
        Note that if 'catch' is True, all exceptions are swallowed, but there may
        be other side-effects such as undefined variables.
        """
        self.log.info("%s: Start %s" % (NODE, operation))
        try:
            yield
        except Exception:
            if catch:
                cls, e, _ = sys.exc_info()
                self.log.warn("%s: Caught %s while %s: %s" % (NODE, cls.__name__, operation, e))
                if trace:
                    self.log.info("%s: Traceback:\n%s" % (NODE, traceback.format_exc()))
                return
            raise
        finally:
            self.log.info("%s: Finished %s" % (NODE, operation))
 
 

parseAndRun() [1/2]

def lsst.pipe.tasks.ingest.IngestTask.parseAndRun (   cls )

inherited

Parse the command-line arguments and run the Task.

Definition at line 428 of file ingest.py.

    def parseAndRun(cls):
        """Parse the command-line arguments and run the Task."""
        task, args = cls._parse()
        task.run(args)
 

parseAndRun() [2/2]

def lsst.pipe.drivers.ingestDriver.PoolIngestTask.parseAndRun	(		cls,
		*	args,
		**	kwargs
	)

Run with a MPI process pool

Definition at line 24 of file ingestDriver.py.

    def parseAndRun(cls, *args, **kwargs):
        """Run with a MPI process pool"""
        pool = startPool()
        config = cls.ConfigClass()
        parser = cls.ArgumentParser(name=cls._DefaultName)
        args = parser.parse_args(config)
        task = cls(config=args.config)
        task.run(args)
        pool.exit()
 

parseAndSubmit()

def lsst.ctrl.pool.parallel.BatchCmdLineTask.parseAndSubmit (   cls,   args = None, **  kwargs  )

inherited

Definition at line 435 of file parallel.py.

    def parseAndSubmit(cls, args=None, **kwargs):
        taskParser = cls._makeArgumentParser(doBatch=True, add_help=False)
        batchParser = BatchArgumentParser(parent=taskParser)
        batchArgs = batchParser.parse_args(config=cls.ConfigClass(), args=args, override=cls.applyOverrides,
                                           **kwargs)
 
        if not cls.RunnerClass(cls, batchArgs.parent).precall(batchArgs.parent):  # Write config, schema
            taskParser.error("Error in task preparation")
 
        setBatchType(batchArgs.batch)
 
        if batchArgs.batch is None:     # don't use a batch system
            sys.argv = [sys.argv[0]] + batchArgs.leftover  # Remove all batch arguments
 
            return cls.parseAndRun()
        else:
            if batchArgs.walltime > 0:
                walltime = batchArgs.walltime
            else:
                numCores = batchArgs.cores if batchArgs.cores > 0 else batchArgs.nodes*batchArgs.procs
                walltime = cls.batchWallTime(batchArgs.time, batchArgs.parent, numCores)
 
            command = cls.batchCommand(batchArgs)
            batchArgs.batch.run(command, walltime=walltime)
 

prepareTask()

def lsst.pipe.tasks.ingest.IngestTask.prepareTask (   cls,   root = None,   dryrun = False,   mode = "move",   create = False,   ignoreIngested = False  )

inherited

Prepare for running the task repeatedly with `ingestFiles`.

Saves the parsed arguments, including the Butler and log, as a
private instance variable.

Parameters
----------
root : `str`, optional
    Repository root pathname.  If None, run the Task using the
    command line arguments, ignoring all other arguments below.
dryrun : `bool`, optional
    If True, don't perform any action; log what would have happened.
mode : `str`, optional
    How files are delivered to their destination.  Default is "move",
    unlike the command-line default of "link".
create : `bool`, optional
    If True, create a new registry, clobbering any old one present.
ignoreIngested : `bool`, optional
    If True, do not complain if the file is already present in the
    registry (and do nothing else).

Returns
-------
task : `IngestTask`
    If `root` was provided, the IngestTask instance

Definition at line 434 of file ingest.py.

                    ignoreIngested=False):
        """Prepare for running the task repeatedly with `ingestFiles`.
 
        Saves the parsed arguments, including the Butler and log, as a
        private instance variable.
 
        Parameters
        ----------
        root : `str`, optional
            Repository root pathname.  If None, run the Task using the
            command line arguments, ignoring all other arguments below.
        dryrun : `bool`, optional
            If True, don't perform any action; log what would have happened.
        mode : `str`, optional
            How files are delivered to their destination.  Default is "move",
            unlike the command-line default of "link".
        create : `bool`, optional
            If True, create a new registry, clobbering any old one present.
        ignoreIngested : `bool`, optional
            If True, do not complain if the file is already present in the
            registry (and do nothing else).
 
        Returns
        -------
        task : `IngestTask`
            If `root` was provided, the IngestTask instance
        """
        sys.argv = ["IngestTask"]
        sys.argv.append(root)
        if dryrun:
            sys.argv.append("--dry-run")
        sys.argv.append("--mode")
        sys.argv.append(mode)
        if create:
            sys.argv.append("--create")
        if ignoreIngested:
            sys.argv.append("--ignore-ingested")
        sys.argv.append("__fakefile__")    # needed for parsing, not used
 
        task, args = cls._parse()
        task._args = args
        return task
 

run()

def lsst.pipe.drivers.ingestDriver.PoolIngestTask.run	(		self,
			args
	)

Run ingest

We read and ingest the files in parallel, and then
stuff the registry database in serial.

Reimplemented from lsst.pipe.tasks.ingest.IngestTask.

Definition at line 60 of file ingestDriver.py.

    def run(self, args):
        """Run ingest
 
        We read and ingest the files in parallel, and then
        stuff the registry database in serial.
        """
        # Parallel
        pool = Pool(None)
        filenameList = self.expandFiles(args.files)
        dataList = [Struct(filename=filename, position=ii) for ii, filename in enumerate(filenameList)]
        infoList = pool.map(self.runFileWrapper, dataList, args)
 
        # Serial
        root = args.input
        context = self.register.openRegistry(root, create=args.create, dryrun=args.dryrun)
        with context as registry:
            for hduInfoList in infoList:
                if hduInfoList is None:
                    continue
                for info in hduInfoList:
                    self.register.addRow(registry, info, dryrun=args.dryrun, create=args.create)
 

runFile()

def lsst.pipe.tasks.ingest.IngestTask.runFile (   self,   infile,   registry,   args,   pos  )

inherited

Examine and ingest a single file.

    @param infile: File to process
    @param registry: Registry into which to insert Butler metadata, or None
    @param args: Parsed command-line arguments
    @param pos: Position number of this file in the input list
    @return parsed information from FITS HDUs if registry is None; or None

Definition at line 576 of file ingest.py.

    def runFile(self, infile, registry, args, pos):
        """!Examine and ingest a single file
 
        @param infile: File to process
        @param registry: Registry into which to insert Butler metadata, or None
        @param args: Parsed command-line arguments
        @param pos: Position number of this file in the input list
        @return parsed information from FITS HDUs if registry is None; or None
        """
        if self.isBadFile(infile, args.badFile):
            self.log.info("Skipping declared bad file %s", infile)
            return None
        try:
            fileInfo, hduInfoList = self.parse.getInfo(infile)
        except Exception as e:
            if not self.config.allowError:
                raise RuntimeError(f"Error parsing {infile}") from e
            self.log.warning("Error parsing %s (%s); skipping", infile, e)
            return None
        if self.isBadId(fileInfo, args.badId.idList):
            self.log.info("Skipping declared bad file %s: %s", infile, fileInfo)
            return None
        if registry is not None and self.register.check(registry, fileInfo):
            if args.ignoreIngested:
                return None
            self.log.warning("%s: already ingested: %s", infile, fileInfo)
        outfile = self.parse.getDestination(args.butler, fileInfo, infile)
        if not self.ingest(infile, outfile, mode=args.mode, dryrun=args.dryrun):
            return None
        if hduInfoList is None:
            return None
        if registry is None:
            return hduInfoList
        for info in hduInfoList:
            try:
                self.register.addRow(registry, info, dryrun=args.dryrun, create=args.create)
            except Exception as exc:
                raise IngestError(f"Failed to register file {infile}", infile, pos) from exc
        return None  # No further registration should be performed
 

runFileWrapper()

def lsst.pipe.drivers.ingestDriver.PoolIngestTask.runFileWrapper	(		self,
			struct,
			args
	)

Run ingest on one file

This is a wrapper method for calling ``runFile``.

Parameters
----------
struct : `lsst.pipe.base.Struct`
    Structure containing ``filename`` (`str`) and ``position`` (`int`).
args : `argparse.Namespace`
    Parsed command-line arguments.

Returns
-------
hduInfoList : `list` of `dict`
    Parsed information from FITS HDUs, or ``None``.

Definition at line 34 of file ingestDriver.py.

    def runFileWrapper(self, struct, args):
        """Run ingest on one file
 
        This is a wrapper method for calling ``runFile``.
 
        Parameters
        ----------
        struct : `lsst.pipe.base.Struct`
            Structure containing ``filename`` (`str`) and ``position`` (`int`).
        args : `argparse.Namespace`
            Parsed command-line arguments.
 
        Returns
        -------
        hduInfoList : `list` of `dict`
            Parsed information from FITS HDUs, or ``None``.
        """
        filename = struct.filename
        position = struct.position
        try:
            return self.runFile(filename, None, args, position)
        except IngestError as exc:
            self.log.warn(f"Unable to ingest {filename}: {exc}")
            return None
 

writeConfig()

def lsst.pipe.drivers.ingestDriver.PoolIngestTask.writeConfig	(		self,
		*	args,
		**	kwargs
	)

Definition at line 82 of file ingestDriver.py.

    def writeConfig(self, *args, **kwargs):
        pass
 

writeMetadata()

def lsst.pipe.drivers.ingestDriver.PoolIngestTask.writeMetadata	(		self,
		*	args,
		**	kwargs
	)

Definition at line 85 of file ingestDriver.py.

    def writeMetadata(self, *args, **kwargs):
        pass

Member Data Documentation

ArgumentParser

lsst.pipe.tasks.ingest.IngestTask.ArgumentParser = IngestArgumentParser

staticinherited

Definition at line 409 of file ingest.py.

ConfigClass

lsst.pipe.tasks.ingest.IngestTask.ConfigClass = IngestConfig

staticinherited

Definition at line 408 of file ingest.py.

---

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

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-0.7.0/Linux64/pipe_drivers/22.0.0+a28f4c53b1/python/lsst/pipe/drivers/ingestDriver.py