应用结构

            

            
import hololink as hololink_module

def main():
    # Get handles to GPU
    cuda.cuInit(0)
    cu_device_ordinal = 0
    cu_device = cuda.cuDeviceGet(cu_device_ordinal)
    cu_context = cuda.cuDevicePrimaryCtxRetain(cu_device)

    # Look for sensor bridge enumeration messages; return only the one we're looking for
    channel_metadata = hololink_module.Enumerator.find_channel(channel_ip="192.168.0.2")
    # Use that enumeration data to instantiate a data receiver object
    hololink_channel = hololink_module.DataChannel(channel_metadata)

    # Now that we can communicate, create the camera controller
    camera = hololink_module.sensors.imx274.dual_imx274.Imx274Cam(hololink_channel, ...)

    # Set up our Holoscan pipeline
    application = HoloscanApplication(cu_context, cu_device_ordinal, camera, hololink_channel, ...)
    application.config(...)

    # Connect and initialize the sensor bridge device
    hololink = hololink_channel.hololink()
    hololink.start()  # Establish a connection to the sensor bridge device
    hololink.reset()  # Drive the sensor bridge to a known state

    # Configure the camera for 4k at 60 frames per second
    camera_mode = imx274_mode.Imx274_Mode.IMX274_MODE_3840X2160_60FPS
    camera.setup_clock()
    camera.configure(camera_mode)

    # Run our Holoscan pipeline
    application.run()  # we don't usually return from this call.
    hololink.stop()
        

            

            
#include <hololink/data_channel.hpp>
#include <hololink/enumerator.hpp>
#include <hololink/hololink.hpp>

int main(int argc, char** argv)
{
  // Get handles to GPU
  cuInit(0);
  int cu_device_ordinal = 0;
  CUdevice cu_device;
  cuDeviceGet(&cu_device, cu_device_ordinal);
  CUcontext cu_context;
  cuCtxCreate(&cu_context, 0, cu_device);

  // Look for sensor bridge enumeration messages; return only the one we're looking for
  hololink::Metadata channel_metadata = hololink::Enumerator::find_channel(hololink_ip);
  // Use that enumeration data to instantiate a data receiver object
  hololink::DataChannel hololink_channel(channel_metadata);

  // Import the IMX274 sensor module and the IMX274 mode
  py::module_ imx274 = py::module_::import("hololink.sensors.imx274");
  py::object Imx274Cam = imx274.attr("dual_imx274").attr("Imx274Cam");

  // Now that we can communicate, create the camera controller
  py::object camera = Imx274Cam("hololink_channel"_a = hololink_channel, ...);

  // Set up our Holoscan pipeline
  auto application = holoscan::make_application<HoloscanApplication>(...)
  application->config(...)

  // Connect and initialize the sensor bridge device
  std::shared_ptr<hololink::Hololink> hololink = hololink_channel.hololink();
  hololink->start(); // Establish a connection to the sensor bridge device
  hololink->reset(); // Drive the sensor bridge to a known state

  // Configure the camera for 4k at 60 frames per second
  camera.attr("setup_clock")();
  camera.attr("configure")(Imx274_Mode(0));

  // Run our Holoscan pipeline
  application->run(); // we don't usually return from this call.
  hololink->stop();
}
        

            

            
class HoloscanApplication(holoscan.core.Application):
    def __init__(self, ..., camera, hololink_channel, ...):
        ...
        self._camera = camera
        self._hololink_channel = hololink_channel
        ...
    def compose(self):
        ...
        # Create the CSI to bayer converter.
        csi_to_bayer_operator = hololink_module.operators.CsiToBayerOp(...)

        # The call to camera.configure(...) earlier set our image dimensions
        # and bytes per pixel. This call asks the camera to configure the
        # converter accordingly.
        self._camera.configure_converter(csi_to_bayer_operator)

        # csi_to_bayer_operator now knows the image dimensions and bytes per pixel,
        # and can compute the overall size of the received image data.
        frame_size = csi_to_bayer_operator.get_csi_length()

        # Create a receiver object that fills out our frame buffer. The receiver
        # operator knows how to configure hololink_channel to send its data
        # to us and to provide an end-of-frame indication at the right time.
        receiver_operator = hololink_module.operators.RoceReceiverOp(
            hololink_channel,
            frame_size, ...)
        ...
        # Use add_flow to connect the operators together:
        ...
        # receiver_operator.compute() will be followed by csi_to_bayer_operator.compute()
        self.add_flow(receiver_operator, csi_to_bayer_operator, {("output", "input")})
        ...
        

            

            
class HoloscanApplication : public holoscan::Application {
public:
    explicit HoloscanApplication(..., py::object camera, hololink::DataChannel& hololink_channel, ...)
        : ...
        , camera_(camera)
        , hololink_channel_(hololink_channel)
        ...
    {
    }

    void compose() override
    {
        ...
        // Create the CSI to bayer converter.
        auto csi_to_bayer_operator = make_operator<hololink::operators::CsiToBayerOp>(...);

        // The call to camera.attr("configure")(...) earlier set our image dimensions
        // and bytes per pixel. This call asks the camera to configure the
        // converter accordingly.
        camera_.attr("configure_converter")(csi_to_bayer_operator);

        // csi_to_bayer_operator now knows the image dimensions and bytes per pixel,
        // and can compute the overall size of the received image data.
        const size_t frame_size = csi_to_bayer_operator->get_csi_length();

        // Create a receiver object that fills out our frame buffer. The receiver
        // operator knows how to configure hololink_channel to send its data
        // to us and to provide an end-of-frame indication at the right time.
        auto receiver_operator = make_operator<hololink::operators::RoceReceiverOp>(
            holoscan::Arg("hololink_channel", &hololink_channel_),
            holoscan::Arg("frame_size", frame_size), ...);
        ...
        // Use add_flow to connect the operators together:
        ...
        // receiver_operator.compute() will be followed by csi_to_bayer_operator.compute()
        add_flow(receiver_operator, csi_to_bayer_operator, { { "output", "input" } });
        ...
    }

private:
    const py::object camera_;
    hololink::DataChannel& hololink_channel_;
};
        

            

            
argus_isp = hololink_module.operators.ArgusIspOp(
        self,
        name="argus_isp",
        bayer_format=bayer_format.value, # RGGB or other Bayer format
        exposure_time_ms=16.67,          # Exposure time in milliseconds. 60fps is 16.67ms
        analog_gain=10.0,                # Minimum Analog Gain
        pixel_bit_depth=10,              # Effective bit depth of input per pixel
        pool=isp_pool,
    )