Using Dust DDS for Edge AI applications – A Face Detection Demo on Jetson Nano

Summary

In Edge AI applications, seamless data communication is essential. Applications that perform real-time tasks need to handle data efficiently, without significant delays or failures. That’s where DDS (Data Distribution Service) comes into play. DDS, with its data-centric architecture and flexible and extensive Quality of Service (QoS) options, is a powerful middleware solution that can meet the demands of such applications. In this article, we’ll walk you through how we used Dust DDS Python bindings, along with GStreamer, to quickly get camera data to an NVIDIA Jetson Nano for a face detection application.

Why DDS for Edge AI

Edge AI typically involves executing machine learning algorithms close to the data source (the “edge” of the network). Devices such as smart cameras, drones, and industrial sensors generate vast amounts of data, and transferring this data to centralized servers for processing can be inefficient or even impractical due to latency and bandwidth concerns.

DDS offers an optimal solution by ensuring low-latency communication between edge devices and processing units. With its distributed topic based system it ensures that the right data is distributed to the right place, avoid unnecessary network traffic. Together with its flexible QoS options it is easy for a user ensure that the right data is used for the AI application and enabling other possibilities such as redundancy and data storage.

Demo: Using Dust DDS for real-time face detection

To demonstrate the interaction between DDS and Jetson Nano for an Edge AI application, we implemented a face detection system. Dust DDS provides a convenient Python API for integrating DDS into your applications. With the bindings, setting up the communication infrastructure is straightforward, allowing you to focus more on the logic and less on the plumbing. The bindings make setting up the communication infrastructure straightforward, allowing you to focus more on your application logic and less on the plumbing. The DDS publish-subscribe model also makes it easier to share data between data publishing devices and processing nodes.

Setting up the system

Single-board computers like the NVIDIA Jetson Nano are a small yet powerful edge computing platform designed for AI applications. With its integrated GPU and efficient processing capabilities, it excels at tasks such as image processing and machine learning. For a real-time face detection system, the Jetson Nano can quickly process video streams from cameras and run inference models without requiring expensive, cloud-based infrastructure.

By using the Jetson Nano in conjunction with DDS, you can create a robust, distributed AI application that makes intelligent decisions right at the edge.

The system we used for this demo is composed of two main components: the video publisher running on an Android phone and the video subscriber running on a Jetson Orin™ Nano developer kit. The image below shows a summary of the system architecture:

Publishing the camera data with Dust DDS and GStreamer

To publish the video data to DDS we used our Video Over DDS (VoDA) application. VoDA makes use of GStreamer to capture the phone camera input allowing you to build efficient media pipelines. The encoded video data is streamed directly to the network with minimal overhead. With GStreamer we encode the data using H.264 and publish it on the VideoStream topic. You can find the full VoDA source code on GitHub

Receiving and processing the data on the Jetson Nano

On the Jetson Nano, we set up a DDS subscriber to receive the camera data and process it for face detection. We utilize the DDS Listener mechanism to minimize latency during data reception. After receiving the data from the DDS reader it is pushed to the GStreamer decoding pipeline. Each decoded sample is transferred to the inference engine for real-time face detection.

The complete code on the subscriber side looks like this.

from dataclasses import dataclass
import dust_dds
import numpy
import sys

from jetson_inference import detectNet
from jetson_utils import cudaFromNumpy, videoOutput

import gi
gi.require_version('Gst', '1.0')
gi.require_version('GstApp', '1.0')
from gi.repository import Gst, GstApp, GLib

# Define the data type used for video transmission
@dataclass
class Video:
    userid: dust_dds.TypeKind.int16
    frame_num: dust_dds.TypeKind.int32
    frame: bytes
    def _key():
        return ["userid"]

# Start the GStreamer pipeline
Gst.init(sys.argv)
pipeline = Gst.parse_launch('appsrc name=app_src ! openh264dec ! videoconvert ! queue leaky=2 max-size-buffers=1 ! appsink name=app_sink sync=false ')
appsrc = pipeline.get_by_name("app_src")
appsrc.set_caps(Gst.Caps.from_string("video/x-h264,stream-format=byte-stream,alignment=au,profile=constrained-baseline"))
appsink = pipeline.get_by_name("app_sink")
appsink.set_caps(Gst.Caps.from_string("video/x-raw,format=RGB"))
appsink.set_emit_signals(True)

output = videoOutput()

# Method to be called when new samples are pushed out of the GStreamer pipeline
def on_new_sample(object, user_data):
    sample = object.pull_sample()
    if sample:
        caps = sample.get_caps()
        gst_buffer = sample.get_buffer()

        try:
            (ret, buffer_map) = gst_buffer.map(Gst.MapFlags.READ)

            height = caps.get_structure(0).get_value("height")
            width = caps.get_structure(0).get_value("width")

            img_numpy = numpy.frombuffer(buffer_map.data, dtype=numpy.uint8)
            img_numpy = img_numpy.reshape((height,width,3))

            img_input = cudaFromNumpy(img_numpy)
            net.Detect(img_input, overlay="box,labels,conf")
            output.Render(img_input)
        finally:
            gst_buffer.unmap(buffer_map)
    return Gst.FlowReturn.OK

# load the segmentation network
net = detectNet("facedetect", 1, 0.5)
appsink.connect("new-sample", on_new_sample, 0)
pipeline.set_state(Gst.State.PLAYING)

# Listener to be installed on the reader to push samples
# into the gstreamer buffer
class MyReaderListener:
    def on_data_available(reader):
        received_data = reader.read(max_samples = 1)
        data = received_data[0].get_data()
        print(f"Received data {data.frame_num}")

        buffer = Gst.Buffer.new_wrapped_bytes(GLib.Bytes(data.frame))
        appsrc.push_buffer(buffer)

# Start the Dust DDS Data Reader
participant_factory = dust_dds.DomainParticipantFactory.get_instance()
participant = participant_factory.create_participant(domain_id = 0)
topic = participant.create_topic(topic_name = "VideoStream", type_ = Video)

subscriber = participant.create_subscriber()
data_reader = subscriber.create_datareader(topic, a_listener = MyReaderListener, mask=[dust_dds.StatusKind.DataAvailable] )

while True:
    pass

The face detection output is displayed directly on a screen connected to the Jetson Nano. Below are examples of the system in action:

Conclusion

By using Dust DDS Python bindings and GStreamer, we were able to quickly and efficiently build a face detection system that processes camera data in real-time on the NVIDIA Jetson Nano. DDS’s data-centric communication model, combined with its flexible QoS options, proved to be a perfect match for the latency-sensitive and resource-constrained nature of edge AI applications.

This setup showcases the power of distributed computing on the edge and provides a scalable approach for building intelligent applications across various industries, from healthcare to IoT devices. Whether you’re processing images, sensor data, or any other form of data in real-time, DDS and GStreamer make it possible to do so with efficiency and reliability.

Contact us if you are looking for support to create your own DDS Edge AI systems or to learn more about how you can use DDS as building block for your edge computing applications.