From 02819b5c047bb354b0ef2374e7c6a6ac4bbd5bba Mon Sep 17 00:00:00 2001
From: wangrong <wangrong@shsening.com>
Date: 星期四, 23 一月 2025 11:42:36 +0800
Subject: [PATCH] add sample
---
nodes/vp_rtsp_ffmpeg_src_node.cpp | 182 +++++++++++++++++++++++++++++++++++----------
1 files changed, 140 insertions(+), 42 deletions(-)
diff --git a/nodes/vp_rtsp_ffmpeg_src_node.cpp b/nodes/vp_rtsp_ffmpeg_src_node.cpp
index ce254df..10e5020 100644
--- a/nodes/vp_rtsp_ffmpeg_src_node.cpp
+++ b/nodes/vp_rtsp_ffmpeg_src_node.cpp
@@ -3,13 +3,19 @@
#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 <iostream>
#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
{
@@ -28,11 +34,7 @@
ffmpeg_pix_fmt(ffmpeg_pix_fmt)
{
assert(skip_interval >= 0 && skip_interval <= 9);
- if (use_gpu)
- {
- this->ffmpeg_template = vp_utils::string_format(this->ffmpeg_template, rtsp_url.c_str(), ffmpeg_format.c_str(), ffmpeg_pix_fmt.c_str(), std::to_string(100 + channel_index).c_str());
- }
- VP_INFO(vp_utils::string_format("[%s] [%s]", node_name.c_str(), ffmpeg_template.c_str()));
+ VP_INFO(vp_utils::string_format("[%s] [%s]", node_name.c_str(), ffmpeg_format.c_str()));
this->initialized();
}
@@ -41,9 +43,7 @@
try
{
// FFmpeg command to probe stream metadata
- std::string ffprobe_command = vp_utils::string_format(
- "ffprobe -v error -select_streams v:0 -show_entries stream=width,height,r_frame_rate -of csv=p=0 %s",
- rtsp_url.c_str());
+ 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");
@@ -59,6 +59,7 @@
result += buffer;
}
pclose(pipe);
+ VP_INFO(vp_utils::string_format("FFPROBE RESULT:%s ", result));
// Parse the output (expected format: width,height,r_frame_rate)
std::stringstream ss(result);
@@ -70,7 +71,7 @@
// Parse width and height
int video_width = std::stoi(width_str);
int video_height = std::stoi(height_str);
- int fps=25;
+ int fps = 25;
// Parse r_frame_rate (e.g., "299/12" or "25")
if (fps_str.find('/') != std::string::npos)
@@ -86,12 +87,12 @@
int denominator = std::stoi(denominator_str);
// Calculate and round to the nearest integer
- int fps = static_cast<int>(std::round(static_cast<float>(numerator) / denominator));
+ fps = static_cast<int>(std::round(static_cast<float>(numerator) / denominator));
}
else
{
// Handle integer frame rates
- int fps = std::stoi(fps_str);
+ fps = std::stoi(fps_str);
}
// Optional: Correct minor deviations (e.g., if fps is close to 25)
@@ -102,10 +103,11 @@
// Debug log for parsed properties
VP_INFO(vp_utils::string_format("Stream URL:%s Stream properties - Width: %d, Height: %d, FPS: %d",
- rtsp_url, video_width, video_height, fps));
+ rtsp_url.c_str(), video_width, video_height, fps));
return std::vector<int>{video_width, video_height, fps};
- }catch (...)
+ }
+ catch (...)
{
return std::vector<int>{0, 0, 0};
}
@@ -116,38 +118,87 @@
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();
- int original_width = stream_infos[0];
- int original_height = stream_infos[1];
- int original_fps = stream_infos[2];
+ 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;
+ cv::Mat frame(original_height, original_width, CV_8UC3);
int skip = 0;
- FILE* ffmpeg_pipe ;
+ int pipe_fd = -1;
- // Create a pipe buffer to read raw data from FFmpeg
- size_t frame_size = original_height * original_width * 3; // Adjust this based on expected frame size (e.g., 1920x1080 with RGB)
- unsigned char *frame_buffer = new unsigned char[frame_size];
+ int frame_size = original_width * original_height * 3; // 鍋囪 BGR24
+ std::vector<unsigned char> frame_buffer(frame_size);
+
+ bool pipe_ok = true;
if (use_gpu)
{
- VP_INFO(this->ffmpeg_template);
- ffmpeg_pipe = popen(this->ffmpeg_template.c_str(), "r");
- if (!ffmpeg_pipe)
+ pipe_fd = start_ffmpeg_thread(rtsp_url, original_width, original_height);
+ VP_INFO(std::to_string(pipe_fd));
+ if (pipe_fd < 0)
{
- VP_WARN("Failed to start FFmpeg process, switching USING CPU.");
+ VP_ERROR(vp_utils::string_format("Failed to start cuda ffmpeg thread for [%s]", rtsp_url.c_str()));
use_gpu = false;
}
}
@@ -159,7 +210,6 @@
if (!use_gpu)
{
-
// try to open capture
if (!rtsp_capture.isOpened())
{
@@ -190,18 +240,60 @@
}
else
{
- // Read the raw video data from the pipe into a frame
- if (fread(frame_buffer, sizeof(unsigned char), frame_size, ffmpeg_pipe) <= 0)
- {
- VP_WARN("Failed to read frame from FFmpeg pipe, retrying...");
- continue;
- }
- // 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);
- // Convert the raw video data into an OpenCV Mat object
- frame = cv::Mat(original_height, original_width, CV_8UC3, frame_buffer);
+ 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())
@@ -232,6 +324,12 @@
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);
@@ -253,7 +351,7 @@
}
if (use_gpu)
{
- fclose(ffmpeg_pipe);
+ close(pipe_fd);
}
// send dead flag for dispatch_thread
@@ -266,4 +364,4 @@
{
return rtsp_url;
}
-}
\ No newline at end of file
+}
--
Gitblit v1.9.1