lsst.pipe.base.cmdLineTask.TaskRunner Class Reference

Inheritance diagram for lsst.pipe.base.cmdLineTask.TaskRunner:

Public Member Functions
def	__init__ (self, TaskClass, parsedCmd, doReturnResults=False)
def	prepareForMultiProcessing (self)
def	run (self, parsedCmd)
def	makeTask (self, parsedCmd=None, args=None)
def	precall (self, parsedCmd)
def	__call__ (self, args)
def	runTask (self, task, dataRef, kwargs)

Static Public Member Functions
def	getTargetList (parsedCmd, **kwargs)

Public Attributes
	TaskClass
	doReturnResults
	config
	log
	doRaise
	clobberConfig
	doBackup
	numProcesses
	timeout

Static Public Attributes
int	TIMEOUT = 3600*24*30

Detailed Description

Run a command-line task, using `multiprocessing` if requested.

Parameters
----------
TaskClass : `lsst.pipe.base.Task` subclass
    The class of the task to run.
parsedCmd : `argparse.Namespace`
    The parsed command-line arguments, as returned by the task's argument
    parser's `~lsst.pipe.base.ArgumentParser.parse_args` method.

    .. warning::

       Do not store ``parsedCmd``, as this instance is pickled (if
       multiprocessing) and parsedCmd may contain non-picklable elements.
       It certainly contains more data than we need to send to each
       instance of the task.
doReturnResults : `bool`, optional
    Should run return the collected result from each invocation of the
    task? This is only intended for unit tests and similar use. It can
    easily exhaust memory (if the task returns enough data and you call it
    enough times) and it will fail when using multiprocessing if the
    returned data cannot be pickled.

    Note that even if ``doReturnResults`` is False a struct with a single
    member "exitStatus" is returned, with value 0 or 1 to be returned to
    the unix shell.

Raises
------
ImportError
    Raised if multiprocessing is requested (and the task supports it) but
    the multiprocessing library cannot be imported.

Notes
-----
Each command-line task (subclass of `lsst.pipe.base.CmdLineTask`) has a
task runner. By default it is this class, but some tasks require a
subclass. See the manual :ref:`creating-a-command-line-task` for more
information. See `CmdLineTask.parseAndRun` to see how a task runner is
used.

You may use this task runner for your command-line task if your task has a
``runDataRef`` method that takes exactly one argument: a butler data
reference. Otherwise you must provide a task-specific subclass of
this runner for your task's ``RunnerClass`` that overrides
`TaskRunner.getTargetList` and possibly
`TaskRunner.__call__`. See `TaskRunner.getTargetList` for details.

This design matches the common pattern for command-line tasks: the
``runDataRef`` method takes a single data reference, of some suitable name.
Additional arguments are rare, and if present, require a subclass of
`TaskRunner` that calls these additional arguments by name.

Instances of this class must be picklable in order to be compatible with
multiprocessing. If multiprocessing is requested
(``parsedCmd.numProcesses > 1``) then `runDataRef` calls
`prepareForMultiProcessing` to jettison optional non-picklable elements.
If your task runner is not compatible with multiprocessing then indicate
this in your task by setting class variable ``canMultiprocess=False``.

Due to a `python bug`__, handling a `KeyboardInterrupt` properly `requires
specifying a timeout`__. This timeout (in sec) can be specified as the
``timeout`` element in the output from `~lsst.pipe.base.ArgumentParser`
(the ``parsedCmd``), if available, otherwise we use `TaskRunner.TIMEOUT`.

By default, we disable "implicit" threading -- ie, as provided by
underlying numerical libraries such as MKL or BLAS. This is designed to
avoid thread contention both when a single command line task spawns
multiple processes and when multiple users are running on a shared system.
Users can override this behaviour by setting the
``LSST_ALLOW_IMPLICIT_THREADS`` environment variable.

.. __: http://bugs.python.org/issue8296
.. __: http://stackoverflow.com/questions/1408356/

Definition at line 94 of file cmdLineTask.py.

Constructor & Destructor Documentation

__init__()

def lsst.pipe.base.cmdLineTask.TaskRunner.__init__	(		self,
			TaskClass,
			parsedCmd,
			doReturnResults = False
	)

Reimplemented in lsst.pipe.drivers.multiBandDriver.MultiBandDriverTaskRunner.

Definition at line 174 of file cmdLineTask.py.

    def __init__(self, TaskClass, parsedCmd, doReturnResults=False):
        self.TaskClass = TaskClass
        self.doReturnResults = bool(doReturnResults)
        self.config = parsedCmd.config
        self.log = parsedCmd.log
        self.doRaise = bool(parsedCmd.doraise)
        self.clobberConfig = bool(parsedCmd.clobberConfig)
        self.doBackup = not bool(parsedCmd.noBackupConfig)
        self.numProcesses = int(getattr(parsedCmd, 'processes', 1))
 
        self.timeout = getattr(parsedCmd, 'timeout', None)
        if self.timeout is None or self.timeout <= 0:
            self.timeout = self.TIMEOUT
 
        if self.numProcesses > 1:
            if not TaskClass.canMultiprocess:
                self.log.warn("This task does not support multiprocessing; using one process")
                self.numProcesses = 1
 

Member Function Documentation

__call__()

def lsst.pipe.base.cmdLineTask.TaskRunner.__call__	(		self,
			args
	)

Run the Task on a single target.

Parameters
----------
args
    Arguments for Task.runDataRef()

Returns
-------
struct : `lsst.pipe.base.Struct`
    Contains these fields if ``doReturnResults`` is `True`:

    - ``dataRef``: the provided data reference.
    - ``metadata``: task metadata after execution of run.
    - ``result``: result returned by task run, or `None` if the task
      fails.
    - ``exitStatus``: 0 if the task completed successfully, 1
      otherwise.

    If ``doReturnResults`` is `False` the struct contains:

    - ``exitStatus``: 0 if the task completed successfully, 1
      otherwise.

Notes
-----
This default implementation assumes that the ``args`` is a tuple
containing a data reference and a dict of keyword arguments.

.. warning::

   If you override this method and wish to return something when
   ``doReturnResults`` is `False`, then it must be picklable to
   support multiprocessing and it should be small enough that pickling
   and unpickling do not add excessive overhead.

Reimplemented in lsst.pipe.drivers.constructCalibs.CalibTaskRunner.

Definition at line 380 of file cmdLineTask.py.

    def __call__(self, args):
        """Run the Task on a single target.
 
        Parameters
        ----------
        args
            Arguments for Task.runDataRef()
 
        Returns
        -------
        struct : `lsst.pipe.base.Struct`
            Contains these fields if ``doReturnResults`` is `True`:
 
            - ``dataRef``: the provided data reference.
            - ``metadata``: task metadata after execution of run.
            - ``result``: result returned by task run, or `None` if the task
              fails.
            - ``exitStatus``: 0 if the task completed successfully, 1
              otherwise.
 
            If ``doReturnResults`` is `False` the struct contains:
 
            - ``exitStatus``: 0 if the task completed successfully, 1
              otherwise.
 
        Notes
        -----
        This default implementation assumes that the ``args`` is a tuple
        containing a data reference and a dict of keyword arguments.
 
        .. warning::
 
           If you override this method and wish to return something when
           ``doReturnResults`` is `False`, then it must be picklable to
           support multiprocessing and it should be small enough that pickling
           and unpickling do not add excessive overhead.
        """
        dataRef, kwargs = args
        if self.log is None:
            self.log = Log.getDefaultLogger()
        if hasattr(dataRef, "dataId"):
            self.log.MDC("LABEL", str(dataRef.dataId))
        elif isinstance(dataRef, (list, tuple)):
            self.log.MDC("LABEL", str([ref.dataId for ref in dataRef if hasattr(ref, "dataId")]))
        task = self.makeTask(args=args)
        result = None                   # in case the task fails
        exitStatus = 0                  # exit status for the shell
        if self.doRaise:
            result = self.runTask(task, dataRef, kwargs)
        else:
            try:
                result = self.runTask(task, dataRef, kwargs)
            except Exception as e:
                # The shell exit value will be the number of dataRefs returning
                # non-zero, so the actual value used here is lost.
                exitStatus = 1
 
                # don't use a try block as we need to preserve the original
                # exception
                eName = type(e).__name__
                if hasattr(dataRef, "dataId"):
                    task.log.fatal("Failed on dataId=%s: %s: %s", dataRef.dataId, eName, e)
                elif isinstance(dataRef, (list, tuple)):
                    task.log.fatal("Failed on dataIds=[%s]: %s: %s",
                                   ", ".join(str(ref.dataId) for ref in dataRef), eName, e)
                else:
                    task.log.fatal("Failed on dataRef=%s: %s: %s", dataRef, eName, e)
 
                if not isinstance(e, TaskError):
                    traceback.print_exc(file=sys.stderr)
 
        # Ensure all errors have been logged and aren't hanging around in a
        # buffer
        sys.stdout.flush()
        sys.stderr.flush()
 
        task.writeMetadata(dataRef)
 
        # remove MDC so it does not show up outside of task context
        self.log.MDCRemove("LABEL")
 
        if self.doReturnResults:
            return Struct(
                exitStatus=exitStatus,
                dataRef=dataRef,
                metadata=task.metadata,
                result=result,
            )
        else:
            return Struct(
                exitStatus=exitStatus,
            )
 

getTargetList()

def lsst.pipe.base.cmdLineTask.TaskRunner.getTargetList (   parsedCmd, **  kwargs  )

static

Get a list of (dataRef, kwargs) for `TaskRunner.__call__`.

Parameters
----------
parsedCmd : `argparse.Namespace`
    The parsed command object returned by
    `lsst.pipe.base.argumentParser.ArgumentParser.parse_args`.
kwargs
    Any additional keyword arguments. In the default `TaskRunner` this
    is an empty dict, but having it simplifies overriding `TaskRunner`
    for tasks whose runDataRef method takes additional arguments
    (see case (1) below).

Notes
-----
The default implementation of `TaskRunner.getTargetList` and
`TaskRunner.__call__` works for any command-line task whose
``runDataRef`` method takes exactly one argument: a data reference.
Otherwise you must provide a variant of TaskRunner that overrides
`TaskRunner.getTargetList` and possibly `TaskRunner.__call__`.
There are two cases.

**Case 1**

If your command-line task has a ``runDataRef`` method that takes one
data reference followed by additional arguments, then you need only
override `TaskRunner.getTargetList` to return the additional
arguments as an argument dict. To make this easier, your overridden
version of `~TaskRunner.getTargetList` may call
`TaskRunner.getTargetList` with the extra arguments as keyword
arguments. For example, the following adds an argument dict containing
a single key: "calExpList", whose value is the list of data IDs for
the calexp ID argument:

.. code-block:: python

    def getTargetList(parsedCmd):
return TaskRunner.getTargetList(
    parsedCmd,
    calExpList=parsedCmd.calexp.idList
)

It is equivalent to this slightly longer version:

.. code-block:: python

    @staticmethod
    def getTargetList(parsedCmd):
argDict = dict(calExpList=parsedCmd.calexp.idList)
return [(dataId, argDict) for dataId in parsedCmd.id.idList]

**Case 2**

If your task does not meet condition (1) then you must override both
TaskRunner.getTargetList and `TaskRunner.__call__`. You may do this
however you see fit, so long as `TaskRunner.getTargetList`
returns a list, each of whose elements is sent to
`TaskRunner.__call__`, which runs your task.

Reimplemented in lsst.pipe.drivers.constructCalibs.CalibTaskRunner, lsst.meas.base.forcedPhotCoadd.ForcedPhotCoaddRunner, lsst.pipe.tasks.multiBandUtils.MergeSourcesRunner, and lsst.pipe.drivers.utils.ButlerTaskRunner.

Definition at line 253 of file cmdLineTask.py.

    def getTargetList(parsedCmd, **kwargs):
        """Get a list of (dataRef, kwargs) for `TaskRunner.__call__`.
 
        Parameters
        ----------
        parsedCmd : `argparse.Namespace`
            The parsed command object returned by
            `lsst.pipe.base.argumentParser.ArgumentParser.parse_args`.
        kwargs
            Any additional keyword arguments. In the default `TaskRunner` this
            is an empty dict, but having it simplifies overriding `TaskRunner`
            for tasks whose runDataRef method takes additional arguments
            (see case (1) below).
 
        Notes
        -----
        The default implementation of `TaskRunner.getTargetList` and
        `TaskRunner.__call__` works for any command-line task whose
        ``runDataRef`` method takes exactly one argument: a data reference.
        Otherwise you must provide a variant of TaskRunner that overrides
        `TaskRunner.getTargetList` and possibly `TaskRunner.__call__`.
        There are two cases.
 
        **Case 1**
 
        If your command-line task has a ``runDataRef`` method that takes one
        data reference followed by additional arguments, then you need only
        override `TaskRunner.getTargetList` to return the additional
        arguments as an argument dict. To make this easier, your overridden
        version of `~TaskRunner.getTargetList` may call
        `TaskRunner.getTargetList` with the extra arguments as keyword
        arguments. For example, the following adds an argument dict containing
        a single key: "calExpList", whose value is the list of data IDs for
        the calexp ID argument:
 
        .. code-block:: python
 
            def getTargetList(parsedCmd):
                return TaskRunner.getTargetList(
                    parsedCmd,
                    calExpList=parsedCmd.calexp.idList
                )
 
        It is equivalent to this slightly longer version:
 
        .. code-block:: python
 
            @staticmethod
            def getTargetList(parsedCmd):
                argDict = dict(calExpList=parsedCmd.calexp.idList)
                return [(dataId, argDict) for dataId in parsedCmd.id.idList]
 
        **Case 2**
 
        If your task does not meet condition (1) then you must override both
        TaskRunner.getTargetList and `TaskRunner.__call__`. You may do this
        however you see fit, so long as `TaskRunner.getTargetList`
        returns a list, each of whose elements is sent to
        `TaskRunner.__call__`, which runs your task.
        """
        return [(ref, kwargs) for ref in parsedCmd.id.refList]
 

makeTask()

def lsst.pipe.base.cmdLineTask.TaskRunner.makeTask	(		self,
			parsedCmd = None,
			args = None
	)

Create a Task instance.

Parameters
----------
parsedCmd
    Parsed command-line options (used for extra task args by some task
    runners).
args
    Args tuple passed to `TaskRunner.__call__` (used for extra task
    arguments by some task runners).

Notes
-----
``makeTask`` can be called with either the ``parsedCmd`` argument or
``args`` argument set to None, but it must construct identical Task
instances in either case.

Subclasses may ignore this method entirely if they reimplement both
`TaskRunner.precall` and `TaskRunner.__call__`.

Reimplemented in lsst.pipe.base.cmdLineTask.ButlerInitializedTaskRunner, lsst.pipe.drivers.multiBandDriver.MultiBandDriverTaskRunner, and lsst.pipe.tasks.multiBandUtils.MergeSourcesRunner.

Definition at line 315 of file cmdLineTask.py.

    def makeTask(self, parsedCmd=None, args=None):
        """Create a Task instance.
 
        Parameters
        ----------
        parsedCmd
            Parsed command-line options (used for extra task args by some task
            runners).
        args
            Args tuple passed to `TaskRunner.__call__` (used for extra task
            arguments by some task runners).
 
        Notes
        -----
        ``makeTask`` can be called with either the ``parsedCmd`` argument or
        ``args`` argument set to None, but it must construct identical Task
        instances in either case.
 
        Subclasses may ignore this method entirely if they reimplement both
        `TaskRunner.precall` and `TaskRunner.__call__`.
        """
        return self.TaskClass(config=self.config, log=self.log)
 

precall()

def lsst.pipe.base.cmdLineTask.TaskRunner.precall	(		self,
			parsedCmd
	)

Hook for code that should run exactly once, before multiprocessing.

Notes
-----
Must return True if `TaskRunner.__call__` should subsequently be
called.

.. warning::

   Implementations must take care to ensure that no unpicklable
   attributes are added to the TaskRunner itself, for compatibility
   with multiprocessing.

The default implementation writes package versions, schemas and
configs, or compares them to existing files on disk if present.

Definition at line 349 of file cmdLineTask.py.

    def precall(self, parsedCmd):
        """Hook for code that should run exactly once, before multiprocessing.
 
        Notes
        -----
        Must return True if `TaskRunner.__call__` should subsequently be
        called.
 
        .. warning::
 
           Implementations must take care to ensure that no unpicklable
           attributes are added to the TaskRunner itself, for compatibility
           with multiprocessing.
 
        The default implementation writes package versions, schemas and
        configs, or compares them to existing files on disk if present.
        """
        task = self.makeTask(parsedCmd=parsedCmd)
 
        if self.doRaise:
            self._precallImpl(task, parsedCmd)
        else:
            try:
                self._precallImpl(task, parsedCmd)
            except Exception as e:
                task.log.fatal("Failed in task initialization: %s", e)
                if not isinstance(e, TaskError):
                    traceback.print_exc(file=sys.stderr)
                return False
        return True
 

prepareForMultiProcessing()

def lsst.pipe.base.cmdLineTask.TaskRunner.prepareForMultiProcessing

(

self

)

Prepare this instance for multiprocessing

Optional non-picklable elements are removed.

This is only called if the task is run under multiprocessing.

Definition at line 193 of file cmdLineTask.py.

    def prepareForMultiProcessing(self):
        """Prepare this instance for multiprocessing
 
        Optional non-picklable elements are removed.
 
        This is only called if the task is run under multiprocessing.
        """
        self.log = None
 

run()

def lsst.pipe.base.cmdLineTask.TaskRunner.run	(		self,
			parsedCmd
	)

Run the task on all targets.

Parameters
----------
parsedCmd : `argparse.Namespace`
    Parsed command `argparse.Namespace`.

Returns
-------
resultList : `list`
    A list of results returned by `TaskRunner.__call__`, or an empty
    list if `TaskRunner.__call__` is not called (e.g. if
    `TaskRunner.precall` returns `False`). See `TaskRunner.__call__`
    for details.

Notes
-----
The task is run under multiprocessing if `TaskRunner.numProcesses`
is more than 1; otherwise processing is serial.

Reimplemented in lsst.ctrl.pool.parallel.BatchTaskRunner.

Definition at line 202 of file cmdLineTask.py.

    def run(self, parsedCmd):
        """Run the task on all targets.
 
        Parameters
        ----------
        parsedCmd : `argparse.Namespace`
            Parsed command `argparse.Namespace`.
 
        Returns
        -------
        resultList : `list`
            A list of results returned by `TaskRunner.__call__`, or an empty
            list if `TaskRunner.__call__` is not called (e.g. if
            `TaskRunner.precall` returns `False`). See `TaskRunner.__call__`
            for details.
 
        Notes
        -----
        The task is run under multiprocessing if `TaskRunner.numProcesses`
        is more than 1; otherwise processing is serial.
        """
        resultList = []
        disableImplicitThreading()  # To prevent thread contention
        if self.numProcesses > 1:
            import multiprocessing
            self.prepareForMultiProcessing()
            pool = multiprocessing.Pool(processes=self.numProcesses, maxtasksperchild=1)
            mapFunc = functools.partial(_runPool, pool, self.timeout)
        else:
            pool = None
            mapFunc = map
 
        if self.precall(parsedCmd):
            profileName = parsedCmd.profile if hasattr(parsedCmd, "profile") else None
            log = parsedCmd.log
            targetList = self.getTargetList(parsedCmd)
            if len(targetList) > 0:
                with profile(profileName, log):
                    # Run the task using self.__call__
                    resultList = list(mapFunc(self, targetList))
            else:
                log.warn("Not running the task because there is no data to process; "
                         "you may preview data using \"--show data\"")
 
        if pool is not None:
            pool.close()
            pool.join()
 
        return resultList
 

runTask()

def lsst.pipe.base.cmdLineTask.TaskRunner.runTask	(		self,
			task,
			dataRef,
			kwargs
	)

Make the actual call to `runDataRef` for this task.

Parameters
----------
task : `lsst.pipe.base.CmdLineTask` class
    The class of the task to run.
dataRef
    Butler data reference that contains the data the task will process.
kwargs
    Any additional keyword arguments.  See `TaskRunner.getTargetList`
    above.

Notes
-----
The default implementation of `TaskRunner.runTask` works for any
command-line task which has a ``runDataRef`` method that takes a data
reference and an optional set of additional keyword arguments.
This method returns the results generated by the task's `runDataRef`
method.

Reimplemented in lsst.pipe.base.cmdLineTask.LegacyTaskRunner.

Definition at line 473 of file cmdLineTask.py.

    def runTask(self, task, dataRef, kwargs):
        """Make the actual call to `runDataRef` for this task.
 
        Parameters
        ----------
        task : `lsst.pipe.base.CmdLineTask` class
            The class of the task to run.
        dataRef
            Butler data reference that contains the data the task will process.
        kwargs
            Any additional keyword arguments.  See `TaskRunner.getTargetList`
            above.
 
        Notes
        -----
        The default implementation of `TaskRunner.runTask` works for any
        command-line task which has a ``runDataRef`` method that takes a data
        reference and an optional set of additional keyword arguments.
        This method returns the results generated by the task's `runDataRef`
        method.
 
        """
        return task.runDataRef(dataRef, **kwargs)
 
 

Member Data Documentation

clobberConfig

lsst.pipe.base.cmdLineTask.TaskRunner.clobberConfig

Definition at line 180 of file cmdLineTask.py.

config

lsst.pipe.base.cmdLineTask.TaskRunner.config

Definition at line 177 of file cmdLineTask.py.

doBackup

lsst.pipe.base.cmdLineTask.TaskRunner.doBackup

Definition at line 181 of file cmdLineTask.py.

doRaise

lsst.pipe.base.cmdLineTask.TaskRunner.doRaise

Definition at line 179 of file cmdLineTask.py.

doReturnResults

lsst.pipe.base.cmdLineTask.TaskRunner.doReturnResults

Definition at line 176 of file cmdLineTask.py.

log

lsst.pipe.base.cmdLineTask.TaskRunner.log

Definition at line 178 of file cmdLineTask.py.

numProcesses

lsst.pipe.base.cmdLineTask.TaskRunner.numProcesses

Definition at line 182 of file cmdLineTask.py.

TaskClass

lsst.pipe.base.cmdLineTask.TaskRunner.TaskClass

Definition at line 175 of file cmdLineTask.py.

TIMEOUT

int lsst.pipe.base.cmdLineTask.TaskRunner.TIMEOUT = 3600*24*30

static

Definition at line 171 of file cmdLineTask.py.

timeout

lsst.pipe.base.cmdLineTask.TaskRunner.timeout

Definition at line 184 of file cmdLineTask.py.

---

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

• /j/snowflake/release/lsstsw/stack/cb4e2dc/Linux64/pipe_base/20.0.0-25-g3dcad98+544a109665/python/lsst/pipe/base/cmdLineTask.py