Program Listing for File gstreamer_pipeline.cpp

↰ Return to documentation for file (Src/gstreamer_pipeline.cpp)

module;

#include "kataglyphis_export.h"
#include <atomic>
#include <cstring>
#include <expected>
#include <gst/analytics/analytics.h>
#include <gst/analytics/gsttensor.h>
#include <gst/analytics/gsttensormeta.h>
#include <gst/app/gstappsink.h>
#include <gst/app/gstappsrc.h>
#include <gst/gst.h>
#include <gst/video/video.h>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

module kataglyphis.gstreamer_pipeline;

import kataglyphis.project_config;

namespace kataglyphis::gstreamer {

namespace {
    std::atomic<bool> g_gstreamer_initialized{ false };
    std::mutex g_gstreamer_mutex;

    auto parse_pipeline_description(const std::string &pipeline_description, const char *failure_context)
      -> std::expected<GstElement *, GStreamerError>
    {
        GError *error = nullptr;
        GstElement *pipeline = gst_parse_launch(pipeline_description.c_str(), &error);
        // gst_parse_launch hands out a floating reference; sinking it makes the later unref the matching release.
        if (pipeline != nullptr) { gst_object_ref_sink(pipeline); }

        if (error != nullptr) {
            g_printerr("%s: %s\n", failure_context, error->message);
            g_error_free(error);
            if (pipeline != nullptr) { gst_object_unref(pipeline); }
            return std::unexpected(GStreamerError::PipelineCreationFailed);
        }

        if (pipeline == nullptr) {
            g_printerr("%s returned null\n", failure_context);
            return std::unexpected(GStreamerError::PipelineCreationFailed);
        }

        return pipeline;
    }

    // The first element of the given type in the pipeline (or the pipeline itself), with a reference the caller owns.
    auto find_element_of_type(GstElement *root, GType type) -> GstElement *
    {
        if (G_TYPE_CHECK_INSTANCE_TYPE(root, type)) { return GST_ELEMENT(gst_object_ref(root)); }
        if (!GST_IS_BIN(root)) { return nullptr; }

        GstElement *found = nullptr;
        GstIterator *iterator = gst_bin_iterate_recurse(GST_BIN(root));
        GValue item = G_VALUE_INIT;
        bool done = false;
        while (!done) {
            switch (gst_iterator_next(iterator, &item)) {
            case GST_ITERATOR_OK: {
                auto *element = GST_ELEMENT(g_value_get_object(&item));
                if (found == nullptr && G_TYPE_CHECK_INSTANCE_TYPE(element, type)) {
                    found = GST_ELEMENT(gst_object_ref(element));
                }
                g_value_reset(&item);
                break;
            }
            case GST_ITERATOR_RESYNC:
                if (found != nullptr) {
                    gst_object_unref(found);
                    found = nullptr;
                }
                gst_iterator_resync(iterator);
                break;
            default:
                done = true;
                break;
            }
        }
        g_value_unset(&item);
        gst_iterator_free(iterator);
        return found;
    }

    void append_tensor_metadata(GstBuffer *buffer, BufferInfo &buffer_info)
    {
        if (auto *meta = gst_buffer_get_tensor_meta(buffer)) {
            buffer_info.tensors.reserve(meta->num_tensors);
            for (gsize i = 0; i < meta->num_tensors; ++i) {
                const GstTensor *tensor = gst_tensor_meta_get(meta, i);
                TensorMeta tensor_meta;
                tensor_meta.tensor_index = i;
                tensor_meta.num_tensors = meta->num_tensors;
                tensor_meta.data_type = static_cast<int>(tensor->data_type);

                gsize num_dims = 0;
                auto *dims = gst_tensor_get_dims(const_cast<GstTensor *>(tensor), &num_dims);
                tensor_meta.dimensions.reserve(num_dims);
                for (gsize j = 0; j < num_dims; ++j) { tensor_meta.dimensions.push_back(dims[j]); }
                buffer_info.tensors.push_back(std::move(tensor_meta));
            }
        }
    }

    void append_frame_metadata(GstCaps *caps, GstBuffer *buffer, BufferInfo &buffer_info)
    {
        if (caps != nullptr) {
            GstStructure *structure = gst_caps_get_structure(caps, 0);
            if (structure != nullptr) {
                // Raw video caps carry width and height as G_TYPE_INT, which gst_structure_get_uint never matches.
                gint width = 0;
                gint height = 0;
                if (gst_structure_get_int(structure, "width", &width) != 0 && width > 0) {
                    buffer_info.metadata.width = static_cast<std::uint32_t>(width);
                }
                if (gst_structure_get_int(structure, "height", &height) != 0 && height > 0) {
                    buffer_info.metadata.height = static_cast<std::uint32_t>(height);
                }

                gint fps_n = 0;
                gint fps_d = 0;
                if (gst_structure_get_fraction(structure, "framerate", &fps_n, &fps_d) != 0 && fps_n >= 0
                    && fps_d > 0) {
                    buffer_info.metadata.fps_n = static_cast<std::uint32_t>(fps_n);
                    buffer_info.metadata.fps_d = static_cast<std::uint32_t>(fps_d);
                }

                const gchar *format = gst_structure_get_string(structure, "format");
                if (format != nullptr) { buffer_info.metadata.format = format; }
            }
        }

        buffer_info.metadata.timestamp_ns = GST_BUFFER_PTS(buffer);
        buffer_info.metadata.duration_ns = GST_BUFFER_DURATION(buffer);
    }

    auto extract_buffer_info(GstSample *sample) -> std::expected<BufferInfo, GStreamerError>
    {
        GstBuffer *buffer = gst_sample_get_buffer(sample);
        if (buffer == nullptr) { return std::unexpected(GStreamerError::BufferAllocationFailed); }

        GstMapInfo map_info;
        if (gst_buffer_map(buffer, &map_info, GST_MAP_READ) == 0) {
            return std::unexpected(GStreamerError::BufferAllocationFailed);
        }

        BufferInfo buffer_info;
        buffer_info.size = map_info.size;
        buffer_info.data.resize(map_info.size);
        std::memcpy(buffer_info.data.data(), map_info.data, map_info.size);

        append_tensor_metadata(buffer, buffer_info);
        append_frame_metadata(gst_sample_get_caps(sample), buffer, buffer_info);

        gst_buffer_unmap(buffer, &map_info);
        return buffer_info;
    }
}// namespace

struct GStreamerPipeline::Impl
{
    GstElement *pipeline{ nullptr };
    GstElement *appsink{ nullptr };
    GstElement *appsrc{ nullptr };

    BufferCallback buffer_callback;
    std::mutex callback_mutex;
    std::atomic<bool> is_playing{ false };
    std::atomic<bool> is_paused{ false };
    std::string caps_string;

    GMainLoop *main_loop{ nullptr };
    std::thread main_loop_thread;
    bool callbacks_installed{ false };

    ~Impl() { cleanup(); }

    void cleanup()
    {
        if (main_loop != nullptr) {
            g_main_loop_quit(main_loop);
            if (main_loop_thread.joinable()) { main_loop_thread.join(); }
            g_main_loop_unref(main_loop);
            main_loop = nullptr;
        }
        if (pipeline != nullptr) {
            gst_element_set_state(pipeline, GST_STATE_NULL);
            wait_for_callbacks();
        }
        // Both hold their own reference, so the pipeline below is only freed once they let go.
        if (appsink != nullptr) {
            if (callbacks_installed) {
                GstAppSinkCallbacks none{};
                gst_app_sink_set_callbacks(GST_APP_SINK(appsink), &none, nullptr, nullptr);
            }
            gst_object_unref(appsink);
            appsink = nullptr;
        }
        callbacks_installed = false;
        if (appsrc != nullptr) {
            gst_object_unref(appsrc);
            appsrc = nullptr;
        }
        if (pipeline != nullptr) {
            gst_object_unref(pipeline);
            pipeline = nullptr;
        }
        is_playing.store(false);
        is_paused.store(false);
    }

    // No callback runs after the NULL state change; the lock makes that ordering visible to TSan as well.
    void wait_for_callbacks() { std::scoped_lock lock(callback_mutex); }

    // The C API, not the appsink signals, which delivered nothing on Windows (run 36902874261).
    void connect_buffer_callback()
    {
        if (appsink == nullptr || callbacks_installed) { return; }
        {
            std::scoped_lock lock(callback_mutex);
            if (!buffer_callback) { return; }
        }
        GstAppSinkCallbacks callbacks{};
        callbacks.new_sample = &Impl::on_new_sample;
        gst_app_sink_set_callbacks(GST_APP_SINK(appsink), &callbacks, this, nullptr);
        callbacks_installed = true;
    }

    // appsink stops the stream on anything but GST_FLOW_OK.
    static auto on_new_sample(GstAppSink *sink, gpointer user_data) -> GstFlowReturn
    {
        auto *self = static_cast<Impl *>(user_data);
        GstSample *sample = gst_app_sink_pull_sample(sink);

        if (sample != nullptr) {
            auto buffer_info = extract_buffer_info(sample);
            if (buffer_info) {
                std::scoped_lock lock(self->callback_mutex);
                if (self->buffer_callback) { self->buffer_callback(*buffer_info); }
            }
            gst_sample_unref(sample);
        }
        return GST_FLOW_OK;
    }

    static auto on_pad_probe([[maybe_unused]] GstPad *pad,
      [[maybe_unused]] GstPadProbeInfo *info,
      [[maybe_unused]] gpointer user_data) -> GstPadProbeReturn
    { return GST_PAD_PROBE_OK; }
};

GStreamerPipeline::GStreamerPipeline() : impl_(std::make_unique<Impl>()) {}

GStreamerPipeline::~GStreamerPipeline() = default;

GStreamerPipeline::GStreamerPipeline(GStreamerPipeline &&) noexcept = default;

auto GStreamerPipeline::operator=(GStreamerPipeline &&) noexcept -> GStreamerPipeline & = default;

auto GStreamerPipeline::initialize_gstreamer(int *argc, char ***argv) -> std::expected<void, GStreamerError>
{

    std::scoped_lock lock(g_gstreamer_mutex);

    if (g_gstreamer_initialized.load()) { return {}; }

    GError *error = nullptr;
    if (gst_init_check(argc, argv, &error) == 0) {
        if (error != nullptr) { g_error_free(error); }
        return std::unexpected(GStreamerError::InitializationFailed);
    }

    g_gstreamer_initialized.store(true);
    return {};
}

auto GStreamerPipeline::deinitialize_gstreamer() -> void
{
    std::scoped_lock lock(g_gstreamer_mutex);
    if (g_gstreamer_initialized.load()) {
        gst_deinit();
        g_gstreamer_initialized.store(false);
    }
}

auto GStreamerPipeline::create_pipeline(const PipelineConfig &config) -> std::expected<void, GStreamerError>
{

    if (!g_gstreamer_initialized.load()) { return std::unexpected(GStreamerError::InitializationFailed); }

    impl_->cleanup();

    auto pipeline = parse_pipeline_description(config.pipeline_description, "Pipeline creation failed");
    if (!pipeline) { return std::unexpected(pipeline.error()); }
    impl_->pipeline = *pipeline;

    // Without these pull_sample, push_buffer and the buffer callback have nothing to talk to.
    impl_->appsink = find_element_of_type(impl_->pipeline, GST_TYPE_APP_SINK);
    impl_->appsrc = find_element_of_type(impl_->pipeline, GST_TYPE_APP_SRC);
    impl_->connect_buffer_callback();

    return {};
}

auto GStreamerPipeline::create_pipeline_from_string(const std::string &description)
  -> std::expected<void, GStreamerError>
{
    PipelineConfig config;
    config.pipeline_description = description;
    return create_pipeline(config);
}

auto GStreamerPipeline::create_inference_pipeline(const std::string &input_source,
  const std::string &model_path,
  const std::vector<std::size_t> & /*input_shape*/,
  const std::string &output_format) -> std::expected<void, GStreamerError>
{

    if (!g_gstreamer_initialized.load()) { return std::unexpected(GStreamerError::InitializationFailed); }

    impl_->cleanup();

    std::string pipeline_desc;

    if (input_source.contains("://")) {
        pipeline_desc = "uridecodebin uri=" + input_source + " ! ";
    } else if (input_source.starts_with("/dev/video")) {
        pipeline_desc = "v4l2src device=" + input_source + " ! ";
    } else {
        pipeline_desc = "filesrc location=" + input_source + " ! decodebin ! ";
    }

    pipeline_desc +=
      "videoconvert ! videoscale ! "
      "video/x-raw,format=RGB,width=640,height=480 ! "
      "tensor_transform mode=transpose option=1:2:0:3 ! "
      "tensor_transform mode=typecast option=float32 ! "
      "onnxruntime model-path="
      + model_path;

    if (output_format == "TENSOR") { pipeline_desc += " ! tensor_decoder mode=direct"; }

    pipeline_desc += " ! appsink name=sink emit-signals=true";

    auto pipeline = parse_pipeline_description(pipeline_desc, "Inference pipeline creation failed");
    if (!pipeline) { return std::unexpected(pipeline.error()); }
    impl_->pipeline = *pipeline;

    impl_->appsink = gst_bin_get_by_name(GST_BIN(impl_->pipeline), "sink");
    if (impl_->appsink == nullptr) { return std::unexpected(GStreamerError::ElementCreationFailed); }

    g_object_set(impl_->appsink, "emit-signals", TRUE, nullptr);
    impl_->connect_buffer_callback();

    return {};
}

auto GStreamerPipeline::start() -> std::expected<void, GStreamerError>
{
    if (impl_->pipeline == nullptr) { return std::unexpected(GStreamerError::PipelineCreationFailed); }

    GstStateChangeReturn ret = gst_element_set_state(impl_->pipeline, GST_STATE_PLAYING);

    if (ret == GST_STATE_CHANGE_FAILURE) { return std::unexpected(GStreamerError::StateChangeFailed); }

    impl_->is_playing.store(true);
    impl_->is_paused.store(false);
    return {};
}

auto GStreamerPipeline::stop() -> std::expected<void, GStreamerError>
{
    if (impl_->pipeline == nullptr) { return {}; }

    GstStateChangeReturn ret = gst_element_set_state(impl_->pipeline, GST_STATE_NULL);
    impl_->wait_for_callbacks();

    if (ret == GST_STATE_CHANGE_FAILURE) { return std::unexpected(GStreamerError::StateChangeFailed); }

    impl_->is_playing.store(false);
    impl_->is_paused.store(false);
    return {};
}

auto GStreamerPipeline::pause() -> std::expected<void, GStreamerError>
{
    if (impl_->pipeline == nullptr) { return std::unexpected(GStreamerError::PipelineCreationFailed); }

    GstStateChangeReturn ret = gst_element_set_state(impl_->pipeline, GST_STATE_PAUSED);

    if (ret == GST_STATE_CHANGE_FAILURE) { return std::unexpected(GStreamerError::StateChangeFailed); }

    impl_->is_paused.store(true);
    return {};
}

auto GStreamerPipeline::resume() -> std::expected<void, GStreamerError> { return start(); }

auto GStreamerPipeline::is_playing() const -> bool { return impl_->is_playing.load() && !impl_->is_paused.load(); }

auto GStreamerPipeline::is_paused() const -> bool { return impl_->is_paused.load(); }

auto GStreamerPipeline::set_buffer_callback(BufferCallback callback) -> void
{
    {
        std::scoped_lock lock(impl_->callback_mutex);
        impl_->buffer_callback = std::move(callback);
    }

    impl_->connect_buffer_callback();
}

auto GStreamerPipeline::pull_sample(std::uint32_t timeout_ms) -> std::expected<BufferInfo, GStreamerError>
{

    if (impl_->appsink == nullptr) { return std::unexpected(GStreamerError::ElementCreationFailed); }

    GstSample *sample = nullptr;

    if (timeout_ms == 0) {
        sample = gst_app_sink_pull_sample(GST_APP_SINK(impl_->appsink));
    } else {
        GstClockTime timeout = static_cast<GstClockTime>(timeout_ms) * GST_MSECOND;
        sample = gst_app_sink_try_pull_sample(GST_APP_SINK(impl_->appsink), timeout);
    }

    if (sample == nullptr) { return std::unexpected(GStreamerError::BufferAllocationFailed); }

    auto buffer_info = extract_buffer_info(sample);
    gst_sample_unref(sample);
    return buffer_info;
}

auto GStreamerPipeline::push_buffer(void *data, std::size_t size, const FrameMetadata &metadata)
  -> std::expected<void, GStreamerError>
{

    if (impl_->appsrc == nullptr) { return std::unexpected(GStreamerError::ElementCreationFailed); }

    GstBuffer *buffer = gst_buffer_new_allocate(nullptr, size, nullptr);
    if (buffer == nullptr) { return std::unexpected(GStreamerError::BufferAllocationFailed); }

    GstMapInfo map_info;
    if (gst_buffer_map(buffer, &map_info, GST_MAP_WRITE) == 0) {
        gst_buffer_unref(buffer);
        return std::unexpected(GStreamerError::BufferAllocationFailed);
    }

    std::memcpy(map_info.data, data, size);
    gst_buffer_unmap(buffer, &map_info);

    GST_BUFFER_PTS(buffer) = metadata.timestamp_ns;
    GST_BUFFER_DURATION(buffer) = metadata.duration_ns;

    // Takes the buffer's reference, unlike the push-buffer signal.
    const GstFlowReturn ret = gst_app_src_push_buffer(GST_APP_SRC(impl_->appsrc), buffer);

    if (ret != GST_FLOW_OK) { return std::unexpected(GStreamerError::StreamError); }

    return {};
}

auto GStreamerPipeline::get_position_ns() const -> std::expected<std::uint64_t, GStreamerError>
{
    if (impl_->pipeline == nullptr) { return std::unexpected(GStreamerError::PipelineCreationFailed); }

    gint64 position = 0;
    if (gst_element_query_position(impl_->pipeline, GST_FORMAT_TIME, &position) == 0) {
        return std::unexpected(GStreamerError::ResourceNotFound);
    }

    return static_cast<std::uint64_t>(position);
}

auto GStreamerPipeline::get_duration_ns() const -> std::expected<std::uint64_t, GStreamerError>
{
    if (impl_->pipeline == nullptr) { return std::unexpected(GStreamerError::PipelineCreationFailed); }

    gint64 duration = 0;
    if (gst_element_query_duration(impl_->pipeline, GST_FORMAT_TIME, &duration) == 0) {
        return std::unexpected(GStreamerError::ResourceNotFound);
    }

    return static_cast<std::uint64_t>(duration);
}

auto GStreamerPipeline::seek(std::uint64_t timestamp_ns) -> std::expected<void, GStreamerError>
{
    if (impl_->pipeline == nullptr) { return std::unexpected(GStreamerError::PipelineCreationFailed); }

    if (gst_element_seek_simple(impl_->pipeline,
          GST_FORMAT_TIME,
          static_cast<GstSeekFlags>(GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_KEY_UNIT),
          static_cast<gint64>(timestamp_ns))
        == 0) {
        return std::unexpected(GStreamerError::ResourceNotFound);
    }

    return {};
}

auto GStreamerPipeline::get_caps_string() const -> std::string { return impl_->caps_string; }

auto GStreamerPipeline::get_current_state() const -> int
{
    if (impl_->pipeline == nullptr) { return GST_STATE_NULL; }

    GstState state;
    gst_element_get_state(impl_->pipeline, &state, nullptr, 0);
    return static_cast<int>(state);
}

auto create_video_inference_pipeline(const std::string &video_source,
  const std::string &model_path,
  std::uint32_t width,
  std::uint32_t height,
  const std::string &output_sink) -> std::expected<GStreamerPipeline, GStreamerError>
{

    auto init_result = GStreamerPipeline::initialize_gstreamer();
    if (!init_result) { return std::unexpected(init_result.error()); }

    GStreamerPipeline pipeline;

    std::string pipeline_desc;

    if (video_source.contains("://")) {
        pipeline_desc = "uridecodebin uri=" + video_source + " ! ";
    } else {
        pipeline_desc = "filesrc location=" + video_source + " ! decodebin ! ";
    }

    pipeline_desc += "videoconvert ! videoscale ! "
        "video/x-raw,format=RGB,width=" + std::to_string(width) +
        ",height=" + std::to_string(height) + " ! "
        "tensor_transform mode=transpose option=1:2:0:3 ! "
        "tensor_transform mode=typecast option=float32 ! "
        "onnxruntime model-path=" + model_path;

    if (output_sink == "appsink") {
        pipeline_desc += " ! appsink name=sink emit-signals=true sync=false";
    } else if (output_sink == "fakesink") {
        pipeline_desc += " ! fakesink sync=false";
    }

    auto result = pipeline.create_pipeline_from_string(pipeline_desc);
    if (!result) { return std::unexpected(result.error()); }

    return pipeline;
}

auto create_camera_inference_pipeline(const std::string &device,
  const std::string &model_path,
  std::uint32_t width,
  std::uint32_t height,
  std::uint32_t fps) -> std::expected<GStreamerPipeline, GStreamerError>
{

    auto init_result = GStreamerPipeline::initialize_gstreamer();
    if (!init_result) { return std::unexpected(init_result.error()); }

    GStreamerPipeline pipeline;

    std::string pipeline_desc =
        "v4l2src device=" + device + " ! "
        "video/x-raw,width=" + std::to_string(width) +
        ",height=" + std::to_string(height) +
        ",framerate=" + std::to_string(fps) + "/1 ! "
        "videoconvert ! "
        "tensor_transform mode=transpose option=1:2:0:3 ! "
        "tensor_transform mode=typecast option=float32 ! "
        "onnxruntime model-path=" + model_path + " ! "
        "appsink name=sink emit-signals=true sync=false";

    auto result = pipeline.create_pipeline_from_string(pipeline_desc);
    if (!result) { return std::unexpected(result.error()); }

    return pipeline;
}

}// namespace kataglyphis::gstreamer