通过 create_op 装饰器简化 Python 运算符的创建

            

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

            

            
from holoscan.decorator import create_op


@create_op(
    inputs=("image", "mask"),
    outputs="out",
)
def mask_and_offset(image, mask, offset=1.5):
    return image * mask + offset
        

            

            
from holoscan.core import Application, Operator


def MyApp(Application):

    def compose(self)
        mask_op = mask_and_offset(self, name="my_scaling_op", offset=0.0)

        # verify that an Operator was generated
        assert(isinstance(mask_op, Operator))

        # now add any additional operators and create the computation graph using add_flow
        

            

            
def MaskAndOffsetOp(Operator):
    def setup(self, spec):
        spec.input("image")
        spec.input("mask")
        spec.output("out")

    def compute(self, op_input, op_output, context):
        # Simplified logic here assumes received values are GPU tensors
        # create_op would add additional logic to handle the inputs
        image = op_input.receive("image")
        image = cp.asarray(image)

        mask = op_input.receive("mask")
        mask = cp.asarray(mask)

        out = image * mask + offset
        op_output.emit(out, "out")
        

            

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

            

            
@create_op(
    inputs=("x1", "x2"),
    outputs=("out1", "ou2"),
)
def scale_and_offset(x1, x2, scale=2.0, offset=1.5):
    y1 = x1 * scale
    y2 = x2 + offset
    return y1, y2
        

            

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

            

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

            

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

            

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

            

            
@create_op(
    outputs=(
        Output("out1", tensor_names=("x", "y")),
        Output("out2", tensor_names=("z",)),
    ),
)
def xyz_generator(nx=32, ny=32, nz=16):
    x = cp.arange(nx, dtype=np.uint32)
    y = cp.arange(ny, dtype=np.uint32)
    z = cp.arange(nz, dtype=np.uint32)

    # must return a dict type when Output arg(s) with `tensor_names` is used
    return dict(x=x, y=y, z=z)
        

            

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

            

            
import os

from holoscan.core import Application
from holoscan.decorator import Input, Output, create_op
from holoscan.operators import HolovizOp, VideoStreamReplayerOp

sample_data_path = os.environ.get("HOLOSCAN_INPUT_PATH", "../data")


@create_op(
    inputs="tensor",
    outputs="out_tensor",
)
def invert(tensor):
    tensor = 255 - tensor
    return tensor


@create_op(inputs=Input("in", arg_map="tensor"), outputs=Output("out", tensor_names=("frame",)))
def tensor_info(tensor):
    print(f"tensor from 'in' port: shape ={tensor.shape}, " f"dtype ={tensor.dtype.name}")
    return tensor

        

            

            
class VideoReplayerApp(Application):
    """Example of an application that uses the operators defined above.

This application has the following operators:

- VideoStreamReplayerOp
- HolovizOp
- invert (created via decorator API)
- tensor_info (created via decorator API)

`VideoStreamReplayerOp` reads a video file and sends the frames to the HolovizOp.
The `invert` operator inverts the color map (the 8-bit `value` in each color channel is
set to `255 - value`).
The `tensor_info` operator prints information about the tensors shape and data type.
`HolovizOp` displays the frames.
"""

    def compose(self):
        video_dir = os.path.join(sample_data_path, "racerx")
        if not os.path.exists(video_dir):
            raise ValueError(f"Could not find video data:{video_dir=}")

        # Define the replayer and holoviz operators
        replayer = VideoStreamReplayerOp(
            self,
            name="replayer",
            directory=video_dir,
            basename="racerx",
            frame_rate=0,  # as specified in timestamps
            repeat=False,  # default: false
            realtime=True,  # default: true
            count=40,  # default: 0 (no frame count restriction)
        )
        invert_op = invert(self, name="image_invert")
        info_op = tensor_info(self, name="tensor_info")
        visualizer = HolovizOp(
            self,
            name="holoviz",
            width=854,
            height=480,
            # name="frame" to match Output argument to create_op for tensor_info
            tensors=[dict(name="frame", type="color", opacity=1.0, priority=0)],
        )
        # Define the workflow
        self.add_flow(replayer, invert_op, {("output", "tensor")})
        self.add_flow(invert_op, info_op, {("out_tensor", "in")})
        self.add_flow(info_op, visualizer, {("out", "receivers")})


def main():
    app = VideoReplayerApp()
    app.run()


if __name__ == "__main__":
    main()
        

            

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

            

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

            

            
@create_op
class NegateEven:
    def __init__(self, start_index=0):
        self.counter = start_index

    def __call__(self, x):
        # cast tensor to 32-bit floating point
        x = x.astype('float32')

        # negate the values if the frame is even
        if self.counter % 2 == 0:
            x = -x
        return x
        

            

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

            

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

            

            
import cupy as cp
import numpy as np


class NegateEvenOp(Operator):

    def __init__(self, fragment, *args, start_index=0, **kwargs):
        self.counter = start_index
        super().__init__(fragment, *args, **kwargs)

    def setup(self, spec):
        spec.input("x")
        spec.output("")

    def compute(op_input, op_output, context):
        x = op_input.receive("x")

        # cast to CuPy or NumPy array
        # (validation that `x` is a holoscan.Tensor is omitted for simplicity)
        if hasattr(x, '__cuda_array_interface__'):
            x = cupy.asarray(x)
        else:
            x = numpy.asarray(x)

        # cast tensor to 32-bit floating point
        x = x.astype('float32')

        # negate the values if the frame is even
        if self.counter % 2 == 0:
            x = -x

        op_output.emit(x, "")