完善minio上传配置等

This commit is contained in:
sladro 2025-12-31 14:32:23 +08:00
parent edb8f48f8b
commit 4412870d32
7 changed files with 427 additions and 232 deletions

View File

@ -1,5 +1,6 @@
#include "clip_action.h"
#include <algorithm>
#include <chrono>
#include <ctime>
#include <filesystem>
@ -7,6 +8,7 @@
#include <iomanip>
#include <iostream>
#include <sstream>
#include <thread>
#if defined(RK3588_ENABLE_FFMPEG)
extern "C" {
@ -23,11 +25,6 @@ extern "C" {
namespace rk3588 {
ClipAction::~ClipAction() {
Drain();
Stop();
}
bool ClipAction::Init(const SimpleJson& config) {
pre_sec_ = config.ValueOr<int>("pre_sec", 5);
post_sec_ = config.ValueOr<int>("post_sec", 10);
@ -45,130 +42,79 @@ bool ClipAction::Init(const SimpleJson& config) {
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> frame) {
ClipTask task;
task.event = event;
task.trigger_frame = frame;
if (!frame_buffer_) {
std::cerr << "[ClipAction] frame buffer not set\n";
return;
}
if (!frame) return;
// Get pre-event frames
if (frame_buffer_) {
task.pre_frames = frame_buffer_->GetPreEventFrames();
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());
}
// Start collecting post-event frames
{
std::lock_guard<std::mutex> lock(post_mutex_);
collecting_post_ = true;
post_frames_.clear();
post_frames_needed_ = post_sec_ * fps_;
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;
// Best-effort: wait for post-event window to elapse so frames are available in ring buffer.
if (post_sec_ > 0) {
std::this_thread::sleep_for(std::chrono::seconds(post_sec_));
}
// Wait for post-event frames (simplified: just queue the task)
// In a real implementation, we'd collect post-event frames asynchronously
{
std::lock_guard<std::mutex> lock(queue_mutex_);
task_queue_.push(std::move(task));
auto frames = frame_buffer_->GetFramesInRange(start_ts, end_ts);
if (frames.empty()) {
frames.push_back(frame);
}
queue_cv_.notify_one();
}
void ClipAction::PushPostEventFrame(std::shared_ptr<Frame> frame) {
std::lock_guard<std::mutex> lock(post_mutex_);
if (!collecting_post_) return;
post_frames_.push_back(frame);
if (static_cast<int>(post_frames_.size()) >= post_frames_needed_) {
collecting_post_ = false;
std::string url = ProcessClip(frames, event);
if (!url.empty()) {
event.clip_url = url;
std::cout << "[ClipAction] clip uploaded: " << url << "\n";
}
}
void ClipAction::Drain() {
if (!running_.load()) return;
std::unique_lock<std::mutex> lock(queue_mutex_);
queue_cv_.wait_for(lock, std::chrono::seconds(30),
[this] { return task_queue_.empty() && in_flight_ == 0; });
}
void ClipAction::Stop() {
running_.store(false);
{
std::lock_guard<std::mutex> lock(post_mutex_);
collecting_post_ = false;
}
queue_cv_.notify_all();
if (worker_.joinable()) worker_.join();
}
void ClipAction::WorkerLoop() {
while (running_.load()) {
ClipTask task;
{
std::unique_lock<std::mutex> 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();
++in_flight_;
}
// 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<std::mutex> 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::lock_guard<std::mutex> lock(queue_mutex_);
if (in_flight_ > 0) --in_flight_;
}
queue_cv_.notify_all();
}
}
std::string ClipAction::ProcessClip(ClipTask& task) {
std::string ClipAction::ProcessClip(const std::vector<std::shared_ptr<Frame>>& frames, const AlarmEvent& event) {
#if HAS_FFMPEG
if (task.pre_frames.empty() && !task.trigger_frame) {
if (frames.empty()) {
std::cerr << "[ClipAction] no frames to process\n";
return "";
}
// Determine frame dimensions from first available frame
std::shared_ptr<Frame> sample_frame =
task.trigger_frame ? task.trigger_frame
: (task.pre_frames.empty() ? nullptr : task.pre_frames[0]);
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;
if (width <= 0 || height <= 0) {
std::cerr << "[ClipAction] invalid frame size\n";
return "";
}
// Create temporary file
std::string tmp_path = "/tmp/clip_" + std::to_string(task.event.timestamp_ms) + "." + format_;
std::filesystem::path tmp_dir;
try {
tmp_dir = std::filesystem::temp_directory_path();
} catch (...) {
tmp_dir = ".";
}
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.c_str()) < 0) {
if (avformat_alloc_output_context2(&fmt_ctx, nullptr, format_.c_str(), tmp_path.string().c_str()) < 0) {
std::cerr << "[ClipAction] failed to create output context\n";
return "";
}
@ -208,7 +154,7 @@ std::string ClipAction::ProcessClip(ClipTask& task) {
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) {
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 "";
@ -231,24 +177,65 @@ std::string ClipAction::ProcessClip(ClipTask& task) {
AVPacket* pkt = av_packet_alloc();
int64_t pts = 0;
auto encode_frame = [&](std::shared_ptr<Frame>& frame) {
if (!frame || !frame->data) return;
auto encode_frame = [&](const std::shared_ptr<Frame>& frame) {
if (!frame) 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;
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) return;
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;
uint8_t* src_data[4] = {nullptr, nullptr, nullptr, nullptr};
int src_linesize[4] = {0, 0, 0, 0};
auto plane_ptr = [&](int idx) -> uint8_t* {
if (idx < 0 || idx >= frame->plane_count) return nullptr;
if (frame->planes[idx].data) return frame->planes[idx].data;
if (!frame->data) return nullptr;
const int off = frame->planes[idx].offset;
if (off < 0 || static_cast<size_t>(off) >= frame->data_size) return nullptr;
return frame->data + off;
};
auto plane_stride = [&](int idx, int fallback) -> int {
if (idx >= 0 && idx < frame->plane_count && frame->planes[idx].stride > 0) {
return frame->planes[idx].stride;
}
if (frame->stride > 0) return frame->stride;
return fallback;
};
if (frame->format == PixelFormat::RGB || frame->format == PixelFormat::BGR) {
src_data[0] = plane_ptr(0) ? plane_ptr(0) : frame->data;
src_linesize[0] = plane_stride(0, width * 3);
} else if (frame->format == PixelFormat::NV12) {
src_data[0] = plane_ptr(0) ? plane_ptr(0) : frame->data;
src_data[1] = plane_ptr(1);
if (!src_data[1] && frame->data) src_data[1] = frame->data + static_cast<size_t>(width) * height;
src_linesize[0] = plane_stride(0, width);
src_linesize[1] = plane_stride(1, width);
} else if (frame->format == PixelFormat::YUV420) {
src_data[0] = plane_ptr(0) ? plane_ptr(0) : frame->data;
src_data[1] = plane_ptr(1);
src_data[2] = plane_ptr(2);
src_linesize[0] = plane_stride(0, width);
src_linesize[1] = plane_stride(1, width / 2);
src_linesize[2] = plane_stride(2, width / 2);
}
if (!src_data[0]) {
sws_freeContext(sws);
return;
}
sws_scale(sws, src_data, src_linesize, 0, height,
av_frame->data, av_frame->linesize);
@ -266,14 +253,10 @@ std::string ClipAction::ProcessClip(ClipTask& task) {
}
};
// Encode pre-event frames
for (auto& f : task.pre_frames) {
for (const auto& f : 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) {
@ -288,9 +271,7 @@ std::string ClipAction::ProcessClip(ClipTask& task) {
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
@ -306,11 +287,12 @@ std::string ClipAction::ProcessClip(ClipTask& task) {
file.read(reinterpret_cast<char*>(file_data.data()), file_size);
file.close();
std::string key = GenerateKey(task.event);
std::string key = GenerateKey(event);
auto result = uploader_->Upload(key, file_data.data(), file_data.size(), "video/mp4");
// Clean up temp file
std::filesystem::remove(tmp_path);
std::error_code ec;
std::filesystem::remove(tmp_path, ec);
if (result.success) {
return result.url;

View File

@ -4,39 +4,26 @@
#include "../uploaders/uploader_base.h"
#include "../frame_ring_buffer.h"
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>
namespace rk3588 {
struct ClipTask {
AlarmEvent event;
std::vector<std::shared_ptr<Frame>> pre_frames;
std::shared_ptr<Frame> trigger_frame;
};
class ClipAction : public IAlarmAction {
public:
~ClipAction();
~ClipAction() override = default;
bool Init(const SimpleJson& config) override;
void Execute(AlarmEvent& event, std::shared_ptr<Frame> frame) override;
void Drain() override;
void Stop() override;
void Drain() override {}
void Stop() override {}
std::string Name() const override { return "clip"; }
void SetFrameBuffer(std::shared_ptr<FrameRingBuffer> buffer) {
frame_buffer_ = buffer;
}
void PushPostEventFrame(std::shared_ptr<Frame> frame);
private:
void WorkerLoop();
std::string ProcessClip(ClipTask& task);
std::string ProcessClip(const std::vector<std::shared_ptr<Frame>>& frames, const AlarmEvent& event);
std::string GenerateKey(const AlarmEvent& event);
int pre_sec_ = 5;
@ -45,20 +32,6 @@ private:
int fps_ = 25;
std::unique_ptr<IUploader> uploader_;
std::shared_ptr<FrameRingBuffer> frame_buffer_;
std::atomic<bool> running_{false};
std::thread worker_;
std::mutex queue_mutex_;
std::condition_variable queue_cv_;
std::queue<ClipTask> task_queue_;
size_t in_flight_ = 0;
// Post-event collection state
std::mutex post_mutex_;
bool collecting_post_ = false;
std::vector<std::shared_ptr<Frame>> post_frames_;
int post_frames_needed_ = 0;
ClipTask* current_task_ = nullptr;
};
} // namespace rk3588

View File

@ -18,6 +18,12 @@
namespace rk3588 {
class AlarmNode : public INode {
private:
struct AlarmJob {
AlarmEvent event;
std::shared_ptr<Frame> frame;
};
public:
std::string Id() const override { return id_; }
std::string Type() const override { return "alarm"; }
@ -40,16 +46,34 @@ public:
}
}
// Get pre-event buffer settings from clip action config
// Get pre/post buffer settings from clip action config
int pre_sec = 5;
int post_sec = 0;
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<int>("pre_sec", 5);
post_sec = clip_cfg->ValueOr<int>("post_sec", 0);
fps_hint = clip_cfg->ValueOr<int>("fps", 25);
}
}
frame_buffer_ = std::make_shared<FrameRingBuffer>(pre_sec, fps_hint);
frame_buffer_ = std::make_shared<FrameRingBuffer>(pre_sec, post_sec, fps_hint);
// Alarm worker queue (avoid blocking Process())
size_t alarm_q_size = 32;
QueueDropStrategy alarm_q_strategy = QueueDropStrategy::DropOldest;
if (const SimpleJson* q = config.Find("event_queue")) {
if (q->IsObject()) {
alarm_q_size = static_cast<size_t>(q->ValueOr<int>("size", static_cast<int>(alarm_q_size)));
const std::string strat = q->ValueOr<std::string>("strategy", "drop_oldest");
if (strat == "drop_newest") alarm_q_strategy = QueueDropStrategy::DropNewest;
else if (strat == "block") alarm_q_strategy = QueueDropStrategy::Block;
else alarm_q_strategy = QueueDropStrategy::DropOldest;
}
}
alarm_queue_size_ = alarm_q_size;
alarm_queue_strategy_ = alarm_q_strategy;
alarm_queue_ = std::make_shared<SpscQueue<AlarmJob>>(alarm_queue_size_, alarm_queue_strategy_);
// Initialize actions
if (const SimpleJson* actions_cfg = config.Find("actions")) {
@ -80,7 +104,6 @@ public:
if (action->Init(*clip_cfg)) {
auto* clip_ptr = static_cast<ClipAction*>(action.get());
clip_ptr->SetFrameBuffer(frame_buffer_);
clip_action_ = clip_ptr;
actions_.push_back(std::move(action));
}
}
@ -114,11 +137,17 @@ public:
}
bool Start() override {
worker_running_.store(true);
worker_ = std::thread(&AlarmNode::WorkerLoop, this);
std::cout << "[alarm] started\n";
return true;
}
void Stop() override {
worker_running_.store(false);
if (alarm_queue_) alarm_queue_->Stop();
if (worker_.joinable()) worker_.join();
// Ensure all actions fully stop (Drain only clears pending work; Stop must reclaim threads/resources).
for (auto& action : actions_) action->Drain();
for (auto& action : actions_) action->Stop();
@ -128,9 +157,11 @@ public:
}
void Drain() override {
for (auto& action : actions_) {
action->Drain();
// First drain alarm jobs (so snapshot/clip are finished before draining HTTP queue).
while (alarm_queue_ && (alarm_queue_->Size() > 0 || in_flight_.load() > 0)) {
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
for (auto& action : actions_) action->Drain();
}
bool UpdateConfig(const SimpleJson& new_config) override {
@ -155,16 +186,18 @@ public:
}
}
// Pre-event buffer settings
// Pre/post buffer settings
int pre_sec = 5;
int post_sec = 0;
int fps_hint = 25;
if (const SimpleJson* actions_cfg = new_config.Find("actions")) {
if (const SimpleJson* clip_cfg = actions_cfg->Find("clip")) {
pre_sec = clip_cfg->ValueOr<int>("pre_sec", 5);
post_sec = clip_cfg->ValueOr<int>("post_sec", 0);
fps_hint = clip_cfg->ValueOr<int>("fps", 25);
}
}
auto new_frame_buffer = std::make_shared<FrameRingBuffer>(pre_sec, fps_hint);
auto new_frame_buffer = std::make_shared<FrameRingBuffer>(pre_sec, post_sec, fps_hint);
// Initialize new actions
std::vector<std::unique_ptr<IAlarmAction>> new_actions;
@ -212,6 +245,18 @@ public:
}
}
// Stop worker thread to avoid executing actions while swapping.
std::shared_ptr<SpscQueue<AlarmJob>> old_queue;
std::thread old_worker;
{
std::lock_guard<std::mutex> lock(mu_);
worker_running_.store(false);
old_queue = alarm_queue_;
old_worker = std::move(worker_);
}
if (old_queue) old_queue->Stop();
if (old_worker.joinable()) old_worker.join();
std::vector<std::unique_ptr<IAlarmAction>> old_actions;
{
std::lock_guard<std::mutex> lock(mu_);
@ -220,13 +265,11 @@ public:
frame_buffer_ = std::move(new_frame_buffer);
old_actions = std::move(actions_);
actions_ = std::move(new_actions);
clip_action_ = nullptr;
for (auto& a : actions_) {
if (auto* p = dynamic_cast<ClipAction*>(a.get())) {
clip_action_ = p;
break;
}
}
in_flight_.store(0);
alarm_queue_ = std::make_shared<SpscQueue<AlarmJob>>(alarm_queue_size_, alarm_queue_strategy_);
worker_running_.store(true);
worker_ = std::thread(&AlarmNode::WorkerLoop, this);
}
for (auto& action : old_actions) action->Drain();
@ -247,8 +290,11 @@ public:
if (!frame) return NodeStatus::DROP;
std::lock_guard<std::mutex> lock(mu_);
if (frame_buffer_) frame_buffer_->Push(frame);
if (clip_action_) clip_action_->PushPostEventFrame(frame);
if (frame_buffer_) {
using namespace std::chrono;
const uint64_t ts_ms = static_cast<uint64_t>(duration_cast<milliseconds>(steady_clock::now().time_since_epoch()).count());
frame_buffer_->Push(frame, ts_ms);
}
auto result = rule_engine_.Evaluate(frame);
if (result.matched) TriggerAlarm(result, frame);
@ -258,6 +304,24 @@ public:
}
private:
void WorkerLoop() {
while (worker_running_.load()) {
AlarmJob job;
if (!alarm_queue_) {
std::this_thread::sleep_for(std::chrono::milliseconds(50));
continue;
}
if (!alarm_queue_->Pop(job, std::chrono::milliseconds(100))) {
continue;
}
in_flight_.fetch_add(1);
for (auto& action : actions_) {
action->Execute(job.event, job.frame);
}
in_flight_.fetch_sub(1);
}
}
void TriggerAlarm(const RuleMatchResult& result, FramePtr frame) {
++alarm_count_;
@ -269,11 +333,12 @@ private:
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);
AlarmJob job;
job.event = std::move(event);
job.frame = std::move(frame);
if (alarm_queue_) {
alarm_queue_->Push(std::move(job));
}
}
@ -282,10 +347,16 @@ private:
RuleEngine rule_engine_;
std::shared_ptr<FrameRingBuffer> frame_buffer_;
std::vector<std::unique_ptr<IAlarmAction>> actions_;
ClipAction* clip_action_ = nullptr;
mutable std::mutex mu_;
std::shared_ptr<SpscQueue<AlarmJob>> alarm_queue_;
size_t alarm_queue_size_ = 32;
QueueDropStrategy alarm_queue_strategy_ = QueueDropStrategy::DropOldest;
std::atomic<bool> worker_running_{false};
std::thread worker_;
std::atomic<size_t> in_flight_{0};
std::shared_ptr<SpscQueue<FramePtr>> input_queue_;
uint64_t processed_frames_ = 0;
uint64_t alarm_count_ = 0;

View File

@ -1,31 +1,70 @@
#include "frame_ring_buffer.h"
#include <algorithm>
#include <chrono>
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<size_t>(pre_event_sec_ * fps_hint_);
namespace {
uint64_t NowSteadyMs() {
using namespace std::chrono;
return static_cast<uint64_t>(duration_cast<milliseconds>(steady_clock::now().time_since_epoch()).count());
}
}
FrameRingBuffer::FrameRingBuffer(int pre_event_sec, int post_event_sec, int fps_hint)
: pre_event_sec_(pre_event_sec), post_event_sec_(post_event_sec), fps_hint_(fps_hint > 0 ? fps_hint : 25) {
const int total_sec = std::max(0, pre_event_sec_) + std::max(0, post_event_sec_) + 2; // +2s margin
max_frames_ = static_cast<size_t>(std::max(1, total_sec) * fps_hint_);
if (max_frames_ < 10) max_frames_ = 10;
}
void FrameRingBuffer::Push(std::shared_ptr<Frame> frame) {
if (!frame) return;
std::lock_guard<std::mutex> lock(mutex_);
buffer_.push_back(frame);
while (buffer_.size() > max_frames_) {
buffer_.pop_front();
}
Push(std::move(frame), NowSteadyMs());
}
std::vector<std::shared_ptr<Frame>> FrameRingBuffer::GetPreEventFrames() const {
void FrameRingBuffer::Push(std::shared_ptr<Frame> frame, uint64_t ts_ms) {
if (!frame) return;
if (ts_ms == 0) ts_ms = NowSteadyMs();
std::lock_guard<std::mutex> lock(mutex_);
return std::vector<std::shared_ptr<Frame>>(buffer_.begin(), buffer_.end());
buffer_.push_back(Entry{ts_ms, std::move(frame)});
last_ts_ms_ = std::max(last_ts_ms_, ts_ms);
while (buffer_.size() > max_frames_) buffer_.pop_front();
}
std::vector<std::shared_ptr<Frame>> FrameRingBuffer::GetFramesInRange(uint64_t start_ts_ms, uint64_t end_ts_ms) const {
std::lock_guard<std::mutex> lock(mutex_);
std::vector<std::shared_ptr<Frame>> out;
for (const auto& e : buffer_) {
if (e.ts_ms < start_ts_ms) continue;
if (e.ts_ms > end_ts_ms) break;
if (e.frame) out.push_back(e.frame);
}
return out;
}
bool FrameRingBuffer::FindTimestampByFrameId(uint64_t frame_id, uint64_t& out_ts_ms) const {
std::lock_guard<std::mutex> lock(mutex_);
for (auto it = buffer_.rbegin(); it != buffer_.rend(); ++it) {
if (!it->frame) continue;
if (it->frame->frame_id == frame_id) {
out_ts_ms = it->ts_ms;
return true;
}
}
return false;
}
uint64_t FrameRingBuffer::LatestTimestampMs() const {
std::lock_guard<std::mutex> lock(mutex_);
return last_ts_ms_;
}
void FrameRingBuffer::Clear() {
std::lock_guard<std::mutex> lock(mutex_);
buffer_.clear();
last_ts_ms_ = 0;
}
size_t FrameRingBuffer::Size() const {

View File

@ -1,6 +1,7 @@
#pragma once
#include <chrono>
#include <cstdint>
#include <deque>
#include <memory>
#include <mutex>
@ -12,18 +13,29 @@ namespace rk3588 {
class FrameRingBuffer {
public:
explicit FrameRingBuffer(int pre_event_sec = 5, int fps_hint = 25);
explicit FrameRingBuffer(int pre_event_sec = 5, int post_event_sec = 0, int fps_hint = 25);
void Push(std::shared_ptr<Frame> frame);
std::vector<std::shared_ptr<Frame>> GetPreEventFrames() const;
void Push(std::shared_ptr<Frame> frame, uint64_t ts_ms);
std::vector<std::shared_ptr<Frame>> GetFramesInRange(uint64_t start_ts_ms, uint64_t end_ts_ms) const;
bool FindTimestampByFrameId(uint64_t frame_id, uint64_t& out_ts_ms) const;
uint64_t LatestTimestampMs() const;
void Clear();
size_t Size() const;
private:
struct Entry {
uint64_t ts_ms = 0;
std::shared_ptr<Frame> frame;
};
mutable std::mutex mutex_;
std::deque<std::shared_ptr<Frame>> buffer_;
std::deque<Entry> buffer_;
uint64_t last_ts_ms_ = 0;
size_t max_frames_;
int pre_event_sec_;
int post_event_sec_;
int fps_hint_;
};

View File

@ -1,9 +1,13 @@
#include "minio_uploader.h"
#include <algorithm>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <sstream>
#include <vector>
#include "utils/simple_json_writer.h"
#if __has_include(<curl/curl.h>)
#include <curl/curl.h>
@ -16,18 +20,115 @@ namespace rk3588 {
namespace {
std::string GetEnvOrDefault(const std::string& env_name, const std::string& default_val) {
(void)env_name;
// 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) : "";
std::string ResolveEnv(const std::string& env_name, const std::string& val) {
// Support explicit "${ENV_VAR}" syntax.
if (val.size() > 3 && val.substr(0, 2) == "${" && val.back() == '}') {
std::string var_name = val.substr(2, val.size() - 3);
const char* v = std::getenv(var_name.c_str());
return v ? std::string(v) : "";
}
return default_val;
// Default behavior: if empty, fall back to env_name.
if (val.empty()) {
const char* v = std::getenv(env_name.c_str());
return v ? std::string(v) : "";
}
return val;
}
#if HAS_CURL
size_t CurlWriteCb(char* ptr, size_t size, size_t nmemb, void* userdata) {
if (!userdata || !ptr) return 0;
auto* out = static_cast<std::string*>(userdata);
out->append(ptr, size * nmemb);
return size * nmemb;
}
struct PresignResult {
std::string upload_url;
std::string public_url;
std::vector<std::string> headers; // "Key: Value"
};
bool GetPresignedPutUrl(const std::string& endpoint,
const std::string& bucket,
const std::string& key,
const std::string& content_type,
size_t size,
int timeout_ms,
PresignResult& out,
std::string& err) {
err.clear();
out = PresignResult{};
SimpleJson::Object req;
req["bucket"] = SimpleJson(bucket);
req["key"] = SimpleJson(key);
req["content_type"] = SimpleJson(content_type);
req["size"] = SimpleJson(static_cast<double>(size));
req["method"] = SimpleJson(std::string("PUT"));
const std::string body = StringifySimpleJson(SimpleJson(std::move(req)));
CURL* curl = curl_easy_init();
if (!curl) {
err = "curl_easy_init failed";
return false;
}
std::string resp;
struct curl_slist* headers = nullptr;
headers = curl_slist_append(headers, "Content-Type: application/json");
curl_easy_setopt(curl, CURLOPT_URL, endpoint.c_str());
curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
curl_easy_setopt(curl, CURLOPT_POSTFIELDS, body.c_str());
curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, static_cast<long>(body.size()));
curl_easy_setopt(curl, CURLOPT_TIMEOUT_MS, timeout_ms);
curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT_MS, std::max(500, timeout_ms / 2));
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, &CurlWriteCb);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &resp);
CURLcode res = curl_easy_perform(curl);
long http_code = 0;
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code);
curl_slist_free_all(headers);
curl_easy_cleanup(curl);
if (res != CURLE_OK) {
err = curl_easy_strerror(res);
return false;
}
if (http_code < 200 || http_code >= 300) {
err = "presign http " + std::to_string(http_code);
return false;
}
SimpleJson j;
std::string jerr;
if (!ParseSimpleJson(resp, j, jerr) || !j.IsObject()) {
err = "presign response json invalid: " + jerr;
return false;
}
const SimpleJson* url = j.Find("upload_url");
if (!url) url = j.Find("url");
if (!url || !url->IsString()) {
err = "presign response missing upload_url";
return false;
}
out.upload_url = url->AsString("");
if (const SimpleJson* pu = j.Find("public_url"); pu && pu->IsString()) {
out.public_url = pu->AsString("");
}
if (const SimpleJson* hs = j.Find("headers"); hs && hs->IsObject()) {
for (const auto& kv : hs->AsObject()) {
if (!kv.second.IsString()) continue;
out.headers.push_back(kv.first + ": " + kv.second.AsString(""));
}
}
return !out.upload_url.empty();
}
#endif
} // namespace
bool MinioUploader::Init(const SimpleJson& config) {
@ -35,16 +136,23 @@ bool MinioUploader::Init(const SimpleJson& config) {
bucket_ = config.ValueOr<std::string>("bucket", "alarms");
region_ = config.ValueOr<std::string>("region", "us-east-1");
presign_endpoint_ = config.ValueOr<std::string>("presign_endpoint", "");
presign_timeout_ms_ = config.ValueOr<int>("presign_timeout_ms", presign_timeout_ms_);
std::string ak = config.ValueOr<std::string>("access_key", "");
std::string sk = config.ValueOr<std::string>("secret_key", "");
access_key_ = GetEnvOrDefault("MINIO_ACCESS_KEY", ak);
secret_key_ = GetEnvOrDefault("MINIO_SECRET_KEY", sk);
access_key_ = ResolveEnv("MINIO_ACCESS_KEY", ak);
secret_key_ = ResolveEnv("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";
#if HAS_CURL
curl_global_init(CURL_GLOBAL_DEFAULT);
#endif
if (presign_endpoint_.empty() && (access_key_.empty() || secret_key_.empty())) {
std::cerr << "[MinioUploader] warning: access_key/secret_key not set; recommended to set presign_endpoint\n";
}
std::cout << "[MinioUploader] initialized, endpoint=" << endpoint_
@ -59,12 +167,19 @@ UploadResult MinioUploader::Upload(const std::string& key,
UploadResult result;
#if HAS_CURL
// Simple S3 PUT request (works with MinIO)
// For production, use proper AWS4 signing
if (presign_endpoint_.empty()) {
result.error = "presign_endpoint is required for S3/MinIO upload (recommended: use presigned PUT URL)";
return result;
}
std::string url = endpoint_;
if (url.back() != '/') url += '/';
url += bucket_ + "/" + key;
const std::string ct = content_type.empty() ? "application/octet-stream" : content_type;
PresignResult presigned;
std::string perr;
if (!GetPresignedPutUrl(presign_endpoint_, bucket_, key, ct, size, presign_timeout_ms_, presigned, perr)) {
result.error = perr;
return result;
}
CURL* curl = curl_easy_init();
if (!curl) {
@ -73,18 +188,12 @@ UploadResult MinioUploader::Upload(const std::string& key,
}
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());
for (const auto& h : presigned.headers) {
headers = curl_slist_append(headers, h.c_str());
}
curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
curl_easy_setopt(curl, CURLOPT_URL, presigned.upload_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<curl_off_t>(size));
@ -110,25 +219,27 @@ UploadResult MinioUploader::Upload(const std::string& key,
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);
}
}
long http_code = 0;
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code);
curl_slist_free_all(headers);
curl_easy_cleanup(curl);
if (res != CURLE_OK) {
result.error = curl_easy_strerror(res);
return result;
}
if (http_code < 200 || http_code >= 300) {
result.error = "HTTP " + std::to_string(http_code);
return result;
}
result.success = true;
result.url = presigned.public_url.empty() ? presigned.upload_url : presigned.public_url;
#else
// Stub for Windows
std::cout << "[MinioUploader] would upload " << size << " bytes to "
<< endpoint_ << "/" << bucket_ << "/" << key << "\n";
std::cout << "[MinioUploader] would upload " << size << " bytes as key=" << key
<< " (configure presign_endpoint to enable real upload)\n";
result.success = true;
result.url = endpoint_ + "/" + bucket_ + "/" + key;
#endif

View File

@ -19,6 +19,13 @@ private:
std::string access_key_;
std::string secret_key_;
std::string region_;
// Recommended: use a presign service to return per-object presigned PUT URLs.
// When set, Upload() will:
// 1) POST to presign_endpoint_ with {bucket,key,content_type,size}
// 2) PUT the payload to returned upload_url
// 3) return public_url (if present) or upload_url
std::string presign_endpoint_;
int presign_timeout_ms_ = 3000;
bool use_ssl_ = false;
};