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
orpythondppp
.
python.class
- type:
string
- doc:
Name of the python class that subclasses the
Step
class. In the example above, this should beMockPyStep
.
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, setmodule
tomockpystep
.