PythonStep

Run a user defined python step. Parameters are passed to python. Python steps can be used both from command-line invocations of DP3 and from the DP3 python bindings.

Note: The documentation on how to call DP3 from python is currently work in progress.

To make a python step work, make sure that the python libraries of DP3 are in the python path, e.g. by setting

export PYTHONPATH=/path/to/dp3/lib64/python3.10/site-packages:$PYTHONPATH

The example below shows a minimal example of a python step.

# Copyright (C) 2020 ASTRON (Netherlands Institute for Radio Astronomy)
# SPDX-License-Identifier: GPL-3.0-or-later

# MockPyStep.py: simple python step, where the MockPyStep inherits from
# the (py)dp3.DPStep class.

from dp3 import parameterset, Fields

# Import raw step from pydp3
# The extra checks in the higher level interface (dp3.Step) are thereby omitted
# The raw Step is needed because some unit tests trigger these additional checks
from dp3.pydp3 import Step

import numpy as np


class MockPyStep(Step):
    """Example python DPStep that multiplies DATA and WEIGHT_SPECTRUM"""

    def __init__(self, parset, prefix):
        """
        Set up the step (constructor). Read the parset here.

        Args:
          parset: Parameter set for the entire pipeline
          prefix: Prefix for this step, e.g. "thisstepname."
        """

        super().__init__()
        self.datafactor = parset.get_double(prefix + "datafactor")
        self.weightsfactor = parset.get_double(prefix + "weightsfactor")

    def show(self):
        """Print a summary of the step and its settings"""
        print("\nMockPyStep")
        print(f"  data factor:    {self.datafactor}")
        print(f"  weights factor: {self.weightsfactor}")

    def get_required_fields(self):
        return Fields.DATA | Fields.WEIGHTS

    def get_provided_fields(self):
        # The Flags field is added as provided fields only for testing purposes.
        return Fields.DATA | Fields.FLAGS | Fields.WEIGHTS

    def process(self, dpbuffer):
        """
        Process one time slot of data. This function MUST call self.get_next_step().process

        Args:
          dpbuffer: DPBuffer object which can contain data, flags and weights
                    for one time slot.
        """

        data = np.array(dpbuffer.get_data(), copy=False)
        weights = np.array(dpbuffer.get_weights(), copy=False)

        # Do the operation on data
        data *= self.datafactor
        weights *= self.weightsfactor

        # Send processed data to the next step
        self.get_next_step().process(dpbuffer)

    def finish(self):
        """
        If there is any remaining data, process it. This can be useful if the
        step accumulates multiple time slots.
        """
        pass

For a complete list of DP3 functions available in python see the source code of pydp3.cc

inputs

type

type:

string

doc:

Case-insensitive step type; must be python or pythondppp .

python.class

type:

string

doc:

Name of the python class that subclasses the Step class. In the example above, this should be MockPyStep.

python.module

type:

string

doc:

Name of the module where the python code lives. Typically this is the file name of the python code, which should be in the python path. E.g. if your code is in mockpystep.py in the working directory, set module to mockpystep.