From 3c67625a4c98da222ec2e3eb7e91703f97509476 Mon Sep 17 00:00:00 2001 From: sladro Date: Fri, 26 Dec 2025 14:22:02 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=9B=E8=A1=8Csprint2=E5=BC=80=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- configs/sample_preprocess.json | 52 +++ include/node.h | 16 + plugins/CMakeLists.txt | 40 ++- plugins/preprocess/preprocess_node.cpp | 423 +++++++++++++++++++++++++ 4 files changed, 530 insertions(+), 1 deletion(-) create mode 100644 configs/sample_preprocess.json create mode 100644 plugins/preprocess/preprocess_node.cpp diff --git a/configs/sample_preprocess.json b/configs/sample_preprocess.json new file mode 100644 index 0000000..db3a07d --- /dev/null +++ b/configs/sample_preprocess.json @@ -0,0 +1,52 @@ +{ + "queue": { "size": 8, "strategy": "drop_oldest" }, + "graphs": [ + { + "name": "cam1_preprocess", + "nodes": [ + { + "id": "in_cam1", + "type": "input_rtsp", + "role": "source", + "enable": true, + "url": "rtsp://10.0.0.9:8554/cam", + "fps": 25, + "width": 1920, + "height": 1080, + "use_mpp": false, + "use_ffmpeg": true, + "force_tcp": true + }, + { + "id": "pre_cam1", + "type": "preprocess", + "role": "filter", + "enable": true, + "dst_w": 640, + "dst_h": 640, + "keep_ratio": true, + "use_rga": true + }, + { + "id": "pub_cam1", + "type": "publish", + "role": "sink", + "enable": true, + "codec": "h264", + "fps": 25, + "gop": 50, + "bitrate_kbps": 2000, + "use_mpp": true, + "use_ffmpeg_mux": true, + "outputs": [ + { "proto": "rtsp_server", "port": 8554, "path": "/live/cam1" } + ] + } + ], + "edges": [ + ["in_cam1", "pre_cam1"], + ["pre_cam1", "pub_cam1"] + ] + } + ] +} diff --git a/include/node.h b/include/node.h index e481d8e..98d19a6 100644 --- a/include/node.h +++ b/include/node.h @@ -12,6 +12,12 @@ namespace rk3588 { using FramePtr = std::shared_ptr; +enum class NodeStatus { + OK, // Normal processing + DROP, // Drop this frame (e.g., condition not met) + ERROR // Error occurred, should log +}; + struct NodeContext { // For nodes with upstream input. std::shared_ptr> input_queue; @@ -27,6 +33,16 @@ public: virtual bool Init(const SimpleJson& config, const NodeContext& ctx) = 0; virtual bool Start() = 0; virtual void Stop() = 0; + + // Process a single frame (for filter/sink nodes driven by GraphMgr). + // Default implementation does nothing; source nodes typically ignore this. + virtual NodeStatus Process(FramePtr /*frame*/) { return NodeStatus::OK; } + + // Dynamic config update without restart. Returns true if update succeeded. + virtual bool UpdateConfig(const SimpleJson& /*new_config*/) { return false; } + + // Called before Stop() to flush internal buffers (e.g., finish writing files). + virtual void Drain() {} }; constexpr int kNodeAbiVersion = 1; diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index 0525f57..afb9b37 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -3,6 +3,11 @@ set(RK_PLUGIN_OUTPUT_DIR ${CMAKE_BINARY_DIR}/plugins) option(RK3588_ENABLE_FFMPEG "Enable FFmpeg-based RTSP input" OFF) option(RK3588_ENABLE_MPP "Enable Rockchip MPP decode/encode" OFF) option(RK3588_ENABLE_ZLMEDIAKIT "Enable embedded ZLMediaKit RTSP server" OFF) +option(RK3588_ENABLE_RGA "Enable Rockchip RGA hardware acceleration" OFF) + +set(RK_RGA_ROOT "${RK_RKNN_ROOT}/examples/3rdparty/rga/RK3588" CACHE PATH "Path to RGA library") +set(RK_RGA_INCLUDE_DIR "${RK_RGA_ROOT}/include" CACHE PATH "RGA include directory") +set(RK_RGA_LIB_DIR "${RK_RGA_ROOT}/lib/Linux/aarch64" CACHE PATH "RGA library directory") set(RK_ZLMEDIAKIT_ROOT "${RK_RKNN_ROOT}/examples/3rdparty/zlmediakit" CACHE PATH "Path to ZLMediaKit mk_api bundle") set(RK_ZLMEDIAKIT_INCLUDE_DIR "${RK_ZLMEDIAKIT_ROOT}/include" CACHE PATH "ZLMediaKit mk_api include directory") @@ -52,6 +57,20 @@ if(RK3588_ENABLE_ZLMEDIAKIT) endif() endif() +if(RK3588_ENABLE_RGA) + find_library(RK_RGA_LIB rga + HINTS + ${RK_RGA_LIB_DIR} + NO_DEFAULT_PATH + ) + if(NOT RK_RGA_LIB) + find_library(RK_RGA_LIB rga) + endif() + if(NOT RK_RGA_LIB) + message(WARNING "RGA enabled but librga not found; disable RK3588_ENABLE_RGA or set RK_RGA_LIB_DIR") + endif() +endif() + add_library(input_rtsp SHARED input_rtsp/input_rtsp_node.cpp) target_include_directories(input_rtsp PRIVATE ${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/third_party) target_link_libraries(input_rtsp PRIVATE project_options Threads::Threads) @@ -106,7 +125,26 @@ set_target_properties(publish PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} ) -install(TARGETS input_rtsp publish +add_library(preprocess SHARED preprocess/preprocess_node.cpp) +target_include_directories(preprocess PRIVATE ${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/third_party) +target_link_libraries(preprocess PRIVATE project_options Threads::Threads) +if(RK3588_ENABLE_FFMPEG) + target_compile_definitions(preprocess PRIVATE RK3588_ENABLE_FFMPEG) + target_include_directories(preprocess PRIVATE ${FFMPEG_INCLUDE_DIRS}) + target_link_libraries(preprocess PRIVATE PkgConfig::FFMPEG) +endif() +if(RK3588_ENABLE_RGA AND RK_RGA_LIB) + target_compile_definitions(preprocess PRIVATE RK3588_ENABLE_RGA) + target_include_directories(preprocess PRIVATE ${RK_RGA_INCLUDE_DIR}) + target_link_libraries(preprocess PRIVATE ${RK_RGA_LIB}) +endif() +set_target_properties(preprocess PROPERTIES + OUTPUT_NAME "preprocess" + LIBRARY_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} + RUNTIME_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} +) + +install(TARGETS input_rtsp publish preprocess LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}/rk3588-media-server/plugins RUNTIME DESTINATION ${CMAKE_INSTALL_LIBDIR}/rk3588-media-server/plugins ) diff --git a/plugins/preprocess/preprocess_node.cpp b/plugins/preprocess/preprocess_node.cpp new file mode 100644 index 0000000..f769c78 --- /dev/null +++ b/plugins/preprocess/preprocess_node.cpp @@ -0,0 +1,423 @@ +#include +#include +#include +#include +#include +#include +#include + +#include "node.h" + +#if defined(RK3588_ENABLE_RGA) +#include "im2d.hpp" +#include "im2d_buffer.h" +#include "im2d_type.h" +#endif + +#if defined(RK3588_ENABLE_FFMPEG) +extern "C" { +#include +#include +} +#endif + +namespace rk3588 { + +namespace { + +inline int Align16(int v) { return (v + 15) & ~15; } + +#if defined(RK3588_ENABLE_RGA) +int ToRgaFormat(PixelFormat fmt) { + switch (fmt) { + case PixelFormat::NV12: return RK_FORMAT_YCbCr_420_SP; + case PixelFormat::YUV420: return RK_FORMAT_YCbCr_420_P; + case PixelFormat::RGB: return RK_FORMAT_RGB_888; + case PixelFormat::BGR: return RK_FORMAT_BGR_888; + default: return RK_FORMAT_UNKNOWN; + } +} +#endif + +PixelFormat ParseFormat(const std::string& s) { + if (s == "nv12" || s == "NV12") return PixelFormat::NV12; + if (s == "yuv420" || s == "YUV420") return PixelFormat::YUV420; + if (s == "rgb" || s == "RGB") return PixelFormat::RGB; + if (s == "bgr" || s == "BGR") return PixelFormat::BGR; + return PixelFormat::UNKNOWN; +} + +size_t CalcImageSize(int w, int h, PixelFormat fmt) { + switch (fmt) { + case PixelFormat::NV12: + case PixelFormat::YUV420: + return static_cast(w) * h * 3 / 2; + case PixelFormat::RGB: + case PixelFormat::BGR: + return static_cast(w) * h * 3; + default: + return 0; + } +} + +} // namespace + +class PreprocessNode : public INode { +public: + std::string Id() const override { return id_; } + std::string Type() const override { return "preprocess"; } + + bool Init(const SimpleJson& config, const NodeContext& ctx) override { + id_ = config.ValueOr("id", "preprocess"); + dst_w_ = config.ValueOr("dst_w", 640); + dst_h_ = config.ValueOr("dst_h", 640); + keep_ratio_ = config.ValueOr("keep_ratio", false); + std::string fmt_str = config.ValueOr("dst_format", ""); + if (!fmt_str.empty()) { + dst_fmt_ = ParseFormat(fmt_str); + } + use_rga_ = config.ValueOr("use_rga", true); + + input_queue_ = ctx.input_queue; + if (!input_queue_) { + std::cerr << "[preprocess] no input queue for node " << id_ << "\n"; + return false; + } + if (ctx.output_queues.empty()) { + std::cerr << "[preprocess] no output queue for node " << id_ << "\n"; + return false; + } + output_queues_ = ctx.output_queues; + +#if !defined(RK3588_ENABLE_RGA) + use_rga_ = false; +#endif +#if !defined(RK3588_ENABLE_FFMPEG) + if (!use_rga_) { + std::cerr << "[preprocess] neither RGA nor FFmpeg enabled\n"; + return false; + } +#endif + return true; + } + + bool Start() override { + if (!input_queue_) return false; + running_.store(true); + +#if defined(RK3588_ENABLE_RGA) + if (use_rga_) { + worker_ = std::thread(&PreprocessNode::LoopRga, this); + } else { + worker_ = std::thread(&PreprocessNode::LoopSwscale, this); + } +#elif defined(RK3588_ENABLE_FFMPEG) + worker_ = std::thread(&PreprocessNode::LoopSwscale, this); +#else + worker_ = std::thread(&PreprocessNode::LoopPassthrough, this); +#endif + std::cout << "[preprocess] start dst=" << dst_w_ << "x" << dst_h_ + << (use_rga_ ? " (rga)" : " (swscale)") << "\n"; + return true; + } + + void Stop() override { + running_.store(false); + if (input_queue_) input_queue_->Stop(); + for (auto& q : output_queues_) q->Stop(); + if (worker_.joinable()) worker_.join(); + } + +private: + void PushToDownstream(FramePtr frame) { + for (auto& q : output_queues_) { + q->Push(frame); + } + } + + void LoopPassthrough() { + using namespace std::chrono; + FramePtr frame; + while (running_.load()) { + if (!input_queue_->Pop(frame, milliseconds(200))) continue; + if (!frame) continue; + PushToDownstream(frame); + ++processed_; + if (processed_ % 100 == 0) { + std::cout << "[preprocess] passthrough frame " << frame->frame_id << "\n"; + } + } + } + +#if defined(RK3588_ENABLE_RGA) + void LoopRga() { + using namespace std::chrono; + FramePtr frame; + while (running_.load()) { + if (!input_queue_->Pop(frame, milliseconds(200))) continue; + if (!frame) continue; + + PixelFormat out_fmt = (dst_fmt_ != PixelFormat::UNKNOWN) ? dst_fmt_ : frame->format; + int out_w = dst_w_; + int out_h = dst_h_; + + if (keep_ratio_ && frame->width > 0 && frame->height > 0) { + float scale = std::min(static_cast(dst_w_) / frame->width, + static_cast(dst_h_) / frame->height); + out_w = static_cast(frame->width * scale); + out_h = static_cast(frame->height * scale); + out_w = (out_w + 1) & ~1; + out_h = (out_h + 1) & ~1; + } + + size_t out_size = CalcImageSize(out_w, out_h, out_fmt); + if (out_size == 0) { + PushToDownstream(frame); + continue; + } + + auto buffer = std::make_shared>(out_size); + int src_fmt_rga = ToRgaFormat(frame->format); + int dst_fmt_rga = ToRgaFormat(out_fmt); + + if (src_fmt_rga == RK_FORMAT_UNKNOWN || dst_fmt_rga == RK_FORMAT_UNKNOWN) { + std::cerr << "[preprocess] unsupported format for RGA\n"; + PushToDownstream(frame); + continue; + } + + int src_stride = frame->planes[0].stride > 0 ? frame->planes[0].stride + : (frame->stride > 0 ? frame->stride : frame->width); + + rga_buffer_t src_buf{}; + rga_buffer_t dst_buf{}; + + if (frame->dma_fd >= 0) { + src_buf = wrapbuffer_fd_t(frame->dma_fd, frame->width, frame->height, + src_stride, Align16(frame->height), src_fmt_rga); + } else if (frame->data) { + src_buf = wrapbuffer_virtualaddr_t(frame->data, frame->width, frame->height, + src_stride, Align16(frame->height), src_fmt_rga); + } else { + PushToDownstream(frame); + continue; + } + + dst_buf = wrapbuffer_virtualaddr_t(buffer->data(), out_w, out_h, + out_w, out_h, dst_fmt_rga); + + IM_STATUS status = IM_STATUS_SUCCESS; + bool need_cvt = (src_fmt_rga != dst_fmt_rga); + bool need_resize = (frame->width != out_w || frame->height != out_h); + + if (need_resize && need_cvt) { + // Resize + color convert in one call using improcess + status = imresize(src_buf, dst_buf); + if (status == IM_STATUS_SUCCESS) { + // imresize doesn't do color conversion; need separate call + // For simplicity, do resize first then cvtcolor + rga_buffer_t tmp = dst_buf; + status = imcvtcolor(tmp, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT); + } + } else if (need_resize) { + status = imresize(src_buf, dst_buf); + } else if (need_cvt) { + status = imcvtcolor(src_buf, dst_buf, src_fmt_rga, dst_fmt_rga, IM_COLOR_SPACE_DEFAULT); + } else { + status = imcopy(src_buf, dst_buf); + } + + if (status != IM_STATUS_SUCCESS) { + std::cerr << "[preprocess] RGA failed: " << imStrError(status) << "\n"; + PushToDownstream(frame); + continue; + } + + auto out_frame = std::make_shared(); + out_frame->width = out_w; + out_frame->height = out_h; + out_frame->format = out_fmt; + out_frame->stride = out_w; + out_frame->data = buffer->data(); + out_frame->data_size = buffer->size(); + out_frame->data_owner = buffer; + out_frame->pts = frame->pts; + out_frame->frame_id = frame->frame_id; + out_frame->det = frame->det; + out_frame->user_meta = frame->user_meta; + + SetupPlanes(*out_frame, out_fmt); + PushToDownstream(out_frame); + ++processed_; + + if (processed_ % 100 == 0) { + std::cout << "[preprocess] rga frame " << out_frame->frame_id + << " " << frame->width << "x" << frame->height + << " -> " << out_w << "x" << out_h << "\n"; + } + } + } +#endif + +#if defined(RK3588_ENABLE_FFMPEG) + void LoopSwscale() { + using namespace std::chrono; + FramePtr frame; + SwsContext* sws_ctx = nullptr; + int last_src_w = 0, last_src_h = 0; + AVPixelFormat last_src_fmt = AV_PIX_FMT_NONE; + AVPixelFormat last_dst_fmt = AV_PIX_FMT_NONE; + + while (running_.load()) { + if (!input_queue_->Pop(frame, milliseconds(200))) continue; + if (!frame) continue; + + PixelFormat out_fmt = (dst_fmt_ != PixelFormat::UNKNOWN) ? dst_fmt_ : frame->format; + int out_w = dst_w_; + int out_h = dst_h_; + + if (keep_ratio_ && frame->width > 0 && frame->height > 0) { + float scale = std::min(static_cast(dst_w_) / frame->width, + static_cast(dst_h_) / frame->height); + out_w = static_cast(frame->width * scale); + out_h = static_cast(frame->height * scale); + out_w = (out_w + 1) & ~1; + out_h = (out_h + 1) & ~1; + } + + AVPixelFormat src_av_fmt = ToAvFormat(frame->format); + AVPixelFormat dst_av_fmt = ToAvFormat(out_fmt); + + if (src_av_fmt == AV_PIX_FMT_NONE || dst_av_fmt == AV_PIX_FMT_NONE) { + PushToDownstream(frame); + continue; + } + + if (!sws_ctx || frame->width != last_src_w || frame->height != last_src_h || + src_av_fmt != last_src_fmt || dst_av_fmt != last_dst_fmt) { + if (sws_ctx) sws_freeContext(sws_ctx); + sws_ctx = sws_getContext(frame->width, frame->height, src_av_fmt, + out_w, out_h, dst_av_fmt, + SWS_BILINEAR, nullptr, nullptr, nullptr); + last_src_w = frame->width; + last_src_h = frame->height; + last_src_fmt = src_av_fmt; + last_dst_fmt = dst_av_fmt; + } + + if (!sws_ctx) { + PushToDownstream(frame); + continue; + } + + size_t out_size = CalcImageSize(out_w, out_h, out_fmt); + auto buffer = std::make_shared>(out_size); + + uint8_t* src_data[4] = {nullptr}; + int src_linesize[4] = {0}; + uint8_t* dst_data[4] = {nullptr}; + int dst_linesize[4] = {0}; + + SetupAvPlanes(frame.get(), src_data, src_linesize); + av_image_fill_arrays(dst_data, dst_linesize, buffer->data(), + dst_av_fmt, out_w, out_h, 1); + + sws_scale(sws_ctx, src_data, src_linesize, 0, frame->height, + dst_data, dst_linesize); + + auto out_frame = std::make_shared(); + out_frame->width = out_w; + out_frame->height = out_h; + out_frame->format = out_fmt; + out_frame->stride = out_w; + out_frame->data = buffer->data(); + out_frame->data_size = buffer->size(); + out_frame->data_owner = buffer; + out_frame->pts = frame->pts; + out_frame->frame_id = frame->frame_id; + out_frame->det = frame->det; + out_frame->user_meta = frame->user_meta; + + SetupPlanes(*out_frame, out_fmt); + PushToDownstream(out_frame); + ++processed_; + + if (processed_ % 100 == 0) { + std::cout << "[preprocess] swscale frame " << out_frame->frame_id + << " " << frame->width << "x" << frame->height + << " -> " << out_w << "x" << out_h << "\n"; + } + } + + if (sws_ctx) sws_freeContext(sws_ctx); + } + + static AVPixelFormat ToAvFormat(PixelFormat fmt) { + switch (fmt) { + case PixelFormat::NV12: return AV_PIX_FMT_NV12; + case PixelFormat::YUV420: return AV_PIX_FMT_YUV420P; + case PixelFormat::RGB: return AV_PIX_FMT_RGB24; + case PixelFormat::BGR: return AV_PIX_FMT_BGR24; + default: return AV_PIX_FMT_NONE; + } + } + + static void SetupAvPlanes(const Frame* f, uint8_t* data[4], int linesize[4]) { + if (!f->data) return; + if (f->format == PixelFormat::NV12) { + data[0] = f->planes[0].data ? f->planes[0].data : f->data; + data[1] = f->planes[1].data ? f->planes[1].data : (f->data + f->width * f->height); + linesize[0] = f->planes[0].stride > 0 ? f->planes[0].stride : f->width; + linesize[1] = f->planes[1].stride > 0 ? f->planes[1].stride : f->width; + } else if (f->format == PixelFormat::YUV420) { + data[0] = f->planes[0].data ? f->planes[0].data : f->data; + int y_size = f->width * f->height; + int uv_size = y_size / 4; + data[1] = f->planes[1].data ? f->planes[1].data : (f->data + y_size); + data[2] = f->planes[2].data ? f->planes[2].data : (f->data + y_size + uv_size); + linesize[0] = f->planes[0].stride > 0 ? f->planes[0].stride : f->width; + linesize[1] = f->planes[1].stride > 0 ? f->planes[1].stride : f->width / 2; + linesize[2] = f->planes[2].stride > 0 ? f->planes[2].stride : f->width / 2; + } else { + data[0] = f->data; + linesize[0] = f->stride > 0 ? f->stride : f->width * 3; + } + } +#endif + + void SetupPlanes(Frame& f, PixelFormat fmt) { + if (fmt == PixelFormat::NV12) { + f.plane_count = 2; + int y_size = f.width * f.height; + f.planes[0] = {f.data, f.width, y_size, 0}; + f.planes[1] = {f.data + y_size, f.width, y_size / 2, y_size}; + } else if (fmt == PixelFormat::YUV420) { + f.plane_count = 3; + int y_size = f.width * f.height; + int uv_size = y_size / 4; + f.planes[0] = {f.data, f.width, y_size, 0}; + f.planes[1] = {f.data + y_size, f.width / 2, uv_size, y_size}; + f.planes[2] = {f.data + y_size + uv_size, f.width / 2, uv_size, y_size + uv_size}; + } else { + f.plane_count = 1; + f.planes[0] = {f.data, f.width * 3, static_cast(f.data_size), 0}; + } + } + + std::string id_; + int dst_w_ = 640; + int dst_h_ = 640; + bool keep_ratio_ = false; + PixelFormat dst_fmt_ = PixelFormat::UNKNOWN; + bool use_rga_ = true; + + std::atomic running_{false}; + std::shared_ptr> input_queue_; + std::vector>> output_queues_; + std::thread worker_; + uint64_t processed_ = 0; +}; + +REGISTER_NODE(PreprocessNode, "preprocess"); + +} // namespace rk3588