From 80e7abcd08367423548ec7fc2994cb8a90623d6a Mon Sep 17 00:00:00 2001 From: sladro Date: Tue, 6 Jan 2026 16:26:39 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=80=A7=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- configs/multi_security_instances.json | 6 +- ..._cam1_strict_minio_alarm_5ch_rtsp_hls.json | 6 +- ...t_cam1_strict_minio_alarm_rtsp_server.json | 6 +- include/media/encoded_video_meta.h | 39 ++ plugins/CMakeLists.txt | 1 + plugins/ai_yolo/ai_yolo_node.cpp | 23 ++ plugins/alarm/actions/clip_action.cpp | 365 +++++------------- plugins/alarm/actions/clip_action.h | 11 +- plugins/alarm/alarm_node.cpp | 196 +++------- plugins/alarm/packet_ring_buffer.cpp | 54 +++ plugins/alarm/packet_ring_buffer.h | 36 ++ plugins/publish/publish_node.cpp | 50 ++- 12 files changed, 357 insertions(+), 436 deletions(-) create mode 100644 include/media/encoded_video_meta.h create mode 100644 plugins/alarm/packet_ring_buffer.cpp create mode 100644 plugins/alarm/packet_ring_buffer.h diff --git a/configs/multi_security_instances.json b/configs/multi_security_instances.json index 1d6551e..8e147cc 100644 --- a/configs/multi_security_instances.json +++ b/configs/multi_security_instances.json @@ -38,6 +38,7 @@ "type": "ai_yolo", "role": "filter", "enable": true, + "infer_fps": 10, "model_path": "${model_path}", "model_version": "v5", "num_classes": 80, @@ -72,6 +73,7 @@ "type": "alarm", "role": "sink", "enable": true, + "eval_fps": 10, "labels": [], "rules": [ { @@ -124,7 +126,7 @@ { "id": "pub", "type": "publish", - "role": "sink", + "role": "filter", "enable": true, "codec": "h264", "fps": "${fps}", @@ -143,7 +145,7 @@ ["ai", "osd"], ["osd", "post"], ["post", "pub"], - ["post", "alarm"] + ["pub", "alarm"] ] } }, diff --git a/configs/stress_cam1_strict_minio_alarm_5ch_rtsp_hls.json b/configs/stress_cam1_strict_minio_alarm_5ch_rtsp_hls.json index 6d585a3..42080fd 100644 --- a/configs/stress_cam1_strict_minio_alarm_5ch_rtsp_hls.json +++ b/configs/stress_cam1_strict_minio_alarm_5ch_rtsp_hls.json @@ -39,6 +39,7 @@ "type": "ai_yolo", "role": "filter", "enable": true, + "infer_fps": 10, "model_path": "${model_path}", "model_version": "v5", "num_classes": 80, @@ -73,6 +74,7 @@ "type": "alarm", "role": "sink", "enable": true, + "eval_fps": 10, "labels": [], "rules": [ { @@ -125,7 +127,7 @@ { "id": "pub", "type": "publish", - "role": "sink", + "role": "filter", "enable": true, "codec": "h264", "fps": "${fps}", @@ -145,7 +147,7 @@ ["ai", "osd"], ["osd", "post"], ["post", "pub"], - ["post", "alarm"] + ["pub", "alarm"] ] } }, diff --git a/configs/test_cam1_strict_minio_alarm_rtsp_server.json b/configs/test_cam1_strict_minio_alarm_rtsp_server.json index 540bcb7..71a383c 100644 --- a/configs/test_cam1_strict_minio_alarm_rtsp_server.json +++ b/configs/test_cam1_strict_minio_alarm_rtsp_server.json @@ -35,6 +35,7 @@ "type": "ai_yolo", "role": "filter", "enable": true, + "infer_fps": 10, "model_path": "./third_party/rknpu2/examples/rknn_yolov5_demo/model/RK3588/yolov5s-640-640.rknn", "model_version": "v5", "num_classes": 80, @@ -69,6 +70,7 @@ "type": "alarm", "role": "sink", "enable": true, + "eval_fps": 10, "labels": [], "rules": [ { @@ -121,7 +123,7 @@ { "id": "pub_cam1", "type": "publish", - "role": "sink", + "role": "filter", "enable": true, "codec": "h264", "fps": 30, @@ -141,7 +143,7 @@ ["ai_cam1", "osd_cam1"], ["osd_cam1", "post_cam1"], ["post_cam1", "pub_cam1"], - ["post_cam1", "alarm_cam1"] + ["pub_cam1", "alarm_cam1"] ] } ] diff --git a/include/media/encoded_video_meta.h b/include/media/encoded_video_meta.h new file mode 100644 index 0000000..ec5e578 --- /dev/null +++ b/include/media/encoded_video_meta.h @@ -0,0 +1,39 @@ +#pragma once + +#include +#include +#include + +namespace rk3588 { + +enum class VideoCodec { + H264, + H265, + UNKNOWN, +}; + +struct StreamCodecInfo { + VideoCodec codec = VideoCodec::UNKNOWN; + int width = 0; + int height = 0; + int fps = 0; + + // Encoder header / codec extradata. For H264/H265, AnnexB SPS/PPS(/VPS) is acceptable. + std::vector extradata; +}; + +struct EncodedVideoPacket { + std::vector data; + bool key = false; + int64_t pts_ms = 0; +}; + +struct EncodedVideoFrameMeta { + static constexpr uint32_t kMagic = 0x45564D31; // 'EVM1' + uint32_t magic = kMagic; + + std::shared_ptr codec; + EncodedVideoPacket pkt; +}; + +} // namespace rk3588 diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index 04e2a3e..a084c19 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -246,6 +246,7 @@ add_library(alarm SHARED alarm/alarm_node.cpp alarm/rule_engine.cpp alarm/frame_ring_buffer.cpp + alarm/packet_ring_buffer.cpp alarm/actions/log_action.cpp alarm/actions/http_action.cpp alarm/actions/snapshot_action.cpp diff --git a/plugins/ai_yolo/ai_yolo_node.cpp b/plugins/ai_yolo/ai_yolo_node.cpp index c810350..6002e91 100644 --- a/plugins/ai_yolo/ai_yolo_node.cpp +++ b/plugins/ai_yolo/ai_yolo_node.cpp @@ -273,6 +273,16 @@ public: model_input_h_ = config.ValueOr("model_h", 640); num_classes_ = config.ValueOr("num_classes", 80); + // Optional inference throttle. 0 = run every frame. + infer_interval_ms_ = std::max(0, static_cast(config.ValueOr("infer_interval_ms", 0))); + if (infer_interval_ms_ <= 0) { + const double infer_fps = config.ValueOr("infer_fps", 0.0); + if (infer_fps > 0.0) { + infer_interval_ms_ = static_cast(1000.0 / infer_fps); + if (infer_interval_ms_ < 1) infer_interval_ms_ = 1; + } + } + std::string ver = config.ValueOr("model_version", "auto"); if (ver == "v5") { yolo_version_ = YoloVersion::V5; @@ -355,6 +365,16 @@ public: NodeStatus Process(FramePtr frame) override { if (!frame) return NodeStatus::DROP; + if (infer_interval_ms_ > 0 && frame->pts > 0) { + const int64_t pts_ms = static_cast(frame->pts / 1000ULL); + if (last_infer_pts_ms_ > 0 && (pts_ms - last_infer_pts_ms_) < infer_interval_ms_) { + PushToDownstream(frame); + ++processed_; + return NodeStatus::OK; + } + last_infer_pts_ms_ = pts_ms; + } + #if defined(RK3588_ENABLE_RKNN) RunInference(frame); #endif @@ -559,6 +579,9 @@ private: std::vector>> output_queues_; uint64_t processed_ = 0; + int64_t infer_interval_ms_ = 0; + int64_t last_infer_pts_ms_ = 0; + #if defined(RK3588_ENABLE_RKNN) ModelHandle model_handle_ = kInvalidModelHandle; uint32_t n_output_ = 0; diff --git a/plugins/alarm/actions/clip_action.cpp b/plugins/alarm/actions/clip_action.cpp index 90cabac..c48cfe9 100644 --- a/plugins/alarm/actions/clip_action.cpp +++ b/plugins/alarm/actions/clip_action.cpp @@ -2,7 +2,6 @@ #include #include -#include #include #include #include @@ -16,8 +15,7 @@ extern "C" { #include #include -#include -#include +#include } #define HAS_FFMPEG 1 #else @@ -26,161 +24,6 @@ extern "C" { namespace rk3588 { -namespace { - -inline uint8_t ClipU8(int v) { - if (v < 0) return 0; - if (v > 255) return 255; - return static_cast(v); -} - -inline const uint8_t* PlanePtr(const Frame& f, int idx) { - if (idx < 0 || idx >= f.plane_count) return nullptr; - if (f.planes[idx].data) return f.planes[idx].data; - if (!f.data) return nullptr; - const int off = f.planes[idx].offset; - if (off < 0) return nullptr; - return f.data + static_cast(off); -} - -inline int PlaneStride(const Frame& f, int idx, int fallback) { - if (idx >= 0 && idx < f.plane_count && f.planes[idx].stride > 0) return f.planes[idx].stride; - if (f.stride > 0) return f.stride; - return fallback; -} - -bool FillYuv420pFromFrame(const Frame& src, int width, int height, AVFrame* dst) { - if (!dst) return false; - if (dst->format != AV_PIX_FMT_YUV420P) return false; - if (width <= 0 || height <= 0) return false; - - uint8_t* y = dst->data[0]; - uint8_t* u = dst->data[1]; - uint8_t* v = dst->data[2]; - const int y_stride = dst->linesize[0]; - const int u_stride = dst->linesize[1]; - const int v_stride = dst->linesize[2]; - if (!y || !u || !v || y_stride <= 0 || u_stride <= 0 || v_stride <= 0) return false; - - if (src.width != width || src.height != height) return false; - - if (src.format == PixelFormat::YUV420) { - const uint8_t* sy = PlanePtr(src, 0); - const uint8_t* su = PlanePtr(src, 1); - const uint8_t* sv = PlanePtr(src, 2); - if (!sy || !su || !sv) return false; - const int sy_stride = PlaneStride(src, 0, width); - const int su_stride = PlaneStride(src, 1, width / 2); - const int sv_stride = PlaneStride(src, 2, width / 2); - - for (int row = 0; row < height; ++row) { - std::memcpy(y + row * y_stride, sy + row * sy_stride, static_cast(width)); - } - const int uv_h = height / 2; - const int uv_w = width / 2; - for (int row = 0; row < uv_h; ++row) { - std::memcpy(u + row * u_stride, su + row * su_stride, static_cast(uv_w)); - std::memcpy(v + row * v_stride, sv + row * sv_stride, static_cast(uv_w)); - } - return true; - } - - if (src.format == PixelFormat::NV12) { - const uint8_t* sy = PlanePtr(src, 0); - const uint8_t* suv = PlanePtr(src, 1); - if (!sy) return false; - const int sy_stride = PlaneStride(src, 0, width); - const int suv_stride = PlaneStride(src, 1, width); - if (!suv) { - if (!src.data) return false; - suv = src.data + static_cast(sy_stride) * static_cast(height); - } - - for (int row = 0; row < height; ++row) { - std::memcpy(y + row * y_stride, sy + row * sy_stride, static_cast(width)); - } - const int uv_h = height / 2; - const int uv_w = width / 2; - for (int row = 0; row < uv_h; ++row) { - const uint8_t* src_uv = suv + row * suv_stride; - uint8_t* dst_u = u + row * u_stride; - uint8_t* dst_v = v + row * v_stride; - for (int col = 0; col < uv_w; ++col) { - dst_u[col] = src_uv[col * 2 + 0]; - dst_v[col] = src_uv[col * 2 + 1]; - } - } - return true; - } - - if (src.format == PixelFormat::RGB || src.format == PixelFormat::BGR) { - const bool is_bgr = (src.format == PixelFormat::BGR); - const uint8_t* s = PlanePtr(src, 0); - if (!s) s = src.data; - if (!s) return false; - const int s_stride = PlaneStride(src, 0, width * 3); - - const int uv_w = width / 2; - const int uv_h = height / 2; - - for (int row = 0; row < height; row += 2) { - const uint8_t* row0 = s + row * s_stride; - const uint8_t* row1 = (row + 1 < height) ? (s + (row + 1) * s_stride) : row0; - uint8_t* y0 = y + row * y_stride; - uint8_t* y1 = (row + 1 < height) ? (y + (row + 1) * y_stride) : y0; - const int uv_row = row / 2; - uint8_t* uu = (uv_row < uv_h) ? (u + uv_row * u_stride) : nullptr; - uint8_t* vv = (uv_row < uv_h) ? (v + uv_row * v_stride) : nullptr; - - for (int col = 0; col < width; col += 2) { - int u_sum = 0; - int v_sum = 0; - int samples = 0; - - auto sample = [&](const uint8_t* p, uint8_t* ydst) { - const int b = is_bgr ? p[0] : p[2]; - const int g = p[1]; - const int r = is_bgr ? p[2] : p[0]; - const int yy = (77 * r + 150 * g + 29 * b + 128) >> 8; - *ydst = ClipU8(yy); - u_sum += (-43 * r - 84 * g + 127 * b); - v_sum += (127 * r - 106 * g - 21 * b); - samples += 1; - }; - - const uint8_t* p00 = row0 + col * 3; - const uint8_t* p01 = (col + 1 < width) ? (row0 + (col + 1) * 3) : p00; - const uint8_t* p10 = row1 + col * 3; - const uint8_t* p11 = (col + 1 < width) ? (row1 + (col + 1) * 3) : p10; - - sample(p00, &y0[col]); - if (col + 1 < width) sample(p01, &y0[col + 1]); - if (row + 1 < height) { - sample(p10, &y1[col]); - if (col + 1 < width) sample(p11, &y1[col + 1]); - } else { - sample(p10, &y1[col]); - if (col + 1 < width) sample(p11, &y1[col + 1]); - } - - const int denom = samples > 0 ? samples : 1; - const int u_val = ((u_sum / denom) + 128 * 256 + 128) >> 8; - const int v_val = ((v_sum / denom) + 128 * 256 + 128) >> 8; - const int uv_col = col / 2; - if (uu && vv && uv_col >= 0 && uv_col < uv_w) { - uu[uv_col] = ClipU8(u_val); - vv[uv_col] = ClipU8(v_val); - } - } - } - return true; - } - - return false; -} - -} // namespace - bool ClipAction::Init(const SimpleJson& config) { pre_sec_ = config.ValueOr("pre_sec", 5); post_sec_ = config.ValueOr("post_sec", 10); @@ -204,62 +47,78 @@ bool ClipAction::Init(const SimpleJson& config) { } void ClipAction::Execute(AlarmEvent& event, std::shared_ptr frame) { - if (!frame_buffer_) { - std::cerr << "[ClipAction] frame buffer not set\n"; + if (!packet_buffer_) { + std::cerr << "[ClipAction] packet buffer not set\n"; return; } if (!frame) return; - uint64_t trigger_ts_ms = 0; - if (!frame_buffer_->FindTimestampByFrameId(frame->frame_id, trigger_ts_ms)) { - using namespace std::chrono; - trigger_ts_ms = static_cast(duration_cast(steady_clock::now().time_since_epoch()).count()); + const int64_t trigger_pts_ms = frame->pts > 0 ? static_cast(frame->pts / 1000ULL) : 0; + if (trigger_pts_ms <= 0) { + std::cerr << "[ClipAction] invalid trigger pts\n"; + return; } - const uint64_t pre_ms = static_cast(std::max(0, pre_sec_)) * 1000ULL; - const uint64_t post_ms = static_cast(std::max(0, post_sec_)) * 1000ULL; - const uint64_t start_ts = (trigger_ts_ms > pre_ms) ? (trigger_ts_ms - pre_ms) : 0; - const uint64_t end_ts = trigger_ts_ms + post_ms; + const int64_t pre_ms = static_cast(std::max(0, pre_sec_)) * 1000LL; + const int64_t post_ms = static_cast(std::max(0, post_sec_)) * 1000LL; + const int64_t start_pts = std::max(0, trigger_pts_ms - pre_ms); + const int64_t end_pts = trigger_pts_ms + post_ms; - // Best-effort: wait for post-event window to elapse so frames are available in ring buffer. + // Best-effort wait for post window to arrive. if (post_sec_ > 0) { - std::this_thread::sleep_for(std::chrono::seconds(post_sec_)); + using namespace std::chrono; + const auto deadline = steady_clock::now() + seconds(post_sec_); + while (steady_clock::now() < deadline) { + if (packet_buffer_->LatestPtsMs() >= end_pts) break; + std::this_thread::sleep_for(milliseconds(20)); + } } - auto frames = frame_buffer_->GetFramesInRange(start_ts, end_ts); - if (frames.empty()) { - frames.push_back(frame); + auto metas = packet_buffer_->GetPacketsInRange(start_pts, end_pts); + if (metas.empty()) { + // Fallback: at least try the latest packet. + metas = packet_buffer_->GetPacketsInRange(trigger_pts_ms, end_pts); + } + if (metas.empty()) { + std::cerr << "[ClipAction] no packets in range\n"; + return; } - std::string url = ProcessClip(frames, event); + std::string url = ProcessClipFromPackets(metas, event); if (!url.empty()) { event.clip_url = url; std::cout << "[ClipAction] clip uploaded: " << url << "\n"; } } -std::string ClipAction::ProcessClip(const std::vector>& frames, const AlarmEvent& event) { +std::string ClipAction::ProcessClipFromPackets(const std::vector>& metas, + const AlarmEvent& event) { #if HAS_FFMPEG - if (frames.empty()) { - std::cerr << "[ClipAction] no frames to process\n"; + std::shared_ptr first; + for (const auto& m : metas) { + if (m && m->magic == EncodedVideoFrameMeta::kMagic && m->codec && !m->pkt.data.empty()) { + first = m; + break; + } + } + if (!first || !first->codec) return ""; + + AVCodecID cid = AV_CODEC_ID_NONE; + const auto codec = first->codec->codec; + if (codec == VideoCodec::H264) cid = AV_CODEC_ID_H264; + else if (codec == VideoCodec::H265) cid = AV_CODEC_ID_HEVC; + else { + std::cerr << "[ClipAction] unsupported codec\n"; return ""; } - std::shared_ptr sample_frame; - for (const auto& f : frames) { - if (f) { sample_frame = f; break; } - } - if (!sample_frame) return ""; - - int width = sample_frame->width; - int height = sample_frame->height; - + const int width = first->codec->width; + const int height = first->codec->height; if (width <= 0 || height <= 0) { - std::cerr << "[ClipAction] invalid frame size\n"; + std::cerr << "[ClipAction] invalid codec size\n"; return ""; } - // Create temporary file std::filesystem::path tmp_dir; try { tmp_dir = std::filesystem::temp_directory_path(); @@ -268,136 +127,110 @@ std::string ClipAction::ProcessClip(const std::vector>& f } std::filesystem::path tmp_path = tmp_dir / ("clip_" + std::to_string(event.timestamp_ms) + "." + format_); - // Set up FFmpeg output AVFormatContext* fmt_ctx = nullptr; - if (avformat_alloc_output_context2(&fmt_ctx, nullptr, format_.c_str(), tmp_path.string().c_str()) < 0) { + if (avformat_alloc_output_context2(&fmt_ctx, nullptr, format_.c_str(), tmp_path.string().c_str()) < 0 || !fmt_ctx) { std::cerr << "[ClipAction] failed to create output context\n"; return ""; } - const AVCodec* codec = avcodec_find_encoder(AV_CODEC_ID_H264); - if (!codec) { - avformat_free_context(fmt_ctx); - return ""; - } - - AVStream* stream = avformat_new_stream(fmt_ctx, codec); + AVStream* stream = avformat_new_stream(fmt_ctx, nullptr); if (!stream) { avformat_free_context(fmt_ctx); return ""; } - AVCodecContext* enc_ctx = avcodec_alloc_context3(codec); - enc_ctx->width = width; - enc_ctx->height = height; - enc_ctx->time_base = AVRational{1, fps_}; - enc_ctx->framerate = AVRational{fps_, 1}; - enc_ctx->pix_fmt = AV_PIX_FMT_YUV420P; - enc_ctx->bit_rate = 2000000; - enc_ctx->gop_size = fps_; + stream->time_base = AVRational{1, 1000}; + stream->codecpar->codec_type = AVMEDIA_TYPE_VIDEO; + stream->codecpar->codec_id = cid; + stream->codecpar->width = width; + stream->codecpar->height = height; + stream->codecpar->format = AV_PIX_FMT_YUV420P; - if (fmt_ctx->oformat->flags & AVFMT_GLOBALHEADER) { - enc_ctx->flags |= AV_CODEC_FLAG_GLOBAL_HEADER; + if (!first->codec->extradata.empty()) { + stream->codecpar->extradata_size = static_cast(first->codec->extradata.size()); + stream->codecpar->extradata = static_cast(av_mallocz(first->codec->extradata.size() + AV_INPUT_BUFFER_PADDING_SIZE)); + std::memcpy(stream->codecpar->extradata, first->codec->extradata.data(), first->codec->extradata.size()); } - if (avcodec_open2(enc_ctx, codec, nullptr) < 0) { - avcodec_free_context(&enc_ctx); - avformat_free_context(fmt_ctx); - return ""; - } - - avcodec_parameters_from_context(stream->codecpar, enc_ctx); - stream->time_base = enc_ctx->time_base; - if (!(fmt_ctx->oformat->flags & AVFMT_NOFILE)) { if (avio_open(&fmt_ctx->pb, tmp_path.string().c_str(), AVIO_FLAG_WRITE) < 0) { - avcodec_free_context(&enc_ctx); avformat_free_context(fmt_ctx); return ""; } } if (avformat_write_header(fmt_ctx, nullptr) < 0) { - avcodec_free_context(&enc_ctx); + if (!(fmt_ctx->oformat->flags & AVFMT_NOFILE)) avio_closep(&fmt_ctx->pb); avformat_free_context(fmt_ctx); return ""; } - // Encode frames - AVFrame* av_frame = av_frame_alloc(); - av_frame->width = width; - av_frame->height = height; - av_frame->format = AV_PIX_FMT_YUV420P; - av_frame_get_buffer(av_frame, 32); + int64_t first_pts = 0; + bool have_first = false; + int64_t last_pts = -1; + const int64_t frame_dur = (fps_ > 0) ? (1000LL / std::max(1, fps_)) : 40; - AVPacket* pkt = av_packet_alloc(); - int64_t pts = 0; + for (const auto& m : metas) { + if (!m || m->magic != EncodedVideoFrameMeta::kMagic) continue; + if (!m->codec || m->pkt.data.empty()) continue; + if (m->pkt.pts_ms <= 0) continue; - auto encode_frame = [&](const std::shared_ptr& frame) { - if (!frame) return; - - if (!FillYuv420pFromFrame(*frame, width, height, av_frame)) return; - - av_frame->pts = pts++; - - avcodec_send_frame(enc_ctx, av_frame); - while (avcodec_receive_packet(enc_ctx, pkt) == 0) { - av_packet_rescale_ts(pkt, enc_ctx->time_base, stream->time_base); - pkt->stream_index = stream->index; - av_interleaved_write_frame(fmt_ctx, pkt); - av_packet_unref(pkt); + if (!have_first) { + first_pts = m->pkt.pts_ms; + have_first = true; } - }; + int64_t pts = m->pkt.pts_ms - first_pts; + if (pts < 0) pts = 0; + if (pts <= last_pts) pts = last_pts + 1; + last_pts = pts; - for (const auto& f : frames) { - encode_frame(f); - } - - // Flush encoder - avcodec_send_frame(enc_ctx, nullptr); - while (avcodec_receive_packet(enc_ctx, pkt) == 0) { - av_packet_rescale_ts(pkt, enc_ctx->time_base, stream->time_base); + AVPacket* pkt = av_packet_alloc(); + if (!pkt) continue; + if (av_new_packet(pkt, static_cast(m->pkt.data.size())) < 0) { + av_packet_free(&pkt); + continue; + } + std::memcpy(pkt->data, m->pkt.data.data(), m->pkt.data.size()); pkt->stream_index = stream->index; - av_interleaved_write_frame(fmt_ctx, pkt); - av_packet_unref(pkt); + pkt->pts = pts; + pkt->dts = pts; + pkt->duration = frame_dur; + if (m->pkt.key) pkt->flags |= AV_PKT_FLAG_KEY; + + (void)av_interleaved_write_frame(fmt_ctx, pkt); + av_packet_free(&pkt); } av_write_trailer(fmt_ctx); - - av_packet_free(&pkt); - av_frame_free(&av_frame); - avcodec_free_context(&enc_ctx); - if (!(fmt_ctx->oformat->flags & AVFMT_NOFILE)) { avio_closep(&fmt_ctx->pb); } + if (!(fmt_ctx->oformat->flags & AVFMT_NOFILE)) avio_closep(&fmt_ctx->pb); avformat_free_context(fmt_ctx); - // Read file and upload std::ifstream file(tmp_path, std::ios::binary | std::ios::ate); if (!file) { std::cerr << "[ClipAction] failed to read temp file\n"; return ""; } - - size_t file_size = file.tellg(); + const size_t file_size = static_cast(file.tellg()); file.seekg(0); std::vector file_data(file_size); - file.read(reinterpret_cast(file_data.data()), file_size); + file.read(reinterpret_cast(file_data.data()), static_cast(file_data.size())); file.close(); std::string key = GenerateKey(event); auto result = uploader_->Upload(key, file_data.data(), file_data.size(), "video/mp4"); - // Clean up temp file std::error_code ec; std::filesystem::remove(tmp_path, ec); - if (result.success) { - return result.url; - } + if (result.success) return result.url; std::cerr << "[ClipAction] upload failed: " << result.error << "\n"; -#else - std::cerr << "[ClipAction] FFmpeg not enabled\n"; -#endif return ""; +#else + (void)metas; + (void)event; + std::cerr << "[ClipAction] FFmpeg not enabled\n"; + return ""; +#endif } std::string ClipAction::GenerateKey(const AlarmEvent& event) { diff --git a/plugins/alarm/actions/clip_action.h b/plugins/alarm/actions/clip_action.h index a747bc0..f0849fc 100644 --- a/plugins/alarm/actions/clip_action.h +++ b/plugins/alarm/actions/clip_action.h @@ -2,7 +2,7 @@ #include "action_base.h" #include "../uploaders/uploader_base.h" -#include "../frame_ring_buffer.h" +#include "../packet_ring_buffer.h" #include #include @@ -18,12 +18,11 @@ public: void Stop() override {} std::string Name() const override { return "clip"; } - void SetFrameBuffer(std::shared_ptr buffer) { - frame_buffer_ = buffer; - } + void SetPacketBuffer(std::shared_ptr buffer) { packet_buffer_ = buffer; } private: - std::string ProcessClip(const std::vector>& frames, const AlarmEvent& event); + std::string ProcessClipFromPackets(const std::vector>& metas, + const AlarmEvent& event); std::string GenerateKey(const AlarmEvent& event); int pre_sec_ = 5; @@ -31,7 +30,7 @@ private: std::string format_ = "mp4"; int fps_ = 25; std::unique_ptr uploader_; - std::shared_ptr frame_buffer_; + std::shared_ptr packet_buffer_; }; } // namespace rk3588 diff --git a/plugins/alarm/alarm_node.cpp b/plugins/alarm/alarm_node.cpp index 812ec41..bb431af 100644 --- a/plugins/alarm/alarm_node.cpp +++ b/plugins/alarm/alarm_node.cpp @@ -2,6 +2,7 @@ #include #include #include +#include #include #include #include @@ -10,13 +11,12 @@ #include "node.h" #include "rule_engine.h" -#include "frame_ring_buffer.h" +#include "packet_ring_buffer.h" #include "actions/action_base.h" #include "actions/log_action.h" #include "actions/http_action.h" #include "actions/snapshot_action.h" #include "actions/clip_action.h" -#include "utils/dma_alloc.h" namespace rk3588 { @@ -27,135 +27,12 @@ private: std::shared_ptr frame; }; - static std::shared_ptr CloneFrameForRingBuffer(const std::shared_ptr& src) { - if (!src) return nullptr; - if (src->width <= 0 || src->height <= 0) return nullptr; - if (!src->data) return nullptr; - - const int dma_fd = src->dma_fd; - if (dma_fd >= 0) DmaSyncStartFd(dma_fd); - struct SyncGuard { - int fd; - ~SyncGuard() { - if (fd >= 0) DmaSyncEndFd(fd); - } - } guard{dma_fd}; - - auto out = std::make_shared(); - out->width = src->width; - out->height = src->height; - out->format = src->format; - out->pts = src->pts; - out->frame_id = src->frame_id; - out->det = src->det; - out->user_meta = src->user_meta; - out->dma_fd = -1; - - const int w = src->width; - const int h = src->height; - - auto plane_ptr = [&](int idx) -> const uint8_t* { - if (idx < 0 || idx >= src->plane_count) return nullptr; - if (src->planes[idx].data) return src->planes[idx].data; - const int off = src->planes[idx].offset; - if (off < 0) return nullptr; - return src->data + static_cast(off); - }; - auto plane_stride = [&](int idx, int fallback) -> int { - if (idx >= 0 && idx < src->plane_count && src->planes[idx].stride > 0) return src->planes[idx].stride; - if (src->stride > 0) return src->stride; - return fallback; - }; - - if (src->format == PixelFormat::NV12) { - const uint8_t* sy = plane_ptr(0); - const uint8_t* suv = plane_ptr(1); - if (!sy) return nullptr; - const int sy_stride = plane_stride(0, w); - const int suv_stride = plane_stride(1, w); - if (!suv) { - suv = src->data + static_cast(sy_stride) * static_cast(h); - } - - const int y_stride = w; - const int uv_stride = w; - const size_t y_bytes = static_cast(y_stride) * static_cast(h); - const size_t uv_bytes = static_cast(uv_stride) * static_cast(h / 2); - auto buf = std::make_shared>(y_bytes + uv_bytes); - out->data_owner = buf; - out->data = buf->data(); - out->data_size = buf->size(); - out->stride = y_stride; - out->plane_count = 2; - out->planes[0] = {out->data, y_stride, static_cast(y_bytes), 0}; - out->planes[1] = {out->data + y_bytes, uv_stride, static_cast(uv_bytes), static_cast(y_bytes)}; - - // Copy Y - for (int row = 0; row < h; ++row) { - std::memcpy(out->planes[0].data + row * y_stride, sy + row * sy_stride, static_cast(w)); - } - // Copy UV (interleaved) - for (int row = 0; row < h / 2; ++row) { - std::memcpy(out->planes[1].data + row * uv_stride, suv + row * suv_stride, static_cast(w)); - } - return out; - } - - if (src->format == PixelFormat::YUV420) { - const uint8_t* sy = plane_ptr(0); - const uint8_t* su = plane_ptr(1); - const uint8_t* sv = plane_ptr(2); - if (!sy || !su || !sv) return nullptr; - const int sy_stride = plane_stride(0, w); - const int su_stride = plane_stride(1, w / 2); - const int sv_stride = plane_stride(2, w / 2); - - const int y_stride = w; - const int uv_stride = w / 2; - const size_t y_bytes = static_cast(y_stride) * static_cast(h); - const size_t u_bytes = static_cast(uv_stride) * static_cast(h / 2); - const size_t v_bytes = u_bytes; - auto buf = std::make_shared>(y_bytes + u_bytes + v_bytes); - out->data_owner = buf; - out->data = buf->data(); - out->data_size = buf->size(); - out->stride = y_stride; - out->plane_count = 3; - out->planes[0] = {out->data, y_stride, static_cast(y_bytes), 0}; - out->planes[1] = {out->data + y_bytes, uv_stride, static_cast(u_bytes), static_cast(y_bytes)}; - out->planes[2] = {out->data + y_bytes + u_bytes, uv_stride, static_cast(v_bytes), static_cast(y_bytes + u_bytes)}; - - for (int row = 0; row < h; ++row) { - std::memcpy(out->planes[0].data + row * y_stride, sy + row * sy_stride, static_cast(w)); - } - for (int row = 0; row < h / 2; ++row) { - std::memcpy(out->planes[1].data + row * uv_stride, su + row * su_stride, static_cast(w / 2)); - std::memcpy(out->planes[2].data + row * uv_stride, sv + row * sv_stride, static_cast(w / 2)); - } - return out; - } - - if (src->format == PixelFormat::RGB || src->format == PixelFormat::BGR) { - const uint8_t* s = plane_ptr(0); - if (!s) s = src->data; - if (!s) return nullptr; - const int s_stride = plane_stride(0, w * 3); - const int dst_stride = w * 3; - const size_t bytes = static_cast(dst_stride) * static_cast(h); - auto buf = std::make_shared>(bytes); - out->data_owner = buf; - out->data = buf->data(); - out->data_size = buf->size(); - out->stride = dst_stride; - out->plane_count = 1; - out->planes[0] = {out->data, dst_stride, static_cast(bytes), 0}; - for (int row = 0; row < h; ++row) { - std::memcpy(out->data + row * dst_stride, s + row * s_stride, static_cast(dst_stride)); - } - return out; - } - - return nullptr; + static std::shared_ptr TryGetEncodedMeta(const std::shared_ptr& frame) { + if (!frame || !frame->user_meta) return nullptr; + auto meta = std::static_pointer_cast(frame->user_meta); + if (!meta || meta->magic != EncodedVideoFrameMeta::kMagic) return nullptr; + if (!meta->codec || meta->pkt.data.empty() || meta->pkt.pts_ms <= 0) return nullptr; + return meta; } public: @@ -165,6 +42,16 @@ public: bool Init(const SimpleJson& config, const NodeContext& ctx) override { id_ = config.ValueOr("id", "alarm"); + // Optional evaluation throttle (for alarm side). 0 = disabled. + eval_interval_ms_ = std::max(0, static_cast(config.ValueOr("eval_interval_ms", 0))); + if (eval_interval_ms_ <= 0) { + const double eval_fps = config.ValueOr("eval_fps", 0.0); + if (eval_fps > 0.0) { + eval_interval_ms_ = static_cast(1000.0 / eval_fps); + if (eval_interval_ms_ < 1) eval_interval_ms_ = 1; + } + } + // Parse labels for class name mapping if (const SimpleJson* labels_cfg = config.Find("labels")) { for (const auto& label : labels_cfg->AsArray()) { @@ -180,7 +67,7 @@ public: } } - // Get pre/post buffer settings from clip action config + // Derive packet ring window from clip action config int pre_sec = 5; int post_sec = 0; int fps_hint = 25; @@ -191,7 +78,8 @@ public: fps_hint = clip_cfg->ValueOr("fps", 25); } } - frame_buffer_ = std::make_shared(pre_sec, post_sec, fps_hint); + const int window_sec = std::max(1, std::max(0, pre_sec) + std::max(0, post_sec) + 2); + packet_buffer_ = std::make_shared(window_sec, fps_hint); // Alarm worker queue (avoid blocking Process()) size_t alarm_q_size = 32; @@ -237,7 +125,7 @@ public: auto action = std::make_unique(); if (action->Init(*clip_cfg)) { auto* clip_ptr = static_cast(action.get()); - clip_ptr->SetFrameBuffer(frame_buffer_); + clip_ptr->SetPacketBuffer(packet_buffer_); actions_.push_back(std::move(action)); } } @@ -320,7 +208,7 @@ public: } } - // Pre/post buffer settings + // Packet ring window settings int pre_sec = 5; int post_sec = 0; int fps_hint = 25; @@ -331,7 +219,8 @@ public: fps_hint = clip_cfg->ValueOr("fps", 25); } } - auto new_frame_buffer = std::make_shared(pre_sec, post_sec, fps_hint); + const int window_sec = std::max(1, std::max(0, pre_sec) + std::max(0, post_sec) + 2); + auto new_packet_buffer = std::make_shared(window_sec, fps_hint); // Initialize new actions std::vector> new_actions; @@ -357,7 +246,7 @@ public: auto action = std::make_unique(); if (action->Init(*clip_cfg)) { auto* clip_ptr = static_cast(action.get()); - clip_ptr->SetFrameBuffer(new_frame_buffer); + clip_ptr->SetPacketBuffer(new_packet_buffer); new_actions.push_back(std::move(action)); } } @@ -396,11 +285,12 @@ public: std::lock_guard lock(mu_); labels_ = std::move(new_labels); rule_engine_ = std::move(new_engine); - frame_buffer_ = std::move(new_frame_buffer); + packet_buffer_ = std::move(new_packet_buffer); old_actions = std::move(actions_); actions_ = std::move(new_actions); in_flight_.store(0); + last_eval_pts_ms_ = 0; alarm_queue_ = std::make_shared>(alarm_queue_size_, alarm_queue_strategy_); worker_running_.store(true); worker_ = std::thread(&AlarmNode::WorkerLoop, this); @@ -423,23 +313,24 @@ public: NodeStatus Process(FramePtr frame) override { if (!frame) return NodeStatus::DROP; - // Copy pointer out of lock; FrameRingBuffer is internally synchronized. - std::shared_ptr fb; - { - std::lock_guard lock(mu_); - fb = frame_buffer_; - } - if (fb) { - using namespace std::chrono; - const uint64_t ts_ms = static_cast(duration_cast(steady_clock::now().time_since_epoch()).count()); - // IMPORTANT: store a CPU copy so the ring buffer doesn't extend DMA-BUF lifetime. - if (auto copied = CloneFrameForRingBuffer(frame)) { - fb->Push(std::move(copied), ts_ms); + if (packet_buffer_) { + if (auto meta = TryGetEncodedMeta(frame)) { + packet_buffer_->Push(std::move(meta)); } } { std::lock_guard lock(mu_); + + if (eval_interval_ms_ > 0 && frame->pts > 0) { + const int64_t pts_ms = static_cast(frame->pts / 1000ULL); + if (last_eval_pts_ms_ > 0 && (pts_ms - last_eval_pts_ms_) < eval_interval_ms_) { + ++processed_frames_; + return NodeStatus::OK; + } + last_eval_pts_ms_ = pts_ms; + } + auto result = rule_engine_.Evaluate(frame); if (result.matched) TriggerAlarm(result, frame); ++processed_frames_; @@ -489,7 +380,7 @@ private: std::string id_; std::vector labels_; RuleEngine rule_engine_; - std::shared_ptr frame_buffer_; + std::shared_ptr packet_buffer_; std::vector> actions_; mutable std::mutex mu_; @@ -504,6 +395,9 @@ private: std::shared_ptr> input_queue_; uint64_t processed_frames_ = 0; uint64_t alarm_count_ = 0; + + int64_t eval_interval_ms_ = 0; + int64_t last_eval_pts_ms_ = 0; }; REGISTER_NODE(AlarmNode, "alarm"); diff --git a/plugins/alarm/packet_ring_buffer.cpp b/plugins/alarm/packet_ring_buffer.cpp new file mode 100644 index 0000000..d7311f5 --- /dev/null +++ b/plugins/alarm/packet_ring_buffer.cpp @@ -0,0 +1,54 @@ +#include "packet_ring_buffer.h" + +#include + +namespace rk3588 { + +PacketRingBuffer::PacketRingBuffer(int window_sec, int fps_hint) { + const int sec = std::max(1, window_sec); + const int fps = std::max(1, fps_hint); + max_packets_ = static_cast(sec * fps); + if (max_packets_ < 50) max_packets_ = 50; +} + +void PacketRingBuffer::Push(std::shared_ptr meta) { + if (!meta || meta->magic != EncodedVideoFrameMeta::kMagic) return; + if (meta->pkt.data.empty()) return; + const int64_t pts_ms = meta->pkt.pts_ms; + if (pts_ms <= 0) return; + + std::lock_guard lock(mu_); + buf_.push_back(Entry{pts_ms, std::move(meta)}); + if (pts_ms > last_pts_ms_) last_pts_ms_ = pts_ms; + while (buf_.size() > max_packets_) buf_.pop_front(); +} + +std::vector> PacketRingBuffer::GetPacketsInRange( + int64_t start_pts_ms, int64_t end_pts_ms) const { + std::lock_guard lock(mu_); + std::vector> out; + for (const auto& e : buf_) { + if (e.pts_ms < start_pts_ms) continue; + if (e.pts_ms > end_pts_ms) break; + if (e.meta) out.push_back(e.meta); + } + return out; +} + +int64_t PacketRingBuffer::LatestPtsMs() const { + std::lock_guard lock(mu_); + return last_pts_ms_; +} + +size_t PacketRingBuffer::Size() const { + std::lock_guard lock(mu_); + return buf_.size(); +} + +void PacketRingBuffer::Clear() { + std::lock_guard lock(mu_); + buf_.clear(); + last_pts_ms_ = 0; +} + +} // namespace rk3588 diff --git a/plugins/alarm/packet_ring_buffer.h b/plugins/alarm/packet_ring_buffer.h new file mode 100644 index 0000000..de8c7e6 --- /dev/null +++ b/plugins/alarm/packet_ring_buffer.h @@ -0,0 +1,36 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include "media/encoded_video_meta.h" + +namespace rk3588 { + +class PacketRingBuffer { +public: + explicit PacketRingBuffer(int window_sec = 20, int fps_hint = 25); + + void Push(std::shared_ptr meta); + std::vector> GetPacketsInRange(int64_t start_pts_ms, + int64_t end_pts_ms) const; + int64_t LatestPtsMs() const; + size_t Size() const; + void Clear(); + +private: + struct Entry { + int64_t pts_ms = 0; + std::shared_ptr meta; + }; + + mutable std::mutex mu_; + std::deque buf_; + size_t max_packets_ = 0; + int64_t last_pts_ms_ = 0; +}; + +} // namespace rk3588 diff --git a/plugins/publish/publish_node.cpp b/plugins/publish/publish_node.cpp index 2b35d49..9b20afc 100644 --- a/plugins/publish/publish_node.cpp +++ b/plugins/publish/publish_node.cpp @@ -16,6 +16,8 @@ #include #include "node.h" +#include "media/encoded_video_meta.h" + #if defined(RK3588_ENABLE_FFMPEG) extern "C" { #include @@ -604,12 +606,14 @@ public: bool Init(const SimpleJson& config, const NodeContext& ctx) override { id_ = config.ValueOr("id", "publish"); input_queue_ = ctx.input_queue; + output_queues_ = ctx.output_queues; codec_ = config.ValueOr("codec", "h264"); fps_ = config.ValueOr("fps", 25); gop_ = config.ValueOr("gop", 50); bitrate_kbps_ = config.ValueOr("bitrate_kbps", 4000); use_mpp_ = config.ValueOr("use_mpp", true); use_ffmpeg_mux_ = config.ValueOr("use_ffmpeg_mux", true); + attach_encoded_meta_ = config.ValueOr("attach_encoded_meta", !output_queues_.empty()); const SimpleJson* outputs = config.Find("outputs"); if (outputs && outputs->IsArray()) { @@ -719,6 +723,7 @@ public: if (mpp_encoder_) mpp_encoder_->Shutdown(); encoder_ready_ = false; #endif + codec_info_.reset(); } return true; @@ -758,21 +763,32 @@ public: NodeStatus Process(FramePtr frame) override { if (!frame) return NodeStatus::DROP; - std::lock_guard lock(mu_); + { + std::lock_guard lock(mu_); #if defined(RK3588_ENABLE_MPP) - if (use_mpp_) { - ProcessMpp(frame); - } else { - ProcessStub(frame); - } + if (use_mpp_) { + ProcessMpp(frame); + } else { + ProcessStub(frame); + } #else - ProcessStub(frame); + ProcessStub(frame); #endif + } + + PushToDownstream(frame); return NodeStatus::OK; } private: + void PushToDownstream(const FramePtr& frame) { + if (output_queues_.empty()) return; + for (auto& q : output_queues_) { + if (q) q->Push(frame); + } + } + #if defined(RK3588_ENABLE_ZLMEDIAKIT) struct ZlmAppStream { std::string app; @@ -1198,7 +1214,23 @@ private: } const bool is_h265 = (codec_ == "h265" || codec_ == "hevc"); + + if (!codec_info_) codec_info_ = std::make_shared(); + codec_info_->codec = is_h265 ? VideoCodec::H265 : VideoCodec::H264; + codec_info_->width = frame->width; + codec_info_->height = frame->height; + codec_info_->fps = fps_; + codec_info_->extradata = mpp_encoder_->Header(); + mpp_encoder_->Encode(frame, [&](const EncodedPacket& pkt) { + if (attach_encoded_meta_ && codec_info_) { + auto meta = std::make_shared(); + meta->codec = codec_info_; + meta->pkt.data = pkt.data; + meta->pkt.key = pkt.key; + meta->pkt.pts_ms = pkt.pts_ms; + frame->user_meta = meta; + } ++encoded_frames_; if (encoded_frames_ % 100 == 0) { std::cout << "[publish] encoded frame " << encoded_frames_ @@ -1232,10 +1264,14 @@ private: std::vector zlm_outputs_; std::shared_ptr> input_queue_; + std::vector>> output_queues_; uint64_t encoded_frames_ = 0; mutable std::mutex mu_; + bool attach_encoded_meta_ = false; + std::shared_ptr codec_info_; + #if defined(RK3588_ENABLE_MPP) std::unique_ptr mpp_encoder_; bool encoder_ready_ = false;