How to Write a Command-Line Task

This document describes how to write a command-line task, which is the LSST version of a complete data processing pipeline. To create a command-line task you will benefit from some background:

• Read the pipe_base documentation to get a basic idea of what tasks are and what classes are used to implement them.
• Read How to Write a Task. A command-line task is an enhanced version of a regular task. Thus all the considerations for writing a normal task also apply to writing a command-line task, and these will not be repeated in this manual.
• Acquire a basic understanding of data repositories and how to use the butler to read and write data (to be written; for now read existing tasks to see how it is done).

Contents

• Introduction
• Run Script
• Reading and Writing Data
  • Adding Dataset Types
• Persisting Config and Metadata
  • Customizing Config and Metadata Dataset Types
  • Prevent Saving Config and Metadata
• Custom Argument Parser
• Custom Task Runner

Introduction

A command-line task is an enhanced version of a regular task (see How to Write a Task). Regular tasks are only intended to be used as relatively self-contained stages in data processing pipelines, whereas command-line tasks can also be used as complete pipelines. As such, command-line tasks include Run Script that run them as pipelines.

Command-line tasks have the following key attributes, in addition to the attributes for regular tasks:

• They are subclasses of lsst.pipe.base.CmdLineTask, whereas regular tasks are subclasses of lsst.pipe.base.Task.
• They have an associated run script to run them from the command-line as pipelines (this is common, but not required, for regular tasks).
• They have a run method which performs the full pipeline data processing.
• By default the run method takes exactly one argument: a data reference for the item of data to be processed. Variations are possible, but require that you provide a Custom Argument Parser and often a Custom Task Runner.
• When run from the command line, most command-line tasks save the configuration used and the metadata generated. See Persisting Config and Metadata for more information.
• They have an additional class variable: `RunnerClass', which specifies a "task runner" for the task. The task runner takes a parsed command and runs the task. The default task runner will work for any script whose run method accepts a single data reference, such as ExampleCmdLineTask. If your task's run method needs something else then you will have to provide a custom task runner.
• They have an additional class variable canMultiprocess, which defaults to True. If your task runner cannot run your task with multiprocessing then set it False. Note: multiprocessing only affects how the task runner calls the top-level task; thus it is ignored when a task is used as a subtask.

Run Script

A command-line task can be run as a pipeline via run script. This is usually a trivial script which merely calls the task's parseAndRun method. parseAndRun does the following:

• Parses the command line, which includes determining the configuration for the task and which data items to process.
• Constructs the task.
• Calls the task's run method once for each data item to process.

The runner script for ExampleCmdLineTask is examples/exampleCmdLineTask.py:

from lsst.pipe.tasks.exampleCmdLineTask import ExampleCmdLineTask
ExampleCmdLineTask.parseAndRun()

For mosts command-line tasks you should put the run script into your package's bin/ directory, so that it is on your $PATH when you setup your package with eups. We did not want the run script for ExampleCmdLineTask to be quite so accessible, so we placed it in the examples/ directory instead of bin/.

Don't forget to and your run script executable using chmod +x.

Reading and Writing Data

The run method typically receives a single data reference, as mentioned above. It read and writes data using this data reference (or the underlying butler, if necessary).

Adding Dataset Types

Every time you write a task that writes a new kind of data (a new "dataset type") you must tell the butler about it. Similarly, if you write a new task for which you want to save configuration and metadata (which is the case for most tasks that process data), you have to tell the butler about it.

To add a dataset, edit the mapper configuration file for each obs_ package on whose data the task can be run. If the task is of general interest (wanted for most or all obs_ packages) then this process of updating all the mapper configuration files can be time consuming.

There are plans to change how mappers are configured. But as of this writing, mapper configuration files are contained in the policy directory of the obs_ package. For instance the configuration for the lsstSim mapper is defined in obs_lsstSim/policy/LsstSimMapper.paf.

Persisting Config and Metadata

Normally when you run a task you want the configuration for the task and the metadata generated by the task to be saved to the data repository. By default, this is done automatically, using dataset types:

• _DefaultName_config for the configuration
• _DefaultName_metadata for the metadata

where _DefaultName is the value of the task's _DefaultName class variable.

Whether you use these default dataset types or customize the dataset types, you will have to add dataset types for the configuration and metadata.

Customizing Config and Metadata Dataset Types

Occasionally the default dataset types for configuration and metadata are not sufficient. For instance in the case of the pipe.tasks.MakeSkyMapTask and various co-addition tasks, the co-add type must be part of the config and metadata dataset type name. To customize the dataset type of a task's config or metadata, define task methods _getConfigName and _getMetadataName to return the desired names.

Prevent Saving Config and Metadata

For some tasks you may wish to not save config and metadata at all. This is appropriate for tasks that simply report information without saving data. To disable saving configuration and metadata, define task methods _getConfigName and _getMetadataName methods to return None.

Custom Argument Parser

The default argument parser returned by CmdLineTask._makeArgumentParser assumes that your task's run method processes raw or calibrated images. If this is not the case you can easily provide a modified argument parser.

Typically this consists of constructing an instance of lsst.pipe.base.ArgumentParser and then adding some ID arguments to it using ArgumentParser.add_id_argument. This is shown in several examples below. Please resist the urge to add other kinds of arguments to the argument parser unless truly needed. One strength of our tasks is how similar they are to each other. Learning one set of arguments suffices to use many tasks.

Warning

If your task requires a custom argument parser to do more than just change the type of the single data reference, then it also require a custom task runner, as well.

Here are some examples:

• A task's run methd requires a data reference of some kind other than a raw or calibrated image. This is a common case, and easily solved. For example the processCoadd.ProcessCoaddTask processes co-adds, which are specified by sky map patch. Here is ProcessCoaddTask._makeArgumentParser:

      @classmethod
      def _makeArgumentParser(cls):
          parser = pipeBase.ArgumentParser(name=cls._DefaultName)
          parser.add_id_argument("--id", "deepCoadd", help="data ID, e.g. --id tract=12345 patch=1,2",
                                 ContainerClass=CoaddDataIdContainer)
          return parser

  • The first argument to pipeBase.ArgumentParser is the name of the ID argument.
  • The second argument is a dataset type, which specifies the keys that are used with this ID argument. The keys associated with a particular dataset type are specified in the mapper configuration file for the obs_ package and camera in question, and thus may vary from camera to camera. In practice, the keys for "raw" and "calexp" dataset types usually do vary from camera to camera, but the keys for coadds do not. However, this is not a fixed rule. For most obs_ packages "deepCoadd" is one of two coadd dataset types, and the other, "goodSeeingCoadd", would work just as well for this argument.
  • A custom "ContainerClass" `CoaddDataIdContainer is provided to support iterating over missing keys (e.g. if you provide a tract but not a patch then the task will iterate over all available patches for that tract). This happens automatically for "raw" and "calexp" dataset types, but not most other dataset types. Examine the code in CoaddDataIdContainer to see how it works.

• A task's run method requires more than one kind of data reference. An example is co-addition, which requires the user to specify the co-add as a sky map patch, and optionally allows the user to specify a list of exposures to co-add. CoaddBaseTask._makeArgumentParser is a straightforward example of specifying two data IDs arguments: one for the sky map patch, and an optional ID argument for which exposures to co-add:

      @classmethod
      def _makeArgumentParser(cls):
          """Create an argument parser
          """
          parser = pipeBase.ArgumentParser(name=cls._DefaultName)
          parser.add_id_argument("--id", "deepCoadd", help="data ID, e.g. --id tract=12345 patch=1,2",
                                 ContainerClass=CoaddDataIdContainer)
          parser.add_id_argument("--selectId", "calexp", help="data ID, e.g. --selectId visit=6789 ccd=0..9",
                                 ContainerClass=SelectDataIdContainer)
          return parser

  In this case the custom container class SelectDataIdContainer adds additional information for the task, to save processing time.

• A task's run method requires no data references at all. An example is makeSkyMap.MakeSkyMapTask, which makes a sky map for a set of co-adds. makeSkyMap.MakeSkyMapTask._makeArgumentParser is trivial:

      @classmethod
      def _makeArgumentParser(cls):
          """Create an argument parser
  
          No identifiers are added because none are used.
          """
          return pipeBase.ArgumentParser(name=cls._DefaultName)

Custom Task Runner

The standard task runner is lsst.pipe.base.TaskRunner. It assumes that your task's run method wants a single data reference and nothing else. If that is not the case then you will have to provide a custom task runner for your task. This involves writing a subclass of lsst.pipe.base.TaskRunner and specifying it in your task using the RunnerClass Class Variables" class variable".

Here are some situations where a custom task runner is required:

• The task's run method requires extra arguments. An example is co-addition, which optionally accepts a list of images to co-add. The custom task runner is coaddBase.CoaddTaskRunner and is pleasantly simple:

  class CoaddTaskRunner(pipeBase.TaskRunner):
      @staticmethod
      def getTargetList(parsedCmd, **kwargs):
          return pipeBase.TaskRunner.getTargetList(parsedCmd, selectDataList=parsedCmd.selectId.dataList,
                                                   **kwargs)

• The task requires no data references, just a butler. An example is makeSkyMap.MakeSkyMapTask, which makes a skymap.SkyMap for a set of co-adds. It uses the custom task runner makeSkyMap.MakeSkyMapRunner, which is more complicated than the previous example because the entire __call__ method must be overridden:

  class MakeSkyMapRunner(pipeBase.TaskRunner):
      """Only need a single butler instance to run on."""
      @staticmethod
      def getTargetList(parsedCmd):
          return [parsedCmd.butler]
  
      def precall(self, parsedCmd):
          # We overload to disable writing/checking of schemas and configs.
          # There's only one SkyMap per rerun anyway, so the config is redundant,
          # and checking it means we can't overwrite or append to one once we've
          # written it.
          return True
  
      def __call__(self, butler):
          task = self.TaskClass(config=self.config, log=self.log)
          if self.doRaise:
              results = task.run(butler)
          else:
              try:
                  results = task.run(butler)
              except Exception, e:
                  task.log.fatal("Failed: %s" % e)
                  if not isinstance(e, pipeBase.TaskError):
                      traceback.print_exc(file=sys.stderr)
          task.writeMetadata(butler)
          if self.doReturnResults:
              return results