|
|
#include <opencv2/core/core.hpp>
|
#include <opencv2/imgproc.hpp>
|
#include <opencv2/videoio.hpp>
|
#include <opencv2/opencv.hpp>
|
#include "vp_rtsp_ffmpeg_src_node.h"
|
#include "../utils/vp_utils.h"
|
#include <cstdio>
|
#include <memory>
|
#include <stdexcept>
|
#include <iostream>
|
#include <vector>
|
#include <thread>
|
#include <unistd.h>
|
#include <fcntl.h>
|
#include <sys/wait.h>
|
#include <cstdlib> // for setting environment variable
|
|
namespace vp_nodes
|
{
|
vp_rtsp_ffmpeg_src_node::vp_rtsp_ffmpeg_src_node(std::string node_name,
|
int channel_index,
|
std::string rtsp_url,
|
float resize_ratio,
|
int skip_interval,
|
bool use_gpu,
|
std::string ffmpeg_format,
|
std::string ffmpeg_pix_fmt) : vp_src_node(node_name, channel_index, resize_ratio),
|
rtsp_url(rtsp_url),
|
skip_interval(skip_interval),
|
use_gpu(use_gpu),
|
ffmpeg_format(ffmpeg_format),
|
ffmpeg_pix_fmt(ffmpeg_pix_fmt)
|
{
|
assert(skip_interval >= 0 && skip_interval <= 9);
|
VP_INFO(vp_utils::string_format("[%s] [%s]", node_name.c_str(), ffmpeg_format.c_str()));
|
this->initialized();
|
}
|
|
std::vector<int> vp_rtsp_ffmpeg_src_node::initialize_stream_properties()
|
{
|
try
|
{
|
// FFmpeg command to probe stream metadata
|
std::string ffprobe_command = vp_utils::string_format(ffprobe_template, rtsp_url.c_str());
|
|
// Execute command and capture output
|
FILE *pipe = popen(ffprobe_command.c_str(), "r");
|
if (!pipe)
|
{
|
throw std::runtime_error("Failed to execute ffprobe command.");
|
}
|
|
char buffer[128];
|
std::string result = "";
|
while (fgets(buffer, sizeof(buffer), pipe) != nullptr)
|
{
|
result += buffer;
|
}
|
pclose(pipe);
|
VP_INFO(vp_utils::string_format("FFPROBE RESULT:%s url: %s", result, rtsp_url.c_str()));
|
|
// Parse the output (expected format: width,height,r_frame_rate)
|
std::stringstream ss(result);
|
std::string width_str, height_str, fps_str;
|
std::getline(ss, width_str, ',');
|
std::getline(ss, height_str, ',');
|
std::getline(ss, fps_str, ',');
|
|
// Parse width and height
|
int video_width = std::stoi(width_str);
|
int video_height = std::stoi(height_str);
|
int fps = 25;
|
|
// Parse r_frame_rate (e.g., "299/12" or "25")
|
if (fps_str.find('/') != std::string::npos)
|
{
|
// Handle fractional frame rates
|
std::stringstream fps_ss(fps_str);
|
std::string numerator_str, denominator_str;
|
std::getline(fps_ss, numerator_str, '/');
|
std::getline(fps_ss, denominator_str, '/');
|
|
// Convert to integer values
|
int numerator = std::stoi(numerator_str);
|
int denominator = std::stoi(denominator_str);
|
|
// Calculate and round to the nearest integer
|
fps = static_cast<int>(std::round(static_cast<float>(numerator) / denominator));
|
}
|
else
|
{
|
// Handle integer frame rates
|
fps = std::stoi(fps_str);
|
}
|
|
// Optional: Correct minor deviations (e.g., if fps is close to 25)
|
if (std::abs(fps - 25) < 1)
|
{
|
fps = 25; // Force fps to 25 if very close
|
}
|
|
// Debug log for parsed properties
|
VP_INFO(vp_utils::string_format("Stream URL:%s Stream properties - Width: %d, Height: %d, FPS: %d",
|
rtsp_url.c_str(), video_width, video_height, fps));
|
|
return std::vector<int>{video_width, video_height, fps};
|
}
|
catch (...)
|
{
|
return std::vector<int>{0, 0, 0};
|
}
|
}
|
|
vp_rtsp_ffmpeg_src_node::~vp_rtsp_ffmpeg_src_node()
|
{
|
deinitialized();
|
}
|
|
// 子进程启动 ffmpeg 并返回管道文件描述符
|
int vp_rtsp_ffmpeg_src_node::start_ffmpeg_thread(const std::string &rtsp_url, int width, int height)
|
{
|
|
int pipe_fd[2];
|
if (pipe(pipe_fd) == -1)
|
{
|
perror("pipe");
|
return -1;
|
}
|
|
pid_t pid = fork();
|
if (pid == -1)
|
{
|
perror("fork");
|
return -1;
|
}
|
|
if (pid == 0) // 子进程
|
{
|
close(pipe_fd[0]); // 关闭读端
|
dup2(pipe_fd[1], STDOUT_FILENO); // 将管道写端绑定到标准输出
|
|
// 根据 channel_index 奇偶来设置 GPU
|
//int gpu_id = channel_index % total_gpus; // 根据 channel_index 和 GPU 数量进行分配
|
// 设置环境变量以选择 GPU
|
//setenv("CUDA_VISIBLE_DEVICES", std::to_string(gpu_id).c_str(), 1);
|
|
// 构造 ffmpeg 命令
|
execlp("ffmpeg", ("ffmpeg-" + std::to_string(channel_index)).c_str(),
|
"-loglevel", "warning",
|
"-hwaccel", "cuda",
|
"-rtsp_transport", "tcp",
|
"-i", rtsp_url.c_str(),
|
"-f", ffmpeg_format.c_str(),
|
"-pix_fmt", ffmpeg_pix_fmt.c_str(),
|
"-s", (std::to_string(width) + "x" + std::to_string(height)).c_str(),
|
"pipe:1",
|
nullptr);
|
exit(EXIT_FAILURE); // 若 execlp 失败
|
}
|
else
|
{ // 父进程
|
close(pipe_fd[1]); // 关闭写端
|
return pipe_fd[0]; // 返回读端文件描述符
|
}
|
}
|
|
// define how to read video from rtsp stream, create frame meta etc.
|
// please refer to the implementation of vp_node::handle_run.
|
void vp_rtsp_ffmpeg_src_node::handle_run()
|
{
|
std::vector<int> stream_infos = initialize_stream_properties();
|
|
original_width = stream_infos[0];
|
original_height = stream_infos[1];
|
original_fps = stream_infos[2];
|
|
|
int video_width = original_width;
|
int video_height = original_height;
|
// set true fps because skip some frames
|
int fps = original_fps / (skip_interval + 1);
|
|
cv::Mat frame(original_height, original_width, CV_8UC3);
|
|
int skip = 0;
|
int pipe_fd = -1;
|
|
int frame_size = original_width * original_height * 3; // 假设 BGR24
|
std::vector<unsigned char> frame_buffer(frame_size);
|
|
bool pipe_ok = true;
|
|
if (use_gpu)
|
{
|
pipe_fd = start_ffmpeg_thread(rtsp_url, original_width, original_height);
|
VP_INFO(std::to_string(pipe_fd));
|
if (pipe_fd < 0)
|
{
|
VP_ERROR(vp_utils::string_format("Failed to start cuda ffmpeg thread for [%s]", rtsp_url.c_str()));
|
use_gpu = false;
|
}
|
}
|
|
while (alive)
|
{
|
// check if need workc
|
gate.knock();
|
|
if (!use_gpu)
|
{
|
// try to open capture
|
if (!rtsp_capture.isOpened())
|
{
|
if (!rtsp_capture.open(this->rtsp_url, cv::CAP_FFMPEG))
|
{
|
VP_WARN(vp_utils::string_format("[%s] open rtsp failed, try again...", node_name.c_str()));
|
continue;
|
}
|
}
|
if (video_width == 0 || video_height == 0 || fps == 0)
|
{
|
video_width = rtsp_capture.get(cv::CAP_PROP_FRAME_WIDTH);
|
video_height = rtsp_capture.get(cv::CAP_PROP_FRAME_HEIGHT);
|
fps = rtsp_capture.get(cv::CAP_PROP_FPS);
|
|
original_fps = fps;
|
original_width = video_width;
|
original_height = video_height;
|
|
// set true fps because skip some frames
|
fps = fps / (skip_interval + 1);
|
}
|
// stream_info_hooker activated if need
|
vp_stream_info stream_info{channel_index, original_fps, original_width, original_height, to_string()};
|
invoke_stream_info_hooker(node_name, stream_info);
|
|
rtsp_capture >> frame;
|
}
|
else
|
{
|
|
if (!pipe_ok) // restart pipe_fd
|
{
|
pipe_fd = start_ffmpeg_thread(rtsp_url, original_width, original_height);
|
|
}
|
|
size_t total_read = 0;
|
|
// 循环读取，确保一帧完整数据
|
while (total_read < frame_size)
|
{
|
ssize_t bytes_read = read(pipe_fd, frame_buffer.data() + total_read, frame_size - total_read);
|
if (bytes_read < 0)
|
{ // 读取错误
|
if (errno == EINTR)
|
{
|
// 系统调用被中断，重试读取
|
VP_DEBUG(vp_utils::string_format("Interrupted read for [%s],re etrying...", rtsp_url.c_str()));
|
continue;
|
}
|
else if (errno == EAGAIN)
|
{
|
// 非阻塞模式下无数据可读
|
VP_DEBUG(vp_utils::string_format("Io data available yet for [%s],re etrying...", rtsp_url.c_str()));
|
std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 等待片刻再试
|
continue;
|
}
|
else
|
{
|
// 其他不可恢复错误，记录日志并跳过帧
|
VP_DEBUG(vp_utils::string_format("Read error for [%s].skipping frame...", rtsp_url.c_str()));
|
break;
|
}
|
}
|
else if (bytes_read == 0)
|
{ // 流结束
|
VP_DEBUG(vp_utils::string_format("Stream ended for [%s].skipping frame...", rtsp_url.c_str()));
|
pipe_ok = false;
|
continue; // 结束读取
|
}
|
|
total_read += bytes_read;
|
}
|
|
// 如果未能成功读取完整帧，记录日志并跳过
|
if (total_read < frame_size)
|
{
|
VP_DEBUG(vp_utils::string_format("Incomplete frame read for [%s].skipping frame...", rtsp_url.c_str()));
|
|
}
|
|
// 将数据拷贝到 OpenCV Mat
|
memcpy(frame.data, frame_buffer.data(), frame_size);
|
}
|
|
if (frame.empty())
|
{
|
VP_WARN(vp_utils::string_format("[%s] reading frame empty, total frame==>%d", node_name.c_str(), frame_index));
|
continue;
|
}
|
|
// need skip
|
if (skip < skip_interval)
|
{
|
skip++;
|
continue;
|
}
|
skip = 0;
|
|
cv::Mat resize_frame;
|
if (this->resize_ratio != 1.0f)
|
{
|
cv::resize(frame, resize_frame, cv::Size(), resize_ratio, resize_ratio);
|
}
|
else
|
{
|
resize_frame = frame.clone(); // clone!;
|
}
|
// set true size because resize
|
video_width = resize_frame.cols;
|
video_height = resize_frame.rows;
|
|
this->frame_index++;
|
// 插入此处
|
if (frame_index % 1000 == 0) {
|
VP_INFO(vp_utils::string_format(
|
"Periodic Log - Channel Index: %d, Frame Index: %d, RTSP URL: %s, Use GPU: %s",
|
channel_index, frame_index, rtsp_url.c_str(), use_gpu ? "true" : "false"));
|
}
|
// create frame meta
|
auto out_meta =
|
std::make_shared<vp_objects::vp_frame_meta>(resize_frame, this->frame_index, this->channel_index, video_width, video_height, fps);
|
|
if (out_meta != nullptr)
|
{
|
this->out_queue.push(out_meta);
|
|
// handled hooker activated if need
|
if (this->meta_handled_hooker)
|
{
|
meta_handled_hooker(node_name, out_queue.size(), out_meta);
|
}
|
|
// important! notify consumer of out_queue in case it is waiting.
|
this->out_queue_semaphore.signal();
|
VP_DEBUG(vp_utils::string_format("[%s] after handling meta, out_queue.size()==>%d", node_name.c_str(), out_queue.size()));
|
}
|
}
|
if (use_gpu)
|
{
|
close(pipe_fd);
|
}
|
|
// send dead flag for dispatch_thread
|
this->out_queue.push(nullptr);
|
this->out_queue_semaphore.signal();
|
}
|
|
// return stream url
|
std::string vp_rtsp_ffmpeg_src_node::to_string()
|
{
|
return rtsp_url;
|
}
|
}
|
