Example demonstrating IOStage to read and write FITS files

This example describes a Pipeline that performs a test of IOStage.InputStage and IOStage.OutputStage. This example is found under pex_harness/examples/inputoutputfits/.

The three stages in the pipeline are

• InputStage
• ImageprocStage
• OutputStage

To configure the pipeline with the three application stages we insert one "appStage" policy for each stage. These will be gathered into an array by policy. Each appStage policy designates the Python package and class of the application stage, the event dependency for the stage, and the stage policy file. The pipeline policy file inputoutputfits_policy.paf for this case is

localLogMode: true
executionMode: "oneloop"
eventBrokerHost: "lsst8.ncsa.uiuc.edu"
logThreshold: -3

appStage: {
     stageName: "lsst.pex.harness.IOStage.InputStage"
     eventTopic: "triggerImageprocEvent"
     stagePolicy: "policy/input_policy.paf"
}

appStage: {
     stageName: "lsst.pexhexamples.apps.ImageprocStage"
     eventTopic: "None"
     stagePolicy: "None"
}

appStage: {
     stageName: "lsst.pex.harness.IOStage.OutputStage"
     eventTopic: "None"
     stagePolicy: "policy/output_policy.paf"
}


dir: {

    defaultRoot:  .
    runDirPattern:  "sample"

    # These indicate the directory that should be used for a named purpose.
    # If relative paths are given, the resulting directory will be relative
    # to the default run directory (determined by defaultRoot and the
    # runDirPattern).  These can be given as patterns specified in the same
    # format as runDirPattern.  (If a directory is given as an absolute path,
    # using a pattern is recommended in order to distinguish between different
    # production runs.)
    #
    work:     .    # the working directory, where the pipeline is started
    input:    .    # the directory to cache/find input data
    output:   .    # the directory to write output data
    update:   .    # the directory where updatable data is deployed
    scratch:  .    # a directory for temporary files that may be deleted
                   # upon completion of the pipeline

}

The attribute localLogMode set to true will give local log files written to disk for the Pipleline and each Slice. Because of the assignment eventTopic: "triggerImageprocEvent" for the first Stage, the pipeline initially blocks to receive an event of topic "triggerImageprocEvent", which carries information such as the exposureId to identify the input image.

The InputStage and OutputStage require appropriate policy files for their effective operation. The policy file input_policy.paf for InputStage is

AdditionalData: "exposureId=triggerImageprocEvent.exposureid"
RunMode: "process"
InputItems: {
    InputImage: {
        Type: "MaskedImageF"
        PythonType: "lsst.afw.image.imageLib.MaskedImageF"
        StoragePolicy: {
            Storage: "FitsStorage"
            Location: "%(input)/%(exposureId)p_1_MI"
        }
    }
}

The InputStage is configured to store data as a PythonType: "lsst.afw.image.imageLib.MaskedImageF" after accessing Storage: "FitsStorage". InputStage is configured to place the input data onto the Clipboard under key "InputImage".

The ImageprocStage is a dummy application stage that shows that mocks an image processing stage. It pulls the "inputImage" from the Clipboard and posts it back to the Clipboard under the key "outputImage".

The policy file output_policy.paf for OutputStage is

AdditionalData: "exposureId=triggerImageprocEvent.exposureid"
RunMode: "process"
OutputItems: {
    OutputImage: {
        required: true
        StoragePolicy: {
            Storage: "FitsStorage"
            Location: "%(output)/image%(exposureId)"
        }
    }
}

The OutputStage is configured to retrieve data from the Clipboard under the key "OutputStage" of PythonType: "lsst.afw.image.imageLib.MaskedImageF" and persist ot using Storage: "FitsStorage". The OutputStage encodes the exposureId within the name of the output file.

By default this example will run with the Pipeline with one Slice on a single node. Edit the MPI machinefile "nodelist.scr" to specify the hostname of the current host.

Prior to running the example, the full path to the directory pex_harness/examples/stages should be added to the PYTHONPATH:

% export PYTHONPATH ${PWD}/../stages:${PYTHONPATH}    (bash)
% setenv PYTHONPATH ${PWD}/../stages:${PYTHONPATH}    (tcsh)

The pipeline is then executed via

% launchPipeline.py inputoutputfits_policy.paf <some-run-id>

such as

% launchPipeline.py inputoutputfits_policy.paf test_3001 

The pipeline blocks for an event that will carry the exposureId of the image to process. This event can be supplied by running with in a separate shell the script:

% python triggerImageprocEventTransmit.py

if an ActiveMQbroker other than the LSST default (lsst8.ncsa.uiuc.edu) is used, this needs to be specified in the pipeline policy file (under "eventBrokerHost") and the event trigger python script.