需求：要仔细分析和审核下面第二部分的第二种方案（3）""""   """"内提供的源代码，现在运行""""   """"代码，出现了$$$  $$$所示问题，怎么修改？请最后请给出修改后的完整代码。
下面第一部分是整体功能描述；第二部分是具体介绍，第二种方案（3）是第一种方案（2）的改进，第二种方案（3）运行后的出现了问题。
第一部分：整体功能描述：根据配置文件camera_config.ini的接入两路相机，实现的是2路视频流正常接入，然后2路视频每隔5秒抽帧图片数据形成一张图片，
然后发送给算法服务分析。
第二部分：具体介绍如下：
（1）项目工程文件的配置文件camera_config.ini如{}内所示；
（2）第一种方案：&&&&   &&&&是代码文件video_capture.h和video_capture.cpp;整体，程序正常运行后，cpu占用太高占用太高，高达36.2%。通过top指令查看程序占用的内存和CPU情
。本程序对应的：进程号：770527；%CPU ：36.2；COMMAND ：algorithm_analy 。为什么高呢？初步分析：高CPU使用率是因为解码器解码每一帧而不是每五秒一帧。
（3）第二种方案：第二方案是第一方案的改进。为了降低CPU的使用率，在解码器之前就丢弃不必要的帧，从而减少解码器的工作量。为此，我们可以在解码器之前添加一个Pad Probe，在该Probe中根据时间间隔来丢弃帧。改进后的代码如""""   """"所示，但运行后出现了问题，问题如$$$  $$$所示。
{
[Camera1]
name=room1
rtsp_url=rtsp://117.71.99.41:5544/live/43130100001320000006_43130100001320000006?token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJVSUQiOjAsIlVzZXJuYW1lIjoiNDMxMzAxMDAwMDEzMjAwMDAwMDZfNDMxMzAxMDAwMDEzMjAwMDAwMDYiLCJHcm91cElEIjowLCJpc3MiOiJ4eEBnb2xhbmcuc3BhY2UiLCJleHAiOjE3Mzc3MTI5MDMsImlhdCI6MTczMTQ5MjEwM30.NT9CkDrZVLRZGjxX_DJiCXH0yvCN64D01gr1eXDJHbI&expired=20250124180143
algorithms=peoplecarboat_detector
frame_interval=5000
alarm_strategy=peoplecarboat_detector:boat_present=true,peoplecarboat_detector:illegal_fishing=true
alarm_storage_path=/data/alarms/camera1

[Camera2]
name=room2
rtsp_url=rtsp://117.71.99.41:5544/live/43130100001320000002_43130100001320000002?token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJVSUQiOjAsIlVzZXJuYW1lIjoiNDMxMzAxMDAwMDEzMjAwMDAwMDJfNDMxMzAxMDAwMDEzMjAwMDAwMDIiLCJHcm91cElEIjowLCJpc3MiOiJ4eEBnb2xhbmcuc3BhY2UiLCJleHAiOjE3Mzc3MTI5NDIsImlhdCI6MTczMTQ5MjE0Mn0.VuU_FjJZ0JqevQt3PJjKMhsJjOn4-WU75xf5XYut5as&expired=20250124180222
algorithms=peoplecarboat_detector
frame_interval=5000
alarm_strategy=peoplecarboat_detector:person_count>2,peoplecarboat_detector:vehicle_present=true
alarm_storage_path=/data/alarms/camera2
}

&&&&
1.video_capture.h
#pragma once

#include <string>
#include <memory>
#include <chrono>
#include <gst/gst.h>
#include <gst/video/video.h>
#include <opencv2/opencv.hpp>
#include "config_manager.h"
#include "frame_processor.h"

class VideoCapture {
public:
VideoCapture(const CameraConfig& config);
~VideoCapture();

bool start();
bool stop();

bool isRunning() const;

void setFrameProcessor(std::shared_ptr<FrameProcessor> processor);

private:
static GstFlowReturn onNewSample(GstElement* sink, gpointer data);
static gboolean onBusMessage(GstBus* bus, GstMessage* msg, gpointer data);

void reconnect();

std::chrono::system_clock::time_point last_frame_time_;

bool is_running_;
GstElement* pipeline_;
std::shared_ptr<FrameProcessor> frame_processor_;
CameraConfig config_;

std::mutex reconnect_mutex_;
int reconnect_attempts_;
std::vector<int> reconnect_intervals_;
};
2.video_capture.cpp
#include "video_capture.h"
#include "logger.h"
#include <gst/app/gstappsink.h>
#include <gst/video/video.h>
#include <iomanip>

VideoCapture::VideoCapture(const CameraConfig &config)
: config_(config), pipeline_(nullptr),
is_running_(false),
reconnect_attempts_(0),
last_frame_time_(std::chrono::system_clock::now())
{
if (!gst_is_initialized())
{
gst_init(NULL, NULL); // Initialize GStreamer if not already initialized
}

reconnect_intervals_ = {5, 10, 30, 180}; // 秒
}

VideoCapture::~VideoCapture()
{
stop();
}

void VideoCapture::setFrameProcessor(std::shared_ptr<FrameProcessor> processor)
{
frame_processor_ = processor;
}

bool VideoCapture::start()
{
std::lock_guard<std::mutex> lock(reconnect_mutex_);

if (!frame_processor_)
{
LOG_ERROR("No frame processor set for camera " + config_.name);
return false;
}

if (is_running_)
{
LOG_WARNING("Camera " + config_.name + " is already running");
return true;
}

// 确保清理之前的pipeline
if (pipeline_)
{
gst_element_set_state(pipeline_, GST_STATE_NULL);
gst_object_unref(pipeline_);
pipeline_ = nullptr;
}

GError *error = nullptr;
std::string sink_name = "sink_" + config_.name; // 为每个摄像头的appsink设置唯一名称

// 计算帧率：每隔 frame_interval 毫秒获取一帧
double interval_seconds = config_.frame_interval / 1000.0;
int numerator = 1;
int denominator = static_cast<int>(interval_seconds);

if (denominator == 0)
{
denominator = 1;
}

// 使用硬件解码器，例如 vaapih264dec，如果不可用，可以尝试其他硬件解码器avdec_h264
std::string decoder = "avdec_h264"; // 根据您的系统调整硬件解码器

// 构建GStreamer管道，在解码器之后使用 videorate 来降低帧率
std::ostringstream pipeline_ss;
pipeline_ss << "rtspsrc location=" << config_.rtsp_url << " protocols=tcp latency=300 ! "
<< "rtph264depay ! h264parse ! "
<< decoder << " ! "
<< "videorate ! video/x-raw,framerate=" << numerator << "/" << denominator << " ! "
<< "videoconvert ! appsink name=" << sink_name << " max-buffers=1 drop=true";

std::string pipeline_str = pipeline_ss.str();

pipeline_ = gst_parse_launch(pipeline_str.c_str(), &error);
if (error)
{
LOG_ERROR("Failed to create pipeline for camera " + config_.name + ": " + error->message);
g_error_free(error);
return false;
}

GstElement *sink = gst_bin_get_by_name(GST_BIN(pipeline_), sink_name.c_str());
if (!sink)
{
LOG_ERROR("Failed to get sink element for camera " + config_.name);
gst_object_unref(pipeline_);
pipeline_ = nullptr;
return false;
}

g_object_set(G_OBJECT(sink),
"emit-signals", TRUE,
"sync", FALSE,
"max-buffers", 1,
"drop", TRUE,
NULL);

g_signal_connect(sink, "new-sample", G_CALLBACK(onNewSample), this);
gst_object_unref(sink);

// 处理GStreamer的bus消息，监听错误和EOS
GstBus *bus = gst_element_get_bus(pipeline_);
gst_bus_add_watch(bus, (GstBusFunc)onBusMessage, this);
gst_object_unref(bus);

// 等待pipeline完全启动
GstStateChangeReturn ret = gst_element_set_state(pipeline_, GST_STATE_PLAYING);
if (ret == GST_STATE_CHANGE_FAILURE)
{
LOG_ERROR("Failed to start pipeline for camera " + config_.name);
gst_object_unref(pipeline_);
pipeline_ = nullptr;
return false;
}

// 验证pipeline是否真正启动
GstState state;
ret = gst_element_get_state(pipeline_, &state, nullptr, 5 * GST_SECOND);
if (ret != GST_STATE_CHANGE_SUCCESS || state != GST_STATE_PLAYING)
{
LOG_ERROR("Pipeline failed to reach PLAYING state for camera " + config_.name);
gst_element_set_state(pipeline_, GST_STATE_NULL);
gst_object_unref(pipeline_);
pipeline_ = nullptr;
return false;
}

is_running_ = true;
LOG_INFO("Started video capture for camera " + config_.name);
return true;
}

bool VideoCapture::stop()
{
std::lock_guard<std::mutex> lock(reconnect_mutex_);
if (!is_running_)
return true;

if (pipeline_)
{
gst_element_set_state(pipeline_, GST_STATE_NULL);
gst_object_unref(pipeline_);
pipeline_ = nullptr;
}

is_running_ = false;
return true;
}

bool VideoCapture::isRunning() const
{
return is_running_;
}

GstFlowReturn VideoCapture::onNewSample(GstElement *sink, gpointer data)
{
VideoCapture *self = static_cast<VideoCapture *>(data);
if (!self)
{
g_print("Error: Invalid VideoCapture instance\n");
return GST_FLOW_ERROR;
}

std::cout << "VideoCapture::onNewSample==1 " << self->config_.name << std::endl;


// 获取示例
GstSample *sample_ptr = nullptr;
g_signal_emit_by_name(sink, "pull-sample", &sample_ptr);
if (!sample_ptr)
{
g_print("Error: Failed to pull sample for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// 使用智能指针管理 GstSample
std::unique_ptr<GstSample, decltype(&gst_sample_unref)> sample(sample_ptr, gst_sample_unref);

// 获取缓冲区
GstBuffer *buffer = gst_sample_get_buffer(sample.get());
if (!buffer)
{
g_print("Error: Failed to get buffer from sample for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

try
{
// 获取 caps 和视频格式信息
GstCaps *caps = gst_sample_get_caps(sample.get());
if (!caps)
{
g_print("Error: No caps in sample for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

GstStructure *structure = gst_caps_get_structure(caps, 0);
if (!structure)
{
g_print("Error: No structure in caps for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// 获取视频格式
const gchar *format = gst_structure_get_string(structure, "format");
if (!format)
{
g_print("Error: No format in caps for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// 获取视频尺寸
int width = 0, height = 0;
if (!gst_structure_get_int(structure, "width", &width) ||
!gst_structure_get_int(structure, "height", &height) ||
width <= 0 || height <= 0)
{
g_print("Error: Invalid dimensions (%dx%d) for camera %s\n",
width, height, self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// Map buffer
GstMapInfo map;
if (!gst_buffer_map(buffer, &map, GST_MAP_READ))
{
g_print("Error: Failed to map buffer for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// 使用RAII确保buffer被正确unmap
struct MapGuard
{
GstBuffer *buffer;
GstMapInfo *map;
~MapGuard()
{
if (buffer && map)
gst_buffer_unmap(buffer, map);
}
} mapGuard{buffer, &map};

// 确定正确的 Mat 类型
int cv_type;
if (g_str_equal(format, "RGB"))
{
cv_type = CV_8UC3;
}
else if (g_str_equal(format, "RGBA"))
{
cv_type = CV_8UC4;
}
else if (g_str_equal(format, "GRAY8"))
{
cv_type = CV_8UC1;
}
else if (g_str_equal(format, "BGR"))
{
cv_type = CV_8UC3;
}
else if (g_str_equal(format, "I420"))
{
cv_type = CV_8UC1;
}
else
{
g_print("Error: Unsupported format %s for camera %s\n", format, self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// 安全地创建 Mat
cv::Mat frame;
{
cv::Mat temp(height, width, cv_type, (void *)map.data);
if (temp.empty())
{
g_print("Error: Failed to create temp frame for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// 确保数据被完全复制
frame = temp.clone();
}

if (frame.empty())
{
g_print("Error: Failed to clone frame for camera %s\n", self->config_.name.c_str());
return GST_FLOW_ERROR;
}

// 如果需要，转换为 BGR 格式
if (!g_str_equal(format, "BGR"))
{
cv::Mat bgr_frame;
if (g_str_equal(format, "RGB"))
{
cv::cvtColor(frame, bgr_frame, cv::COLOR_RGB2BGR);
}
else if (g_str_equal(format, "RGBA"))
{
cv::cvtColor(frame, bgr_frame, cv::COLOR_RGBA2BGR);
}
else if (g_str_equal(format, "GRAY8"))
{
cv::cvtColor(frame, bgr_frame, cv::COLOR_GRAY2BGR);
}
else if (g_str_equal(format, "I420"))
{
cv::cvtColor(frame, bgr_frame, cv::COLOR_YUV2BGR_I420);
}
frame = bgr_frame;
}

// 处理帧
if (self->frame_processor_)
{
try
{
self->frame_processor_->processFrame(frame);
}
catch (const std::exception &e)
{
g_print("Error: Frame processing exception for camera %s: %s\n",
self->config_.name.c_str(), e.what());
return GST_FLOW_ERROR;
}
}

return GST_FLOW_OK;
}
catch (const cv::Exception &e)
{
g_print("OpenCV Error in onNewSample for camera %s: %s\n",
self->config_.name.c_str(), e.what());
return GST_FLOW_ERROR;
}
catch (const std::exception &e)
{
g_print("Fatal: Unhandled exception in onNewSample for camera %s: %s\n",
self->config_.name.c_str(), e.what());
return GST_FLOW_ERROR;
}
catch (...)
{
g_print("Fatal: Unknown exception in onNewSample for camera %s\n",
self->config_.name.c_str());
return GST_FLOW_ERROR;
}
}

gboolean VideoCapture::onBusMessage(GstBus *bus, GstMessage *msg, gpointer data)
{
VideoCapture *self = static_cast<VideoCapture *>(data);

switch (GST_MESSAGE_TYPE(msg))
{
case GST_MESSAGE_EOS:
LOG_WARNING("Stream reached EOS for camera " + self->config_.name);
self->reconnect();
break;
case GST_MESSAGE_ERROR:
{
GError *err;
gchar *debug_info;
gst_message_parse_error(msg, &err, &debug_info);
LOG_ERROR("GStreamer error for camera " + self->config_.name + ": " + err->message);
self->reconnect();
g_clear_error(&err);
g_free(debug_info);
break;
}
default:
break;
}
return TRUE;
}

void VideoCapture::reconnect()
{
std::lock_guard<std::mutex> lock(reconnect_mutex_);
if (is_running_)
{
gst_element_set_state(pipeline_, GST_STATE_NULL);
gst_object_unref(pipeline_);
pipeline_ = nullptr;
is_running_ = false;
}

int wait_time = reconnect_attempts_ < reconnect_intervals_.size() ? reconnect_intervals_[reconnect_attempts_] : reconnect_intervals_.back();
LOG_INFO("Attempting to reconnect camera " + config_.name + " in " + std::to_string(wait_time) + " seconds");
std::this_thread::sleep_for(std::chrono::seconds(wait_time));

reconnect_attempts_++;
start();
}
&&&&


【
// video_capture.cpp

#include "video_capture.h"
#include "logger.h"
#include <gst/app/gstappsink.h>
#include <gst/video/video.h>
#include <iomanip>

VideoCapture::VideoCapture(const CameraConfig &config)
    : config_(config), pipeline_(nullptr),
      is_running_(false),
      reconnect_attempts_(0),
      last_frame_time_(std::chrono::steady_clock::now())
{
    if (!gst_is_initialized())
    {
        gst_init(NULL, NULL); // Initialize GStreamer if not already initialized
    }

    reconnect_intervals_ = {5, 10, 30, 180}; // 秒
}

VideoCapture::~VideoCapture()
{
    stop();
}

void VideoCapture::setFrameProcessor(std::shared_ptr<FrameProcessor> processor)
{
    frame_processor_ = processor;
}

bool VideoCapture::start()
{
    std::lock_guard<std::mutex> lock(reconnect_mutex_);

    if (!frame_processor_)
    {
        LOG_ERROR("No frame processor set for camera " + config_.name);
        return false;
    }

    if (is_running_)
    {
        LOG_WARNING("Camera " + config_.name + " is already running");
        return true;
    }

    // 确保清理之前的pipeline
    if (pipeline_)
    {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
    }

    GError *error = nullptr;
    std::string sink_name = "sink_" + config_.name; // 为每个摄像头的appsink设置唯一名称

    // 计算帧率：每隔 frame_interval 毫秒获取一帧
    double interval_seconds = config_.frame_interval / 1000.0;
    int numerator = 1;
    int denominator = static_cast<int>(interval_seconds);

    if (denominator == 0)
    {
        denominator = 1;
    }

    // 使用 decodebin 自动选择解码器
    std::string decoder = "decodebin"; // 使用 decodebin 自动选择解码器

    // **修改点：在管道字符串中对 rtsp_url 进行双引号处理，并使用 decodebin 解码器**
    std::ostringstream pipeline_ss;
    pipeline_ss << "rtspsrc location=\"" << config_.rtsp_url << "\" protocols=tcp latency=300 ! "
                << "rtph264depay ! h264parse ! "
                << decoder << " ! "
                << "videoconvert ! "
                << "videorate ! video/x-raw,framerate=" << numerator << "/" << denominator << " ! "
                << "appsink name=" << sink_name << " max-buffers=1 drop=true";

    std::string pipeline_str = pipeline_ss.str();

    pipeline_ = gst_parse_launch(pipeline_str.c_str(), &error);
    if (error)
    {
        LOG_ERROR("Failed to create pipeline for camera " + config_.name + ": " + error->message);
        g_error_free(error);
        return false;
    }

    GstElement *sink = gst_bin_get_by_name(GST_BIN(pipeline_), sink_name.c_str());
    if (!sink)
    {
        LOG_ERROR("Failed to get sink element for camera " + config_.name);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        return false;
    }

    g_object_set(G_OBJECT(sink),
                 "emit-signals", TRUE,
                 "sync", FALSE,
                 "max-buffers", 1,
                 "drop", TRUE,
                 NULL);

    g_signal_connect(sink, "new-sample", G_CALLBACK(onNewSample), this);
    gst_object_unref(sink);

    // 处理GStreamer的bus消息，监听错误和EOS
    GstBus *bus = gst_element_get_bus(pipeline_);
    gst_bus_add_watch(bus, (GstBusFunc)onBusMessage, this);
    gst_object_unref(bus);

    // 等待pipeline完全启动
    GstStateChangeReturn ret = gst_element_set_state(pipeline_, GST_STATE_PLAYING);
    if (ret == GST_STATE_CHANGE_FAILURE)
    {
        LOG_ERROR("Failed to start pipeline for camera " + config_.name);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        return false;
    }

    // 验证pipeline是否真正启动
    GstState state;
    ret = gst_element_get_state(pipeline_, &state, nullptr, 5 * GST_SECOND);
    if (ret != GST_STATE_CHANGE_SUCCESS || state != GST_STATE_PLAYING)
    {
        LOG_ERROR("Pipeline failed to reach PLAYING state for camera " + config_.name);
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        return false;
    }

    is_running_ = true;
    LOG_INFO("Started video capture for camera " + config_.name);
    return true;
}

bool VideoCapture::stop()
{
    std::lock_guard<std::mutex> lock(reconnect_mutex_);
    if (!is_running_)
        return true;

    if (pipeline_)
    {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
    }

    is_running_ = false;
    return true;
}

bool VideoCapture::isRunning() const
{
    return is_running_;
}

GstFlowReturn VideoCapture::onNewSample(GstElement *sink, gpointer data)
{
    VideoCapture *self = static_cast<VideoCapture *>(data);
    if (!self)
    {
        g_print("Error: Invalid VideoCapture instance\n");
        return GST_FLOW_ERROR;
    }
    
    std::cout << "VideoCapture::onNewSample==1 " << self->config_.name << std::endl;

    // 获取示例
    GstSample *sample_ptr = nullptr;
    g_signal_emit_by_name(sink, "pull-sample", &sample_ptr);
    if (!sample_ptr)
    {
        g_print("Error: Failed to pull sample for camera %s\n", self->config_.name.c_str());
        return GST_FLOW_ERROR;
    }

    // 使用智能指针管理 GstSample
    std::unique_ptr<GstSample, decltype(&gst_sample_unref)> sample(sample_ptr, gst_sample_unref);

    // 获取缓冲区
    GstBuffer *buffer = gst_sample_get_buffer(sample.get());
    if (!buffer)
    {
        g_print("Error: Failed to get buffer from sample for camera %s\n", self->config_.name.c_str());
        return GST_FLOW_ERROR;
    }

    try
    {
        // 获取 caps 和视频格式信息
        GstCaps *caps = gst_sample_get_caps(sample.get());
        if (!caps)
        {
            g_print("Error: No caps in sample for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        GstStructure *structure = gst_caps_get_structure(caps, 0);
        if (!structure)
        {
            g_print("Error: No structure in caps for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 获取视频格式
        const gchar *format = gst_structure_get_string(structure, "format");
        if (!format)
        {
            g_print("Error: No format in caps for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 获取视频尺寸
        int width = 0, height = 0;
        if (!gst_structure_get_int(structure, "width", &width) ||
            !gst_structure_get_int(structure, "height", &height) ||
            width <= 0 || height <= 0)
        {
            g_print("Error: Invalid dimensions (%dx%d) for camera %s\n",
                    width, height, self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // Map buffer
        GstMapInfo map;
        if (!gst_buffer_map(buffer, &map, GST_MAP_READ))
        {
            g_print("Error: Failed to map buffer for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 使用RAII确保buffer被正确unmap
        struct MapGuard
        {
            GstBuffer *buffer;
            GstMapInfo *map;
            ~MapGuard()
            {
                if (buffer && map)
                    gst_buffer_unmap(buffer, map);
            }
        } mapGuard{buffer, &map};

        // 确定正确的 Mat 类型
        int cv_type;
        if (g_str_equal(format, "RGB"))
        {
            cv_type = CV_8UC3;
        }
        else if (g_str_equal(format, "RGBA"))
        {
            cv_type = CV_8UC4;
        }
        else if (g_str_equal(format, "GRAY8"))
        {
            cv_type = CV_8UC1;
        }
        else if (g_str_equal(format, "BGR"))
        {
            cv_type = CV_8UC3;
        }
        else if (g_str_equal(format, "I420"))
        {
            cv_type = CV_8UC1;
        }
        else
        {
            g_print("Error: Unsupported format %s for camera %s\n", format, self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 安全地创建 Mat
        cv::Mat frame;
        {
            cv::Mat temp(height, width, cv_type, (void *)map.data);
            if (temp.empty())
            {
                g_print("Error: Failed to create temp frame for camera %s\n", self->config_.name.c_str());
                return GST_FLOW_ERROR;
            }

            // 确保数据被完全复制
            frame = temp.clone();
        }

        if (frame.empty())
        {
            g_print("Error: Failed to clone frame for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 如果需要，转换为 BGR 格式
        if (!g_str_equal(format, "BGR"))
        {
            cv::Mat bgr_frame;
            if (g_str_equal(format, "RGB"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_RGB2BGR);
            }
            else if (g_str_equal(format, "RGBA"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_RGBA2BGR);
            }
            else if (g_str_equal(format, "GRAY8"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_GRAY2BGR);
            }
            else if (g_str_equal(format, "I420"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_YUV2BGR_I420);
            }
            frame = bgr_frame;
        }

        // 处理帧
        if (self->frame_processor_)
        {
            try
            {
                self->frame_processor_->processFrame(frame);
            }
            catch (const std::exception &e)
            {
                g_print("Error: Frame processing exception for camera %s: %s\n",
                        self->config_.name.c_str(), e.what());
                return GST_FLOW_ERROR;
            }
        }

        return GST_FLOW_OK;
    }
    catch (const cv::Exception &e)
    {
        g_print("OpenCV Error in onNewSample for camera %s: %s\n",
                self->config_.name.c_str(), e.what());
        return GST_FLOW_ERROR;
    }
    catch (const std::exception &e)
    {
        g_print("Fatal: Unhandled exception in onNewSample for camera %s: %s\n",
                self->config_.name.c_str(), e.what());
        return GST_FLOW_ERROR;
    }
    catch (...)
    {
        g_print("Fatal: Unknown exception in onNewSample for camera %s\n",
                self->config_.name.c_str());
        return GST_FLOW_ERROR;
    }
}

gboolean VideoCapture::onBusMessage(GstBus *bus, GstMessage *msg, gpointer data)
{
    VideoCapture *self = static_cast<VideoCapture *>(data);

    switch (GST_MESSAGE_TYPE(msg))
    {
    case GST_MESSAGE_EOS:
        LOG_WARNING("Stream reached EOS for camera " + self->config_.name);
        self->reconnect();
        break;
    case GST_MESSAGE_ERROR:
    {
        GError *err;
        gchar *debug_info;
        gst_message_parse_error(msg, &err, &debug_info);
        LOG_ERROR("GStreamer error for camera " + self->config_.name + ": " + err->message);
        self->reconnect();
        g_clear_error(&err);
        g_free(debug_info);
        break;
    }
    default:
        break;
    }
    return TRUE;
}

void VideoCapture::reconnect()
{
    std::lock_guard<std::mutex> lock(reconnect_mutex_);
    if (is_running_)
    {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        is_running_ = false;
    }

    int wait_time = reconnect_attempts_ < reconnect_intervals_.size() ? reconnect_intervals_[reconnect_attempts_] : reconnect_intervals_.back();
    LOG_INFO("Attempting to reconnect camera " + config_.name + " in " + std::to_string(wait_time) + " seconds");
    std::this_thread::sleep_for(std::chrono::seconds(wait_time));

    reconnect_attempts_++;
    start();
}

GstPadProbeReturn VideoCapture::buffer_probe_callback(GstPad *pad, GstPadProbeInfo *info, gpointer user_data)
{
    VideoCapture *self = static_cast<VideoCapture *>(user_data);

    GstPadProbeReturn ret = GST_PAD_PROBE_OK;

    GstBuffer *buffer = GST_PAD_PROBE_INFO_BUFFER(info);

    if (buffer)
    {
        auto now = std::chrono::steady_clock::now();
        auto time_since_last_frame = std::chrono::duration_cast<std::chrono::milliseconds>(now - self->last_frame_time_);

        if (time_since_last_frame.count() < self->config_.frame_interval)
        {
            // 丢弃该buffer
            return GST_PAD_PROBE_DROP;
        }
        else
        {
            self->last_frame_time_ = now;
            return GST_PAD_PROBE_OK;
        }
    }

    return ret;
}
】

""""
1.video_capture.h

#pragma once

#include <string>
#include <memory>
#include <chrono>
#include <gst/gst.h>
#include <gst/video/video.h>
#include <opencv2/opencv.hpp>
#include "config_manager.h"
#include "frame_processor.h"
#include <mutex>

class VideoCapture {
public:
    VideoCapture(const CameraConfig& config);
    ~VideoCapture();

    bool start();
    bool stop();

    bool isRunning() const;

    void setFrameProcessor(std::shared_ptr<FrameProcessor> processor);

private:
    static GstFlowReturn onNewSample(GstElement* sink, gpointer data);
    static gboolean onBusMessage(GstBus* bus, GstMessage* msg, gpointer data);
    static GstPadProbeReturn buffer_probe_callback(GstPad* pad, GstPadProbeInfo* info, gpointer user_data);

    void reconnect();

    std::chrono::steady_clock::time_point last_frame_time_;

    bool is_running_;
    GstElement* pipeline_;
    std::shared_ptr<FrameProcessor> frame_processor_;
    CameraConfig config_;

    std::mutex reconnect_mutex_;
    int reconnect_attempts_;
    std::vector<int> reconnect_intervals_;
};

2.video_capture.cpp

#include "video_capture.h"
#include "logger.h"
#include <gst/app/gstappsink.h>
#include <gst/video/video.h>
#include <iomanip>

VideoCapture::VideoCapture(const CameraConfig &config)
    : config_(config), pipeline_(nullptr),
      is_running_(false),
      reconnect_attempts_(0),
      last_frame_time_(std::chrono::steady_clock::now())
{
    if (!gst_is_initialized())
    {
        gst_init(NULL, NULL); // Initialize GStreamer if not already initialized
    }

    reconnect_intervals_ = {5, 10, 30, 180}; // 秒
}

VideoCapture::~VideoCapture()
{
    stop();
}

void VideoCapture::setFrameProcessor(std::shared_ptr<FrameProcessor> processor)
{
    frame_processor_ = processor;
}

bool VideoCapture::start()
{
    std::lock_guard<std::mutex> lock(reconnect_mutex_);

    if (!frame_processor_)
    {
        LOG_ERROR("No frame processor set for camera " + config_.name);
        return false;
    }

    if (is_running_)
    {
        LOG_WARNING("Camera " + config_.name + " is already running");
        return true;
    }

    // 确保清理之前的pipeline
    if (pipeline_)
    {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
    }

    GError *error = nullptr;
    std::string sink_name = "sink_" + config_.name; // 为每个摄像头的appsink设置唯一名称

    // 使用硬件解码器，例如 vaapih264dec，如果不可用，可以尝试其他硬件解码器avdec_h264
    std::string decoder = "avdec_h264"; // 根据您的系统调整硬件解码器

    // 构建GStreamer管道，在解码器之前添加Pad Probe来降低帧率
    std::ostringstream pipeline_ss;
    pipeline_ss << "rtspsrc location=" << config_.rtsp_url << " protocols=tcp latency=300 ! "
                << "rtph264depay name=depay ! h264parse ! "
                << decoder << " ! "
                << "videoconvert ! appsink name=" << sink_name << " max-buffers=1 drop=true";

    std::string pipeline_str = pipeline_ss.str();

    pipeline_ = gst_parse_launch(pipeline_str.c_str(), &error);
    if (error)
    {
        LOG_ERROR("Failed to create pipeline for camera " + config_.name + ": " + error->message);
        g_error_free(error);
        return false;
    }

    GstElement *sink = gst_bin_get_by_name(GST_BIN(pipeline_), sink_name.c_str());
    if (!sink)
    {
        LOG_ERROR("Failed to get sink element for camera " + config_.name);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        return false;
    }

    g_object_set(G_OBJECT(sink),
                 "emit-signals", TRUE,
                 "sync", FALSE,
                 "max-buffers", 1,
                 "drop", TRUE,
                 NULL);

    g_signal_connect(sink, "new-sample", G_CALLBACK(onNewSample), this);
    gst_object_unref(sink);

    // 处理GStreamer的bus消息，监听错误和EOS
    GstBus *bus = gst_element_get_bus(pipeline_);
    gst_bus_add_watch(bus, (GstBusFunc)onBusMessage, this);
    gst_object_unref(bus);

    // 添加Pad Probe来降低帧率
    GstElement *depay = gst_bin_get_by_name(GST_BIN(pipeline_), "depay");
    if (!depay)
    {
        LOG_ERROR("Failed to get depay element for camera " + config_.name);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        return false;
    }

    GstPad *depay_src_pad = gst_element_get_static_pad(depay, "src");
    gst_pad_add_probe(depay_src_pad, GST_PAD_PROBE_TYPE_BUFFER, buffer_probe_callback, this, NULL);
    gst_object_unref(depay_src_pad);
    gst_object_unref(depay);

    // 等待pipeline完全启动
    GstStateChangeReturn ret = gst_element_set_state(pipeline_, GST_STATE_PLAYING);
    if (ret == GST_STATE_CHANGE_FAILURE)
    {
        LOG_ERROR("Failed to start pipeline for camera " + config_.name);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        return false;
    }

    // 验证pipeline是否真正启动
    GstState state;
    ret = gst_element_get_state(pipeline_, &state, nullptr, 5 * GST_SECOND);
    if (ret != GST_STATE_CHANGE_SUCCESS || state != GST_STATE_PLAYING)
    {
        LOG_ERROR("Pipeline failed to reach PLAYING state for camera " + config_.name);
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        return false;
    }

    is_running_ = true;
    LOG_INFO("Started video capture for camera " + config_.name);
    return true;
}

bool VideoCapture::stop()
{
    std::lock_guard<std::mutex> lock(reconnect_mutex_);
    if (!is_running_)
        return true;

    if (pipeline_)
    {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
    }

    is_running_ = false;
    return true;
}

bool VideoCapture::isRunning() const
{
    return is_running_;
}

GstFlowReturn VideoCapture::onNewSample(GstElement *sink, gpointer data)
{
    VideoCapture *self = static_cast<VideoCapture *>(data);
    if (!self)
    {
        g_print("Error: Invalid VideoCapture instance\n");
        return GST_FLOW_ERROR;
    }

    // 获取示例
    GstSample *sample_ptr = nullptr;
    g_signal_emit_by_name(sink, "pull-sample", &sample_ptr);
    if (!sample_ptr)
    {
        g_print("Error: Failed to pull sample for camera %s\n", self->config_.name.c_str());
        return GST_FLOW_ERROR;
    }

    // 使用智能指针管理 GstSample
    std::unique_ptr<GstSample, decltype(&gst_sample_unref)> sample(sample_ptr, gst_sample_unref);

    // 获取缓冲区
    GstBuffer *buffer = gst_sample_get_buffer(sample.get());
    if (!buffer)
    {
        g_print("Error: Failed to get buffer from sample for camera %s\n", self->config_.name.c_str());
        return GST_FLOW_ERROR;
    }

    try
    {
        // 获取 caps 和视频格式信息
        GstCaps *caps = gst_sample_get_caps(sample.get());
        if (!caps)
        {
            g_print("Error: No caps in sample for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        GstStructure *structure = gst_caps_get_structure(caps, 0);
        if (!structure)
        {
            g_print("Error: No structure in caps for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 获取视频格式
        const gchar *format = gst_structure_get_string(structure, "format");
        if (!format)
        {
            g_print("Error: No format in caps for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 获取视频尺寸
        int width = 0, height = 0;
        if (!gst_structure_get_int(structure, "width", &width) ||
            !gst_structure_get_int(structure, "height", &height) ||
            width <= 0 || height <= 0)
        {
            g_print("Error: Invalid dimensions (%dx%d) for camera %s\n",
                    width, height, self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // Map buffer
        GstMapInfo map;
        if (!gst_buffer_map(buffer, &map, GST_MAP_READ))
        {
            g_print("Error: Failed to map buffer for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 使用RAII确保buffer被正确unmap
        struct MapGuard
        {
            GstBuffer *buffer;
            GstMapInfo *map;
            ~MapGuard()
            {
                if (buffer && map)
                    gst_buffer_unmap(buffer, map);
            }
        } mapGuard{buffer, &map};

        // 确定正确的 Mat 类型
        int cv_type;
        if (g_str_equal(format, "RGB"))
        {
            cv_type = CV_8UC3;
        }
        else if (g_str_equal(format, "RGBA"))
        {
            cv_type = CV_8UC4;
        }
        else if (g_str_equal(format, "GRAY8"))
        {
            cv_type = CV_8UC1;
        }
        else if (g_str_equal(format, "BGR"))
        {
            cv_type = CV_8UC3;
        }
        else if (g_str_equal(format, "I420"))
        {
            cv_type = CV_8UC1;
        }
        else
        {
            g_print("Error: Unsupported format %s for camera %s\n", format, self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 安全地创建 Mat
        cv::Mat frame;
        {
            cv::Mat temp(height, width, cv_type, (void *)map.data);
            if (temp.empty())
            {
                g_print("Error: Failed to create temp frame for camera %s\n", self->config_.name.c_str());
                return GST_FLOW_ERROR;
            }

            // 确保数据被完全复制
            frame = temp.clone();
        }

        if (frame.empty())
        {
            g_print("Error: Failed to clone frame for camera %s\n", self->config_.name.c_str());
            return GST_FLOW_ERROR;
        }

        // 如果需要，转换为 BGR 格式
        if (!g_str_equal(format, "BGR"))
        {
            cv::Mat bgr_frame;
            if (g_str_equal(format, "RGB"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_RGB2BGR);
            }
            else if (g_str_equal(format, "RGBA"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_RGBA2BGR);
            }
            else if (g_str_equal(format, "GRAY8"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_GRAY2BGR);
            }
            else if (g_str_equal(format, "I420"))
            {
                cv::cvtColor(frame, bgr_frame, cv::COLOR_YUV2BGR_I420);
            }
            frame = bgr_frame;
        }

        // 处理帧
        if (self->frame_processor_)
        {
            try
            {
                self->frame_processor_->processFrame(frame);
            }
            catch (const std::exception &e)
            {
                g_print("Error: Frame processing exception for camera %s: %s\n",
                        self->config_.name.c_str(), e.what());
                return GST_FLOW_ERROR;
            }
        }

        return GST_FLOW_OK;
    }
    catch (const cv::Exception &e)
    {
        g_print("OpenCV Error in onNewSample for camera %s: %s\n",
                self->config_.name.c_str(), e.what());
        return GST_FLOW_ERROR;
    }
    catch (const std::exception &e)
    {
        g_print("Fatal: Unhandled exception in onNewSample for camera %s: %s\n",
                self->config_.name.c_str(), e.what());
        return GST_FLOW_ERROR;
    }
    catch (...)
    {
        g_print("Fatal: Unknown exception in onNewSample for camera %s\n",
                self->config_.name.c_str());
        return GST_FLOW_ERROR;
    }
}

gboolean VideoCapture::onBusMessage(GstBus *bus, GstMessage *msg, gpointer data)
{
    VideoCapture *self = static_cast<VideoCapture *>(data);

    switch (GST_MESSAGE_TYPE(msg))
    {
    case GST_MESSAGE_EOS:
        LOG_WARNING("Stream reached EOS for camera " + self->config_.name);
        self->reconnect();
        break;
    case GST_MESSAGE_ERROR:
    {
        GError *err;
        gchar *debug_info;
        gst_message_parse_error(msg, &err, &debug_info);
        LOG_ERROR("GStreamer error for camera " + self->config_.name + ": " + err->message);
        self->reconnect();
        g_clear_error(&err);
        g_free(debug_info);
        break;
    }
    default:
        break;
    }
    return TRUE;
}

void VideoCapture::reconnect()
{
    std::lock_guard<std::mutex> lock(reconnect_mutex_);
    if (is_running_)
    {
        gst_element_set_state(pipeline_, GST_STATE_NULL);
        gst_object_unref(pipeline_);
        pipeline_ = nullptr;
        is_running_ = false;
    }

    int wait_time = reconnect_attempts_ < reconnect_intervals_.size() ? reconnect_intervals_[reconnect_attempts_] : reconnect_intervals_.back();
    LOG_INFO("Attempting to reconnect camera " + config_.name + " in " + std::to_string(wait_time) + " seconds");
    std::this_thread::sleep_for(std::chrono::seconds(wait_time));

    reconnect_attempts_++;
    start();
}

GstPadProbeReturn VideoCapture::buffer_probe_callback(GstPad *pad, GstPadProbeInfo *info, gpointer user_data)
{
    VideoCapture *self = static_cast<VideoCapture *>(user_data);

    GstPadProbeReturn ret = GST_PAD_PROBE_OK;

    GstBuffer *buffer = GST_PAD_PROBE_INFO_BUFFER(info);

    if (buffer)
    {
        auto now = std::chrono::steady_clock::now();
        auto time_since_last_frame = std::chrono::duration_cast<std::chrono::milliseconds>(now - self->last_frame_time_);

        if (time_since_last_frame.count() < self->config_.frame_interval)
        {
            // 丢弃该buffer
            return GST_PAD_PROBE_DROP;
        }
        else
        {
            self->last_frame_time_ = now;
            return GST_PAD_PROBE_OK;
        }
    }

    return ret;
}

""""
运行日志：
$$$
(triton) (base) chen@chen-Precision-3660:~/data/algorithm_analysis_service/build$ export GST_DEBUG=3
(triton) (base) chen@chen-Precision-3660:~/data/algorithm_analysis_service/build$ ./algorithm_analysis_service
2024-11-17 23:28:04 [INFO] Starting Video Analysis Service...
2024-11-17 23:28:04 [INFO] Configuration loaded successfully
2024-11-17 23:28:04 [INFO] Loaded camera config:
2024-11-17 23:28:04 [INFO]   Camera Name: room1
2024-11-17 23:28:04 [INFO]   RTSP URL: rtsp://117.71.99.41:5544/live/43130100001320000006_43130100001320000006?token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJVSUQiOjAsIlVzZXJuYW1lIjoiNDMxMzAxMDAwMDEzMjAwMDAwMDZfNDMxMzAxMDAwMDEzMjAwMDAwMDYiLCJHcm91cElEIjowLCJpc3MiOiJ4eEBnb2xhbmcuc3BhY2UiLCJleHAiOjE3Mzc3MTI5MDMsImlhdCI6MTczMTQ5MjEwM30.NT9CkDrZVLRZGjxX_DJiCXH0yvCN64D01gr1eXDJHbI&expired=20250124180143
constuct FrameProcessor: triton_client_
0:00:00.858985266 788174 0x7f7e50006520 FIXME               basesink gstbasesink.c:3246:gst_base_sink_default_event:<sink_room1> stream-start event without group-id. Consider implementing group-id handling in the upstream elements
2024-11-17 23:28:09 [ERROR] Pipeline failed to reach PLAYING state for camera room1
0:00:05.032163744 788174 0x7f7e50006520 ERROR                  libav gstavviddec.c:1319:gst_ffmpegviddec_negotiate:<avdec_h264-0> negotiation failed
0:00:05.033823021 788174 0x55a599059980 WARN                 rtspsrc gstrtspsrc.c:6326:gst_rtsp_src_receive_response:<rtspsrc0> receive interrupted
0:00:05.033849328 788174 0x55a599059980 WARN                 rtspsrc gstrtspsrc.c:6424:gst_rtspsrc_try_send:<rtspsrc0> receive interrupted
0:00:05.033858894 788174 0x55a599059980 WARN                 rtspsrc gstrtspsrc.c:8672:gst_rtspsrc_pause:<rtspsrc0> PAUSE interrupted
0:00:05.058641415 788174 0x55a599059980 WARN                 rtspsrc gstrtspsrc.c:6385:gst_rtspsrc_try_send:<rtspsrc0> server closed connection
0:00:05.058671709 788174 0x55a599059980 WARN                 rtspsrc gstrtspsrc.c:6421:gst_rtspsrc_try_send:<rtspsrc0> error: Could not receive message. (Received end-of-file)
0:00:05.058713985 788174 0x55a599059980 WARN                 rtspsrc gstrtspsrc.c:8096:gst_rtspsrc_close:<rtspsrc0> error: Could not send message. (Received end-of-file)
2024-11-17 23:28:09 [ERROR] Failed to start video capture for camera: room1
2024-11-17 23:28:09 [INFO] Loaded camera config:
2024-11-17 23:28:09 [INFO]   Camera Name: room2
2024-11-17 23:28:09 [INFO]   RTSP URL: rtsp://117.71.99.41:5544/live/43130100001320000002_43130100001320000002?token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJVSUQiOjAsIlVzZXJuYW1lIjoiNDMxMzAxMDAwMDEzMjAwMDAwMDJfNDMxMzAxMDAwMDEzMjAwMDAwMDIiLCJHcm91cElEIjowLCJpc3MiOiJ4eEBnb2xhbmcuc3BhY2UiLCJleHAiOjE3Mzc3MTI5NDIsImlhdCI6MTczMTQ5MjE0Mn0.VuU_FjJZ0JqevQt3PJjKMhsJjOn4-WU75xf5XYut5as&expired=20250124180222
constuct FrameProcessor: triton_client_
0:00:05.967763857 788174 0x7f7e50006520 FIXME               basesink gstbasesink.c:3246:gst_base_sink_default_event:<sink_room2> stream-start event without group-id. Consider implementing group-id handling in the upstream elements
2024-11-17 23:28:14 [ERROR] Pipeline failed to reach PLAYING state for camera room2
0:00:10.066900078 788174 0x7f7e5006f0c0 WARN                 rtspsrc gstrtspsrc.c:6326:gst_rtsp_src_receive_response:<rtspsrc1> receive interrupted
0:00:10.066927084 788174 0x7f7e5006f0c0 WARN                 rtspsrc gstrtspsrc.c:6424:gst_rtspsrc_try_send:<rtspsrc1> receive interrupted
0:00:10.066935788 788174 0x7f7e5006f0c0 WARN                 rtspsrc gstrtspsrc.c:8672:gst_rtspsrc_pause:<rtspsrc1> PAUSE interrupted
0:00:10.078103630 788174 0x7f7e5006f0c0 WARN                 rtspsrc gstrtspsrc.c:6385:gst_rtspsrc_try_send:<rtspsrc1> server closed connection
0:00:10.078139389 788174 0x7f7e5006f0c0 WARN                 rtspsrc gstrtspsrc.c:6421:gst_rtspsrc_try_send:<rtspsrc1> error: Could not receive message. (Received end-of-file)
0:00:10.078173754 788174 0x7f7e5006f0c0 WARN                 rtspsrc gstrtspsrc.c:8096:gst_rtspsrc_close:<rtspsrc1> error: Could not send message. (Received end-of-file)
2024-11-17 23:28:14 [ERROR] Failed to start video capture for camera: room2
^C
(triton) (base) chen@chen-Precision-3660:~/data/algorithm_analysis_service/build$ 
$$$