Event stream slicing

Events are a sparse representation for brightness changes registered by an event camera. It is useful to group incoming events from a stream in certain time-windows or by a certain number. The dv-processing library provides an efficient, yet powerful tools to approach continuous slicing of events. Following chapters will provide samples on how to use available event stream slicer implementations to group events in specific batches.

EventStreamSlicer

dv::EventStreamSlicer implements event slicing for a single event stream. Incoming events are passed into the slicer using the accept() method, the underlying implementation applies required slicing approach and resulting sliced events are passed into registered callback methods. The slicing can be performed using fixed size time-windows or fixed size number of events. According callback function can be registered using dv::EventStreamSlicer::doEveryNumberOfEvents() or dv::EventStreamSlicer::doEveryTimeInterval() methods. The registered callbacks and their calling parameters can also be modified later on.

The approach can be visualized by looking at a stream of events coming from a camera as a time series. The slicing can be performed by splitting the stream by a fixed time window as shown in a diagram below:

The other approach is to slice by a fixed number of events instead, as shown in the diagram below:

Slicing a stream

Following sample shows how to perform time-based slicing of an event stream:

C++Python

#include <dv-processing/core/core.hpp>
#include <dv-processing/data/generate.hpp>

int main() {
    // Use this namespace to enable literal time expression from the chrono library
    using namespace std::chrono_literals;

    // Initialize slicer, it will have no jobs at this time
    dv::EventStreamSlicer slicer;

    // Register this method to be called every 33 millisecond worth of event data
    slicer.doEveryTimeInterval(33ms, [](const dv::EventStore &events) {
        std::cout << "* Received events time-based slicing: " << events << std::endl;
    });

    // Register this method to be called every 100 events
    slicer.doEveryNumberOfElements(100, [](const dv::EventStore &events) {
        std::cout << "# Received events in number-based slicing: " << events << std::endl;
    });

    // Generate 1000 events within 2 second interval. These will be sliced correctly by the slicer.
    const dv::EventStore store = dv::data::generate::uniformEventsWithinTimeRange(0, 2s, cv::Size(100, 100), 1000);

    // Now push the store into the slicer, the data contents within the store
    // can be arbitrary, the slicer implementation takes care of correct slicing
    // algorithm and calls the previously registered callbacks accordingly.
    slicer.accept(store);

    return 0;
}

Modifying slicing parameters

Slicing parameters can also be modified during runtime if needed. Following code sample is a modified version of the sample from previous chapter, which shows how to use the parameter modification methods:

C++Python

#include <dv-processing/core/core.hpp>
#include <dv-processing/data/generate.hpp>

int main() {
    // Use this namespace to enable literal time expression from the chrono library
    using namespace std::chrono_literals;

    // Initialize slicer, it will have no jobs at this time
    dv::EventStreamSlicer slicer;

    // Register this method to be called every 33 millisecond worth of event data
    int timeJobId = slicer.doEveryTimeInterval(33ms, [](const dv::EventStore &events) {
        std::cout << "* Received events time-based slicing: " << events << std::endl;
    });

    // Register this method to be called every 100 events
    int numberJobId = slicer.doEveryNumberOfElements(100, [](const dv::EventStore &events) {
        std::cout << "# Received events in number-based slicing: " << events << std::endl;
    });

    // Implement data generation; The following loop will generate 10 packets of events,
    // each containing 100 events within 20 millisecond duration.
    for (int i = 0; i < 10; i++) {
        // Generate 100 events within 20 millisecond interval. These will be sliced correctly by the slicer.
        const auto store = dv::data::generate::uniformEventsWithinTimeRange(i * 20'000, 20ms, cv::Size(100, 100), 100);

        // Now push the store into the slicer, the data contents within the store
        // can be arbitrary, the slicer implementation takes care of correct slicing
        // algorithm and calls the previously registered callbacks accordingly.
        slicer.accept(store);

        // When a packet with index 5 is reached, modify the parameters
        if (i == 5) {
            // Modify time range to 10 milliseconds instead of 33
            slicer.modifyTimeInterval(timeJobId, 10ms);
            // Modify number to 200 instead of 100
            slicer.modifyNumberInterval(numberJobId, 200);
        }
    }
}

Stereo stream slicer

A dual stream slicer is required to perform synchronized time slicing from a stereo camera setup. This is implemented in the dv::StereoEventStreamSlicer class. It implements stereo event stream slicing by applying regular stream slicing on one of the input streams and performing time-based slicing within the same time-window. An illustration of the approach for slicing by time is displayed below:

Since the input data can be coming at different rates, in case of slicing by number, the according event from right camera are selected within the same time window. This case is shown below:

In the image above, the slicer is assumed to have a setting to slice every 5 events, these events are sliced from the left camera and their exact timestamps are used to slice events from the right camera.

A sample code on how to use stereo event stream slicer with a live camera is shown below:

C++Python

#include <dv-processing/core/stereo_event_stream_slicer.hpp>
#include <dv-processing/io/stereo_capture.hpp>

int main() {
    using namespace std::chrono_literals;

    // Discover connected camera to the system
    const auto cameras = dv::io::discoverDevices();
    if (cameras.size() < 2) {
        throw dv::exceptions::RuntimeError("Unable to discover two cameras");
    }

    // Open the cameras, just use first detected cameras
    dv::io::StereoCapture stereo(cameras[0], cameras[1]);

    // Initialize a stereo stream slicer
    dv::StereoEventStreamSlicer slicer;

    // Register a job to be performed every 33 milliseconds
    slicer.doEveryTimeInterval(33ms, [](const dv::EventStore &leftEvents, const dv::EventStore &rightEvents) {
        // Print durations for time-based slicing callback
        std::cout << fmt::format(
            "* Received events with duration: left[{}] - right[{}]", leftEvents.duration(), rightEvents.duration())
                  << std::endl;
    });

    // Register a job to be performed every 1000 events
    slicer.doEveryNumberOfEvents(
        // Here we receive events from two camera, time-synchronized
        1000, [](const dv::EventStore &leftEvents, const dv::EventStore &rightEvents) {
            // Print event store sizes for number based slicing callback
            std::cout << fmt::format("# Received events in number-based slicing, counts: left[{}] - right[{}]",
                leftEvents.size(), rightEvents.size())
                      << std::endl;
        });

    // Continue the loop while both cameras are connected
    while (stereo.left.isRunning() && stereo.right.isRunning()) {
        // Initialize
        dv::EventStore left, right;

        // Handle left camera events
        if (const auto batch = stereo.left.getNextEventBatch(); batch.has_value()) {
            left = *batch;
        }

        // Handle right camera events
        if (const auto batch = stereo.right.getNextEventBatch(); batch.has_value()) {
            right = *batch;
        }

        // Pass all events into the slicer
        slicer.accept(left, right);
    }

    return 0;
}

Generic data stream slicing

Event stream slicers described previously are intended for efficient and specific slicing approach to event stream data. The dv-processing library also provides a generic time-series data stream slicer class dv::MultiStreamSlicer that applies the same approach, but supports arbitrary number of streams and arbitrary data-types. The only requirement for data types is the timestamp data, that can be accessed through a predefined API. The predefined API requirement is implemented using C++20 concepts and templates, so the requirement can be satisfied by any type that can provide a microsecond timestamp expressed in a signed 64-bit integer.

The dv::MultiStreamSlicer class provides same slicing capabilities with methods allowing to slice by time dv::MultiStreamSlicer::doEveryTimeInterval() or by number dv::MultiStreamSlicer::doEveryNumberOfElements(). Since the dv::MultiStreamSlicer supports an arbitrary number of streams, the streams are managed by assigning a unique string name to each stream by using the dv::MultiStreamSlicer::addStream() method. By default, the slicer accepts the timestamped data types from the DV flatbuffer type system, such as: dv::EventPacket, dv::IMUPacket, and other packet data types. If a data type is provided without a container, it can be used with STL containers, such as std::vector or dv-processing provided dv::cvector. In a case an image frame dv::Frame stream needs to be sliced, it has to be wrapped in a container, e.g. using dv::cvector: dv::cvector<dv::Frame>. Such a contained and timestamped data can be sliced alongside other data type streams.

To be clear about how the MultiStreamSlicer manages data slicing, it uses a convention of “main stream” and “secondary streams”. Main stream is declared in the constructor of the class and all other stream are added using the MultiStreamSlicer::addStream() method. The main stream is the driving data stream slicing; while secondary streams are following the time-ranges that are sliced from the main stream, similar to how dv::StereoEventStreamSlicer works.

Following sample shows the use of the MultiStreamSlicer to synchronously slice incoming frame and event streams from a DAVIS346 camera and show a preview. The preview below

#include <dv-processing/core/multi_stream_slicer.hpp>
#include <dv-processing/io/camera_capture.hpp>
#include <dv-processing/visualization/event_visualizer.hpp>

#include <opencv2/highgui.hpp>
#include <opencv2/imgproc.hpp>

int main() {
    // Open a DAVIS camera, the sample will slice incoming frames and events synchronously using the generic data
    // stream slicer. The camera name is an empty string to open any DAVIS camera detected on the system
    dv::io::CameraCapture camera("", dv::io::CameraCapture::CameraType::DAVIS);

    // Declare the main stream type to be dv::cvector<dv::Frame> with stream name "frames"
    dv::MultiStreamSlicer<dv::cvector<dv::Frame>> slicer("frames");

    // Add a secondary stream of events
    slicer.addStream<dv::EventStore>("events");

    // It's possible to add additional secondary streams for slicing here, e.g. adding a trigger stream
    // would look like:
    // slicer.addStream<dv::TriggerPacket>("triggers");

    // Use visualizer to overlay events on a frame
    dv::visualization::EventVisualizer visualizer(camera.getEventResolution().value(), dv::visualization::colors::white,
        dv::visualization::colors::green, dv::visualization::colors::red);

    // Declare display windows for frames
    cv::namedWindow("Preview", cv::WINDOW_NORMAL);

    // Register a job to be performed every frame, the event will be sliced in-between the main stream frames
    slicer.doEveryNumberOfElements(
        // Here we receive time-synchronized frames and events; sliced data is provided in a std::unordered_map
        // for easy access using the stream name.
        1, [&visualizer](const auto &data) {
            // Extract events and pass them to the slicer, data is retrieved using the helper method `get()`
            // which accepts the stream name and the type. Stream name and type has to match, otherwise
            // an exception will be thrown
            const auto events = data.template get<dv::EventStore>("events");

            // Retrieve frames, although we get one frame per slice, it is stored in the configured container
            const auto frames = data.template get<dv::cvector<dv::Frame>>("frames");

            // The container is non-empty, we expect only one frame in the container, so just display
            // the first frame in the container
            const dv::Frame &frame = frames.at(0);

            // Convert the frame into a grayscale image for overlay preview
            cv::Mat preview;
            if (frame.image.channels() == 3) {
                // The image is already grayscale, no conversion is needed
                preview = frame.image;
            }
            else {
                // The image coming from the camera is multichannel image, so we can safely assume
                // it's a grayscale image, convert it into BGR for overlay drawing
                cv::cvtColor(frame.image, preview, cv::COLOR_GRAY2BGR);
            }

            // Overlay events on top of preview image
            visualizer.generateImage(events, preview);

            // Display the overlayed image
            cv::imshow("Preview", preview);

            // If escape button is pressed (code 27 is escape key), exit the program cleanly
            if (cv::waitKey(2) == 27) {
                exit(0);
            }
        });

    // Continue the loop while both cameras are connected
    while (camera.isRunning()) {
        // Handle events
        if (const auto events = camera.getNextEventBatch(); events.has_value()) {
            slicer.accept("events", *events);
        }

        // Handle frames
        if (const auto frame = camera.getNextFrame(); frame.has_value()) {
            slicer.accept("frames", *frame);
        }
    }

    return 0;
}

Output from the multi stream sample usage - frames with synchronized events overlayed in the frame image.

The multi stream slicer is also available in python, but due to missing template functionality in python, the python version is more limited. dv.EventMultiStreamSlicer class is provided in python, which only supports events as main stream and has slightly different API, but it supports arbitrary number of secondary streams, which can be of any type that is built-in in dv-processing. Secondary streams can be added using named methods for types.

The example below shows the use of the multi-stream slicer in python which uses events as main stream and slices frames as secondary stream:

import dv_processing as dv
import cv2 as cv
from datetime import timedelta

# Open the camera, just use first detected DAVIS camera
camera = dv.io.CameraCapture("", dv.io.CameraCapture.CameraType.DAVIS)

# Initialize a multi-stream slicer
slicer = dv.EventMultiStreamSlicer("events")

# Add a frame stream to the slicer
slicer.addFrameStream("frames")

# Initialize a visualizer for the overlay
visualizer = dv.visualization.EventVisualizer(camera.getEventResolution(), dv.visualization.colors.white(),
                                              dv.visualization.colors.green(), dv.visualization.colors.red())

# Create a window for image display
cv.namedWindow("Preview", cv.WINDOW_NORMAL)


# Callback method for time based slicing
def display_preview(data):
    # Retrieve frame data using the named method and stream name
    frames = data.getFrames("frames")

    # Retrieve event data
    events = data.getEvents("events")

    # Retrieve and color convert the latest frame of retrieved frames
    latest_image = None
    if len(frames) > 0:
        if len(frames[-1].image.shape) == 3:
            # We already have colored image, no conversion
            latest_image = frames[-1].image
        else:
            # Image is grayscale, convert to color (BGR image)
            latest_image = cv.cvtColor(frames[-1].image, cv.COLOR_GRAY2BGR)
    else:
        return

    # Generate a preview and show the final image
    cv.imshow("Preview", visualizer.generateImage(events, latest_image))

    # If escape button is pressed (code 27 is escape key), exit the program cleanly
    if cv.waitKey(2) == 27:
        exit(0)


# Register a job to be performed every 33 milliseconds
slicer.doEveryTimeInterval(timedelta(milliseconds=33), display_preview)

# Continue the loop while both cameras are connected
while camera.isRunning():
    events = camera.getNextEventBatch()
    if events is not None:
        slicer.accept("events", events)

    frame = camera.getNextFrame()
    if frame is not None:
        slicer.accept("frames", [frame])

This sample will result in similar output to the C++ version, but it rather synchronizes events to frames, so output of the samples are comparable, but not exact.