将 opencv 处理添加到 gstreamer 应用程序

Question

我正在尝试执行以下操作：使用 gstreamer 接收视频流并使用 opencv 处理它。我找到了几个解决方案，其中之一是将视频写入（来自gstreamer）fifo，然后使用opencv读取它。 （此处为选项 3 MJPEG streaming and decoding）。问题是我打不开管道。 cvCreateFileCapture 永远不会返回。这是我写的部分代码：

if(mkfifo("fifo.avi", S_IRUSR| S_IWUSR) == -1)
{
    cout<<"Cant create fifo"<<endl;
    cout<<errno<<endl;
}

loop = g_main_loop_new(NULL, false);

fsink = gst_element_factory_make("filesink", "fsink");
g_object_set(G_OBJECT(fsink), "location", "fifo.avi", NULL);

playbin = gst_element_factory_make("playbin2", "play");    
g_object_set(G_OBJECT(playbin), "uri", uri.c_str(), NULL);
g_object_set(G_OBJECT(playbin), "video-sink", fsink, NULL);

bus = gst_pipeline_get_bus(GST_PIPELINE(playbin));
gst_bus_add_signal_watch(bus);

g_signal_connect(bus, "message::buffering", G_CALLBACK(&set_playbin_state), playbin);
gst_object_unref(bus);

cvNamedWindow("output", CV_WINDOW_AUTOSIZE);
capture = cvCreateFileCapture("fifo.avi");

程序堆栈在最后一行。 PS：我使用的是opencv 2.3.1。

Answer 1

我无法评论 Mikael 的帖子（没有足够的积分）。 我在程序中做了一些修改和更正，你可以在这里查看我的要点。 https://gist.github.com/patrickelectric/5dca1cb7cef4ffa7fbb6fb70dd9f9edc

/**
* Based on:
* https://stackoverflow.com/questions/10403588/adding-opencv-processing-to-gstreamer-application
*/

// Include atomic std library
#include <atomic>

// Include gstreamer library
#include <gst/gst.h>
#include <gst/app/app.h>

// Include OpenCV library
#include <opencv.hpp>

// Share frame between main loop and gstreamer callback
std::atomic<cv::Mat*> atomicFrame;

/**
* @brief Check preroll to get a new frame using callback
*  https://gstreamer.freedesktop.org/documentation/design/preroll.html
* @return GstFlowReturn
*/
GstFlowReturn new_preroll(GstAppSink* /*appsink*/, gpointer /*data*/)
{
    return GST_FLOW_OK;
}

/**
* @brief This is a callback that get a new frame when a preroll exist
*
* @param appsink
* @return GstFlowReturn
*/
GstFlowReturn new_sample(GstAppSink *appsink, gpointer /*data*/)
{
    static int framecount = 0;

    // Get caps and frame
    GstSample *sample = gst_app_sink_pull_sample(appsink);
    GstCaps *caps = gst_sample_get_caps(sample);
    GstBuffer *buffer = gst_sample_get_buffer(sample);
    GstStructure *structure = gst_caps_get_structure(caps, 0);
    const int width = g_value_get_int(gst_structure_get_value(structure, "width"));
    const int height = g_value_get_int(gst_structure_get_value(structure, "height"));

    // Print dot every 30 frames
    if(!(framecount%30)) {
        g_print(".");
    }

    // Show caps on first frame
    if(!framecount) {
        g_print("caps: %s\n", gst_caps_to_string(caps));
    }
    framecount++;

    // Get frame data
    GstMapInfo map;
    gst_buffer_map(buffer, &map, GST_MAP_READ);

    // Convert gstreamer data to OpenCV Mat
    cv::Mat* prevFrame;
    prevFrame = atomicFrame.exchange(new cv::Mat(cv::Size(width, height), CV_8UC3, (char*)map.data, cv::Mat::AUTO_STEP));
    if(prevFrame) {
        delete prevFrame;
    }

    gst_buffer_unmap(buffer, &map);
    gst_sample_unref(sample);

    return GST_FLOW_OK;
}

/**
* @brief Bus callback
*  Print important messages
*
* @param bus
* @param message
* @param data
* @return gboolean
*/
static gboolean my_bus_callback(GstBus *bus, GstMessage *message, gpointer data)
{
    // Debug message
    //g_print("Got %s message\n", GST_MESSAGE_TYPE_NAME(message));
    switch(GST_MESSAGE_TYPE(message)) {
        case GST_MESSAGE_ERROR: {
            GError *err;
            gchar *debug;

            gst_message_parse_error(message, &err, &debug);
            g_print("Error: %s\n", err->message);
            g_error_free(err);
            g_free(debug);
            break;
        }
        case GST_MESSAGE_EOS:
            /* end-of-stream */
            break;
        default:
            /* unhandled message */
            break;
    }
    /* we want to be notified again the next time there is a message
    * on the bus, so returning TRUE (FALSE means we want to stop watching
    * for messages on the bus and our callback should not be called again)
    */
    return true;
}

int main(int argc, char *argv[]) {
    gst_init(&argc, &argv);

    gchar *descr = g_strdup(
        "udpsrc port=5600 "
        "! application/x-rtp, payload=96 ! rtph264depay ! h264parse ! avdec_h264 "
        "! decodebin ! videoconvert ! video/x-raw,format=(string)BGR ! videoconvert "
        "! appsink name=sink emit-signals=true sync=false max-buffers=1 drop=true"
    );

    // Check pipeline
    GError *error = nullptr;
    GstElement *pipeline = gst_parse_launch(descr, &error);

    if(error) {
        g_print("could not construct pipeline: %s\n", error->message);
        g_error_free(error);
        exit(-1);
    }

    // Get sink
    GstElement *sink = gst_bin_get_by_name(GST_BIN(pipeline), "sink");

    /**
    * @brief Get sink signals and check for a preroll
    *  If preroll exists, we do have a new frame
    */
    gst_app_sink_set_emit_signals((GstAppSink*)sink, true);
    gst_app_sink_set_drop((GstAppSink*)sink, true);
    gst_app_sink_set_max_buffers((GstAppSink*)sink, 1);
    GstAppSinkCallbacks callbacks = { nullptr, new_preroll, new_sample };
    gst_app_sink_set_callbacks(GST_APP_SINK(sink), &callbacks, nullptr, nullptr);

    // Declare bus
    GstBus *bus;
    bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
    gst_bus_add_watch(bus, my_bus_callback, nullptr);
    gst_object_unref(bus);

    gst_element_set_state(GST_ELEMENT(pipeline), GST_STATE_PLAYING);

    // Main loop
    while(1) {
        g_main_iteration(false);

        cv::Mat* frame = atomicFrame.load();
        if(frame) {
            cv::imshow("Frame", atomicFrame.load()[0]);
            cv::waitKey(30);
        }
    }

    gst_element_set_state(GST_ELEMENT(pipeline), GST_STATE_NULL);
    gst_object_unref(GST_OBJECT(pipeline));
    return 0;
}

Answer 2

这是我针对 Gstreamer 1.4.0 和 OpenCV 2.4.9 的完整源代码解决方案。

它使用gst_parse_launch() 解析您将提供给gst-launch 的普通命令行。 Gstreamer 管道将帧转换为 RGB888 格式，然后再将其输入 OpenCV，以便尽可能轻松地进行转换。

OpenCV 帧处理不在new_sample() 回调中完成，但它只从 gstreamer 抓取帧并将其推送到队列，然后将在主线程中使用。这样我们就可以调用例如来自 OpenCV 的 imshow() 将图像实际渲染到屏幕。

~150 行...通过删除调试打印等可以减少到

可能应该在双端队列读/写周围添加线程同步

#include <gst/gst.h>
#include <gst/app/gstappsink.h>
#include <stdlib.h>

#include "opencv2/opencv.hpp"
using namespace cv;

// TODO: use synchronized deque
std::deque<Mat> frameQueue;

GstFlowReturn
new_preroll(GstAppSink *appsink, gpointer data) {
  g_print ("Got preroll!\n");
  return GST_FLOW_OK;
}

GstFlowReturn
new_sample(GstAppSink *appsink, gpointer data) {
  static int framecount = 0;
  framecount++;

  GstSample *sample = gst_app_sink_pull_sample(appsink);
  GstCaps *caps = gst_sample_get_caps(sample);
  GstBuffer *buffer = gst_sample_get_buffer(sample);
  const GstStructure *info = gst_sample_get_info(sample);

  // ---- Read frame and convert to opencv format ---------------

  GstMapInfo map;
  gst_buffer_map (buffer, &map, GST_MAP_READ);

  // convert gstreamer data to OpenCV Mat, you could actually
  // resolve height / width from caps...
  Mat frame(Size(320, 240), CV_8UC3, (char*)map.data, Mat::AUTO_STEP);
  int frameSize = map.size;

  // TODO: synchronize this....
  frameQueue.push_back(frame);

  gst_buffer_unmap(buffer, &map);

  // ------------------------------------------------------------

  // print dot every 30 frames
  if (framecount%30 == 0) {
    g_print (".");
  }

  // show caps on first frame
  if (framecount == 1) {
    g_print ("%s\n", gst_caps_to_string(caps));
  }

  gst_sample_unref (sample);
  return GST_FLOW_OK;
}

static gboolean
my_bus_callback (GstBus *bus, GstMessage *message, gpointer data) {
  g_print ("Got %s message\n", GST_MESSAGE_TYPE_NAME (message));
  switch (GST_MESSAGE_TYPE (message)) {
    case GST_MESSAGE_ERROR: {
      GError *err;
      gchar *debug;

      gst_message_parse_error (message, &err, &debug);
      g_print ("Error: %s\n", err->message);
      g_error_free (err);
      g_free (debug);    
      break;
    }
    case GST_MESSAGE_EOS:
      /* end-of-stream */
      break;
    default:
      /* unhandled message */
      break;
  }
  /* we want to be notified again the next time there is a message
   * on the bus, so returning TRUE (FALSE means we want to stop watching
   * for messages on the bus and our callback should not be called again)
   */
  return TRUE;
}

int
main (int argc, char *argv[])
{
  GError *error = NULL;

  gst_init (&argc, &argv);

  gchar *descr = g_strdup(
    "videotestsrc  pattern=ball ! "
    "video/x-raw,format=RGB ! "
    "videoconvert ! "
    "appsink name=sink sync=true"
  );
  GstElement *pipeline = gst_parse_launch (descr, &error);

  if (error != NULL) {
    g_print ("could not construct pipeline: %s\n", error->message);
    g_error_free (error);
    exit (-1);
  }

  /* get sink */
  GstElement *sink = gst_bin_get_by_name (GST_BIN (pipeline), "sink");

  gst_app_sink_set_emit_signals((GstAppSink*)sink, true);
  gst_app_sink_set_drop((GstAppSink*)sink, true);
  gst_app_sink_set_max_buffers((GstAppSink*)sink, 1);
  GstAppSinkCallbacks callbacks = { NULL, new_preroll, new_sample };
  gst_app_sink_set_callbacks (GST_APP_SINK(sink), &callbacks, NULL, NULL);

  GstBus *bus;
  guint bus_watch_id;
  bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
  bus_watch_id = gst_bus_add_watch (bus, my_bus_callback, NULL);
  gst_object_unref (bus);

  gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_PLAYING);

  namedWindow("edges",1);
  while(1) {
    g_main_iteration(false);

      // TODO: synchronize...
    if (frameQueue.size() > 0) {
      // this lags pretty badly even when grabbing frames from webcam
      Mat frame = frameQueue.front();
      Mat edges;
      cvtColor(frame, edges, CV_RGB2GRAY);
      GaussianBlur(edges, edges, Size(7,7), 1.5, 1.5);
      Canny(edges, edges, 0, 30, 3);
      imshow("edges", edges);
      cv::waitKey(30);
      frameQueue.clear();
    }
  }  

  gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_NULL);
  gst_object_unref (GST_OBJECT (pipeline));

  return 0;  
}

要在 OSX/Linux 上编译这样的 make 文件应该这样做：

GST_FLAGS=$(shell pkg-config --cflags --libs gstreamer-gl-1.0 gstreamer-tag-1.0 gstreamer-net-1.0 gstreamer-sdp-1.0 \
  gstreamer-1.0 gstreamer-allocators-1.0 gstreamer-insertbin-1.0 gstreamer-plugins-base-1.0 \
  gstreamer-codecparsers-1.0 gstreamer-base-1.0 gstreamer-app-1.0 gstreamer-check-1.0 \
  gstreamer-controller-1.0 gstreamer-video-1.0 gstreamer-fft-1.0 gstreamer-mpegts-1.0 \
  gstreamer-pbutils-1.0 gstreamer-rtp-1.0 gstreamer-rtsp-1.0 \
  gstreamer-riff-1.0 gstreamer-audio-1.0 gstreamer-plugins-bad-1.0 opencv)

OPENCV_FLAGS=$(shell pkg-config --cflags --libs opencv)

all: gst_opencv

gst_opencv: gst_opencv
    g++ $(GST_FLAGS) $(OPENCV_FLAGS) gst_opencv -o gst_opencv

clean:
    rm -f gst_opencv

Answer 3

所以。经过一段时间的搜索，我找到了一个解决方案，它涉及从缓冲区中检索数据。所以想法是创建playbin并将appsink设置为“video-sink”。这是代码示例：

cout<<"Creating appsink"<<endl;
appsink = gst_element_factory_make("appsink", "asink");
gst_app_sink_set_emit_signals((GstAppSink*)appsink, true);
gst_app_sink_set_drop((GstAppSink*)appsink, true);
gst_app_sink_set_max_buffers((GstAppSink*)appsink, 1);

//creating and initialising pipeline

g_object_set(G_OBJECT(playbin), "video-sink", appsink, NULL);

g_signal_connect(appsink, "new-buffer", G_CALLBACK(DisplayFrame), (gpointer) mark);

//callback function looks like this

gboolean Core::DisplayFrame(GstAppSink *fks, gpointer mark)
{
static bool init = false;
static IplImage *frame;
GstBuffer* buf;
Mark* mk = (Mark*) mark;

if(!init)
{
    init = true;
    frame = cvCreateImage(cvSize(mk->frame_w, mk->frame_h), IPL_DEPTH_8U, 1);
}
buf = gst_app_sink_pull_buffer(fks);
frame->imageData = (char*)GST_BUFFER_DATA(buf);

ProcessFrame(frame);
gst_buffer_unref(buf);
return true;
}

这行得通。 PS。有很多关于这种方法的信息，但我花了很多时间寻找它。所以我决定把它贴在这里，以便提供至少一些关键字供搜索。

更新。 还有更多关于连接 gstreamer 和 opencv 的信息。现在是关于将缓冲区转换为 iplimage 的问题。 首先，我们需要接收 rgb 缓冲区，以使转换尽可能简单。为了做到这一点，我们将用连接到 ffmpegcolorspace 的应用程序池替换应用程序池

cout<<"Creating appsink"<<endl;
appsink = gst_element_factory_make("appsink", "asink");
gst_app_sink_set_emit_signals((GstAppSink*)appsink, true);
gst_app_sink_set_drop((GstAppSink*)appsink, true);
gst_app_sink_set_max_buffers((GstAppSink*)appsink, 1);
csp = gst_element_factory_make("ffmpegcolorspace", "csp");
sinkpipe = gst_pipeline_new("sinkp");
gst_bin_add_many(GST_BIN(sinkpipe), csp, appsink, NULL);
gst_element_link_filtered(csp, appsink, gst_caps_new_simple("video/x-raw-rgb", NULL));
pad = gst_element_get_static_pad(csp, "sink");
gst_element_add_pad(sinkpipe, gst_ghost_pad_new("ghost", pad));
g_object_unref(pad);

//...

g_object_set(G_OBJECT(playbin), "video-sink", sinkpipe, NULL);

//...

g_signal_connect(appsink, "new-buffer", G_CALLBACK(GetFrame), (gpointer) mark);

//...

//caps_struct can be retrieved via writing data probe
//search for it in streamer manual

cout<<"Getting frame resolution"<<endl;
gst_structure_get_int(caps_struct, "width", &(mark->frame_w));
gst_structure_get_int(caps_struct, "height", &(mark->frame_h));
gst_structure_get_int(caps_struct, "depth", &depth);

mark->GeneratePoints();
frame = cvCreateImage(cvSize(mark->frame_w, mark->frame_h), depth/3, 3);


//callback function

gboolean Core::GetFrame(GstAppSink *fks, gpointer frame)
{

IplImage* frame_temp = frame
IplImage* frame_temp_two = cvCloneImage(frame_temp);

GstBuffer* buf;
buf = gst_app_sink_pull_buffer(fks);
frame_temp_two->imageData = (char*) GST_BUFFER_DATA(buf);
cvConvertImage(frame_temp_two, frame_temp, CV_CVTIMG_SWAP_RB);
ProcessFrame(frame_temp);
gst_buffer_unref(buf);
return true;
}

我希望这会对某人有所帮助。