创建算子

            

            
#include "holoscan/holoscan.hpp"

using holoscan::Operator;
using holoscan::OperatorSpec;
using holoscan::InputContext;
using holoscan::OutputContext;
using holoscan::ExecutionContext;
using holoscan::Arg;
using holoscan::ArgList;

class MyOp : public Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(MyOp)

  MyOp() = default;

  void setup(OperatorSpec& spec) override {
  }

  void start() override {
    HOLOSCAN_LOG_TRACE("MyOp::start()");
  }

  void compute(InputContext&, OutputContext& op_output, ExecutionContext&) override {
    HOLOSCAN_LOG_TRACE("MyOp::compute()");
  };

  void stop() override {
    HOLOSCAN_LOG_TRACE("MyOp::stop()");
  }
};
        

            

            
#include "holoscan/holoscan.hpp"

class ValueData {
 public:
  ValueData() = default;
  explicit ValueData(int value) : data_(value) {
    HOLOSCAN_LOG_TRACE("ValueData::ValueData(): {}", data_);
  }
  ~ValueData() { HOLOSCAN_LOG_TRACE("ValueData::~ValueData(): {}", data_); }

  void data(int value) { data_ = value; }

  int data() const { return data_; }

 private:
  int data_;
};

namespace holoscan::ops {

class PingTxOp : public Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(PingTxOp)

  PingTxOp() = default;

  void setup(OperatorSpec& spec) override {
    spec.output<std::shared_ptr<ValueData>>("out1");
    spec.output<std::shared_ptr<ValueData>>("out2");
  }

  void compute(InputContext&, OutputContext& op_output, ExecutionContext&) override {
    auto value1 = std::make_shared<ValueData>(index_++);
    op_output.emit(value1, "out1");

    auto value2 = std::make_shared<ValueData>(index_++);
    op_output.emit(value2, "out2");
  };
  int index_ = 1;
};

class PingMxOp : public Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(PingMxOp)

  PingMxOp() = default;

  void setup(OperatorSpec& spec) override {
    spec.input<std::shared_ptr<ValueData>>("in1");
    spec.input<std::shared_ptr<ValueData>>("in2");
    spec.output<std::shared_ptr<ValueData>>("out1");
    spec.output<std::shared_ptr<ValueData>>("out2");
    spec.param(multiplier_, "multiplier", "Multiplier", "Multiply the input by this value", 2);
  }

  void compute(InputContext& op_input, OutputContext& op_output, ExecutionContext&) override {
    auto value1 = op_input.receive<std::shared_ptr<ValueData>>("in1").value();
    auto value2 = op_input.receive<std::shared_ptr<ValueData>>("in2").value();

    HOLOSCAN_LOG_INFO("Middle message received (count: {})", count_++);

    HOLOSCAN_LOG_INFO("Middle message value1: {}", value1->data());
    HOLOSCAN_LOG_INFO("Middle message value2: {}", value2->data());

    // Multiply the values by the multiplier parameter
    value1->data(value1->data() * multiplier_);
    value2->data(value2->data() * multiplier_);

    op_output.emit(value1, "out1");
    op_output.emit(value2, "out2");
  };

 private:
  int count_ = 1;
  Parameter<int> multiplier_;
};

class PingRxOp : public Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(PingRxOp)

  PingRxOp() = default;

  void setup(OperatorSpec& spec) override {
    // // Since Holoscan SDK v2.3, users can define a multi-receiver input port using 'spec.input()'
    // // with 'IOSpec::kAnySize'.
    // // The old way is to use 'spec.param()' with 'Parameter<std::vector<IOSpec*>> receivers_;'.
    // spec.param(receivers_, "receivers", "Input Receivers", "List of input receivers.", {});
    spec.input<std::vector<std::shared_ptr<ValueData>>>("receivers", IOSpec::kAnySize);
  }

  void compute(InputContext& op_input, OutputContext&, ExecutionContext&) override {
    auto value_vector =
        op_input.receive<std::vector<std::shared_ptr<ValueData>>>("receivers").value();

    HOLOSCAN_LOG_INFO("Rx message received (count: {}, size: {})", count_++, value_vector.size());

    HOLOSCAN_LOG_INFO("Rx message value1: {}", value_vector[0]->data());
    HOLOSCAN_LOG_INFO("Rx message value2: {}", value_vector[1]->data());
  };

 private:
  // // Since Holoscan SDK v2.3, the following line is no longer needed.
  // Parameter<std::vector<IOSpec*>> receivers_;
  int count_ = 1;
};

}  // namespace holoscan::ops

class MyPingApp : public holoscan::Application {
 public:
  void compose() override {
    using namespace holoscan;

    // Define the tx, mx, rx operators, allowing the tx operator to execute 10 times
    auto tx = make_operator<ops::PingTxOp>("tx", make_condition<CountCondition>(10));
    auto mx = make_operator<ops::PingMxOp>("mx", Arg("multiplier", 3));
    auto rx = make_operator<ops::PingRxOp>("rx");

    // Define the workflow
    add_flow(tx, mx, {{"out1", "in1"}, {"out2", "in2"}});
    add_flow(mx, rx, {{"out1", "receivers"}, {"out2", "receivers"}});
  }
};

int main(int argc, char** argv) {
  auto app = holoscan::make_application<MyPingApp>();
  app->run();

  return 0;
}
        

            

            
  void initialize() override {
    // Register custom type and codec for serialization
    register_converter<std::array<float, 3>>();
    register_codec<std::vector<InputSpec>>("std::vector<holoscan::ops::HolovizOp::InputSpec>", true);

    // Set up prerequisite parameters before calling Operator::initialize()
    auto frag = fragment();

    // Check if an argument for 'allocator' exists
    auto has_allocator = std::find_if(
        args().begin(), args().end(), [](const auto& arg) { return (arg.name() == "allocator"); });
    // Create the allocator if no argument is provided
    if (has_allocator == args().end()) {
      allocator_ = frag->make_resource<UnboundedAllocator>("allocator");
      add_arg(allocator_.get());
    }

    // Call the parent class's initialize() method to complete the initialization.
    // Operator::initialize must occur after all arguments have been added.
    Operator::initialize();

    // After Operator::initialize(), the operator is ready for use and the parameters are set
    int multiplier = multiplier_;
    HOLOSCAN_LOG_INFO("Multiplier: {}", multiplier);
  }
        

            

            
Parameter<int> multiplier_;
        

            

            
// Provide key, and optionally other information
spec.param(multiplier_, "multiplier", "Multiplier", "Multiply the input by this value", 2);
        

            

            
  void setup(OperatorSpec& spec) override {
    spec.input<std::shared_ptr<ValueData>>("in");
    // Above statement is equivalent to:
    // spec.input<std::shared_ptr<ValueData>>("in")
    // .condition(ConditionType::kMessageAvailable, Arg("min_size") = static_cast<uint64_t>(1));

    spec.output<std::shared_ptr<ValueData>>("out");
    // Above statement is equivalent to:
    // spec.output<std::shared_ptr<ValueData>>("out")
    // .condition(ConditionType::kDownstreamMessageAffordable, Arg("min_size") = static_cast<uint64_t>(1));
    ...
  }
        

            

            
  void setup(OperatorSpec& spec) override {
    spec.input<std::shared_ptr<ValueData>>("in")
        .condition(ConditionType::kNone);

    spec.output<std::shared_ptr<ValueData>>("out")
        .condition(ConditionType::kNone);
    // ...
  }
        

            

            
// ...

class PingRxOp : public holoscan::Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(PingRxOp)
  PingRxOp() = default;

  void setup(holoscan::OperatorSpec& spec) override {
    spec.input<std::shared_ptr<ValueData>>("in");
  }

  void compute(holoscan::InputContext& op_input, holoscan::OutputContext&,
               holoscan::ExecutionContext&) override {
    auto maybe_value = op_input.receive<std::shared_ptr<ValueData>>("in");

    if (!maybe_value) {
      HOLOSCAN_LOG_ERROR("Failed to receive message - {}", maybe_value.error().what());
      // [error] Failed to receive message - InputContext receive() Error: No message is received from the input port with name 'in'
      throw maybe_value.error(); // or `return;`
    }

    auto& value = maybe_value.value();
    HOLOSCAN_LOG_INFO("Message received (value: {})", value->data());
  }
};
        

            

            
// ...

class TensorRxOp : public holoscan::Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(TensorRxOp)
  TensorRxOp() = default;

  void setup(holoscan::OperatorSpec& spec) override {
    spec.input<holoscan::gxf::Entity>("in");
  }

  void compute(holoscan::InputContext& op_input, holoscan::OutputContext&,
               holoscan::ExecutionContext&) override {
    // Type of 'maybe_entity' is holoscan::expected<holoscan::gxf::Entity, holoscan::RuntimeError>
    auto maybe_entity = op_input.receive<holoscan::gxf::Entity>("in");
    if (maybe_entity) {
      auto& entity = maybe_entity.value();  // holoscan::gxf::Entity&
      // Get a tensor from the entity if it exists.
      // Can pass a tensor name as an argument to get a specific tensor.
      auto tensor = entity.get<holoscan::Tensor>();  // std::shared_ptr<holoscan::Tensor>
      if (tensor) {
        HOLOSCAN_LOG_INFO("tensor nbytes: {}", tensor->nbytes());
      }
    }
  }
};
        

            

            
// ...

class TensorRxOp : public holoscan::Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(TensorRxOp)
  TensorRxOp() = default;

  void setup(holoscan::OperatorSpec& spec) override {
    spec.input<holoscan::TensorMap>("in");
  }

  void compute(holoscan::InputContext& op_input, holoscan::OutputContext&,
               holoscan::ExecutionContext&) override {
    // Type of 'maybe_entity' is holoscan::expected<holoscan::TensorMap, holoscan::RuntimeError>
    auto maybe_tensor_map = op_input.receive<holoscan::TensorMap>("in");
    if (maybe_tensor_map) {
      auto& tensor_map = maybe_tensor_map.value();  // holoscan::TensorMap&
      for (const auto& [name, tensor] : tensor_map) {
        HOLOSCAN_LOG_INFO("tensor name: {}", name);
        HOLOSCAN_LOG_INFO("tensor nbytes: {}", tensor->nbytes());
      }
    }
  }
};
        

            

            
// ...

class AnyRxOp : public holoscan::Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS_SUPER(AnyRxOp, holoscan::ops::GXFOperator)
  AnyRxOp() = default;
  void setup(holoscan::OperatorSpec& spec) override {
    spec.input<std::any>("in");
  }
  void compute(holoscan::InputContext& op_input, holoscan::OutputContext&, holoscan::ExecutionContext&) override {
    auto maybe_any = op_input.receive<std::any>("in");
    if (!maybe_any) {
      HOLOSCAN_LOG_ERROR("Failed to receive message - {}", maybe_any.error().what());
      return;
    }

    auto& in_any = maybe_any.value();
    const auto& in_any_type = in_any.type();

    try {
      if (in_any_type == typeid(holoscan::gxf::Entity)) {
        auto in_entity = std::any_cast<holoscan::gxf::Entity>(in_any);
        auto tensor = in_entity.get<holoscan::Tensor>();  // std::shared_ptr<holoscan::Tensor>
        if (tensor) {
          HOLOSCAN_LOG_INFO("tensor nbytes: {}", tensor->nbytes());
        }
      } else if (in_any_type == typeid(std::shared_ptr<ValueData>)) {
        auto in_value = std::any_cast<std::shared_ptr<ValueData>>(in_any);
        HOLOSCAN_LOG_INFO("Received value: {}", in_value->data());
      } else {
        HOLOSCAN_LOG_ERROR("Invalid message type: {}", in_any_type.name());
      }
    } catch (const std::bad_any_cast& e) {
      HOLOSCAN_LOG_ERROR("Failed to cast message - {}", e.what());
    }
  }
};
        

            

            
  void setup(OperatorSpec& spec) override {
    spec.input<std::vector<std::shared_ptr<ValueData>>>("receivers", IOSpec::kAnySize);
  }
        

            

            
class PingRxOp : public Operator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS(PingRxOp)

  PingRxOp() = default;

  void setup(OperatorSpec& spec) override {
    // // Since Holoscan SDK v2.3, users can define a multi-receiver input port using 'spec.input()'
    // // with 'IOSpec::kAnySize'.
    // // The old way is to use 'spec.param()' with 'Parameter<std::vector<IOSpec*>> receivers_;'.
    // spec.param(receivers_, "receivers", "Input Receivers", "List of input receivers.", {});
    spec.input<std::vector<std::shared_ptr<ValueData>>>("receivers", IOSpec::kAnySize);
  }

  void compute(InputContext& op_input, OutputContext&, ExecutionContext&) override {
    auto value_vector =
        op_input.receive<std::vector<std::shared_ptr<ValueData>>>("receivers").value();

    HOLOSCAN_LOG_INFO("Rx message received (count: {}, size: {})", count_++, value_vector.size());

    HOLOSCAN_LOG_INFO("Rx message value1: {}", value_vector[0]->data());
    HOLOSCAN_LOG_INFO("Rx message value2: {}", value_vector[1]->data());
  };

 private:
  // // Since Holoscan SDK v2.3, the following line is no longer needed.
  // Parameter<std::vector<IOSpec*>> receivers_;
  int count_ = 1;
};

}  // namespace holoscan::ops

class MyPingApp : public holoscan::Application {
 public:
  void compose() override {
    using namespace holoscan;

    // Define the tx, mx, rx operators, allowing the tx operator to execute 10 times
    auto tx = make_operator<ops::PingTxOp>("tx", make_condition<CountCondition>(10));
    auto mx = make_operator<ops::PingMxOp>("mx", Arg("multiplier", 3));
    auto rx = make_operator<ops::PingRxOp>("rx");

    // Define the workflow
    add_flow(tx, mx, {{"out1", "in1"}, {"out2", "in2"}});
    add_flow(mx, rx, {{"out1", "receivers"}, {"out2", "receivers"}});
  }
};
        

            

            
    add_flow(mx, rx, {{"out1", "receivers"}, {"out2", "receivers"}});
        

            

            
    auto value_vector =
        op_input.receive<std::vector<std::shared_ptr<ValueData>>>("receivers").value();
        

            

            
  void setup(holoscan::OperatorSpec& spec) override {
    spec.input<std::shared_ptr<ValueData>>("receivers")
        .connector(holoscan::IOSpec::ConnectorType::kDoubleBuffer,
                   holoscan::Arg("capacity", static_cast<uint64_t>(2)))
        .condition(holoscan::ConditionType::kMessageAvailable,
                   holoscan::Arg("min_size", static_cast<uint64_t>(2)));
  }
        

            

            
  void compute(holoscan::InputContext& op_input, holoscan::OutputContext&,
               holoscan::ExecutionContext&) override {
    std::vector<std::shared_ptr<ValueData>> value_vector;
    auto maybe_value = op_input.receive<std::shared_ptr<ValueData>>("receivers");
    while (maybe_value) {
      value_vector.push_back(maybe_value.value());
      maybe_value = op_input.receive<std::shared_ptr<ValueData>>("receivers");
    }

    HOLOSCAN_LOG_INFO("Rx message received (size: {})", value_vector.size());
  }
        

            

            
  void setup(holoscan::OperatorSpec& spec) override {
    spec.input<std::vector<std::shared_ptr<ValueData>>>("receivers", holoscan::IOSpec::kPrecedingCount);
  }
        

            

            
  void compute(holoscan::InputContext& op_input, holoscan::OutputContext&,
               holoscan::ExecutionContext&) override {
    auto value_vector =
        op_input.receive<std::vector<std::shared_ptr<ValueData>>>("receivers").value();

    HOLOSCAN_LOG_INFO("Rx message received (size: {})", value_vector.size());

    HOLOSCAN_LOG_INFO("Rx message value1: {}", value_vector[0]->data());
    HOLOSCAN_LOG_INFO("Rx message value2: {}", value_vector[1]->data());
  }
        

            

            
  void setup(holoscan::OperatorSpec& spec) override {
    spec.input<std::vector<std::shared_ptr<ValueData>>>("receivers", holoscan::IOSpec::IOSize(2));
  }
        

            

            
  void compute(holoscan::InputContext& op_input, holoscan::OutputContext&,
               holoscan::ExecutionContext&) override {
    while (true) {
      auto maybe_value = op_input.receive<std::shared_ptr<ValueData>>("receivers");
      if (!maybe_value) { break; }
      auto& value = maybe_value.value();
      // Process the input data
      HOLOSCAN_LOG_INFO("Rx message received (value: {})", value->data());
    }
  }
        

            

            
# Your CMake project
cmake_minimum_required(VERSION 3.20)
project(my_project CXX)

# Finds the holoscan SDK
find_package(holoscan REQUIRED CONFIG PATHS "/opt/nvidia/holoscan")

# Create a library for your operator
add_library(my_operator SHARED my_operator.cpp)

# Link your operator against holoscan::core
target_link_libraries(my_operator
    PUBLIC holoscan::core
)
        

            

            
# Configure
cmake -S <src_dir> -B <build_dir> -D Holoscan_ROOT="/opt/nvidia/holoscan"
# Build
cmake --build <build_dir> -j
        

            

            
#ifndef APPS_MY_RECORDER_APP_MY_RECORDER_OP_HPP
#define APPS_MY_RECORDER_APP_MY_RECORDER_OP_HPP

#include "holoscan/core/gxf/gxf_operator.hpp"

namespace holoscan::ops {

class MyRecorderOp : public holoscan::ops::GXFOperator {
 public:
  HOLOSCAN_OPERATOR_FORWARD_ARGS_SUPER(MyRecorderOp, holoscan::ops::GXFOperator)

  MyRecorderOp() = default;

  const char* gxf_typename() const override { return "MyRecorder"; }

  void setup(OperatorSpec& spec) override;

  void initialize() override;

 private:
  Parameter<holoscan::IOSpec*> receiver_;
  Parameter<std::shared_ptr<holoscan::Resource>> my_serializer_;
  Parameter<std::string> directory_;
  Parameter<std::string> basename_;
  Parameter<bool> flush_on_tick_;
};

}  // namespace holoscan::ops

#endif/* APPS_MY_RECORDER_APP_MY_RECORDER_OP_HPP */
        

            

            
  nvidia::gxf::Parameter<nvidia::gxf::Handle<nvidia::gxf::Receiver>> receiver_;
  nvidia::gxf::Parameter<nvidia::gxf::Handle<nvidia::gxf::EntitySerializer>> my_serializer_;
  nvidia::gxf::Parameter<std::string> directory_;
  nvidia::gxf::Parameter<std::string> basename_;
  nvidia::gxf::Parameter<bool> flush_on_tick_;
        

            

            
#include "./my_recorder_op.hpp"

#include "holoscan/core/fragment.hpp"
#include "holoscan/core/gxf/entity.hpp"
#include "holoscan/core/operator_spec.hpp"

#include "holoscan/core/resources/gxf/video_stream_serializer.hpp"

namespace holoscan::ops {

void MyRecorderOp::setup(OperatorSpec& spec) {
  auto& input = spec.input<holoscan::gxf::Entity>("input");
  // Above is same with the following two lines (a default condition is assigned to the input port if not specified):
  //
  // auto& input = spec.input<holoscan::gxf::Entity>("input")
  // .condition(ConditionType::kMessageAvailable, Arg("min_size") = static_cast<uint64_t>(1));

  spec.param(receiver_, "receiver", "Entity receiver", "Receiver channel to log", &input);
  spec.param(my_serializer_,
             "serializer",
             "Entity serializer",
             "Serializer for serializing input data");
  spec.param(directory_, "out_directory", "Output directory path", "Directory path to store received output");
  spec.param(basename_, "basename", "File base name", "User specified file name without extension");
  spec.param(flush_on_tick_,
             "flush_on_tick",
             "Boolean to flush on tick",
             "Flushes output buffer on every `tick` when true",
             false);
}

void MyRecorderOp::initialize() {...}

}  // namespace holoscan::ops
        

            

            
name: recorder
components:
 - name: input
   type: nvidia::gxf::DoubleBufferReceiver
 - name: allocator
   type: nvidia::gxf::UnboundedAllocator
 - name: component_serializer
   type: nvidia::gxf::StdComponentSerializer
   parameters:
     allocator: allocator
 - name: entity_serializer
   type: nvidia::gxf::StdEntitySerializer
   parameters:
     component_serializers: [component_serializer]
 - type: MyRecorder
   parameters:
     receiver: input
     serializer: entity_serializer
     out_directory: "/tmp"
     basename: "tensor_out"
 - type: nvidia::gxf::MessageAvailableSchedulingTerm
   parameters:
     receiver: input
     min_size: 1

        

            

            
  auto& output = spec.output<holoscan::gxf::Entity>("output");
  // Above is same with the following two lines (a default condition is assigned to the output port if not specified):
  //
  // auto& output = spec.output<holoscan::gxf::Entity>("output")
  // .condition(ConditionType::kDownstreamMessageAffordable, Arg("min_size") = static_cast<uint64_t>(1));
        

            

            
#include "./my_recorder_op.hpp"

#include "holoscan/core/fragment.hpp"
#include "holoscan/core/gxf/entity.hpp"
#include "holoscan/core/operator_spec.hpp"

#include "holoscan/core/resources/gxf/video_stream_serializer.hpp"

namespace holoscan::ops {

void MyRecorderOp::setup(OperatorSpec& spec) {...}

void MyRecorderOp::initialize() {
  // Set up prerequisite parameters before calling GXFOperator::initialize()
  auto frag = fragment();
  auto serializer =
      frag->make_resource<holoscan::StdEntitySerializer>("serializer");
  add_arg(Arg("serializer") = serializer);

  GXFOperator::initialize();
}

}  // namespace holoscan::ops
        

            

            
name: recorder
components:
 - name: input
   type: nvidia::gxf::DoubleBufferReceiver
 - name: allocator
   type: nvidia::gxf::UnboundedAllocator
 - name: component_serializer
   type: nvidia::gxf::StdComponentSerializer
   parameters:
     allocator: allocator
 - name: entity_serializer
   type: nvidia::gxf::StdEntitySerializer
   parameters:
     component_serializers: [component_serializer]
 - type: MyRecorder
   parameters:
     receiver: input
     serializer: entity_serializer
     out_directory: "/tmp"
     basename: "tensor_out"
 - type: nvidia::gxf::MessageAvailableSchedulingTerm
   parameters:
     receiver: input
     min_size: 1
        

            

            
  void compute(InputContext& op_input, OutputContext& op_output,
               ExecutionContext& context) override {
    // The type of `in_message` is 'holoscan::TensorMap'.
    auto in_message = op_input.receive<holoscan::TensorMap>("in").value();
    // The type of out_message is TensorMap
    TensorMap out_message;

    for (auto& [key, tensor] : in_message) {  // Process with 'tensor' here.
      cudaError_t cuda_status;
      size_t data_size = tensor->nbytes();
      std::vector<uint8_t> in_data(data_size);
      CUDA_TRY(cudaMemcpy(in_data.data(), tensor->data(), data_size, cudaMemcpyDeviceToHost));
      HOLOSCAN_LOG_INFO("ProcessTensorOp Before key: '{}', shape: ({}), data: [{}]",
                        key,
                        fmt::join(tensor->shape(), ","),
                        fmt::join(in_data, ","));
      for (size_t i = 0; i < data_size; i++) { in_data[i] *= 2; }
      HOLOSCAN_LOG_INFO("ProcessTensorOp After key: '{}', shape: ({}), data: [{}]",
                        key,
                        fmt::join(tensor->shape(), ","),
                        fmt::join(in_data, ","));
      CUDA_TRY(cudaMemcpy(tensor->data(), in_data.data(), data_size, cudaMemcpyHostToDevice));
      out_message.insert({key, tensor});
    }
    // Send the processed message.
    op_output.emit(out_message);
  };
        

            

            
from holoscan.core import (
    ExecutionContext,
    InputContext,
    Operator,
    OperatorSpec,
    OutputContext,
)


class MyOp(Operator):

    def __init__(self, fragment, *args, **kwargs):
        super().__init__(fragment, *args, **kwargs)

    def setup(self, spec: OperatorSpec):
        pass

    def start(self):
        pass

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
        pass

    def stop(self):
        pass
        

            

            
def initialize(self):
        pass
    
        

            

            
import os

from holoscan.conditions import CountCondition
from holoscan.core import Application, Operator, OperatorSpec
from holoscan.logger import LogLevel, set_log_level

import numpy as np


class SignalGeneratorOp(Operator):
    """Generate a time-varying impulse.

Transmits an array of zeros with a single non-zero entry of a
specified `height`. The position of the non-zero entry shifts
to the right (in a periodic fashion) each time `compute` is
called.

Parameters
----------
fragment : holoscan.core.Fragment
The Fragment (or Application) the operator belongs to.
height : number
The height of the signal impulse.
size : number
The total number of samples in the generated 1d signal.
dtype : numpy.dtype or str
The data type of the generated signal.
"""

    def __init__(self, fragment, *args, height=1, size=10, dtype=np.int32, **kwargs):
        self.count = 0
        self.height = height
        self.dtype = dtype
        self.size = size
        super().__init__(fragment, *args, **kwargs)

    def setup(self, spec: OperatorSpec):
        spec.output("signal")

    def compute(self, op_input, op_output, context):

        # single sample wide impulse at a time-varying position
        signal = np.zeros((self.size,), dtype=self.dtype)
        signal[self.count % signal.size] = self.height
        self.count += 1

        op_output.emit(signal, "signal")


class ConvolveOp(Operator):
    """Apply convolution to a tensor.

Convolves an input signal with a "boxcar" (i.e. "rect") kernel.

Parameters
----------
fragment : holoscan.core.Fragment
The Fragment (or Application) the operator belongs to.
width : number
The width of the boxcar kernel used in the convolution.
unit_area : bool, optional
Whether or not to normalize the convolution kernel to unit area.
If False, all samples have implitude one and the dtype of the
kernel will match that of the signal. When True the sum over
the kernel is one and a 32-bit floating point data type is used
for the kernel.
"""

    def __init__(self, fragment, *args, width=4, unit_area=False, **kwargs):
        self.count = 0
        self.width = width
        self.unit_area = unit_area
        super().__init__(fragment, *args, **kwargs)

    def setup(self, spec: OperatorSpec):
        spec.input("signal_in")
        spec.output("signal_out")

    def compute(self, op_input, op_output, context):

        signal = op_input.receive("signal_in")
        assert isinstance(signal, np.ndarray)

        if self.unit_area:
            kernel = np.full((self.width,), 1/self.width, dtype=np.float32)
        else:
            kernel = np.ones((self.width,), dtype=signal.dtype)

        convolved = np.convolve(signal, kernel, mode='same')

        op_output.emit(convolved, "signal_out")


class PrintSignalOp(Operator):
    """Print the received signal to the terminal."""

    def setup(self, spec: OperatorSpec):
        spec.input("signal")

    def compute(self, op_input, op_output, context):
        signal = op_input.receive("signal")
        print(signal)


class ConvolveApp(Application):
    """Minimal signal processing application.

Generates a time-varying impulse, convolves it with a boxcar kernel, and
prints the result to the terminal.

A `CountCondition` is applied to the generate to terminate execution
after a specific number of steps.
"""

    def compose(self):
        signal_generator = SignalGeneratorOp(
            self,
            CountCondition(self, count=24),
            name="generator",
            **self.kwargs("generator"),
        )
        convolver = ConvolveOp(self, name="conv", **self.kwargs("convolve"))
        printer = PrintSignalOp(self, name="printer")
        self.add_flow(signal_generator, convolver)
        self.add_flow(convolver, printer)


def main(config_file):
    app = ConvolveApp()
    # if the --config command line argument was provided, it will override this config_file`
    app.config(config_file)
    app.run()


if __name__ == "__main__":
    config_file = os.path.join(os.path.dirname(__file__), 'convolve.yaml')
    main(config_file=config_file)
        

            

            
signal_generator:
  height: 1
  size: 20
  dtype: int32

convolve:
  width: 4
  unit_area: false
        

            

            
def __init__(self, fragment, *args, width=4, unit_area=False, **kwargs):
    # Internal counter for the time-dependent signal generation
    self.count = 0

    # Parameters
    self.width = width
    self.unit_area = unit_area

    # To forward remaining arguments to any underlying C++ Operator class
    super().__init__(fragment, *args, **kwargs)
        

            

            
def setup(self, spec: OperatorSpec):
        spec.input("in")
        # Above statement is equivalent to:
        # spec.input("in")
        # .condition(ConditionType.MESSAGE_AVAILABLE, min_size = 1)
        spec.output("out")
        # Above statement is equivalent to:
        # spec.output("out")
        # .condition(ConditionType.DOWNSTREAM_MESSAGE_AFFORDABLE, min_size = 1)
    
        

            

            
from holoscan.core import ConditionType, OperatorSpec
# ...
    def setup(self, spec: OperatorSpec):
        spec.input("in").condition(ConditionType.NONE)
        spec.output("out").condition(ConditionType.NONE)
        

            

            
# ...
    def compute(self, op_input, op_output, context):
        msg = op_input.receive("in")
        if msg:
            # Do something with msg
        

            

            
def setup(self, spec: OperatorSpec):
        spec.input("receivers", size=IOSpec.ANY_SIZE)
    
        

            

            
class PingRxOp(Operator):
    """Simple receiver operator.

This operator has:
input: "receivers"

This is an example of a native operator that can dynamically have any
number of inputs connected to is "receivers" port.
"""

    def __init__(self, fragment, *args, **kwargs):
        self.count = 1
        # Need to call the base class constructor last
        super().__init__(fragment, *args, **kwargs)

    def setup(self, spec: OperatorSpec):
        # # Since Holoscan SDK v2.3, users can define a multi-receiver input port using
        # # 'spec.input()' with 'size=IOSpec.ANY_SIZE'.
        # # The old way is to use 'spec.param()' with 'kind="receivers"'.
        # spec.param("receivers", kind="receivers")
        spec.input("receivers", size=IOSpec.ANY_SIZE)

    def compute(self, op_input, op_output, context):
        values = op_input.receive("receivers")
        print(f"Rx message received (count:{self.count}, size:{len(values)})")
        self.count += 1
        print(f"Rx message value1:{values[0].data}")
        print(f"Rx message value2:{values[1].data}")


# Now define a simple application using the operators defined above


class MyPingApp(Application):
    def compose(self):
        # Define the tx, mx, rx operators, allowing the tx operator to execute 10 times
        tx = PingTxOp(self, CountCondition(self, 10), name="tx")
        mx = PingMxOp(self, name="mx", multiplier=3)
        rx = PingRxOp(self, name="rx")

        # Define the workflow
        self.add_flow(tx, mx, {("out1", "in1"), ("out2", "in2")})
        self.add_flow(mx, rx, {("out1", "receivers"), ("out2", "receivers")})

        

            

            
    self.add_flow(mx, rx, {("out1", "receivers"), ("out2", "receivers")})
        

            

            
values = op_input.receive("receivers")
    
        

            

            
def setup(self, spec: OperatorSpec):
        spec.input("receivers").connector(IOSpec.ConnectorType.DOUBLE_BUFFER, capacity=2).condition(
            ConditionType.MESSAGE_AVAILABLE, min_size=2
        )
    
        

            

            
def compute(self, op_input, op_output, context):
        values = []
        value = op_input.receive("receivers")
        while value:
            values.append(value)
            value = op_input.receive("receivers")

        print(f"Rx message received (size:{len(values)})")
    
        

            

            
def setup(self, spec: OperatorSpec):
        spec.input("receivers", size=IOSpec.PRECEDING_COUNT)
    
        

            

            
def compute(self, op_input, op_output, context):
        values = op_input.receive("receivers")

        print(f"Rx message received (size:{len(values)})")

        print(f"Rx message value1:{values[0].data}")
        print(f"Rx message value2:{values[1].data}")
    
        

            

            
def setup(self, spec: OperatorSpec):
        spec.input("receivers", size=IOSpec.IOSize(2))
    
        

            

            
def compute(self, op_input, op_output, context):
        while True:
            value = op_input.receive("receivers", kind="single")
            if value is None:
                break
            # Process the input data
            print(f"Rx message received (value:{value.data})")
    
        

            

            
    def compute(self, op_input, op_output, context):
        # in_message is a dict of tensors
        in_message = op_input.receive("input_tensor")

        # smooth along first two axes, but not the color channels
        sigma = (self.sigma, self.sigma, 0)

        # out_message will be a dict of tensors
        out_message = dict()

        for key, value in in_message.items():
            print(f"message received (count:{self.count})")
            self.count += 1

            cp_array = cp.asarray(value)

            # process cp_array
            cp_array = ndi.gaussian_filter(cp_array, sigma)

            out_message[key] = cp_array

        op_output.emit(out_message, "output_tensor")
        

            

            
out_message = {
            "video": output_array,
            "labels": labels,
            "bbox_coords": bbox_coords,
        }

        # emit the tensors
        op_output.emit(out_message, "outputs")
           
        

            

            
in_message = op_input.receive("inputs")
    # Tensors and tensor names
    video_tensor = in_message["video"]
    labels_tensor = in_message["labels"]
    bbox_coords_tensor = in_message["bbox_coords"]