优化性能
This commit is contained in:
parent
490e5db0e5
commit
80e7abcd08
@ -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"]
|
||||
]
|
||||
}
|
||||
},
|
||||
|
||||
@ -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"]
|
||||
]
|
||||
}
|
||||
},
|
||||
|
||||
@ -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"]
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
39
include/media/encoded_video_meta.h
Normal file
39
include/media/encoded_video_meta.h
Normal file
@ -0,0 +1,39 @@
|
||||
#pragma once
|
||||
|
||||
#include <cstdint>
|
||||
#include <memory>
|
||||
#include <vector>
|
||||
|
||||
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<uint8_t> extradata;
|
||||
};
|
||||
|
||||
struct EncodedVideoPacket {
|
||||
std::vector<uint8_t> 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<StreamCodecInfo> codec;
|
||||
EncodedVideoPacket pkt;
|
||||
};
|
||||
|
||||
} // namespace rk3588
|
||||
@ -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
|
||||
|
||||
@ -273,6 +273,16 @@ public:
|
||||
model_input_h_ = config.ValueOr<int>("model_h", 640);
|
||||
num_classes_ = config.ValueOr<int>("num_classes", 80);
|
||||
|
||||
// Optional inference throttle. 0 = run every frame.
|
||||
infer_interval_ms_ = std::max<int64_t>(0, static_cast<int64_t>(config.ValueOr<int>("infer_interval_ms", 0)));
|
||||
if (infer_interval_ms_ <= 0) {
|
||||
const double infer_fps = config.ValueOr<double>("infer_fps", 0.0);
|
||||
if (infer_fps > 0.0) {
|
||||
infer_interval_ms_ = static_cast<int64_t>(1000.0 / infer_fps);
|
||||
if (infer_interval_ms_ < 1) infer_interval_ms_ = 1;
|
||||
}
|
||||
}
|
||||
|
||||
std::string ver = config.ValueOr<std::string>("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<int64_t>(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<std::shared_ptr<SpscQueue<FramePtr>>> 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;
|
||||
|
||||
@ -2,7 +2,6 @@
|
||||
|
||||
#include <algorithm>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <ctime>
|
||||
#include <filesystem>
|
||||
@ -16,8 +15,7 @@
|
||||
extern "C" {
|
||||
#include <libavcodec/avcodec.h>
|
||||
#include <libavformat/avformat.h>
|
||||
#include <libavutil/imgutils.h>
|
||||
#include <libavutil/opt.h>
|
||||
#include <libavutil/avutil.h>
|
||||
}
|
||||
#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<uint8_t>(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<size_t>(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<size_t>(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<size_t>(uv_w));
|
||||
std::memcpy(v + row * v_stride, sv + row * sv_stride, static_cast<size_t>(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<size_t>(sy_stride) * static_cast<size_t>(height);
|
||||
}
|
||||
|
||||
for (int row = 0; row < height; ++row) {
|
||||
std::memcpy(y + row * y_stride, sy + row * sy_stride, static_cast<size_t>(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<int>("pre_sec", 5);
|
||||
post_sec_ = config.ValueOr<int>("post_sec", 10);
|
||||
@ -204,62 +47,78 @@ bool ClipAction::Init(const SimpleJson& config) {
|
||||
}
|
||||
|
||||
void ClipAction::Execute(AlarmEvent& event, std::shared_ptr<Frame> 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<uint64_t>(duration_cast<milliseconds>(steady_clock::now().time_since_epoch()).count());
|
||||
const int64_t trigger_pts_ms = frame->pts > 0 ? static_cast<int64_t>(frame->pts / 1000ULL) : 0;
|
||||
if (trigger_pts_ms <= 0) {
|
||||
std::cerr << "[ClipAction] invalid trigger pts\n";
|
||||
return;
|
||||
}
|
||||
|
||||
const uint64_t pre_ms = static_cast<uint64_t>(std::max(0, pre_sec_)) * 1000ULL;
|
||||
const uint64_t post_ms = static_cast<uint64_t>(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<int64_t>(std::max(0, pre_sec_)) * 1000LL;
|
||||
const int64_t post_ms = static_cast<int64_t>(std::max(0, post_sec_)) * 1000LL;
|
||||
const int64_t start_pts = std::max<int64_t>(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<std::shared_ptr<Frame>>& frames, const AlarmEvent& event) {
|
||||
std::string ClipAction::ProcessClipFromPackets(const std::vector<std::shared_ptr<EncodedVideoFrameMeta>>& metas,
|
||||
const AlarmEvent& event) {
|
||||
#if HAS_FFMPEG
|
||||
if (frames.empty()) {
|
||||
std::cerr << "[ClipAction] no frames to process\n";
|
||||
std::shared_ptr<EncodedVideoFrameMeta> 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<Frame> 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<std::shared_ptr<Frame>>& 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<int>(first->codec->extradata.size());
|
||||
stream->codecpar->extradata = static_cast<uint8_t*>(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>& 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<int>(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<size_t>(file.tellg());
|
||||
file.seekg(0);
|
||||
std::vector<uint8_t> file_data(file_size);
|
||||
file.read(reinterpret_cast<char*>(file_data.data()), file_size);
|
||||
file.read(reinterpret_cast<char*>(file_data.data()), static_cast<std::streamsize>(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) {
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
#include "action_base.h"
|
||||
#include "../uploaders/uploader_base.h"
|
||||
#include "../frame_ring_buffer.h"
|
||||
#include "../packet_ring_buffer.h"
|
||||
|
||||
#include <memory>
|
||||
#include <vector>
|
||||
@ -18,12 +18,11 @@ public:
|
||||
void Stop() override {}
|
||||
std::string Name() const override { return "clip"; }
|
||||
|
||||
void SetFrameBuffer(std::shared_ptr<FrameRingBuffer> buffer) {
|
||||
frame_buffer_ = buffer;
|
||||
}
|
||||
void SetPacketBuffer(std::shared_ptr<PacketRingBuffer> buffer) { packet_buffer_ = buffer; }
|
||||
|
||||
private:
|
||||
std::string ProcessClip(const std::vector<std::shared_ptr<Frame>>& frames, const AlarmEvent& event);
|
||||
std::string ProcessClipFromPackets(const std::vector<std::shared_ptr<EncodedVideoFrameMeta>>& 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<IUploader> uploader_;
|
||||
std::shared_ptr<FrameRingBuffer> frame_buffer_;
|
||||
std::shared_ptr<PacketRingBuffer> packet_buffer_;
|
||||
};
|
||||
|
||||
} // namespace rk3588
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <algorithm>
|
||||
#include <iostream>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
@ -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> frame;
|
||||
};
|
||||
|
||||
static std::shared_ptr<Frame> CloneFrameForRingBuffer(const std::shared_ptr<Frame>& 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<Frame>();
|
||||
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<size_t>(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<size_t>(sy_stride) * static_cast<size_t>(h);
|
||||
}
|
||||
|
||||
const int y_stride = w;
|
||||
const int uv_stride = w;
|
||||
const size_t y_bytes = static_cast<size_t>(y_stride) * static_cast<size_t>(h);
|
||||
const size_t uv_bytes = static_cast<size_t>(uv_stride) * static_cast<size_t>(h / 2);
|
||||
auto buf = std::make_shared<std::vector<uint8_t>>(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<int>(y_bytes), 0};
|
||||
out->planes[1] = {out->data + y_bytes, uv_stride, static_cast<int>(uv_bytes), static_cast<int>(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<size_t>(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<size_t>(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<size_t>(y_stride) * static_cast<size_t>(h);
|
||||
const size_t u_bytes = static_cast<size_t>(uv_stride) * static_cast<size_t>(h / 2);
|
||||
const size_t v_bytes = u_bytes;
|
||||
auto buf = std::make_shared<std::vector<uint8_t>>(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<int>(y_bytes), 0};
|
||||
out->planes[1] = {out->data + y_bytes, uv_stride, static_cast<int>(u_bytes), static_cast<int>(y_bytes)};
|
||||
out->planes[2] = {out->data + y_bytes + u_bytes, uv_stride, static_cast<int>(v_bytes), static_cast<int>(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<size_t>(w));
|
||||
}
|
||||
for (int row = 0; row < h / 2; ++row) {
|
||||
std::memcpy(out->planes[1].data + row * uv_stride, su + row * su_stride, static_cast<size_t>(w / 2));
|
||||
std::memcpy(out->planes[2].data + row * uv_stride, sv + row * sv_stride, static_cast<size_t>(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<size_t>(dst_stride) * static_cast<size_t>(h);
|
||||
auto buf = std::make_shared<std::vector<uint8_t>>(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<int>(bytes), 0};
|
||||
for (int row = 0; row < h; ++row) {
|
||||
std::memcpy(out->data + row * dst_stride, s + row * s_stride, static_cast<size_t>(dst_stride));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
return nullptr;
|
||||
static std::shared_ptr<EncodedVideoFrameMeta> TryGetEncodedMeta(const std::shared_ptr<Frame>& frame) {
|
||||
if (!frame || !frame->user_meta) return nullptr;
|
||||
auto meta = std::static_pointer_cast<EncodedVideoFrameMeta>(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<std::string>("id", "alarm");
|
||||
|
||||
// Optional evaluation throttle (for alarm side). 0 = disabled.
|
||||
eval_interval_ms_ = std::max<int64_t>(0, static_cast<int64_t>(config.ValueOr<int>("eval_interval_ms", 0)));
|
||||
if (eval_interval_ms_ <= 0) {
|
||||
const double eval_fps = config.ValueOr<double>("eval_fps", 0.0);
|
||||
if (eval_fps > 0.0) {
|
||||
eval_interval_ms_ = static_cast<int64_t>(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<int>("fps", 25);
|
||||
}
|
||||
}
|
||||
frame_buffer_ = std::make_shared<FrameRingBuffer>(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<PacketRingBuffer>(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<ClipAction>();
|
||||
if (action->Init(*clip_cfg)) {
|
||||
auto* clip_ptr = static_cast<ClipAction*>(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<int>("fps", 25);
|
||||
}
|
||||
}
|
||||
auto new_frame_buffer = std::make_shared<FrameRingBuffer>(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<PacketRingBuffer>(window_sec, fps_hint);
|
||||
|
||||
// Initialize new actions
|
||||
std::vector<std::unique_ptr<IAlarmAction>> new_actions;
|
||||
@ -357,7 +246,7 @@ public:
|
||||
auto action = std::make_unique<ClipAction>();
|
||||
if (action->Init(*clip_cfg)) {
|
||||
auto* clip_ptr = static_cast<ClipAction*>(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<std::mutex> 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<SpscQueue<AlarmJob>>(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<FrameRingBuffer> fb;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
fb = frame_buffer_;
|
||||
}
|
||||
if (fb) {
|
||||
using namespace std::chrono;
|
||||
const uint64_t ts_ms = static_cast<uint64_t>(duration_cast<milliseconds>(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<std::mutex> lock(mu_);
|
||||
|
||||
if (eval_interval_ms_ > 0 && frame->pts > 0) {
|
||||
const int64_t pts_ms = static_cast<int64_t>(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<std::string> labels_;
|
||||
RuleEngine rule_engine_;
|
||||
std::shared_ptr<FrameRingBuffer> frame_buffer_;
|
||||
std::shared_ptr<PacketRingBuffer> packet_buffer_;
|
||||
std::vector<std::unique_ptr<IAlarmAction>> actions_;
|
||||
|
||||
mutable std::mutex mu_;
|
||||
@ -504,6 +395,9 @@ private:
|
||||
std::shared_ptr<SpscQueue<FramePtr>> 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");
|
||||
|
||||
54
plugins/alarm/packet_ring_buffer.cpp
Normal file
54
plugins/alarm/packet_ring_buffer.cpp
Normal file
@ -0,0 +1,54 @@
|
||||
#include "packet_ring_buffer.h"
|
||||
|
||||
#include <algorithm>
|
||||
|
||||
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<size_t>(sec * fps);
|
||||
if (max_packets_ < 50) max_packets_ = 50;
|
||||
}
|
||||
|
||||
void PacketRingBuffer::Push(std::shared_ptr<EncodedVideoFrameMeta> 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<std::mutex> 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<std::shared_ptr<EncodedVideoFrameMeta>> PacketRingBuffer::GetPacketsInRange(
|
||||
int64_t start_pts_ms, int64_t end_pts_ms) const {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
std::vector<std::shared_ptr<EncodedVideoFrameMeta>> 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<std::mutex> lock(mu_);
|
||||
return last_pts_ms_;
|
||||
}
|
||||
|
||||
size_t PacketRingBuffer::Size() const {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
return buf_.size();
|
||||
}
|
||||
|
||||
void PacketRingBuffer::Clear() {
|
||||
std::lock_guard<std::mutex> lock(mu_);
|
||||
buf_.clear();
|
||||
last_pts_ms_ = 0;
|
||||
}
|
||||
|
||||
} // namespace rk3588
|
||||
36
plugins/alarm/packet_ring_buffer.h
Normal file
36
plugins/alarm/packet_ring_buffer.h
Normal file
@ -0,0 +1,36 @@
|
||||
#pragma once
|
||||
|
||||
#include <cstdint>
|
||||
#include <deque>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <vector>
|
||||
|
||||
#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<EncodedVideoFrameMeta> meta);
|
||||
std::vector<std::shared_ptr<EncodedVideoFrameMeta>> 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<EncodedVideoFrameMeta> meta;
|
||||
};
|
||||
|
||||
mutable std::mutex mu_;
|
||||
std::deque<Entry> buf_;
|
||||
size_t max_packets_ = 0;
|
||||
int64_t last_pts_ms_ = 0;
|
||||
};
|
||||
|
||||
} // namespace rk3588
|
||||
@ -16,6 +16,8 @@
|
||||
#include <unordered_set>
|
||||
#include "node.h"
|
||||
|
||||
#include "media/encoded_video_meta.h"
|
||||
|
||||
#if defined(RK3588_ENABLE_FFMPEG)
|
||||
extern "C" {
|
||||
#include <libavformat/avformat.h>
|
||||
@ -604,12 +606,14 @@ public:
|
||||
bool Init(const SimpleJson& config, const NodeContext& ctx) override {
|
||||
id_ = config.ValueOr<std::string>("id", "publish");
|
||||
input_queue_ = ctx.input_queue;
|
||||
output_queues_ = ctx.output_queues;
|
||||
codec_ = config.ValueOr<std::string>("codec", "h264");
|
||||
fps_ = config.ValueOr<int>("fps", 25);
|
||||
gop_ = config.ValueOr<int>("gop", 50);
|
||||
bitrate_kbps_ = config.ValueOr<int>("bitrate_kbps", 4000);
|
||||
use_mpp_ = config.ValueOr<bool>("use_mpp", true);
|
||||
use_ffmpeg_mux_ = config.ValueOr<bool>("use_ffmpeg_mux", true);
|
||||
attach_encoded_meta_ = config.ValueOr<bool>("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<std::mutex> lock(mu_);
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<StreamCodecInfo>();
|
||||
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<EncodedVideoFrameMeta>();
|
||||
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<OutputConfig> zlm_outputs_;
|
||||
|
||||
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
|
||||
std::vector<std::shared_ptr<SpscQueue<FramePtr>>> output_queues_;
|
||||
uint64_t encoded_frames_ = 0;
|
||||
|
||||
mutable std::mutex mu_;
|
||||
|
||||
bool attach_encoded_meta_ = false;
|
||||
std::shared_ptr<StreamCodecInfo> codec_info_;
|
||||
|
||||
#if defined(RK3588_ENABLE_MPP)
|
||||
std::unique_ptr<MppVencEncoder> mpp_encoder_;
|
||||
bool encoder_ready_ = false;
|
||||
|
||||
Loading…
Reference in New Issue
Block a user