Create an Operator via Decorator

For convenience, a holoscan.decorator.create_op decorator is provided which can be used to automatically convert a simple Python function/generator or a class into a native Python holoscan.core.Operator. The wrapped function body (or the __call__ method if create_op is applied to a class) will correspond to the computation to be done in the holoscan.core.Operator.compute method, but without any need to explicitly make any calls to holoscan.core.InputContext.receive to receive inputs or holoscan.core.OutputContext.emit to transmit the output. Any necessary input or output ports will have been automatically generated.

Consider first a simple Python function named mask_and_offset that takes image and mask tensors as input and multiplies them, followed by adding some scalar offset.

1
def mask_and_offset(image, mask, offset=1.5):
2
    return image * mask + offset

To turn this into an function that returns a corresponding operator we can add the create_op decorator like this:

1
from holoscan.decorator import create_op
2
 
3
 
4
@create_op(
5
    inputs=("image", "mask"),
6
    outputs="out",
7
)
8
def mask_and_offset(image, mask, offset=1.5):
9
    return image * mask + offset

By supplying the inputs argument we are specifying that there are two input ports, named “image” and “mask”. By setting outputs="out" we are indicating that the output will be transmitted on a port named “out”. When inputs are specified by simple strings in this way, the names used must map to variable names in the wrapped function’s signature. We will see later that it is possible to use the holoscan.decorator.Input class to provide more control over how inputs are mapped to function arguments. Similarly, we will see that the holoscan.decorator.Output class can be used to provide more control over how the function output is mapped to any output port(s).

There is also an optional, cast_tensors argument to create_op. For convenience, this defaults to True, which results in any tensor-like objects being automatically cast to a NumPy or CuPy array (for host or device tensors, respectively) before they are passed on to the function. If this is not desired (e.g. due to working with a different third party tensor framework than NumPy or CuPy), the user can set cast_tensors=False, and manually handle casting of any holoscan.Tensor objects to the desired form in the function body. This casting option applies to either single tensors or a tensor map (dict[Tensor]).

This decorated function can then be used within the compose method of an Application to create an operator corresponding to this computation:

1
from holoscan.core import Application, Operator
2
 
3
 
4
def MyApp(Application):
5
 
6
    def compose(self)
7
        mask_op = mask_and_offset(self, name="my_scaling_op", offset=0.0)
8
 
9
        # verify that an Operator was generated
10
        assert(isinstance(mask_op, Operator))
11
 
12
        # now add any additional operators and create the computation graph using add_flow

Note that as for all other Operator classes, it is required to supply the application (or fragment) as the first argument (self here). The name kwarg is always supported and is the name that will be assigned to the operator. Due to the use of this kwarg to specify the operator name, the wrapped function (mask_and_offset in this case) should not use name as an argument name. In this case, we specified offset=0.0 which would override the default value of offset=1.5 in the function signature.

For completeness, the use of the create_op decorator on mask_and_offset is equivalent to if the user had defined the following MaskAndOffsetOp class and used it in MyApp.compose:

1
def MaskAndOffsetOp(Operator):
2
    def setup(self, spec):
3
        spec.input("image")
4
        spec.input("mask")
5
        spec.output("out")
6
 
7
    def compute(self, op_input, op_output, context):
8
        # Simplified logic here assumes received values are GPU tensors
9
        # create_op would add additional logic to handle the inputs
10
        image = op_input.receive("image")
11
        image = cp.asarray(image)
12
 
13
        mask = op_input.receive("mask")
14
        mask = cp.asarray(mask)
15
 
16
        out = image * mask + offset
17
        op_output.emit(out, "out")

Decorating a function that returns a tuple of arrays

Let’s consider another example where function takes in multiple arrays, processes them, and returns a tuple of updated arrays:

1
def scale_and_offset(x1, x2, scale=2.0, offset=1.5):
2
    y1 = x1 * scale
3
    y2 = x2 + offset
4
    return y1, y2

To turn this into a corresponding operator we can add the create_op decorator like this:

1
@create_op(
2
    inputs=("x1", "x2"),
3
    outputs=("out1", "ou2"),
4
)
5
def scale_and_offset(x1, x2, scale=2.0, offset=1.5):
6
    y1 = x1 * scale
7
    y2 = x2 + offset
8
    return y1, y2

As before, the messages received through the ports defined by inputs, “x1” and “x2”, will be mapped to respective variables x1 and x2. Likewise, the elements of the output tuple, arrays y1 and y2, will be emitted through ports “out1” and “out2”, respectively. In contrast to input mapping, which is determined by the naming of ports and variables, the output mapping is determined by the ordering of output ports and elements in the tuple returned by the function.

Using the Input class for more control over input ports

This section will cover additional use cases where using a str or Tuple[str] for the inputs argument is insufficient.

Scenario 1: Assume that the upstream operator sends a tensormap to a given input port and we need to specify which tensor(s) in the tensormap will map to which input port.

For a concrete example, suppose we want to print a tensor’s shape using a function like:

1
def print_shape(tensor):
2
    print(f"{tensor.shape = }")

but the upstream operator outputs a dictionary containing two tensors named “image” and “labels”. We could use this operator by specifying which tensor name on a particular input port would map to the function’s “tensor” argument. For example:

1
@create_op(inputs=Input("input_tensor", arg_map={"image": "tensor"}))
2
def print_shape(tensor):
3
    print(f"{tensor.shape = }")

would create an operator with a single input port named “input_tensor” and no output port. The input port may receive a tensormap with any number of tensors, but will only use the tensor named “image”, mapping it to the “tensor” argument of the wrapped function. In general, the arg_map is a dictionary mapping tensor names found on the port to their corresponding function argument names.

Scenario 2: we want to override the scheduling condition present on a port. This can be done by specifying Input with the condition and optionally condition_kwargs arguments. For example, to override the MessageAvailableCondition that is added to the port by default and allow it to call compute even when no input message is available:

1
@create_op(inputs=Input("input_tensor", condition=ConditionType.NONE, condition_kwargs={}))

Scenario 3: we want to override the parameters of the receiver present on a port. For example, we could specify a different policy for the double buffer receiver that is used by default (policy=1 corresponds to discarding incoming messages when the queue is already full)

1
@create_op(inputs=Input("input_tensor", connector=ConditionType.DOUBLE_BUFFER, connector_kwargs=dict(capacity=1, policy=1)))

Using the Output class for more control over output ports

To support a case where multiple output ports should be present, the user must have the function return a dict. The holoscan.decorator.Output class then has a tensor_names keyword argument that can be specified to indicate which items in the dictionary are to be transmitted on a given output port.

For example, assume we have a function that generates three tensors, x, y and z and we want to transmit x and y on port “out1” while z will be transmitted on port “out2”. This can be done by specifying outputs as follows in the create_op call:

1
@create_op(
2
    outputs=(
3
        Output("out1", tensor_names=("x", "y")),
4
        Output("out2", tensor_names=("z",)),
5
    ),
6
)
7
def xyz_generator(nx=32, ny=32, nz=16):
8
    x = cp.arange(nx, dtype=np.uint32)
9
    y = cp.arange(ny, dtype=np.uint32)
10
    z = cp.arange(nz, dtype=np.uint32)
11
 
12
    # must return a dict type when Output arg(s) with `tensor_names` is used
13
    return dict(x=x, y=y, z=z)

This operator has no input ports and three optional keyword arguments. It splits the output tensors across two ports as described above. All names used in tensor_names must correspond to keys present in the dict emitted by the object. Often the dict values are tensors, but that is not a requirement.

The holoscan.decorator.Output class also supports condition, condition_kwargs, connector and connector_kwargs that work in the same way as shown for holoscan.decorator.Input above. For example, to override the transmitter queue policy for a single output port named “output_tensor”

1
@create_op(outputs=Output("output_tensor",
2
                           connector=IOSpec.ConnectorType.DOUBLE_BUFFER,
3
                           connector_kwargs=dict(capacity=1, policy=1)))

note that tensor_names was not specified which means that the returned object does not need to be a dict. The object itself will be emitted on the “output_tensor” port.

Configuring the queue size and policy for an input or output port

When using the decorator approach to create operators, you can configure the queue size and policy for input and output ports using the Input and Output classes. Here’s how to configure these parameters:

1
from holoscan.core import ConditionType, IOSpec
2
from holoscan.decorator import Input, Output, create_op
3
 
4
# Example with input port configuration
5
@create_op(
6
    op_param="self",
7
    inputs=Input(
8
        "input",
9
        arg_map="value",
10
        size=IOSpec.SIZE_ONE,  # Set queue size to 1
11
        policy=IOSpec.QueuePolicy.POP  # Pop oldest item if queue is full
12
    ),
13
    outputs="output"
14
)
15
def my_operator(self, value):
16
    return value * 2
17
 
18
# Example with output port configuration
19
@create_op(
20
    op_param="self",
21
    inputs="input",
22
    outputs=Output(
23
        "output",
24
        size=2,  # Set queue size to 2
25
        policy=IOSpec.QueuePolicy.REJECT,  # Reject new items if queue is full
26
        condition_type=ConditionType.NONE
27
    )
28
)
29
def another_operator(self, value):
30
    return value

Queue Size Options

The size parameter can be set to:

• IOSpec.SIZE_ONE: Queue size of 1 (default)
• IOSpec.ANY_SIZE: Any size queue
• IOSpec.PRECEDING_COUNT: Size based on number of preceding connections
• An integer value: Custom queue size

Queue Policy Options

The policy parameter accepts these values:

• IOSpec.QueuePolicy.POP: Pop oldest item when queue is full
• IOSpec.QueuePolicy.REJECT: Reject new items when queue is full
• IOSpec.QueuePolicy.FAULT: Log warning and reject when queue is full

Example Use Cases

1. Throttling execution with POP policy:

1
@create_op(
2
    op_param="self",
3
    inputs=Input("input", arg_map="value"),
4
    outputs=Output(
5
        "output",
6
        policy=IOSpec.QueuePolicy.POP,
7
        condition_type=ConditionType.NONE
8
    )
9
)
10
def execution_throttler_op(self, value):
11
    if value is not None:
12
        return value

1. Multiple input handling:

1
@create_op(
2
    op_param="self",
3
    inputs=Input("input", arg_map="value", size=IOSpec.ANY_SIZE),
4
    outputs="output"
5
)
6
def multi_input_op(self, value):
7
    return value

Using the op_param argument to access the operator instance

The op_param argument to create_op can be used to access the operator instance within the function body. This is useful if the operator needs to access its own name or other attributes.

1
@create_op(op_param="self")
2
def simple_op(self, param1, param2=5):
3
    print(f"I am here - {self.name} (param1: {param1}, param2: {param2})")

The operator can then be used in an application like this:

1
class OpParamApp(Application):
2
    def compose(self):
3
        node1 = simple_op(self, param1=1, name="node1")
4
        node2 = simple_op(self, param1=2, name="node2")
5
        node3 = simple_op(self, param1=3, name="node3")
6
 
7
        self.add_flow(self.start_op(), node1)
8
        self.add_flow(node1, node2)
9
        self.add_flow(node2, node3)

The output of this application will be:

I am here - node1 (param1: 1, param2: 5)
I am here - node2 (param1: 2, param2: 5)
I am here - node3 (param1: 3, param2: 5)

When this application runs, each operator instance will print its name along with its parameter values. The op_param argument allows the function to access operator attributes like name through the specified parameter (in this case self). This is particularly useful when you need to access operator-specific information or methods within your function’s implementation.

Interoperability with wrapped C++ operators

The SDK includes a python_decorator example showing interoperability of wrapped C++ operators (VideoStreamReplayerOp and HolovizOp) alongside native Python operators created via the create_op decorator.

The start of this application imports a couple of the built in C++-based operators with Python bindings (HolovizOp and VideoStreamReplayerOp). In addition to these, two new operators are created via the create_op decorator APIs.

1
import os
2
 
3
from holoscan.core import Application
4
from holoscan.decorator import Input, Output, create_op
5
from holoscan.operators import HolovizOp, VideoStreamReplayerOp
6
 
7
sample_data_path = os.environ.get("HOLOSCAN_INPUT_PATH", "../data")
8
 
9
 
10
@create_op(
11
    inputs="tensor",
12
    outputs="out_tensor",
13
)
14
def invert(tensor):
15
    tensor = 255 - tensor
16
    return tensor
17
 
18
 
19
@create_op(inputs=Input("in", arg_map="tensor"), outputs=Output("out", tensor_names=("frame",)))
20
def tensor_info(tensor):
21
    print(f"tensor from 'in' port: shape = {tensor.shape}, " f"dtype = {tensor.dtype.name}")
22
    return tensor

The first is created by adding the decorator to a function named invert which just inverts the (8-bit RGB) color space values. A second operator, is created by adding the decorator to a function named tensor_info, which assumes that the input is a CuPy or NumPy tensor, and prints its shape and data type. Note that create_op’s default cast_tensors=True option ensures that any host or device tensors are cast to NumPy or CuPy arrays, respectively. This is why it is safe to use NumPy APIs in the function bodies. If the user wants to receive the holoscan.Tensor object directly and manually handle the casting to a different type of object in the function body, then cast_tensors=False should be specified in the keyword arguments to create_op.

Now that we have defined or imported all of the operators, we can build an application in the usual way by inheriting from the holoscan.core.Application class and implementing the compose method. The remainder of the code for this example is shown below.

1
class VideoReplayerApp(Application):
2
    """Example of an application that uses the operators defined above.
3
 
4
    This application has the following operators:
5
 
6
    - VideoStreamReplayerOp
7
    - HolovizOp
8
    - invert (created via decorator API)
9
    - tensor_info (created via decorator API)
10
 
11
    `VideoStreamReplayerOp` reads a video file and sends the frames to the HolovizOp.
12
    The `invert` operator inverts the color map (the 8-bit `value` in each color channel is
13
    set to `255 - value`).
14
    The `tensor_info` operator prints information about the tensors shape and data type.
15
    `HolovizOp` displays the frames.
16
    """
17
 
18
    def compose(self):
19
        video_dir = os.path.join(sample_data_path, "racerx")
20
        if not os.path.exists(video_dir):
21
            raise ValueError(f"Could not find video data: {video_dir=}")
22
 
23
        # Define the replayer and holoviz operators
24
        replayer = VideoStreamReplayerOp(
25
            self,
26
            name="replayer",
27
            directory=video_dir,
28
            basename="racerx",
29
            frame_rate=0,  # as specified in timestamps
30
            repeat=False,  # default: false
31
            realtime=True,  # default: true
32
            count=40,  # default: 0 (no frame count restriction)
33
        )
34
        invert_op = invert(self, name="image_invert")
35
        info_op = tensor_info(self, name="tensor_info")
36
        visualizer = HolovizOp(
37
            self,
38
            name="holoviz",
39
            width=854,
40
            height=480,
41
            # name="frame" to match Output argument to create_op for tensor_info
42
            tensors=[dict(name="frame", type="color", opacity=1.0, priority=0)],
43
        )
44
        # Define the workflow
45
        self.add_flow(replayer, invert_op, {("output", "tensor")})
46
        self.add_flow(invert_op, info_op, {("out_tensor", "in")})
47
        self.add_flow(info_op, visualizer, {("out", "receivers")})
48
 
49
 
50
def main():
51
    app = VideoReplayerApp()
52
    app.run()
53
 
54
 
55
if __name__ == "__main__":
56
    main()

The highlighted lines show how Operators corresponding to the invert and tensor_info functions are created by passing the application itself as the first argument. The invert_op and info_op variables now correspond to a holoscan.core.Operator class and can be connected in the usual way using add_flow to define the computation. Note that a name was provided for these operators, via the optional name keyword argument. In this case each operator is only used once, but if the same operator is to be used more than once in an application, each should be given a unique name.

Using create_op to turn a generator into an Operator

The create_op decorator can be applied to a generator in the same way as for a function. In this case, a BooleanCondition will automatically be added to the operator that will stop it from trying to call compute again once the generator is exhausted (has no more values to yield). The following is a basic example of decorating a generator for integers from 1 to count:

1
@create_op(outputs="out")
2
def source_generator(count):
3
    yield from range(1, count + 1)

The compose method can then create an operator from this decorated generator as follows

1
count_op = source_generator(self, count=100, name="int_source")

Using create_op to turn a class into an Operator

The create_op decorator can also be applied to a class implementing the __call__ method, to turn it into an holoscan.core.Operator. One reason to choose a class vs. a function is if there is some internal state that needs to be maintained across calls. For example, the operator defined below casts the input data to 32-bit floating point and on even frames also negates the values.

1
@create_op
2
class NegateEven:
3
    def __init__(self, start_index=0):
4
        self.counter = start_index
5
 
6
    def __call__(self, x):
7
        # cast tensor to 32-bit floating point
8
        x = x.astype('float32')
9
 
10
        # negate the values if the frame is even
11
        if self.counter % 2 == 0:
12
            x = -x
13
        return x

In this case, since there is only a single input and output for the function, we can omit the inputs and outputs arguments in the call to create_op. In this case the input port will have name "x", as determined from the variable name in the function signature. The output port will be have an empty name "". To use different port names, the inputs and/or outputs arguments should be specified.

The compose method can then create an operator from this decorated generator as follows. Note that any positional or keyword arguments in the __init__ method would be supplied during the NegateEven call. This returns a function (not yet an operator) that can then be called to generate the operator. This is shown below

1
negate_op_creation_func = NegateEven(start_index=0)                 # `negate_op_creation_func` is a function that returns an Operator
2
negate_even_op = negate_op_creation_func(self, name="negate_even")  # call the function to create an instance of NegateEvenOp

or more concisely as just

1
negate_even_op = NegateEven(start_index=0)(self, name="negate_even")

Note that the operator class as defined above is approximately equivalent to the Python native operator defined below. We show it here explicitly for reference.

1
import cupy as cp
2
import numpy as np
3
 
4
 
5
class NegateEvenOp(Operator):
6
 
7
    def __init__(self, fragment, *args, start_index=0, **kwargs):
8
        self.counter = start_index
9
        super().__init__(fragment, *args, **kwargs)
10
 
11
    def setup(self, spec):
12
        spec.input("x")
13
        spec.output("")
14
 
15
    def compute(op_input, op_output, context):
16
        x = op_input.receive("x")
17
 
18
        # cast to CuPy or NumPy array
19
        # (validation that `x` is a holoscan.Tensor is omitted for simplicity)
20
        if hasattr(x, '__cuda_array_interface__'):
21
            x = cupy.asarray(x)
22
        else:
23
            x = numpy.asarray(x)
24
 
25
        # cast tensor to 32-bit floating point
26
        x = x.astype('float32')
27
 
28
        # negate the values if the frame is even
29
        if self.counter % 2 == 0:
30
            x = -x
31
 
32
        op_output.emit(x, "")

The primary differences between this NegateEvenOp class and the decorated NegateEven above are:

• NegateEven does not need to define a setup method
• NegateEven does not inherit from Operator and so does not call its __init__ from the constructor.
• The NegateEven::__call__ method is simpler than the NegateEvenOp::compute method as receive and emit methods do not need to be explicitly called and casting to a NumPy or CuPy array is automatically handled for NegateEven.