diff --git a/Readme.md b/Readme.md index 683decd..92fe3e6 100644 --- a/Readme.md +++ b/Readme.md @@ -497,13 +497,125 @@ GraphMgr 负责： | 节点类型 | 功能描述 | 关键参数 | 硬件依赖 | | :---------- | :--------------------------------- | :----------------------------------- | :----------------- | | `osd` | 绘制检测框、文字等 | `draw_bbox`, `draw_text`, `labels`, `line_width`, `font_scale` | RGA / CPU | -| `alarm` | 报警规则（区域、时间、目标类型） | `rules`（含 ROI、时间段、阈值等） | CPU | -| `storage` | 录像（MP4/TS 切片） | `mode`, `segment_sec`, `path` | CPU（IO） | +| `alarm` | 报警规则 + 多种报警动作 | `rules`, `actions`（见下文详细说明） | CPU | +| `storage` | 持续录像（MP4/TS 切片） | `mode`, `segment_sec`, `path` | MPP VENC + IO | | `publish` | 编码并推流（RTSP/HLS） | `codec`, `bitrate`, `outputs` | MPP VENC + 网络 | **osd 参数说明：** - `labels`: 自定义类别标签数组，如 `["cat", "dog", "bird"]`，为空则使用 COCO 默认标签 +### 6.4 alarm 插件详细说明 + +alarm 插件采用**模块化 Actions 架构**，支持多种报警动作的灵活组合： + +**架构设计：** +``` +┌─────────────────────────────────────────────────────────┐ +│ AlarmNode │ +│ ┌──────────────┐ ┌──────────────────────────────────┐ │ +│ │ RuleEngine │ │ FrameRingBuffer (pre-event) │ │ +│ │ - 类别过滤 │ │ - 保存最近 N 秒的帧 │ │ +│ │ - ROI 检测 │ └──────────────────────────────────┘ │ +│ │ - 持续时间 │ │ +│ │ - 冷却控制 │ ┌──────────────────────────────────┐ │ +│ └──────────────┘ │ AlarmActions (可配置) │ │ +│ │ - LogAction → 日志输出 │ │ +│ │ - HttpAction → HTTP 回调 │ │ +│ │ - SnapshotAction → 截图+上传 │ │ +│ │ - ClipAction → 视频片段+上传 │ │ +│ └──────────────────────────────────┘ │ +└─────────────────────────────────────────────────────────┘ +``` + +**完整配置示例：** +```json +{ + "id": "alarm_cam1", + "type": "alarm", + "role": "sink", + "labels": ["矿体A", "矿体B", "异物"], + "rules": [ + { + "name": "ore_detection", + "class_ids": [0, 1], + "objects": ["矿体A"], + "roi": { "x": 0.1, "y": 0.1, "w": 0.8, "h": 0.8 }, + "min_duration_ms": 500, + "cooldown_ms": 5000, + "schedule": "08:00-20:00" + } + ], + "actions": { + "log": { "enable": true, "level": "info" }, + "http": { + "enable": true, + "url": "http://业务系统/api/alarm", + "timeout_ms": 3000, + "include_media_url": true + }, + "snapshot": { + "enable": true, + "format": "jpg", + "quality": 85, + "upload": { + "type": "minio", + "endpoint": "http://minio:9000", + "bucket": "alarms", + "access_key": "${MINIO_ACCESS_KEY}", + "secret_key": "${MINIO_SECRET_KEY}" + } + }, + "clip": { + "enable": true, + "pre_sec": 5, + "post_sec": 10, + "format": "mp4", + "upload": { "type": "local", "path": "/tmp/alarms" } + } + } +} +``` + +**规则参数说明：** +- `class_ids`: 要监控的类别 ID 数组（与模型输出对应） +- `objects`: 要监控的类别名称数组（需配合 `labels` 使用） +- `roi`: 感兴趣区域，归一化坐标 (0-1) +- `min_duration_ms`: 目标在 ROI 内持续多久才触发报警 +- `cooldown_ms`: 同一规则连续触发的冷却时间 +- `schedule`: 有效时间段，格式 "HH:MM-HH:MM" + +**上传器类型：** +- `local`: 保存到本地文件系统 +- `minio`: 上传到 MinIO 对象存储 +- `s3`: 上传到 AWS S3（兼容 MinIO） + +### 6.5 storage 插件详细说明 + +storage 插件用于**持续录像**（7x24 小时），与 alarm 的事件录像功能互补。 + +**配置示例：** +```json +{ + "id": "storage_cam1", + "type": "storage", + "role": "sink", + "mode": "continuous", + "format": "mp4", + "codec": "h264", + "segment_sec": 300, + "path": "/rec/cam1", + "filename_pattern": "%Y%m%d/%H%M%S", + "fps": 25, + "bitrate_kbps": 2000 +} +``` + +**参数说明：** +- `mode`: 录像模式，当前支持 `continuous`（持续录像） +- `format`: 输出格式，`mp4` 或 `ts` +- `segment_sec`: 单个文件时长（秒） +- `filename_pattern`: 文件名模式，支持 strftime 格式 + --- ## 七、线程模型（Thread Model） diff --git a/configs/sample_security_pipeline.json b/configs/sample_security_pipeline.json new file mode 100644 index 0000000..59f7c45 --- /dev/null +++ b/configs/sample_security_pipeline.json @@ -0,0 +1,151 @@ +{ + "queue": { "size": 8, "strategy": "drop_oldest" }, + "graphs": [ + { + "name": "cam1_security_pipeline", + "nodes": [ + { + "id": "in_cam1", + "type": "input_rtsp", + "role": "source", + "enable": true, + "url": "rtsp://192.168.1.100:554/stream", + "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, + "dst_format": "rgb", + "keep_ratio": false, + "use_rga": true + }, + { + "id": "ai_cam1", + "type": "ai_yolo", + "role": "filter", + "enable": true, + "model_path": "/models/yolov8n.rknn", + "model_version": "v8", + "num_classes": 80, + "conf": 0.5, + "nms": 0.45, + "class_filter": [] + }, + { + "id": "alarm_cam1", + "type": "alarm", + "role": "sink", + "enable": true, + "labels": [], + "rules": [ + { + "name": "object_detection", + "class_ids": [0, 1, 2], + "roi": { "x": 0.1, "y": 0.1, "w": 0.8, "h": 0.8 }, + "min_duration_ms": 500, + "cooldown_ms": 5000 + } + ], + "actions": { + "log": { "enable": true, "level": "info" }, + "http": { + "enable": false, + "url": "http://localhost:8080/api/alarm", + "timeout_ms": 3000, + "include_media_url": true + }, + "snapshot": { + "enable": true, + "format": "jpg", + "quality": 85, + "upload": { + "type": "local", + "path": "/tmp/alarms" + } + }, + "clip": { + "enable": false, + "pre_sec": 5, + "post_sec": 10, + "format": "mp4", + "fps": 25, + "upload": { + "type": "local", + "path": "/tmp/alarms" + } + } + } + }, + { + "id": "osd_cam1", + "type": "osd", + "role": "filter", + "enable": true, + "draw_bbox": true, + "draw_text": true, + "line_width": 2, + "font_scale": 1, + "labels": [] + }, + { + "id": "post_cam1", + "type": "preprocess", + "role": "filter", + "enable": true, + "dst_w": 1920, + "dst_h": 1080, + "dst_format": "nv12", + "keep_ratio": false, + "use_rga": true + }, + { + "id": "storage_cam1", + "type": "storage", + "role": "sink", + "enable": true, + "mode": "continuous", + "format": "mp4", + "codec": "h264", + "segment_sec": 300, + "path": "/rec/cam1", + "filename_pattern": "%Y%m%d/%H%M%S", + "fps": 25, + "bitrate_kbps": 2000 + }, + { + "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", "ai_cam1"], + ["ai_cam1", "alarm_cam1"], + ["ai_cam1", "osd_cam1"], + ["osd_cam1", "post_cam1"], + ["post_cam1", "storage_cam1"], + ["post_cam1", "pub_cam1"] + ] + } + ] +} diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index d8cf6c8..a89a3bf 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -203,7 +203,73 @@ set_target_properties(osd PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} ) -install(TARGETS input_rtsp publish preprocess ai_yolo osd ai_scheduler +# alarm plugin (rule-based alarm with actions: log, http, snapshot, clip) +add_library(alarm SHARED + alarm/alarm_node.cpp + alarm/rule_engine.cpp + alarm/frame_ring_buffer.cpp + alarm/actions/log_action.cpp + alarm/actions/http_action.cpp + alarm/actions/snapshot_action.cpp + alarm/actions/clip_action.cpp + alarm/uploaders/local_uploader.cpp + alarm/uploaders/minio_uploader.cpp + alarm/uploaders/uploader_factory.cpp +) +target_include_directories(alarm PRIVATE + ${CMAKE_SOURCE_DIR}/include + ${CMAKE_SOURCE_DIR}/third_party + ${CMAKE_CURRENT_SOURCE_DIR}/alarm +) +target_link_libraries(alarm PRIVATE project_options Threads::Threads) +if(RK3588_ENABLE_FFMPEG) + target_compile_definitions(alarm PRIVATE RK3588_ENABLE_FFMPEG) + target_include_directories(alarm PRIVATE ${FFMPEG_INCLUDE_DIRS}) + target_link_libraries(alarm PRIVATE PkgConfig::FFMPEG) + # Also need libswscale for image conversion + pkg_check_modules(SWSCALE IMPORTED_TARGET libswscale) + if(SWSCALE_FOUND) + target_link_libraries(alarm PRIVATE PkgConfig::SWSCALE) + endif() +endif() +# Optional: libcurl for HTTP action (on Linux) +find_package(CURL QUIET) +if(CURL_FOUND) + target_link_libraries(alarm PRIVATE CURL::libcurl) +endif() +set_target_properties(alarm PROPERTIES + OUTPUT_NAME "alarm" + LIBRARY_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} + RUNTIME_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} +) + +# storage plugin (continuous recording with segment management) +add_library(storage SHARED storage/storage_node.cpp) +target_include_directories(storage PRIVATE ${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/third_party) +target_link_libraries(storage PRIVATE project_options Threads::Threads) +if(RK3588_ENABLE_FFMPEG) + target_compile_definitions(storage PRIVATE RK3588_ENABLE_FFMPEG) + target_include_directories(storage PRIVATE ${FFMPEG_INCLUDE_DIRS}) + target_link_libraries(storage PRIVATE PkgConfig::FFMPEG) + if(SWSCALE_FOUND) + target_link_libraries(storage PRIVATE PkgConfig::SWSCALE) + endif() +endif() +if(RK3588_ENABLE_MPP AND RK_MPP_LIB) + target_compile_definitions(storage PRIVATE RK3588_ENABLE_MPP) + target_include_directories(storage PRIVATE + ${RK_MPP_ROOT}/inc + ${RK_RKNN_ROOT}/examples/3rdparty/mpp/include + ) + target_link_libraries(storage PRIVATE ${RK_MPP_LIB}) +endif() +set_target_properties(storage PROPERTIES + OUTPUT_NAME "storage" + LIBRARY_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} + RUNTIME_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} +) + +install(TARGETS input_rtsp publish preprocess ai_yolo osd alarm storage ai_scheduler LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}/rk3588-media-server/plugins RUNTIME DESTINATION ${CMAKE_INSTALL_LIBDIR}/rk3588-media-server/plugins ) diff --git a/plugins/alarm/actions/action_base.h b/plugins/alarm/actions/action_base.h new file mode 100644 index 0000000..85197b1 --- /dev/null +++ b/plugins/alarm/actions/action_base.h @@ -0,0 +1,31 @@ +#pragma once + +#include +#include +#include + +#include "frame/frame.h" +#include "utils/simple_json.h" + +namespace rk3588 { + +struct AlarmEvent { + std::string rule_name; + std::string node_id; + uint64_t timestamp_ms; + uint64_t frame_id; + std::vector detections; + std::string snapshot_url; + std::string clip_url; +}; + +class IAlarmAction { +public: + virtual ~IAlarmAction() = default; + virtual bool Init(const SimpleJson& config) = 0; + virtual void Execute(AlarmEvent& event, std::shared_ptr frame) = 0; + virtual void Drain() {} + virtual std::string Name() const = 0; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/actions/clip_action.cpp b/plugins/alarm/actions/clip_action.cpp new file mode 100644 index 0000000..05312b9 --- /dev/null +++ b/plugins/alarm/actions/clip_action.cpp @@ -0,0 +1,321 @@ +#include "clip_action.h" + +#include +#include +#include +#include +#include +#include + +#if defined(RK3588_ENABLE_FFMPEG) +extern "C" { +#include +#include +#include +#include +#include +} +#define HAS_FFMPEG 1 +#else +#define HAS_FFMPEG 0 +#endif + +namespace rk3588 { + +ClipAction::~ClipAction() { + Drain(); + running_.store(false); + queue_cv_.notify_all(); + if (worker_.joinable()) worker_.join(); +} + +bool ClipAction::Init(const SimpleJson& config) { + pre_sec_ = config.ValueOr("pre_sec", 5); + post_sec_ = config.ValueOr("post_sec", 10); + format_ = config.ValueOr("format", "mp4"); + fps_ = config.ValueOr("fps", 25); + + if (const SimpleJson* upload_cfg = config.Find("upload")) { + uploader_ = CreateUploader(*upload_cfg); + if (!uploader_) { + std::cerr << "[ClipAction] failed to create uploader\n"; + return false; + } + } else { + SimpleJson default_cfg; + uploader_ = CreateUploader(default_cfg); + } + + running_.store(true); + worker_ = std::thread(&ClipAction::WorkerLoop, this); + + std::cout << "[ClipAction] initialized, pre=" << pre_sec_ << "s post=" << post_sec_ + << "s format=" << format_ << "\n"; + return true; +} + +void ClipAction::Execute(AlarmEvent& event, std::shared_ptr frame) { + ClipTask task; + task.event = event; + task.trigger_frame = frame; + + // Get pre-event frames + if (frame_buffer_) { + task.pre_frames = frame_buffer_->GetPreEventFrames(); + } + + // Start collecting post-event frames + { + std::lock_guard lock(post_mutex_); + collecting_post_ = true; + post_frames_.clear(); + post_frames_needed_ = post_sec_ * fps_; + } + + // Wait for post-event frames (simplified: just queue the task) + // In a real implementation, we'd collect post-event frames asynchronously + + { + std::lock_guard lock(queue_mutex_); + task_queue_.push(std::move(task)); + } + queue_cv_.notify_one(); +} + +void ClipAction::PushPostEventFrame(std::shared_ptr frame) { + std::lock_guard lock(post_mutex_); + if (!collecting_post_) return; + + post_frames_.push_back(frame); + if (static_cast(post_frames_.size()) >= post_frames_needed_) { + collecting_post_ = false; + } +} + +void ClipAction::Drain() { + std::unique_lock lock(queue_mutex_); + queue_cv_.wait_for(lock, std::chrono::seconds(30), + [this] { return task_queue_.empty(); }); +} + +void ClipAction::WorkerLoop() { + while (running_.load()) { + ClipTask task; + { + std::unique_lock lock(queue_mutex_); + queue_cv_.wait_for(lock, std::chrono::milliseconds(100), + [this] { return !task_queue_.empty() || !running_.load(); }); + if (!running_.load() && task_queue_.empty()) break; + if (task_queue_.empty()) continue; + task = std::move(task_queue_.front()); + task_queue_.pop(); + } + + // Wait a bit for post-event frames if needed + if (post_sec_ > 0) { + std::this_thread::sleep_for(std::chrono::seconds(post_sec_)); + } + + // Get collected post-event frames + { + std::lock_guard lock(post_mutex_); + // In real impl, we'd associate frames with the task + collecting_post_ = false; + } + + std::string url = ProcessClip(task); + if (!url.empty()) { + std::cout << "[ClipAction] clip uploaded: " << url << "\n"; + } + } +} + +std::string ClipAction::ProcessClip(ClipTask& task) { +#if HAS_FFMPEG + if (task.pre_frames.empty() && !task.trigger_frame) { + std::cerr << "[ClipAction] no frames to process\n"; + return ""; + } + + // Determine frame dimensions from first available frame + std::shared_ptr sample_frame = + task.trigger_frame ? task.trigger_frame + : (task.pre_frames.empty() ? nullptr : task.pre_frames[0]); + if (!sample_frame) return ""; + + int width = sample_frame->width; + int height = sample_frame->height; + + // Create temporary file + std::string tmp_path = "/tmp/clip_" + std::to_string(task.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.c_str()) < 0) { + 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); + 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_; + + if (fmt_ctx->oformat->flags & AVFMT_GLOBALHEADER) { + enc_ctx->flags |= AV_CODEC_FLAG_GLOBAL_HEADER; + } + + 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.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); + 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); + + AVPacket* pkt = av_packet_alloc(); + int64_t pts = 0; + + auto encode_frame = [&](std::shared_ptr& frame) { + if (!frame || !frame->data) return; + + // Convert to YUV420P (simplified - assumes RGB input) + AVPixelFormat src_fmt = AV_PIX_FMT_RGB24; + if (frame->format == PixelFormat::BGR) src_fmt = AV_PIX_FMT_BGR24; + else if (frame->format == PixelFormat::NV12) src_fmt = AV_PIX_FMT_NV12; + + SwsContext* sws = sws_getContext(width, height, src_fmt, + width, height, AV_PIX_FMT_YUV420P, + SWS_BILINEAR, nullptr, nullptr, nullptr); + if (sws) { + uint8_t* src_data[4] = {frame->data, nullptr, nullptr, nullptr}; + int src_linesize[4] = {width * 3, 0, 0, 0}; + if (frame->format == PixelFormat::NV12) { + src_data[1] = frame->data + width * height; + src_linesize[0] = width; + src_linesize[1] = width; + } + sws_scale(sws, src_data, src_linesize, 0, height, + av_frame->data, av_frame->linesize); + sws_freeContext(sws); + } + + 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); + } + }; + + // Encode pre-event frames + for (auto& f : task.pre_frames) { + encode_frame(f); + } + + // Encode trigger frame + encode_frame(task.trigger_frame); + + // 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); + pkt->stream_index = stream->index; + av_interleaved_write_frame(fmt_ctx, pkt); + av_packet_unref(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); + } + 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(); + file.seekg(0); + std::vector file_data(file_size); + file.read(reinterpret_cast(file_data.data()), file_size); + file.close(); + + std::string key = GenerateKey(task.event); + auto result = uploader_->Upload(key, file_data.data(), file_data.size(), "video/mp4"); + + // Clean up temp file + std::filesystem::remove(tmp_path); + + if (result.success) { + return result.url; + } + std::cerr << "[ClipAction] upload failed: " << result.error << "\n"; +#else + std::cerr << "[ClipAction] FFmpeg not enabled\n"; +#endif + return ""; +} + +std::string ClipAction::GenerateKey(const AlarmEvent& event) { + auto now = std::chrono::system_clock::now(); + auto time_t_now = std::chrono::system_clock::to_time_t(now); + std::tm* tm = std::localtime(&time_t_now); + + std::ostringstream oss; + oss << std::put_time(tm, "%Y%m%d") << "/" + << event.node_id << "_" + << std::put_time(tm, "%H%M%S") << "_" + << event.frame_id << "." << format_; + return oss.str(); +} + +} // namespace rk3588 diff --git a/plugins/alarm/actions/clip_action.h b/plugins/alarm/actions/clip_action.h new file mode 100644 index 0000000..2da5ff4 --- /dev/null +++ b/plugins/alarm/actions/clip_action.h @@ -0,0 +1,62 @@ +#pragma once + +#include "action_base.h" +#include "../uploaders/uploader_base.h" +#include "../frame_ring_buffer.h" + +#include +#include +#include +#include +#include +#include + +namespace rk3588 { + +struct ClipTask { + AlarmEvent event; + std::vector> pre_frames; + std::shared_ptr trigger_frame; +}; + +class ClipAction : public IAlarmAction { +public: + ~ClipAction(); + bool Init(const SimpleJson& config) override; + void Execute(AlarmEvent& event, std::shared_ptr frame) override; + void Drain() override; + std::string Name() const override { return "clip"; } + + void SetFrameBuffer(std::shared_ptr buffer) { + frame_buffer_ = buffer; + } + + void PushPostEventFrame(std::shared_ptr frame); + +private: + void WorkerLoop(); + std::string ProcessClip(ClipTask& task); + std::string GenerateKey(const AlarmEvent& event); + + int pre_sec_ = 5; + int post_sec_ = 10; + std::string format_ = "mp4"; + int fps_ = 25; + std::unique_ptr uploader_; + std::shared_ptr frame_buffer_; + + std::atomic running_{false}; + std::thread worker_; + std::mutex queue_mutex_; + std::condition_variable queue_cv_; + std::queue task_queue_; + + // Post-event collection state + std::mutex post_mutex_; + bool collecting_post_ = false; + std::vector> post_frames_; + int post_frames_needed_ = 0; + ClipTask* current_task_ = nullptr; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/actions/http_action.cpp b/plugins/alarm/actions/http_action.cpp new file mode 100644 index 0000000..58e0590 --- /dev/null +++ b/plugins/alarm/actions/http_action.cpp @@ -0,0 +1,145 @@ +#include "http_action.h" + +#include +#include +#include +#include + +#if defined(__linux__) || defined(__APPLE__) +#include +#define HAS_CURL 1 +#else +#define HAS_CURL 0 +#endif + +namespace rk3588 { + +HttpAction::~HttpAction() { + Drain(); + running_.store(false); + queue_cv_.notify_all(); + if (worker_.joinable()) worker_.join(); +} + +bool HttpAction::Init(const SimpleJson& config) { + url_ = config.ValueOr("url", ""); + timeout_ms_ = config.ValueOr("timeout_ms", 3000); + include_media_url_ = config.ValueOr("include_media_url", true); + method_ = config.ValueOr("method", "POST"); + + if (url_.empty()) { + std::cerr << "[HttpAction] url is required\n"; + return false; + } + +#if HAS_CURL + curl_global_init(CURL_GLOBAL_DEFAULT); +#endif + + running_.store(true); + worker_ = std::thread(&HttpAction::WorkerLoop, this); + std::cout << "[HttpAction] initialized, url=" << url_ << "\n"; + return true; +} + +void HttpAction::Execute(AlarmEvent& event, std::shared_ptr /*frame*/) { + // Build JSON body + std::ostringstream oss; + oss << "{"; + oss << "\"node_id\":\"" << event.node_id << "\","; + oss << "\"rule_name\":\"" << event.rule_name << "\","; + oss << "\"timestamp\":" << event.timestamp_ms << ","; + oss << "\"frame_id\":" << event.frame_id << ","; + + if (include_media_url_) { + if (!event.snapshot_url.empty()) { + oss << "\"snapshot_url\":\"" << event.snapshot_url << "\","; + } + if (!event.clip_url.empty()) { + oss << "\"clip_url\":\"" << event.clip_url << "\","; + } + } + + oss << "\"detections\":["; + for (size_t i = 0; i < event.detections.size(); ++i) { + const auto& det = event.detections[i]; + if (i > 0) oss << ","; + oss << "{\"cls_id\":" << det.cls_id + << ",\"score\":" << det.score + << ",\"bbox\":{\"x\":" << det.bbox.x + << ",\"y\":" << det.bbox.y + << ",\"w\":" << det.bbox.w + << ",\"h\":" << det.bbox.h << "}" + << ",\"track_id\":" << det.track_id << "}"; + } + oss << "]}"; + + std::string json_body = oss.str(); + + { + std::lock_guard lock(queue_mutex_); + request_queue_.push(json_body); + } + queue_cv_.notify_one(); +} + +void HttpAction::Drain() { + // Wait for queue to empty + std::unique_lock lock(queue_mutex_); + queue_cv_.wait_for(lock, std::chrono::seconds(5), + [this] { return request_queue_.empty(); }); +} + +void HttpAction::WorkerLoop() { + while (running_.load()) { + std::string body; + { + std::unique_lock lock(queue_mutex_); + queue_cv_.wait_for(lock, std::chrono::milliseconds(100), + [this] { return !request_queue_.empty() || !running_.load(); }); + if (!running_.load() && request_queue_.empty()) break; + if (request_queue_.empty()) continue; + body = std::move(request_queue_.front()); + request_queue_.pop(); + } + + if (!body.empty()) { + SendRequest(body); + } + } +} + +bool HttpAction::SendRequest(const std::string& json_body) { +#if HAS_CURL + CURL* curl = curl_easy_init(); + if (!curl) { + std::cerr << "[HttpAction] curl_easy_init failed\n"; + return false; + } + + struct curl_slist* headers = nullptr; + headers = curl_slist_append(headers, "Content-Type: application/json"); + + curl_easy_setopt(curl, CURLOPT_URL, url_.c_str()); + curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers); + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str()); + curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, timeout_ms_); + curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, timeout_ms_ / 2); + + CURLcode res = curl_easy_perform(curl); + if (res != CURLE_OK) { + std::cerr << "[HttpAction] request failed: " << curl_easy_strerror(res) << "\n"; + } + + curl_slist_free_all(headers); + curl_easy_cleanup(curl); + return res == CURLE_OK; +#else + // Stub implementation for Windows without curl + std::cout << "[HttpAction] HTTP POST to " << url_ << "\n" + << " Body: " << json_body << "\n"; + return true; +#endif +} + +} // namespace rk3588 diff --git a/plugins/alarm/actions/http_action.h b/plugins/alarm/actions/http_action.h new file mode 100644 index 0000000..ced29ed --- /dev/null +++ b/plugins/alarm/actions/http_action.h @@ -0,0 +1,37 @@ +#pragma once + +#include "action_base.h" + +#include +#include +#include +#include +#include + +namespace rk3588 { + +class HttpAction : public IAlarmAction { +public: + ~HttpAction(); + bool Init(const SimpleJson& config) override; + void Execute(AlarmEvent& event, std::shared_ptr frame) override; + void Drain() override; + std::string Name() const override { return "http"; } + +private: + void WorkerLoop(); + bool SendRequest(const std::string& json_body); + + std::string url_; + int timeout_ms_ = 3000; + bool include_media_url_ = true; + std::string method_ = "POST"; + + std::atomic running_{false}; + std::thread worker_; + std::mutex queue_mutex_; + std::condition_variable queue_cv_; + std::queue request_queue_; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/actions/log_action.cpp b/plugins/alarm/actions/log_action.cpp new file mode 100644 index 0000000..8e5a927 --- /dev/null +++ b/plugins/alarm/actions/log_action.cpp @@ -0,0 +1,55 @@ +#include "log_action.h" + +#include +#include +#include +#include +#include + +namespace rk3588 { + +bool LogAction::Init(const SimpleJson& config) { + level_ = config.ValueOr("level", "info"); + include_detections_ = config.ValueOr("include_detections", true); + std::cout << "[LogAction] initialized, level=" << level_ << "\n"; + return true; +} + +void LogAction::Execute(AlarmEvent& event, std::shared_ptr /*frame*/) { + auto now = std::chrono::system_clock::now(); + auto time_t_now = std::chrono::system_clock::to_time_t(now); + std::tm* tm_now = std::localtime(&time_t_now); + + std::ostringstream oss; + oss << "[ALARM][" << level_ << "] " + << std::put_time(tm_now, "%Y-%m-%d %H:%M:%S") + << " node=" << event.node_id + << " rule=" << event.rule_name + << " frame=" << event.frame_id; + + if (include_detections_ && !event.detections.empty()) { + oss << " detections=["; + for (size_t i = 0; i < event.detections.size(); ++i) { + const auto& det = event.detections[i]; + if (i > 0) oss << ", "; + oss << "{cls=" << det.cls_id + << " score=" << std::fixed << std::setprecision(2) << det.score + << " bbox=(" << static_cast(det.bbox.x) << "," + << static_cast(det.bbox.y) << "," + << static_cast(det.bbox.w) << "," + << static_cast(det.bbox.h) << ")}"; + } + oss << "]"; + } + + if (!event.snapshot_url.empty()) { + oss << " snapshot=" << event.snapshot_url; + } + if (!event.clip_url.empty()) { + oss << " clip=" << event.clip_url; + } + + std::cout << oss.str() << "\n"; +} + +} // namespace rk3588 diff --git a/plugins/alarm/actions/log_action.h b/plugins/alarm/actions/log_action.h new file mode 100644 index 0000000..57b075f --- /dev/null +++ b/plugins/alarm/actions/log_action.h @@ -0,0 +1,18 @@ +#pragma once + +#include "action_base.h" + +namespace rk3588 { + +class LogAction : public IAlarmAction { +public: + bool Init(const SimpleJson& config) override; + void Execute(AlarmEvent& event, std::shared_ptr frame) override; + std::string Name() const override { return "log"; } + +private: + std::string level_ = "info"; + bool include_detections_ = true; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/actions/snapshot_action.cpp b/plugins/alarm/actions/snapshot_action.cpp new file mode 100644 index 0000000..ea0a2a7 --- /dev/null +++ b/plugins/alarm/actions/snapshot_action.cpp @@ -0,0 +1,193 @@ +#include "snapshot_action.h" + +#include +#include +#include +#include +#include + +#if defined(RK3588_ENABLE_FFMPEG) +extern "C" { +#include +#include +#include +} +#define HAS_FFMPEG 1 +#else +#define HAS_FFMPEG 0 +#endif + +namespace rk3588 { + +bool SnapshotAction::Init(const SimpleJson& config) { + format_ = config.ValueOr("format", "jpg"); + quality_ = config.ValueOr("quality", 85); + + if (const SimpleJson* upload_cfg = config.Find("upload")) { + uploader_ = CreateUploader(*upload_cfg); + if (!uploader_) { + std::cerr << "[SnapshotAction] failed to create uploader\n"; + return false; + } + } else { + // Default to local storage + SimpleJson default_cfg; + uploader_ = CreateUploader(default_cfg); + } + + std::cout << "[SnapshotAction] initialized, format=" << format_ + << " quality=" << quality_ << "\n"; + return true; +} + +void SnapshotAction::Execute(AlarmEvent& event, std::shared_ptr frame) { + if (!frame || !frame->data) { + std::cerr << "[SnapshotAction] no frame data\n"; + return; + } + + auto jpeg_data = EncodeJpeg(frame); + if (jpeg_data.empty()) { + std::cerr << "[SnapshotAction] failed to encode JPEG\n"; + return; + } + + std::string key = GenerateKey(event); + auto result = uploader_->Upload(key, jpeg_data.data(), jpeg_data.size(), "image/jpeg"); + + if (result.success) { + event.snapshot_url = result.url; + std::cout << "[SnapshotAction] uploaded: " << result.url << "\n"; + } else { + std::cerr << "[SnapshotAction] upload failed: " << result.error << "\n"; + } +} + +std::string SnapshotAction::GenerateKey(const AlarmEvent& event) { + auto now = std::chrono::system_clock::now(); + auto time_t_now = std::chrono::system_clock::to_time_t(now); + std::tm* tm = std::localtime(&time_t_now); + + std::ostringstream oss; + oss << std::put_time(tm, "%Y%m%d") << "/" + << event.node_id << "_" + << std::put_time(tm, "%H%M%S") << "_" + << event.frame_id << "." << format_; + return oss.str(); +} + +std::vector SnapshotAction::EncodeJpeg(const std::shared_ptr& frame) { + std::vector output; + +#if HAS_FFMPEG + const AVCodec* codec = avcodec_find_encoder(AV_CODEC_ID_MJPEG); + if (!codec) { + std::cerr << "[SnapshotAction] MJPEG encoder not found\n"; + return output; + } + + AVCodecContext* ctx = avcodec_alloc_context3(codec); + if (!ctx) return output; + + ctx->width = frame->width; + ctx->height = frame->height; + ctx->time_base = AVRational{1, 25}; + ctx->pix_fmt = AV_PIX_FMT_YUVJ420P; + + // Set quality (1-31, lower is better for MJPEG) + int q = 31 - (quality_ * 30 / 100); + ctx->qmin = q; + ctx->qmax = q; + + if (avcodec_open2(ctx, codec, nullptr) < 0) { + avcodec_free_context(&ctx); + return output; + } + + AVFrame* av_frame = av_frame_alloc(); + av_frame->width = frame->width; + av_frame->height = frame->height; + av_frame->format = AV_PIX_FMT_YUVJ420P; + + if (av_frame_get_buffer(av_frame, 32) < 0) { + av_frame_free(&av_frame); + avcodec_free_context(&ctx); + return output; + } + + // Convert input frame to YUV420P + AVPixelFormat src_fmt = AV_PIX_FMT_NONE; + if (frame->format == PixelFormat::RGB) { + src_fmt = AV_PIX_FMT_RGB24; + } else if (frame->format == PixelFormat::BGR) { + src_fmt = AV_PIX_FMT_BGR24; + } else if (frame->format == PixelFormat::NV12) { + src_fmt = AV_PIX_FMT_NV12; + } else if (frame->format == PixelFormat::YUV420) { + src_fmt = AV_PIX_FMT_YUV420P; + } + + if (src_fmt == AV_PIX_FMT_NONE) { + av_frame_free(&av_frame); + avcodec_free_context(&ctx); + return output; + } + + SwsContext* sws = sws_getContext( + frame->width, frame->height, src_fmt, + frame->width, frame->height, AV_PIX_FMT_YUVJ420P, + SWS_BILINEAR, nullptr, nullptr, nullptr); + + if (!sws) { + av_frame_free(&av_frame); + avcodec_free_context(&ctx); + return output; + } + + // Set up source pointers + uint8_t* src_data[4] = {nullptr}; + int src_linesize[4] = {0}; + + if (frame->format == PixelFormat::RGB || frame->format == PixelFormat::BGR) { + src_data[0] = frame->data; + src_linesize[0] = frame->width * 3; + } else if (frame->format == PixelFormat::NV12) { + src_data[0] = frame->planes[0].data ? frame->planes[0].data : frame->data; + src_data[1] = frame->planes[1].data ? frame->planes[1].data + : frame->data + frame->width * frame->height; + src_linesize[0] = frame->planes[0].stride > 0 ? frame->planes[0].stride : frame->width; + src_linesize[1] = frame->planes[1].stride > 0 ? frame->planes[1].stride : frame->width; + } else if (frame->format == PixelFormat::YUV420) { + src_data[0] = frame->planes[0].data ? frame->planes[0].data : frame->data; + src_data[1] = frame->planes[1].data; + src_data[2] = frame->planes[2].data; + src_linesize[0] = frame->planes[0].stride > 0 ? frame->planes[0].stride : frame->width; + src_linesize[1] = frame->planes[1].stride > 0 ? frame->planes[1].stride : frame->width / 2; + src_linesize[2] = frame->planes[2].stride > 0 ? frame->planes[2].stride : frame->width / 2; + } + + sws_scale(sws, src_data, src_linesize, 0, frame->height, + av_frame->data, av_frame->linesize); + sws_freeContext(sws); + + av_frame->pts = 0; + + AVPacket* pkt = av_packet_alloc(); + if (avcodec_send_frame(ctx, av_frame) == 0) { + if (avcodec_receive_packet(ctx, pkt) == 0) { + output.assign(pkt->data, pkt->data + pkt->size); + } + } + + av_packet_free(&pkt); + av_frame_free(&av_frame); + avcodec_free_context(&ctx); +#else + // Stub: just create a minimal valid JPEG header for testing + std::cerr << "[SnapshotAction] FFmpeg not enabled, cannot encode JPEG\n"; +#endif + + return output; +} + +} // namespace rk3588 diff --git a/plugins/alarm/actions/snapshot_action.h b/plugins/alarm/actions/snapshot_action.h new file mode 100644 index 0000000..18fbe31 --- /dev/null +++ b/plugins/alarm/actions/snapshot_action.h @@ -0,0 +1,25 @@ +#pragma once + +#include "action_base.h" +#include "../uploaders/uploader_base.h" + +#include + +namespace rk3588 { + +class SnapshotAction : public IAlarmAction { +public: + bool Init(const SimpleJson& config) override; + void Execute(AlarmEvent& event, std::shared_ptr frame) override; + std::string Name() const override { return "snapshot"; } + +private: + std::vector EncodeJpeg(const std::shared_ptr& frame); + std::string GenerateKey(const AlarmEvent& event); + + std::string format_ = "jpg"; + int quality_ = 85; + std::unique_ptr uploader_; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/alarm_node.cpp b/plugins/alarm/alarm_node.cpp new file mode 100644 index 0000000..76f3266 --- /dev/null +++ b/plugins/alarm/alarm_node.cpp @@ -0,0 +1,205 @@ +#include +#include +#include +#include +#include +#include + +#include "node.h" +#include "rule_engine.h" +#include "frame_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" + +namespace rk3588 { + +class AlarmNode : public INode { +public: + std::string Id() const override { return id_; } + std::string Type() const override { return "alarm"; } + + bool Init(const SimpleJson& config, const NodeContext& ctx) override { + id_ = config.ValueOr("id", "alarm"); + + // Parse labels for class name mapping + if (const SimpleJson* labels_cfg = config.Find("labels")) { + for (const auto& label : labels_cfg->AsArray()) { + labels_.push_back(label.AsString("")); + } + } + + // Initialize rule engine + if (const SimpleJson* rules_cfg = config.Find("rules")) { + if (!rule_engine_.Init(*rules_cfg, labels_)) { + std::cerr << "[alarm] failed to init rule engine\n"; + return false; + } + } + + // Get pre-event buffer settings from clip action config + int pre_sec = 5; + int fps_hint = 25; + if (const SimpleJson* actions_cfg = config.Find("actions")) { + if (const SimpleJson* clip_cfg = actions_cfg->Find("clip")) { + pre_sec = clip_cfg->ValueOr("pre_sec", 5); + fps_hint = clip_cfg->ValueOr("fps", 25); + } + } + frame_buffer_ = std::make_shared(pre_sec, fps_hint); + + // Initialize actions + if (const SimpleJson* actions_cfg = config.Find("actions")) { + // Log action + if (const SimpleJson* log_cfg = actions_cfg->Find("log")) { + if (log_cfg->ValueOr("enable", true)) { + auto action = std::make_unique(); + if (action->Init(*log_cfg)) { + actions_.push_back(std::move(action)); + } + } + } + + // Snapshot action (must be before HTTP to fill snapshot_url) + if (const SimpleJson* snap_cfg = actions_cfg->Find("snapshot")) { + if (snap_cfg->ValueOr("enable", false)) { + auto action = std::make_unique(); + if (action->Init(*snap_cfg)) { + actions_.push_back(std::move(action)); + } + } + } + + // Clip action (must be before HTTP to fill clip_url) + if (const SimpleJson* clip_cfg = actions_cfg->Find("clip")) { + if (clip_cfg->ValueOr("enable", false)) { + auto action = std::make_unique(); + if (action->Init(*clip_cfg)) { + auto* clip_ptr = static_cast(action.get()); + clip_ptr->SetFrameBuffer(frame_buffer_); + clip_action_ = clip_ptr; + actions_.push_back(std::move(action)); + } + } + } + + // HTTP action (should be last to include media URLs) + if (const SimpleJson* http_cfg = actions_cfg->Find("http")) { + if (http_cfg->ValueOr("enable", false)) { + auto action = std::make_unique(); + if (action->Init(*http_cfg)) { + actions_.push_back(std::move(action)); + } + } + } + } else { + // Default: just log action + auto action = std::make_unique(); + SimpleJson empty_cfg; + action->Init(empty_cfg); + actions_.push_back(std::move(action)); + } + + input_queue_ = ctx.input_queue; + if (!input_queue_) { + std::cerr << "[alarm] no input queue for node " << id_ << "\n"; + return false; + } + + std::cout << "[alarm] initialized with " << actions_.size() << " actions\n"; + return true; + } + + bool Start() override { + if (!input_queue_) return false; + running_.store(true); + worker_ = std::thread(&AlarmNode::WorkerLoop, this); + std::cout << "[alarm] started\n"; + return true; + } + + void Stop() override { + running_.store(false); + if (input_queue_) input_queue_->Stop(); + if (worker_.joinable()) worker_.join(); + + // Drain all actions + for (auto& action : actions_) { + action->Drain(); + } + + std::cout << "[alarm] stopped, processed " << processed_frames_ + << " frames, triggered " << alarm_count_ << " alarms\n"; + } + + void Drain() override { + for (auto& action : actions_) { + action->Drain(); + } + } + +private: + void WorkerLoop() { + using namespace std::chrono; + FramePtr frame; + + while (running_.load()) { + if (!input_queue_->Pop(frame, milliseconds(200))) continue; + if (!frame) continue; + + // Always push to ring buffer for pre-event recording + frame_buffer_->Push(frame); + + // Push to clip action for post-event collection if active + if (clip_action_) { + clip_action_->PushPostEventFrame(frame); + } + + // Evaluate rules + auto result = rule_engine_.Evaluate(frame); + if (result.matched) { + TriggerAlarm(result, frame); + } + + ++processed_frames_; + } + } + + void TriggerAlarm(const RuleMatchResult& result, FramePtr frame) { + ++alarm_count_; + + AlarmEvent event; + event.node_id = id_; + event.rule_name = result.rule_name; + event.timestamp_ms = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count(); + event.frame_id = frame->frame_id; + event.detections = result.matched_detections; + + // Execute all actions in order + // Snapshot and Clip actions will fill snapshot_url and clip_url + // HTTP action should be last to include all URLs + for (auto& action : actions_) { + action->Execute(event, frame); + } + } + + std::string id_; + std::vector labels_; + RuleEngine rule_engine_; + std::shared_ptr frame_buffer_; + std::vector> actions_; + ClipAction* clip_action_ = nullptr; + + std::atomic running_{false}; + std::shared_ptr> input_queue_; + std::thread worker_; + uint64_t processed_frames_ = 0; + uint64_t alarm_count_ = 0; +}; + +REGISTER_NODE(AlarmNode, "alarm"); + +} // namespace rk3588 diff --git a/plugins/alarm/frame_ring_buffer.cpp b/plugins/alarm/frame_ring_buffer.cpp new file mode 100644 index 0000000..71a1a1d --- /dev/null +++ b/plugins/alarm/frame_ring_buffer.cpp @@ -0,0 +1,36 @@ +#include "frame_ring_buffer.h" + +namespace rk3588 { + +FrameRingBuffer::FrameRingBuffer(int pre_event_sec, int fps_hint) + : pre_event_sec_(pre_event_sec), fps_hint_(fps_hint > 0 ? fps_hint : 25) { + max_frames_ = static_cast(pre_event_sec_ * fps_hint_); + if (max_frames_ < 10) max_frames_ = 10; +} + +void FrameRingBuffer::Push(std::shared_ptr frame) { + if (!frame) return; + + std::lock_guard lock(mutex_); + buffer_.push_back(frame); + while (buffer_.size() > max_frames_) { + buffer_.pop_front(); + } +} + +std::vector> FrameRingBuffer::GetPreEventFrames() const { + std::lock_guard lock(mutex_); + return std::vector>(buffer_.begin(), buffer_.end()); +} + +void FrameRingBuffer::Clear() { + std::lock_guard lock(mutex_); + buffer_.clear(); +} + +size_t FrameRingBuffer::Size() const { + std::lock_guard lock(mutex_); + return buffer_.size(); +} + +} // namespace rk3588 diff --git a/plugins/alarm/frame_ring_buffer.h b/plugins/alarm/frame_ring_buffer.h new file mode 100644 index 0000000..7b83d4d --- /dev/null +++ b/plugins/alarm/frame_ring_buffer.h @@ -0,0 +1,30 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include "frame/frame.h" + +namespace rk3588 { + +class FrameRingBuffer { +public: + explicit FrameRingBuffer(int pre_event_sec = 5, int fps_hint = 25); + + void Push(std::shared_ptr frame); + std::vector> GetPreEventFrames() const; + void Clear(); + size_t Size() const; + +private: + mutable std::mutex mutex_; + std::deque> buffer_; + size_t max_frames_; + int pre_event_sec_; + int fps_hint_; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/rule_engine.cpp b/plugins/alarm/rule_engine.cpp new file mode 100644 index 0000000..022cc15 --- /dev/null +++ b/plugins/alarm/rule_engine.cpp @@ -0,0 +1,206 @@ +#include "rule_engine.h" + +#include +#include +#include +#include + +namespace rk3588 { + +bool RuleEngine::Init(const SimpleJson& rules_config, const std::vector& labels) { + labels_ = labels; + rules_.clear(); + duration_start_.clear(); + last_trigger_.clear(); + + if (!rules_config.IsArray()) { + std::cerr << "[RuleEngine] rules must be an array\n"; + return false; + } + + for (const auto& rule_json : rules_config.AsArray()) { + AlarmRule rule; + rule.name = rule_json.ValueOr("name", "unnamed_rule"); + rule.min_duration_ms = rule_json.ValueOr("min_duration_ms", 0); + rule.cooldown_ms = rule_json.ValueOr("cooldown_ms", 5000); + rule.schedule = rule_json.ValueOr("schedule", ""); + + // Parse class_ids + if (const SimpleJson* ids = rule_json.Find("class_ids")) { + for (const auto& id_val : ids->AsArray()) { + rule.class_ids.insert(id_val.AsInt(-1)); + } + } + + // Parse objects (class names) and convert to class_ids + if (const SimpleJson* objects = rule_json.Find("objects")) { + for (const auto& obj_val : objects->AsArray()) { + std::string obj_name = obj_val.AsString(""); + for (size_t i = 0; i < labels_.size(); ++i) { + if (labels_[i] == obj_name) { + rule.class_ids.insert(static_cast(i)); + break; + } + } + } + } + + // Parse ROI + if (const SimpleJson* roi_json = rule_json.Find("roi")) { + rule.roi.x = roi_json->ValueOr("x", 0.0f); + rule.roi.y = roi_json->ValueOr("y", 0.0f); + rule.roi.w = roi_json->ValueOr("w", 1.0f); + rule.roi.h = roi_json->ValueOr("h", 1.0f); + } + + rules_.push_back(rule); + std::cout << "[RuleEngine] loaded rule: " << rule.name + << " class_ids=" << rule.class_ids.size() + << " roi=(" << rule.roi.x << "," << rule.roi.y << "," + << rule.roi.w << "," << rule.roi.h << ")" + << " min_duration=" << rule.min_duration_ms << "ms" + << " cooldown=" << rule.cooldown_ms << "ms\n"; + } + + return true; +} + +RuleMatchResult RuleEngine::Evaluate(const std::shared_ptr& frame) { + RuleMatchResult result; + if (!frame || !frame->det) return result; + + const auto& detections = frame->det->items; + int img_w = frame->det->img_w > 0 ? frame->det->img_w : frame->width; + int img_h = frame->det->img_h > 0 ? frame->det->img_h : frame->height; + + for (auto& rule : rules_) { + // Check schedule + if (!rule.schedule.empty() && !IsInSchedule(rule.schedule)) { + duration_start_.erase(rule.name); + continue; + } + + // Check cooldown + if (!CheckCooldown(rule.name)) { + continue; + } + + // Find matching detections + std::vector matched; + for (const auto& det : detections) { + // Check class_id (if empty, match all) + if (!rule.class_ids.empty() && + rule.class_ids.find(det.cls_id) == rule.class_ids.end()) { + continue; + } + + // Check ROI + if (!IsInRoi(det.bbox, rule.roi, img_w, img_h)) { + continue; + } + + matched.push_back(det); + } + + bool currently_matched = !matched.empty(); + + // Check duration + if (CheckDuration(rule.name, currently_matched)) { + result.matched = true; + result.rule_name = rule.name; + result.matched_detections = matched; + TriggerCooldown(rule.name); + duration_start_.erase(rule.name); + return result; + } + } + + return result; +} + +void RuleEngine::Reset() { + duration_start_.clear(); + last_trigger_.clear(); +} + +bool RuleEngine::IsInRoi(const Rect& bbox, const RoiRect& roi, int img_w, int img_h) const { + // Convert bbox to normalized coordinates + float bbox_cx = (bbox.x + bbox.w / 2.0f) / img_w; + float bbox_cy = (bbox.y + bbox.h / 2.0f) / img_h; + + // Check if center is in ROI + return bbox_cx >= roi.x && bbox_cx <= (roi.x + roi.w) && + bbox_cy >= roi.y && bbox_cy <= (roi.y + roi.h); +} + +bool RuleEngine::IsInSchedule(const std::string& schedule) const { + // Parse "HH:MM-HH:MM" format + if (schedule.length() != 11 || schedule[5] != '-') { + return true; // Invalid format, assume always active + } + + int start_h = 0, start_m = 0, end_h = 0, end_m = 0; + std::sscanf(schedule.c_str(), "%d:%d-%d:%d", &start_h, &start_m, &end_h, &end_m); + + std::time_t now = std::time(nullptr); + std::tm* local = std::localtime(&now); + int curr_min = local->tm_hour * 60 + local->tm_min; + int start_min = start_h * 60 + start_m; + int end_min = end_h * 60 + end_m; + + if (start_min <= end_min) { + return curr_min >= start_min && curr_min <= end_min; + } else { + // Overnight schedule (e.g., "22:00-06:00") + return curr_min >= start_min || curr_min <= end_min; + } +} + +bool RuleEngine::CheckDuration(const std::string& rule_name, bool currently_matched) { + auto it = std::find_if(rules_.begin(), rules_.end(), + [&](const AlarmRule& r) { return r.name == rule_name; }); + if (it == rules_.end()) return false; + + int min_duration_ms = it->min_duration_ms; + auto now = std::chrono::steady_clock::now(); + + if (!currently_matched) { + duration_start_.erase(rule_name); + return false; + } + + auto start_it = duration_start_.find(rule_name); + if (start_it == duration_start_.end()) { + duration_start_[rule_name] = now; + if (min_duration_ms <= 0) { + return true; // No duration requirement + } + return false; + } + + auto elapsed = std::chrono::duration_cast( + now - start_it->second); + return elapsed.count() >= min_duration_ms; +} + +bool RuleEngine::CheckCooldown(const std::string& rule_name) { + auto it = std::find_if(rules_.begin(), rules_.end(), + [&](const AlarmRule& r) { return r.name == rule_name; }); + if (it == rules_.end()) return false; + + auto last_it = last_trigger_.find(rule_name); + if (last_it == last_trigger_.end()) { + return true; // Never triggered + } + + auto now = std::chrono::steady_clock::now(); + auto elapsed = std::chrono::duration_cast( + now - last_it->second); + return elapsed.count() >= it->cooldown_ms; +} + +void RuleEngine::TriggerCooldown(const std::string& rule_name) { + last_trigger_[rule_name] = std::chrono::steady_clock::now(); +} + +} // namespace rk3588 diff --git a/plugins/alarm/rule_engine.h b/plugins/alarm/rule_engine.h new file mode 100644 index 0000000..a84da34 --- /dev/null +++ b/plugins/alarm/rule_engine.h @@ -0,0 +1,59 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include "frame/frame.h" +#include "utils/simple_json.h" + +namespace rk3588 { + +struct RoiRect { + float x = 0.0f; + float y = 0.0f; + float w = 1.0f; + float h = 1.0f; +}; + +struct AlarmRule { + std::string name; + std::set class_ids; + RoiRect roi; + int min_duration_ms = 0; + int cooldown_ms = 5000; + std::string schedule; // "HH:MM-HH:MM" format, empty = always active +}; + +struct RuleMatchResult { + bool matched = false; + std::string rule_name; + std::vector matched_detections; +}; + +class RuleEngine { +public: + bool Init(const SimpleJson& rules_config, const std::vector& labels); + RuleMatchResult Evaluate(const std::shared_ptr& frame); + void Reset(); + +private: + bool IsInRoi(const Rect& bbox, const RoiRect& roi, int img_w, int img_h) const; + bool IsInSchedule(const std::string& schedule) const; + bool CheckDuration(const std::string& rule_name, bool currently_matched); + bool CheckCooldown(const std::string& rule_name); + void TriggerCooldown(const std::string& rule_name); + + std::vector rules_; + std::vector labels_; + + // Duration tracking: rule_name -> first match time + std::map duration_start_; + + // Cooldown tracking: rule_name -> last trigger time + std::map last_trigger_; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/uploaders/local_uploader.cpp b/plugins/alarm/uploaders/local_uploader.cpp new file mode 100644 index 0000000..5210d63 --- /dev/null +++ b/plugins/alarm/uploaders/local_uploader.cpp @@ -0,0 +1,62 @@ +#include "local_uploader.h" + +#include +#include +#include + +namespace rk3588 { + +bool LocalUploader::Init(const SimpleJson& config) { + base_path_ = config.ValueOr("path", "/tmp/alarms"); + url_prefix_ = config.ValueOr("url_prefix", "file://"); + + try { + std::filesystem::create_directories(base_path_); + } catch (const std::exception& e) { + std::cerr << "[LocalUploader] failed to create directory: " << e.what() << "\n"; + return false; + } + + std::cout << "[LocalUploader] initialized, path=" << base_path_ << "\n"; + return true; +} + +UploadResult LocalUploader::Upload(const std::string& key, + const uint8_t* data, + size_t size, + const std::string& /*content_type*/) { + UploadResult result; + + std::string full_path = base_path_ + "/" + key; + + // Create parent directories if needed + try { + std::filesystem::path file_path(full_path); + if (file_path.has_parent_path()) { + std::filesystem::create_directories(file_path.parent_path()); + } + } catch (const std::exception& e) { + result.error = std::string("failed to create directory: ") + e.what(); + return result; + } + + std::ofstream file(full_path, std::ios::binary); + if (!file) { + result.error = "failed to open file: " + full_path; + return result; + } + + file.write(reinterpret_cast(data), static_cast(size)); + file.close(); + + if (!file) { + result.error = "failed to write file: " + full_path; + return result; + } + + result.success = true; + result.url = url_prefix_ + full_path; + return result; +} + +} // namespace rk3588 diff --git a/plugins/alarm/uploaders/local_uploader.h b/plugins/alarm/uploaders/local_uploader.h new file mode 100644 index 0000000..cd5ab87 --- /dev/null +++ b/plugins/alarm/uploaders/local_uploader.h @@ -0,0 +1,21 @@ +#pragma once + +#include "uploader_base.h" + +namespace rk3588 { + +class LocalUploader : public IUploader { +public: + bool Init(const SimpleJson& config) override; + UploadResult Upload(const std::string& key, + const uint8_t* data, + size_t size, + const std::string& content_type = "") override; + std::string Type() const override { return "local"; } + +private: + std::string base_path_; + std::string url_prefix_; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/uploaders/minio_uploader.cpp b/plugins/alarm/uploaders/minio_uploader.cpp new file mode 100644 index 0000000..4709c98 --- /dev/null +++ b/plugins/alarm/uploaders/minio_uploader.cpp @@ -0,0 +1,137 @@ +#include "minio_uploader.h" + +#include +#include +#include + +#if defined(__linux__) || defined(__APPLE__) +#include +#define HAS_CURL 1 +#else +#define HAS_CURL 0 +#endif + +namespace rk3588 { + +namespace { + +std::string GetEnvOrDefault(const std::string& env_name, const std::string& default_val) { + // Handle ${ENV_VAR} syntax + if (default_val.size() > 3 && default_val.substr(0, 2) == "${" && + default_val.back() == '}') { + std::string var_name = default_val.substr(2, default_val.size() - 3); + const char* val = std::getenv(var_name.c_str()); + return val ? std::string(val) : ""; + } + return default_val; +} + +} // namespace + +bool MinioUploader::Init(const SimpleJson& config) { + endpoint_ = config.ValueOr("endpoint", "http://localhost:9000"); + bucket_ = config.ValueOr("bucket", "alarms"); + region_ = config.ValueOr("region", "us-east-1"); + + std::string ak = config.ValueOr("access_key", ""); + std::string sk = config.ValueOr("secret_key", ""); + + access_key_ = GetEnvOrDefault("MINIO_ACCESS_KEY", ak); + secret_key_ = GetEnvOrDefault("MINIO_SECRET_KEY", sk); + + use_ssl_ = endpoint_.find("https://") == 0; + + if (access_key_.empty() || secret_key_.empty()) { + std::cerr << "[MinioUploader] warning: access_key or secret_key not set\n"; + } + + std::cout << "[MinioUploader] initialized, endpoint=" << endpoint_ + << " bucket=" << bucket_ << "\n"; + return true; +} + +UploadResult MinioUploader::Upload(const std::string& key, + const uint8_t* data, + size_t size, + const std::string& content_type) { + UploadResult result; + +#if HAS_CURL + // Simple S3 PUT request (works with MinIO) + // For production, use proper AWS4 signing + + std::string url = endpoint_; + if (url.back() != '/') url += '/'; + url += bucket_ + "/" + key; + + CURL* curl = curl_easy_init(); + if (!curl) { + result.error = "curl_easy_init failed"; + return result; + } + + struct curl_slist* headers = nullptr; + std::string ct = content_type.empty() ? "application/octet-stream" : content_type; + headers = curl_slist_append(headers, ("Content-Type: " + ct).c_str()); + + // Simple auth (MinIO supports this for internal use) + // For proper S3 compatibility, implement AWS4 signing + if (!access_key_.empty()) { + // This is a simplified approach - real implementation should use AWS4 signing + std::string auth = "Authorization: AWS " + access_key_ + ":" + secret_key_; + headers = curl_slist_append(headers, auth.c_str()); + } + + curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L); + curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers); + curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, static_cast(size)); + curl_easy_setopt(curl, CURLOPT_TIMEOUT, 30L); + + // Set up data to send + struct ReadData { + const uint8_t* data; + size_t remaining; + } read_data{data, size}; + + curl_easy_setopt(curl, CURLOPT_READFUNCTION, + +[](char* buffer, size_t s, size_t n, void* userp) -> size_t { + auto* rd = static_cast(userp); + size_t to_copy = std::min(s * n, rd->remaining); + if (to_copy > 0) { + std::memcpy(buffer, rd->data, to_copy); + rd->data += to_copy; + rd->remaining -= to_copy; + } + return to_copy; + }); + curl_easy_setopt(curl, CURLOPT_READDATA, &read_data); + + CURLcode res = curl_easy_perform(curl); + if (res != CURLE_OK) { + result.error = curl_easy_strerror(res); + } else { + long http_code = 0; + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code); + if (http_code >= 200 && http_code < 300) { + result.success = true; + result.url = url; + } else { + result.error = "HTTP " + std::to_string(http_code); + } + } + + curl_slist_free_all(headers); + curl_easy_cleanup(curl); +#else + // Stub for Windows + std::cout << "[MinioUploader] would upload " << size << " bytes to " + << endpoint_ << "/" << bucket_ << "/" << key << "\n"; + result.success = true; + result.url = endpoint_ + "/" + bucket_ + "/" + key; +#endif + + return result; +} + +} // namespace rk3588 diff --git a/plugins/alarm/uploaders/minio_uploader.h b/plugins/alarm/uploaders/minio_uploader.h new file mode 100644 index 0000000..d51291a --- /dev/null +++ b/plugins/alarm/uploaders/minio_uploader.h @@ -0,0 +1,25 @@ +#pragma once + +#include "uploader_base.h" + +namespace rk3588 { + +class MinioUploader : public IUploader { +public: + bool Init(const SimpleJson& config) override; + UploadResult Upload(const std::string& key, + const uint8_t* data, + size_t size, + const std::string& content_type = "") override; + std::string Type() const override { return "minio"; } + +private: + std::string endpoint_; + std::string bucket_; + std::string access_key_; + std::string secret_key_; + std::string region_; + bool use_ssl_ = false; +}; + +} // namespace rk3588 diff --git a/plugins/alarm/uploaders/uploader_base.h b/plugins/alarm/uploaders/uploader_base.h new file mode 100644 index 0000000..30f58d8 --- /dev/null +++ b/plugins/alarm/uploaders/uploader_base.h @@ -0,0 +1,30 @@ +#pragma once + +#include +#include +#include + +#include "utils/simple_json.h" + +namespace rk3588 { + +struct UploadResult { + bool success = false; + std::string url; + std::string error; +}; + +class IUploader { +public: + virtual ~IUploader() = default; + virtual bool Init(const SimpleJson& config) = 0; + virtual UploadResult Upload(const std::string& key, + const uint8_t* data, + size_t size, + const std::string& content_type = "") = 0; + virtual std::string Type() const = 0; +}; + +std::unique_ptr CreateUploader(const SimpleJson& config); + +} // namespace rk3588 diff --git a/plugins/alarm/uploaders/uploader_factory.cpp b/plugins/alarm/uploaders/uploader_factory.cpp new file mode 100644 index 0000000..a776031 --- /dev/null +++ b/plugins/alarm/uploaders/uploader_factory.cpp @@ -0,0 +1,27 @@ +#include "uploader_base.h" +#include "local_uploader.h" +#include "minio_uploader.h" + +#include + +namespace rk3588 { + +std::unique_ptr CreateUploader(const SimpleJson& config) { + std::string type = config.ValueOr("type", "local"); + + std::unique_ptr uploader; + if (type == "minio" || type == "s3") { + uploader = std::make_unique(); + } else { + uploader = std::make_unique(); + } + + if (uploader && !uploader->Init(config)) { + std::cerr << "[CreateUploader] failed to init uploader type=" << type << "\n"; + return nullptr; + } + + return uploader; +} + +} // namespace rk3588 diff --git a/plugins/storage/storage_node.cpp b/plugins/storage/storage_node.cpp new file mode 100644 index 0000000..e8bcce8 --- /dev/null +++ b/plugins/storage/storage_node.cpp @@ -0,0 +1,502 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "node.h" + +#if defined(RK3588_ENABLE_FFMPEG) +extern "C" { +#include +#include +#include +#include +#include +} +#define HAS_FFMPEG 1 +#else +#define HAS_FFMPEG 0 +#endif + +#if defined(RK3588_ENABLE_MPP) +extern "C" { +#include +#include +#include +} +#define HAS_MPP 1 +#else +#define HAS_MPP 0 +#endif + +namespace rk3588 { + +namespace { +inline int Align16(int v) { return (v + 15) & ~15; } + +std::string FormatTime(const std::string& pattern) { + auto now = std::chrono::system_clock::now(); + auto time_t_now = std::chrono::system_clock::to_time_t(now); + std::tm* tm = std::localtime(&time_t_now); + + std::ostringstream oss; + for (size_t i = 0; i < pattern.size(); ++i) { + if (pattern[i] == '%' && i + 1 < pattern.size()) { + char spec = pattern[++i]; + switch (spec) { + case 'Y': oss << std::put_time(tm, "%Y"); break; + case 'm': oss << std::put_time(tm, "%m"); break; + case 'd': oss << std::put_time(tm, "%d"); break; + case 'H': oss << std::put_time(tm, "%H"); break; + case 'M': oss << std::put_time(tm, "%M"); break; + case 'S': oss << std::put_time(tm, "%S"); break; + default: oss << '%' << spec; break; + } + } else { + oss << pattern[i]; + } + } + return oss.str(); +} +} // namespace + +#if HAS_MPP +class MppStorageEncoder { +public: + ~MppStorageEncoder() { Shutdown(); } + + bool Init(int width, int height, PixelFormat fmt, const std::string& codec, + int fps, int bitrate_kbps) { + width_ = width; + height_ = height; + hor_stride_ = Align16(width); + ver_stride_ = Align16(height); + fps_ = fps > 0 ? fps : 25; + bitrate_bps_ = (bitrate_kbps > 0 ? bitrate_kbps : 2000) * 1000; + + if (fmt == PixelFormat::NV12) { + mpp_fmt_ = MPP_FMT_YUV420SP; + } else if (fmt == PixelFormat::YUV420) { + mpp_fmt_ = MPP_FMT_YUV420P; + } else { + std::cerr << "[storage] unsupported pixel format for MPP encoder\n"; + return false; + } + + coding_ = (codec == "h265" || codec == "hevc") ? MPP_VIDEO_CodingHEVC + : MPP_VIDEO_CodingAVC; + + if (mpp_create(&ctx_, &mpi_) != MPP_OK) return false; + if (mpp_init(ctx_, MPP_CTX_ENC, coding_) != MPP_OK) return false; + if (mpp_enc_cfg_init(&cfg_) != MPP_OK) return false; + if (mpi_->control(ctx_, MPP_ENC_GET_CFG, cfg_) != MPP_OK) return false; + + mpp_enc_cfg_set_s32(cfg_, "prep:width", width_); + mpp_enc_cfg_set_s32(cfg_, "prep:height", height_); + mpp_enc_cfg_set_s32(cfg_, "prep:hor_stride", hor_stride_); + mpp_enc_cfg_set_s32(cfg_, "prep:ver_stride", ver_stride_); + mpp_enc_cfg_set_s32(cfg_, "prep:format", mpp_fmt_); + mpp_enc_cfg_set_s32(cfg_, "rc:mode", MPP_ENC_RC_MODE_CBR); + mpp_enc_cfg_set_s32(cfg_, "rc:gop", fps_ * 2); + mpp_enc_cfg_set_s32(cfg_, "rc:fps_in_num", fps_); + mpp_enc_cfg_set_s32(cfg_, "rc:fps_in_denorm", 1); + mpp_enc_cfg_set_s32(cfg_, "rc:fps_out_num", fps_); + mpp_enc_cfg_set_s32(cfg_, "rc:fps_out_denorm", 1); + mpp_enc_cfg_set_s32(cfg_, "rc:bps_target", bitrate_bps_); + mpp_enc_cfg_set_s32(cfg_, "rc:bps_max", bitrate_bps_ * 3 / 2); + mpp_enc_cfg_set_s32(cfg_, "rc:bps_min", bitrate_bps_ / 2); + mpp_enc_cfg_set_s32(cfg_, "codec:type", coding_); + + if (mpi_->control(ctx_, MPP_ENC_SET_CFG, cfg_) != MPP_OK) return false; + + MppEncHeaderMode header_mode = MPP_ENC_HEADER_MODE_EACH_IDR; + mpi_->control(ctx_, MPP_ENC_SET_HEADER_MODE, &header_mode); + + if (mpp_buffer_group_get_internal(&frm_grp_, MPP_BUFFER_TYPE_DRM) != MPP_OK) { + mpp_buffer_group_get_internal(&frm_grp_, MPP_BUFFER_TYPE_NORMAL); + } + + // Get header + MppPacket hdr = nullptr; + if (mpi_->control(ctx_, MPP_ENC_GET_HDR_SYNC, &hdr) == MPP_OK && hdr) { + auto* data = static_cast(mpp_packet_get_pos(hdr)); + size_t len = mpp_packet_get_length(hdr); + header_.assign(data, data + len); + mpp_packet_deinit(&hdr); + } + + initialized_ = true; + return true; + } + + void Shutdown() { + if (frm_grp_) { mpp_buffer_group_put(frm_grp_); frm_grp_ = nullptr; } + if (cfg_) { mpp_enc_cfg_deinit(cfg_); cfg_ = nullptr; } + if (ctx_) { + if (mpi_) mpi_->reset(ctx_); + mpp_destroy(ctx_); + ctx_ = nullptr; mpi_ = nullptr; + } + initialized_ = false; + } + + bool Encode(const FramePtr& frame, std::vector& out, bool& is_key) { + out.clear(); + is_key = false; + if (!initialized_ || !frame) return false; + + size_t frame_size = static_cast(hor_stride_) * ver_stride_ * 3 / 2; + MppBuffer buf = nullptr; + + if (frame->dma_fd >= 0 && frame->data_size > 0) { + MppBufferInfo info{}; + info.type = MPP_BUFFER_TYPE_EXT_DMA; + info.size = frame->data_size; + info.fd = frame->dma_fd; + mpp_buffer_import(&buf, &info); + } + + if (!buf) { + if (!frame->data) return false; + if (mpp_buffer_get(frm_grp_, &buf, frame_size) != MPP_OK) return false; + + auto* dst = static_cast(mpp_buffer_get_ptr(buf)); + std::memset(dst, 0, frame_size); + + // Copy Y plane + for (int row = 0; row < height_; ++row) { + const uint8_t* src = frame->planes[0].data + ? frame->planes[0].data + row * frame->planes[0].stride + : frame->data + row * frame->width; + std::memcpy(dst + row * hor_stride_, src, width_); + } + + // Copy UV plane + uint8_t* dst_uv = dst + hor_stride_ * ver_stride_; + int uv_rows = height_ / 2; + for (int row = 0; row < uv_rows; ++row) { + const uint8_t* src = frame->planes[1].data + ? frame->planes[1].data + row * frame->planes[1].stride + : frame->data + width_ * height_ + row * width_; + std::memcpy(dst_uv + row * hor_stride_, src, width_); + } + } + + MppFrame mpp_frame = nullptr; + mpp_frame_init(&mpp_frame); + mpp_frame_set_width(mpp_frame, width_); + mpp_frame_set_height(mpp_frame, height_); + mpp_frame_set_hor_stride(mpp_frame, hor_stride_); + mpp_frame_set_ver_stride(mpp_frame, ver_stride_); + mpp_frame_set_fmt(mpp_frame, mpp_fmt_); + mpp_frame_set_pts(mpp_frame, frame->pts); + mpp_frame_set_buffer(mpp_frame, buf); + + if (mpi_->encode_put_frame(ctx_, mpp_frame) != MPP_OK) { + mpp_frame_deinit(&mpp_frame); + mpp_buffer_put(buf); + return false; + } + mpp_frame_deinit(&mpp_frame); + + MppPacket packet = nullptr; + while (true) { + MPP_RET ret = mpi_->encode_get_packet(ctx_, &packet); + if (ret == MPP_ERR_TIMEOUT) { + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + continue; + } + if (ret != MPP_OK || !packet) break; + + auto* pos = static_cast(mpp_packet_get_pos(packet)); + size_t len = mpp_packet_get_length(packet); + out.assign(pos, pos + len); + + RK_U32 flag = mpp_packet_get_flag(packet); + is_key = (flag & 0x08) != 0; + mpp_packet_deinit(&packet); + break; + } + + mpp_buffer_put(buf); + return !out.empty(); + } + + const std::vector& Header() const { return header_; } + bool IsH265() const { return coding_ == MPP_VIDEO_CodingHEVC; } + +private: + MppCtx ctx_ = nullptr; + MppApi* mpi_ = nullptr; + MppEncCfg cfg_ = nullptr; + MppBufferGroup frm_grp_ = nullptr; + MppCodingType coding_ = MPP_VIDEO_CodingAVC; + MppFrameFormat mpp_fmt_ = MPP_FMT_YUV420SP; + int width_ = 0, height_ = 0; + int hor_stride_ = 0, ver_stride_ = 0; + int fps_ = 25; + int bitrate_bps_ = 2000000; + std::vector header_; + bool initialized_ = false; +}; +#endif + +class StorageNode : public INode { +public: + std::string Id() const override { return id_; } + std::string Type() const override { return "storage"; } + + bool Init(const SimpleJson& config, const NodeContext& ctx) override { + id_ = config.ValueOr("id", "storage"); + mode_ = config.ValueOr("mode", "continuous"); + format_ = config.ValueOr("format", "mp4"); + codec_ = config.ValueOr("codec", "h264"); + segment_sec_ = config.ValueOr("segment_sec", 300); + base_path_ = config.ValueOr("path", "/rec"); + filename_pattern_ = config.ValueOr("filename_pattern", "%Y%m%d/%H%M%S"); + fps_ = config.ValueOr("fps", 25); + bitrate_kbps_ = config.ValueOr("bitrate_kbps", 2000); + + input_queue_ = ctx.input_queue; + if (!input_queue_) { + std::cerr << "[storage] no input queue\n"; + return false; + } + + try { + std::filesystem::create_directories(base_path_); + } catch (const std::exception& e) { + std::cerr << "[storage] failed to create directory: " << e.what() << "\n"; + return false; + } + + std::cout << "[storage] initialized, path=" << base_path_ + << " segment=" << segment_sec_ << "s format=" << format_ << "\n"; + return true; + } + + bool Start() override { + running_.store(true); + worker_ = std::thread(&StorageNode::WorkerLoop, this); + std::cout << "[storage] started\n"; + return true; + } + + void Stop() override { + running_.store(false); + if (input_queue_) input_queue_->Stop(); + if (worker_.joinable()) worker_.join(); + CloseCurrentFile(); + std::cout << "[storage] stopped, recorded " << total_frames_ << " frames\n"; + } + + void Drain() override { + CloseCurrentFile(); + } + +private: + void WorkerLoop() { + using namespace std::chrono; + FramePtr frame; + + while (running_.load()) { + if (!input_queue_->Pop(frame, milliseconds(200))) continue; + if (!frame) continue; + + ProcessFrame(frame); + ++total_frames_; + } + } + + void ProcessFrame(FramePtr frame) { + // Check if we need to start a new segment + auto now = std::chrono::steady_clock::now(); + bool need_new_segment = false; + + if (!file_open_) { + need_new_segment = true; + } else if (segment_sec_ > 0) { + auto elapsed = std::chrono::duration_cast( + now - segment_start_); + if (elapsed.count() >= segment_sec_) { + need_new_segment = true; + } + } + + if (need_new_segment) { + CloseCurrentFile(); + OpenNewFile(frame); + } + + WriteFrame(frame); + } + + void OpenNewFile(FramePtr frame) { +#if HAS_FFMPEG + std::string filename = FormatTime(filename_pattern_) + "." + format_; + current_path_ = base_path_ + "/" + filename; + + try { + std::filesystem::path p(current_path_); + if (p.has_parent_path()) { + std::filesystem::create_directories(p.parent_path()); + } + } catch (...) {} + + const char* fmt_name = format_ == "ts" ? "mpegts" : "mp4"; + if (avformat_alloc_output_context2(&fmt_ctx_, nullptr, fmt_name, + current_path_.c_str()) < 0) { + std::cerr << "[storage] failed to create output context\n"; + return; + } + + AVCodecID codec_id = (codec_ == "h265" || codec_ == "hevc") + ? AV_CODEC_ID_HEVC : AV_CODEC_ID_H264; + const AVCodec* encoder = avcodec_find_encoder(codec_id); + stream_ = avformat_new_stream(fmt_ctx_, encoder); + if (!stream_) { + avformat_free_context(fmt_ctx_); + fmt_ctx_ = nullptr; + return; + } + + stream_->time_base = AVRational{1, fps_}; + stream_->codecpar->codec_type = AVMEDIA_TYPE_VIDEO; + stream_->codecpar->codec_id = codec_id; + stream_->codecpar->width = frame->width; + stream_->codecpar->height = frame->height; + stream_->codecpar->format = AV_PIX_FMT_YUV420P; + +#if HAS_MPP + if (!mpp_encoder_) { + mpp_encoder_ = std::make_unique(); + if (!mpp_encoder_->Init(frame->width, frame->height, frame->format, + codec_, fps_, bitrate_kbps_)) { + mpp_encoder_.reset(); + } + } + + if (mpp_encoder_ && !mpp_encoder_->Header().empty()) { + const auto& hdr = mpp_encoder_->Header(); + stream_->codecpar->extradata_size = static_cast(hdr.size()); + stream_->codecpar->extradata = static_cast( + av_mallocz(hdr.size() + AV_INPUT_BUFFER_PADDING_SIZE)); + std::memcpy(stream_->codecpar->extradata, hdr.data(), hdr.size()); + } +#endif + + if (!(fmt_ctx_->oformat->flags & AVFMT_NOFILE)) { + if (avio_open(&fmt_ctx_->pb, current_path_.c_str(), AVIO_FLAG_WRITE) < 0) { + avformat_free_context(fmt_ctx_); + fmt_ctx_ = nullptr; + return; + } + } + + if (avformat_write_header(fmt_ctx_, nullptr) < 0) { + if (fmt_ctx_->pb) avio_closep(&fmt_ctx_->pb); + avformat_free_context(fmt_ctx_); + fmt_ctx_ = nullptr; + return; + } + + file_open_ = true; + segment_start_ = std::chrono::steady_clock::now(); + segment_frames_ = 0; + std::cout << "[storage] opened: " << current_path_ << "\n"; +#else + std::cerr << "[storage] FFmpeg not enabled\n"; +#endif + } + + void WriteFrame(FramePtr frame) { +#if HAS_FFMPEG && HAS_MPP + if (!file_open_ || !fmt_ctx_ || !mpp_encoder_) return; + + std::vector encoded; + bool is_key = false; + + if (!mpp_encoder_->Encode(frame, encoded, is_key)) return; + if (encoded.empty()) return; + + AVPacket* pkt = av_packet_alloc(); + if (av_new_packet(pkt, static_cast(encoded.size())) < 0) { + av_packet_free(&pkt); + return; + } + + std::memcpy(pkt->data, encoded.data(), encoded.size()); + pkt->stream_index = stream_->index; + pkt->flags = is_key ? AV_PKT_FLAG_KEY : 0; + pkt->pts = segment_frames_; + pkt->dts = segment_frames_; + pkt->duration = 1; + + av_packet_rescale_ts(pkt, AVRational{1, fps_}, stream_->time_base); + av_interleaved_write_frame(fmt_ctx_, pkt); + av_packet_free(&pkt); + ++segment_frames_; +#elif HAS_FFMPEG + // Software encoding fallback would go here + (void)frame; +#endif + } + + void CloseCurrentFile() { +#if HAS_FFMPEG + if (!file_open_ || !fmt_ctx_) return; + + av_write_trailer(fmt_ctx_); + if (!(fmt_ctx_->oformat->flags & AVFMT_NOFILE)) { + avio_closep(&fmt_ctx_->pb); + } + avformat_free_context(fmt_ctx_); + fmt_ctx_ = nullptr; + stream_ = nullptr; + file_open_ = false; + + std::cout << "[storage] closed: " << current_path_ + << " (" << segment_frames_ << " frames)\n"; +#endif + } + + std::string id_; + std::string mode_; + std::string format_; + std::string codec_; + int segment_sec_ = 300; + std::string base_path_; + std::string filename_pattern_; + int fps_ = 25; + int bitrate_kbps_ = 2000; + + std::atomic running_{false}; + std::shared_ptr> input_queue_; + std::thread worker_; + uint64_t total_frames_ = 0; + + // Current file state + bool file_open_ = false; + std::string current_path_; + std::chrono::steady_clock::time_point segment_start_; + int64_t segment_frames_ = 0; + +#if HAS_FFMPEG + AVFormatContext* fmt_ctx_ = nullptr; + AVStream* stream_ = nullptr; +#endif + +#if HAS_MPP + std::unique_ptr mpp_encoder_; +#endif +}; + +REGISTER_NODE(StorageNode, "storage"); + +} // namespace rk3588