我们可以通过gstreamer代码监控OPENCV打开的gstreamer管道吗？

Question

我正在使用 opencv 打开 gstreamer 管道。现在，只要数据到来，一切都正常。 但是，当由于任何原因，如果数据停止传入，那么管道就会被卡住，并且 opencv 也会因此被卡住。 以下是我正在使用的代码：

#define UDP_URL "udpsrc port=15004 buffer-size=5000000 ! watchdog timeout=1000 ! tsdemux latency=0 ! h264parse ! v4l2h264dec ! imxvideoconvert_g2d ! video/x-raw,format=BGRA,width=1280,height=960 ! appsink max-buffers=2"
int main()
{

    cv::VideoCapture video;
    cv::Mat frame;
    video.open(Stream_URL, cv::CAP_GSTREAMER);
    if (!video.isOpened()) {
        printf("Error in opening.\n");
        return -1;
    }

    while(1) {

        if(video.read(frame))
        {
            // some operation on frame.
        }
        else
            break;
    }

    video.release();
    return 0;
}

在上面的代码中，当端口 15004 上没有数据时，video.read(frame) 函数会卡住，尤其是 v4l2h264dec 解码器。我认为这个解码被卡住了。 当数据再次开始出现时，它仍然停留在相同的功能上。 在 gstreamer 调试期间，我得到“gstreamer 管道处于停止状态”，尽管我可以借助 tcpdump 命令得知数据来自端口 15004。 我正在考虑使用 gstreamer 代码来监视管道，但我不知道如何监视。 我也使用IMX8QM板，gstreamer版本是1.0，opencv版本是4.6.0。

我没有尝试任何监控代码，因为我不知道如何从源文件访问opencv后端。

Answer 1

据我所知，大部分都没有。 OpenCv Gstreamer 后端不会将底层管道的内部公开到 cv VideoCapture 中。 对 VideoCapture 的读取（或抓取）的任何调用都将被阻止。

不过，您可以从 gstreamer 管道获取 cv Mat，而无需使用 cv::VideoCapture。

以下示例在队列上使用 src pad prode 回调将帧放入 cv::Mat 中。它不做任何事情（只是在标准输出上输出一个点），您可以将 Mat 推入缓冲区以供另一个线程进一步处理。

#include <unistd.h>
#include <iostream>

#include <gst/gst.h>
#include <opencv2/opencv.hpp>


const char* gstStr = "udpsrc port=15004 buffer-size=1000000 ! queue ! watchdog timeout=1000 ! queue ! tsdemux ! h264parse ! avdec_h264 ! videorate ! videoconvert ! video/x-raw,format=BGRA,width=1280,height=960,framerate=30/1 ! queue name=my_sink ! fakesink";

static GstPadProbeReturn
cb_have_data (GstPad          *pad,
              GstPadProbeInfo *info,
              gpointer         user_data)
{
  GstMapInfo map;
  GstBuffer *buffer;
  buffer = GST_PAD_PROBE_INFO_BUFFER (info);
  if (gst_buffer_map (buffer, &map, GST_MAP_READ)) {
    cv::Mat frame(960,1280, CV_8UC4, map.data);
    std::cout << "." << std::flush;
    gst_buffer_unmap (buffer, &map);
  }

  return GST_PAD_PROBE_OK;
}


int main(int argc, char **argv)
{
    /* init GStreamer */
    gst_init (&argc, &argv);

    while(1) {
        GstElement *pipeline = gst_parse_launch (gstStr, NULL);
        GstBus *bus = gst_element_get_bus (pipeline);
        GstElement *my_sink = gst_bin_get_by_name(GST_BIN(pipeline), "my_sink");
        GstPad *pad = gst_element_get_static_pad (my_sink, "src");
        gulong probe_id = gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BUFFER,(GstPadProbeCallback) cb_have_data, NULL, NULL);
        gst_object_unref (my_sink);
        
        /* run */
        gst_element_set_state (pipeline, GST_STATE_PLAYING);

        /* wait until it's up and running or failed */
        if (gst_element_get_state (pipeline, NULL, NULL, -1) == GST_STATE_CHANGE_FAILURE) {
          g_print ("Failed to go into PLAYING state... Source may be down\n");
          gst_pad_remove_probe (pad, probe_id);
          gst_object_unref (pad);
          gst_element_set_state (pipeline, GST_STATE_NULL);
          gst_object_unref (bus);
          gst_object_unref (pipeline);
          sleep(3);
          std::cout << "Restarting" << std::endl;
          continue;
        }
        
        g_print ("Running ...\n");
        GstMessage *msg = gst_bus_timed_pop_filtered (bus, GST_CLOCK_TIME_NONE, (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS));
        if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_ERROR) {
           g_print ("An error occurred! May be source is down and watchdog detected that\n");
        }
        gst_pad_remove_probe (pad, probe_id);
        gst_object_unref (pad);
        gst_element_set_state (pipeline, GST_STATE_NULL);
        gst_object_unref (bus);
        gst_object_unref (pipeline);
        sleep(3);
        std::cout << "Restarting" << std::endl;
    }

    return 0;
}