From 5c72a489a55454b5ec6d7744872a4c2d5cf2b3ac Mon Sep 17 00:00:00 2001 From: sladro Date: Mon, 12 Jan 2026 21:51:36 +0800 Subject: [PATCH] =?UTF-8?q?=E6=80=A7=E8=83=BD=E4=BC=98=E5=8C=965?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/ai_face_det/ai_face_det_node.cpp | 22 +++-- plugins/ai_face_recog/ai_face_recog_node.cpp | 13 ++- plugins/ai_yolo/ai_yolo_node.cpp | 18 ++--- plugins/alarm/actions/clip_action.cpp | 56 ++++++++----- plugins/alarm/actions/external_api_action.cpp | 55 +++++++++---- plugins/alarm/actions/http_action.cpp | 80 ++++++++++++++++--- plugins/alarm/actions/log_action.cpp | 38 +++++++-- plugins/alarm/actions/snapshot_action.cpp | 44 +++++++--- plugins/alarm/alarm_node.cpp | 28 ++++--- plugins/alarm/rule_engine.cpp | 50 +++++++++--- plugins/alarm/uploaders/local_uploader.cpp | 7 +- plugins/alarm/uploaders/minio_uploader.cpp | 16 ++-- plugins/alarm/uploaders/uploader_factory.cpp | 4 +- plugins/det_post/det_post_node.cpp | 14 ++-- plugins/input_file/input_file_node.cpp | 45 ++++++----- plugins/input_rtsp/input_rtsp_node.cpp | 49 ++++++------ plugins/osd/osd_node.cpp | 5 +- plugins/preprocess/preprocess_node.cpp | 22 ++--- plugins/publish/publish_node.cpp | 33 ++++---- plugins/storage/storage_node.cpp | 69 +++++++++++----- plugins/zlm_http/zlm_http_node.cpp | 12 +-- 21 files changed, 448 insertions(+), 232 deletions(-) diff --git a/plugins/ai_face_det/ai_face_det_node.cpp b/plugins/ai_face_det/ai_face_det_node.cpp index 7a4c558..185f944 100644 --- a/plugins/ai_face_det/ai_face_det_node.cpp +++ b/plugins/ai_face_det/ai_face_det_node.cpp @@ -5,7 +5,6 @@ #include #include #include -#include #include #include #include @@ -424,29 +423,31 @@ public: { std::lock_guard lock(mu_); cfg_ = std::move(snap); +#if defined(RK3588_ENABLE_RKNN) priors_cache_ = {}; +#endif } input_queue_ = ctx.input_queue; output_queues_ = ctx.output_queues; if (!input_queue_) { - std::cerr << "[ai_face_det] no input queue for node " << id_ << "\n"; + LogError("[ai_face_det] no input queue for node " + id_); return false; } if (output_queues_.empty()) { - std::cerr << "[ai_face_det] no output queue for node " << id_ << "\n"; + LogError("[ai_face_det] no output queue for node " + id_); return false; } #if defined(RK3588_ENABLE_RKNN) if (model_path_.empty()) { - std::cerr << "[ai_face_det] model_path is required\n"; + LogError("[ai_face_det] model_path is required"); return false; } std::string err; model_handle_ = AiScheduler::Instance().LoadModel(model_path_, err); if (model_handle_ == kInvalidModelHandle) { - std::cerr << "[ai_face_det] failed to load model: " << err << "\n"; + LogError("[ai_face_det] failed to load model: " + err); return false; } ModelInfo info; @@ -498,7 +499,9 @@ public: { std::lock_guard lock(mu_); cfg_ = std::move(snap); +#if defined(RK3588_ENABLE_RKNN) priors_cache_ = {}; +#endif } return true; } @@ -532,7 +535,7 @@ private: void Run(FramePtr frame) { if (!frame->data || frame->data_size == 0) return; if (frame->format != PixelFormat::RGB && frame->format != PixelFormat::BGR) { - std::cerr << "[ai_face_det] input must be RGB/BGR\n"; + LogWarn("[ai_face_det] input must be RGB/BGR"); return; } @@ -614,7 +617,7 @@ private: auto r = AiScheduler::Instance().InferBorrowed(model_handle_, input); if (!r.success) { - std::cerr << "[ai_face_det] inference failed: " << r.error << "\n"; + LogWarn("[ai_face_det] inference failed: " + r.error); return; } @@ -733,7 +736,8 @@ private: const std::vector& priors = priors_ptr ? *priors_ptr : empty_priors; if (!priors.empty() && static_cast(priors.size()) != n) { // Mismatch: can't reliably decode. - std::cerr << "[ai_face_det] prior mismatch: priors=" << priors.size() << " n=" << n << "\n"; + LogWarn("[ai_face_det] prior mismatch: priors=" + std::to_string(priors.size()) + + " n=" + std::to_string(n)); return; } @@ -838,7 +842,9 @@ private: mutable std::mutex mu_; std::shared_ptr cfg_; +#if defined(RK3588_ENABLE_RKNN) PriorsCache priors_cache_; +#endif std::shared_ptr> input_queue_; std::vector>> output_queues_; diff --git a/plugins/ai_face_recog/ai_face_recog_node.cpp b/plugins/ai_face_recog/ai_face_recog_node.cpp index e56d650..3754b0b 100644 --- a/plugins/ai_face_recog/ai_face_recog_node.cpp +++ b/plugins/ai_face_recog/ai_face_recog_node.cpp @@ -5,7 +5,6 @@ #include #include #include -#include #include #include #include @@ -655,23 +654,23 @@ public: input_queue_ = ctx.input_queue; output_queues_ = ctx.output_queues; if (!input_queue_) { - std::cerr << "[ai_face_recog] no input queue for node " << id_ << "\n"; + LogError("[ai_face_recog] no input queue for node " + id_); return false; } if (output_queues_.empty()) { - std::cerr << "[ai_face_recog] no output queue for node " << id_ << "\n"; + LogError("[ai_face_recog] no output queue for node " + id_); return false; } #if defined(RK3588_ENABLE_RKNN) if (model_path_.empty()) { - std::cerr << "[ai_face_recog] model_path is required\n"; + LogError("[ai_face_recog] model_path is required"); return false; } std::string err; model_handle_ = AiScheduler::Instance().LoadModel(model_path_, err); if (model_handle_ == kInvalidModelHandle) { - std::cerr << "[ai_face_recog] failed to load model: " << err << "\n"; + LogError("[ai_face_recog] failed to load model: " + err); return false; } ModelInfo info; @@ -804,7 +803,7 @@ private: if (!frame->face_det || frame->face_det->faces.empty()) return; if (!frame->data || frame->data_size == 0) return; if (frame->format != PixelFormat::RGB && frame->format != PixelFormat::BGR) { - std::cerr << "[ai_face_recog] input must be RGB/BGR\n"; + LogWarn("[ai_face_recog] input must be RGB/BGR"); return; } @@ -892,7 +891,7 @@ private: auto r = AiScheduler::Instance().InferBorrowed(model_handle_, in); if (!r.success || r.outputs.empty()) { - std::cerr << "[ai_face_recog] inference failed: " << (r.error.empty() ? "unknown" : r.error) << "\n"; + LogWarn(std::string("[ai_face_recog] inference failed: ") + (r.error.empty() ? "unknown" : r.error)); continue; } diff --git a/plugins/ai_yolo/ai_yolo_node.cpp b/plugins/ai_yolo/ai_yolo_node.cpp index a9aa72a..337a9b9 100644 --- a/plugins/ai_yolo/ai_yolo_node.cpp +++ b/plugins/ai_yolo/ai_yolo_node.cpp @@ -4,7 +4,6 @@ #include #include #include -#include #include #include #include @@ -310,25 +309,25 @@ public: input_queue_ = ctx.input_queue; if (!input_queue_) { - std::cerr << "[ai_yolo] no input queue for node " << id_ << "\n"; + LogError("[ai_yolo] no input queue for node " + id_); return false; } if (ctx.output_queues.empty()) { - std::cerr << "[ai_yolo] no output queue for node " << id_ << "\n"; + LogError("[ai_yolo] no output queue for node " + id_); return false; } output_queues_ = ctx.output_queues; #if defined(RK3588_ENABLE_RKNN) if (model_path_.empty()) { - std::cerr << "[ai_yolo] model_path is required\n"; + LogError("[ai_yolo] model_path is required"); return false; } std::string err; model_handle_ = AiScheduler::Instance().LoadModel(model_path_, err); if (model_handle_ == kInvalidModelHandle) { - std::cerr << "[ai_yolo] failed to load model: " << err << "\n"; + LogError("[ai_yolo] failed to load model: " + err); return false; } @@ -410,7 +409,7 @@ private: bool is_rgb = (frame->format == PixelFormat::RGB || frame->format == PixelFormat::BGR); if (!is_rgb) { - std::cerr << "[ai_yolo] input must be RGB/BGR, got other format\n"; + LogWarn("[ai_yolo] input must be RGB/BGR, got other format"); return; } @@ -439,8 +438,9 @@ private: } } else { if (frame->data_size < static_cast(src_stride) * static_cast(h)) { - std::cerr << "[ai_yolo] invalid RGB buffer size/stride (data_size=" << frame->data_size - << ", stride=" << src_stride << ", h=" << h << ")\n"; + LogWarn("[ai_yolo] invalid RGB buffer size/stride (data_size=" + std::to_string(frame->data_size) + + ", stride=" + std::to_string(src_stride) + + ", h=" + std::to_string(h) + ")"); return; } rgb_tmp_.resize(packed_size); @@ -458,7 +458,7 @@ private: auto result = AiScheduler::Instance().InferBorrowed(model_handle_, input); if (!result.success) { - std::cerr << "[ai_yolo] inference failed: " << result.error << "\n"; + LogWarn("[ai_yolo] inference failed: " + result.error); return; } diff --git a/plugins/alarm/actions/clip_action.cpp b/plugins/alarm/actions/clip_action.cpp index 6fe25fb..70d64c0 100644 --- a/plugins/alarm/actions/clip_action.cpp +++ b/plugins/alarm/actions/clip_action.cpp @@ -7,10 +7,12 @@ #include #include #include -#include +#include #include #include +#include "utils/logger.h" + #if defined(RK3588_ENABLE_FFMPEG) extern "C" { #include @@ -26,6 +28,21 @@ namespace rk3588 { namespace { +bool SafeLocalTime(std::time_t t, std::tm& out) { +#if defined(_WIN32) + return localtime_s(&out, &t) == 0; +#elif defined(__unix__) || defined(__APPLE__) + return localtime_r(&t, &out) != nullptr; +#else + static std::mutex mu; + std::lock_guard lock(mu); + std::tm* p = std::localtime(&t); + if (!p) return false; + out = *p; + return true; +#endif +} + static bool HasAnnexBStartCode(const uint8_t* d, size_t n) { if (!d || n < 4) return false; for (size_t i = 0; i + 3 < n; ++i) { @@ -145,7 +162,7 @@ bool ClipAction::Init(const SimpleJson& config) { if (const SimpleJson* upload_cfg = config.Find("upload")) { uploader_ = CreateUploader(*upload_cfg); if (!uploader_) { - std::cerr << "[ClipAction] failed to create uploader\n"; + LogError("[ClipAction] failed to create uploader"); return false; } } else { @@ -153,21 +170,22 @@ bool ClipAction::Init(const SimpleJson& config) { uploader_ = CreateUploader(default_cfg); } - std::cout << "[ClipAction] initialized, pre=" << pre_sec_ << "s post=" << post_sec_ - << "s format=" << format_ << "\n"; + LogInfo("[ClipAction] initialized, pre=" + std::to_string(pre_sec_) + + "s post=" + std::to_string(post_sec_) + + "s format=" + format_); return true; } void ClipAction::Execute(AlarmEvent& event, std::shared_ptr frame) { if (!packet_buffer_) { - std::cerr << "[ClipAction] packet buffer not set\n"; + LogError("[ClipAction] packet buffer not set"); return; } if (!frame) return; const int64_t trigger_pts_ms = frame->pts > 0 ? static_cast(frame->pts / 1000ULL) : 0; if (trigger_pts_ms <= 0) { - std::cerr << "[ClipAction] invalid trigger pts\n"; + LogWarn("[ClipAction] invalid trigger pts"); return; } @@ -192,14 +210,14 @@ void ClipAction::Execute(AlarmEvent& event, std::shared_ptr frame) { metas = packet_buffer_->GetPacketsInRange(trigger_pts_ms, end_pts); } if (metas.empty()) { - std::cerr << "[ClipAction] no packets in range\n"; + LogWarn("[ClipAction] no packets in range"); return; } std::string url = ProcessClipFromPackets(metas, event); if (!url.empty()) { event.clip_url = url; - std::cout << "[ClipAction] clip uploaded: " << url << "\n"; + LogInfo("[ClipAction] clip uploaded: " + url); } } @@ -220,14 +238,14 @@ std::string ClipAction::ProcessClipFromPackets(const std::vectorcodec->width; const int height = first->codec->height; if (width <= 0 || height <= 0) { - std::cerr << "[ClipAction] invalid codec size\n"; + LogError("[ClipAction] invalid codec size"); return ""; } @@ -235,13 +253,14 @@ std::string ClipAction::ProcessClipFromPackets(const std::vector(file.tellg()); @@ -372,12 +391,12 @@ std::string ClipAction::ProcessClipFromPackets(const std::vector #include +#include #include #include -#include +#include #include #include #include "utils/simple_json.h" #include "utils/simple_json_writer.h" +#include "utils/logger.h" #if __has_include() #include @@ -21,7 +23,31 @@ namespace rk3588 { namespace { +bool SafeLocalTime(std::time_t t, std::tm& out) { +#if defined(_WIN32) + return localtime_s(&out, &t) == 0; +#elif defined(__unix__) || defined(__APPLE__) + return localtime_r(&t, &out) != nullptr; +#else + static std::mutex mu; + std::lock_guard lock(mu); + std::tm* p = std::localtime(&t); + if (!p) return false; + out = *p; + return true; +#endif +} + #if HAS_CURL + +void EnsureCurlGlobalInitOnce() { + static std::once_flag once; + std::call_once(once, []() { + curl_global_init(CURL_GLOBAL_DEFAULT); + std::atexit([]() { curl_global_cleanup(); }); + }); +} + size_t CurlWriteCb(char* ptr, size_t size, size_t nmemb, void* userdata) { if (!userdata) return 0; auto* out = static_cast(userdata); @@ -122,30 +148,29 @@ bool ExternalApiAction::Init(const SimpleJson& config) { retry_backoff_ms_ = std::max(0, config.ValueOr("retry_backoff_ms", 200)); if (get_token_url_.empty()) { - std::cerr << "[ExternalApiAction] getTokenUrl is required\n"; + LogError("[ExternalApiAction] getTokenUrl is required"); return false; } if (put_message_url_.empty()) { - std::cerr << "[ExternalApiAction] putMessageUrl is required\n"; + LogError("[ExternalApiAction] putMessageUrl is required"); return false; } if (tenant_code_.empty()) { - std::cerr << "[ExternalApiAction] tenantCode is required\n"; + LogError("[ExternalApiAction] tenantCode is required"); return false; } if (channel_no_.empty()) { - std::cerr << "[ExternalApiAction] channelNo is required\n"; + LogError("[ExternalApiAction] channelNo is required"); return false; } #if HAS_CURL - curl_global_init(CURL_GLOBAL_DEFAULT); + EnsureCurlGlobalInitOnce(); #endif running_.store(true); worker_ = std::thread(&ExternalApiAction::WorkerLoop, this); - std::cout << "[ExternalApiAction] initialized, token_url=" << get_token_url_ - << " msg_url=" << put_message_url_ << "\n"; + LogInfo("[ExternalApiAction] initialized, token_url=" + get_token_url_ + " msg_url=" + put_message_url_); return true; } @@ -214,7 +239,7 @@ void ExternalApiAction::WorkerLoop() { std::string err; if (!EnsureToken(err)) { - std::cerr << "[ExternalApiAction] ensure token failed: " << err << "\n"; + LogWarn("[ExternalApiAction] ensure token failed: " + err); } else { std::string token; { @@ -247,10 +272,10 @@ void ExternalApiAction::WorkerLoop() { token = token_; continue; } - std::cerr << "[ExternalApiAction] refresh token failed: " << terr << "\n"; + LogWarn("[ExternalApiAction] refresh token failed: " + terr); } - std::cerr << "[ExternalApiAction] send failed: " << send_err << " (http=" << http_code << ")\n"; + LogWarn("[ExternalApiAction] send failed: " + send_err + " (http=" + std::to_string(http_code) + ")"); } } @@ -362,17 +387,17 @@ bool ExternalApiAction::SendMessageWithToken(const Job& job, const std::string& #else (void)job; (void)token; - std::cout << "[ExternalApiAction] curl not available; would POST to " << put_message_url_ << "\n"; + LogInfo("[ExternalApiAction] curl not available; would POST to " + put_message_url_); return true; #endif } std::string ExternalApiAction::FormatAlarmTime(uint64_t timestamp_ms) const { const std::time_t t = static_cast(timestamp_ms / 1000ULL); - std::tm* tm = std::localtime(&t); - if (!tm) return ""; + std::tm tm{}; + if (!SafeLocalTime(t, tm)) return ""; std::ostringstream oss; - oss << std::put_time(tm, "%Y-%m-%d %H:%M:%S"); + oss << std::put_time(&tm, "%Y-%m-%d %H:%M:%S"); return oss.str(); } diff --git a/plugins/alarm/actions/http_action.cpp b/plugins/alarm/actions/http_action.cpp index 77c87d2..2a6c7c7 100644 --- a/plugins/alarm/actions/http_action.cpp +++ b/plugins/alarm/actions/http_action.cpp @@ -2,9 +2,14 @@ #include #include +#include +#include #include -#include #include +#include +#include + +#include "utils/logger.h" #if __has_include() #include @@ -15,6 +20,43 @@ namespace rk3588 { +namespace { + +std::string ToLower(std::string s) { + for (char& c : s) c = static_cast(std::tolower(static_cast(c))); + return s; +} + +std::string JsonEscape(std::string_view s) { + std::string out; + out.reserve(s.size() + 8); + for (char c : s) { + switch (c) { + case '\\': out += "\\\\"; break; + case '"': out += "\\\""; break; + case '\n': out += "\\n"; break; + case '\r': out += "\\r"; break; + case '\t': out += "\\t"; break; + default: + if (static_cast(c) < 0x20) out += "?"; + else out.push_back(c); + } + } + return out; +} + +#if HAS_CURL +void EnsureCurlGlobalInitOnce() { + static std::once_flag once; + std::call_once(once, []() { + curl_global_init(CURL_GLOBAL_DEFAULT); + std::atexit([]() { curl_global_cleanup(); }); + }); +} +#endif + +} // namespace + HttpAction::~HttpAction() { Drain(); Stop(); @@ -32,17 +74,17 @@ bool HttpAction::Init(const SimpleJson& config) { else queue_policy_ = QueuePolicy::DropOldest; if (url_.empty()) { - std::cerr << "[HttpAction] url is required\n"; + LogError("[HttpAction] url is required"); return false; } #if HAS_CURL - curl_global_init(CURL_GLOBAL_DEFAULT); + EnsureCurlGlobalInitOnce(); #endif running_.store(true); worker_ = std::thread(&HttpAction::WorkerLoop, this); - std::cout << "[HttpAction] initialized, url=" << url_ << "\n"; + LogInfo("[HttpAction] initialized, url=" + url_ + ", method=" + method_); return true; } @@ -50,17 +92,17 @@ 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 << "\"node_id\":\"" << JsonEscape(event.node_id) << "\","; + oss << "\"rule_name\":\"" << JsonEscape(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 << "\","; + oss << "\"snapshot_url\":\"" << JsonEscape(event.snapshot_url) << "\","; } if (!event.clip_url.empty()) { - oss << "\"clip_url\":\"" << event.clip_url << "\","; + oss << "\"clip_url\":\"" << JsonEscape(event.clip_url) << "\","; } } @@ -142,7 +184,7 @@ 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"; + LogError("[HttpAction] curl_easy_init failed"); return false; } @@ -151,13 +193,26 @@ bool HttpAction::SendRequest(const std::string& json_body) { 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()); + + const std::string method_lc = ToLower(method_); + if (method_lc == "post") { + curl_easy_setopt(curl, CURLOPT_POST, 1L); + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str()); + } else if (method_lc == "put" || method_lc == "patch") { + curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, method_.c_str()); + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, json_body.c_str()); + } else { + LogWarn("[HttpAction] unsupported method='" + method_ + "', falling back to POST"); + curl_easy_setopt(curl, CURLOPT_POST, 1L); + 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"; + LogWarn(std::string("[HttpAction] request failed: ") + curl_easy_strerror(res)); } curl_slist_free_all(headers); @@ -165,8 +220,7 @@ bool HttpAction::SendRequest(const std::string& json_body) { return res == CURLE_OK; #else // Stub implementation for Windows without curl - std::cout << "[HttpAction] HTTP POST to " << url_ << "\n" - << " Body: " << json_body << "\n"; + LogInfo("[HttpAction] (stub) method=" + method_ + " url=" + url_ + " body=" + json_body); return true; #endif } diff --git a/plugins/alarm/actions/log_action.cpp b/plugins/alarm/actions/log_action.cpp index 8e5a927..7500952 100644 --- a/plugins/alarm/actions/log_action.cpp +++ b/plugins/alarm/actions/log_action.cpp @@ -3,26 +3,54 @@ #include #include #include -#include +#include #include +#include "utils/logger.h" + namespace rk3588 { +namespace { + +bool SafeLocalTime(std::time_t t, std::tm& out) { +#if defined(_WIN32) + return localtime_s(&out, &t) == 0; +#elif defined(__unix__) || defined(__APPLE__) + return localtime_r(&t, &out) != nullptr; +#else + static std::mutex mu; + std::lock_guard lock(mu); + std::tm* p = std::localtime(&t); + if (!p) return false; + out = *p; + return true; +#endif +} + +LogLevel LogLevelFromStringOrInfo(const std::string& s) { + LogLevel lvl = LogLevel::Info; + (void)ParseLogLevel(s, lvl); + return lvl; +} + +} // namespace + 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"; + LogInfo("[LogAction] initialized, level=" + level_); 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::tm tm_now{}; + (void)SafeLocalTime(time_t_now, tm_now); std::ostringstream oss; oss << "[ALARM][" << level_ << "] " - << std::put_time(tm_now, "%Y-%m-%d %H:%M:%S") + << std::put_time(&tm_now, "%Y-%m-%d %H:%M:%S") << " node=" << event.node_id << " rule=" << event.rule_name << " frame=" << event.frame_id; @@ -49,7 +77,7 @@ void LogAction::Execute(AlarmEvent& event, std::shared_ptr /*frame*/) { oss << " clip=" << event.clip_url; } - std::cout << oss.str() << "\n"; + Logger::Instance().Log(LogLevelFromStringOrInfo(level_), oss.str()); } } // namespace rk3588 diff --git a/plugins/alarm/actions/snapshot_action.cpp b/plugins/alarm/actions/snapshot_action.cpp index e977c27..588f118 100644 --- a/plugins/alarm/actions/snapshot_action.cpp +++ b/plugins/alarm/actions/snapshot_action.cpp @@ -5,10 +5,11 @@ #include #include #include -#include +#include #include #include "utils/dma_alloc.h" +#include "utils/logger.h" #if defined(RK3588_ENABLE_FFMPEG) extern "C" { @@ -25,6 +26,21 @@ namespace rk3588 { namespace { +bool SafeLocalTime(std::time_t t, std::tm& out) { +#if defined(_WIN32) + return localtime_s(&out, &t) == 0; +#elif defined(__unix__) || defined(__APPLE__) + return localtime_r(&t, &out) != nullptr; +#else + static std::mutex mu; + std::lock_guard lock(mu); + std::tm* p = std::localtime(&t); + if (!p) return false; + out = *p; + return true; +#endif +} + inline uint8_t ClipU8(int v) { if (v < 0) return 0; if (v > 255) return 255; @@ -46,6 +62,7 @@ inline int PlaneStride(const Frame& f, int idx, int fallback) { return fallback; } +#if HAS_FFMPEG bool FillYuv420pFromFrame(const Frame& src, AVFrame* dst) { if (!dst) return false; if (dst->format != AV_PIX_FMT_YUV420P && dst->format != AV_PIX_FMT_YUVJ420P) return false; @@ -179,6 +196,7 @@ bool FillYuv420pFromFrame(const Frame& src, AVFrame* dst) { return false; } +#endif } // namespace @@ -189,7 +207,7 @@ bool SnapshotAction::Init(const SimpleJson& config) { if (const SimpleJson* upload_cfg = config.Find("upload")) { uploader_ = CreateUploader(*upload_cfg); if (!uploader_) { - std::cerr << "[SnapshotAction] failed to create uploader\n"; + LogError("[SnapshotAction] failed to create uploader"); return false; } } else { @@ -198,14 +216,13 @@ bool SnapshotAction::Init(const SimpleJson& config) { uploader_ = CreateUploader(default_cfg); } - std::cout << "[SnapshotAction] initialized, format=" << format_ - << " quality=" << quality_ << "\n"; + LogInfo("[SnapshotAction] initialized, format=" + format_ + " quality=" + std::to_string(quality_)); return true; } void SnapshotAction::Execute(AlarmEvent& event, std::shared_ptr frame) { if (!frame || !frame->data) { - std::cerr << "[SnapshotAction] no frame data\n"; + LogWarn("[SnapshotAction] no frame data"); return; } @@ -214,7 +231,7 @@ void SnapshotAction::Execute(AlarmEvent& event, std::shared_ptr frame) { auto jpeg_data = EncodeJpeg(frame); if (dma_fd >= 0) DmaSyncEndFd(dma_fd); if (jpeg_data.empty()) { - std::cerr << "[SnapshotAction] failed to encode JPEG\n"; + LogWarn("[SnapshotAction] failed to encode JPEG"); return; } @@ -223,21 +240,22 @@ void SnapshotAction::Execute(AlarmEvent& event, std::shared_ptr frame) { if (result.success) { event.snapshot_url = result.url; - std::cout << "[SnapshotAction] uploaded: " << result.url << "\n"; + LogInfo("[SnapshotAction] uploaded: " + result.url); } else { - std::cerr << "[SnapshotAction] upload failed: " << result.error << "\n"; + LogWarn("[SnapshotAction] upload failed: " + result.error); } } 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::tm tm{}; + (void)SafeLocalTime(time_t_now, tm); std::ostringstream oss; - oss << std::put_time(tm, "%Y%m%d") << "/" + oss << std::put_time(&tm, "%Y%m%d") << "/" << event.node_id << "_" - << std::put_time(tm, "%H%M%S") << "_" + << std::put_time(&tm, "%H%M%S") << "_" << event.frame_id << "." << format_; return oss.str(); } @@ -248,7 +266,7 @@ std::vector SnapshotAction::EncodeJpeg(const std::shared_ptr& fr #if HAS_FFMPEG const AVCodec* codec = avcodec_find_encoder(AV_CODEC_ID_MJPEG); if (!codec) { - std::cerr << "[SnapshotAction] MJPEG encoder not found\n"; + LogError("[SnapshotAction] MJPEG encoder not found"); return output; } @@ -304,7 +322,7 @@ std::vector SnapshotAction::EncodeJpeg(const std::shared_ptr& fr 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"; + LogError("[SnapshotAction] FFmpeg not enabled, cannot encode JPEG"); #endif return output; diff --git a/plugins/alarm/alarm_node.cpp b/plugins/alarm/alarm_node.cpp index 28c1053..c2bf3a2 100644 --- a/plugins/alarm/alarm_node.cpp +++ b/plugins/alarm/alarm_node.cpp @@ -4,7 +4,6 @@ #include #include #include -#include #include #include #include @@ -12,6 +11,8 @@ #include #include +#include "utils/logger.h" + #include "face/face_result.h" #include "node.h" #include "rule_engine.h" @@ -51,11 +52,16 @@ public: const uint64_t now = NowEpochMs(); uint64_t last = last_exec_ms_.load(std::memory_order_relaxed); - if (last != 0 && now > last && (now - last) < min_interval_ms_) { - return; + while (true) { + if (last != 0 && now > last && (now - last) < min_interval_ms_) { + return; + } + // Best-effort: only one thread should win the update. + if (last_exec_ms_.compare_exchange_weak(last, now, std::memory_order_relaxed)) { + break; + } + // last is updated by compare_exchange_weak; loop and re-check. } - // Best-effort: only one thread should update; compare-exchange avoids lost updates. - last_exec_ms_.store(now, std::memory_order_relaxed); inner_->Execute(event, std::move(frame)); } @@ -131,7 +137,7 @@ public: // 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"; + LogError("[alarm] failed to init rule engine"); return false; } } @@ -229,18 +235,18 @@ public: input_queue_ = ctx.input_queue; if (!input_queue_) { - std::cerr << "[alarm] no input queue for node " << id_ << "\n"; + LogError("[alarm] no input queue for node " + id_); return false; } - std::cout << "[alarm] initialized with " << actions_.size() << " actions\n"; + LogInfo("[alarm] initialized with " + std::to_string(actions_.size()) + " actions"); return true; } bool Start() override { worker_running_.store(true); worker_ = std::thread(&AlarmNode::WorkerLoop, this); - std::cout << "[alarm] started\n"; + LogInfo("[alarm] started"); return true; } @@ -253,8 +259,8 @@ public: for (auto& action : actions_) action->Drain(); for (auto& action : actions_) action->Stop(); - std::cout << "[alarm] stopped, processed " << processed_frames_ - << " frames, triggered " << alarm_count_ << " alarms\n"; + LogInfo("[alarm] stopped, processed " + std::to_string(processed_frames_) + + " frames, triggered " + std::to_string(alarm_count_) + " alarms"); } void Drain() override { diff --git a/plugins/alarm/rule_engine.cpp b/plugins/alarm/rule_engine.cpp index 022cc15..2b7b591 100644 --- a/plugins/alarm/rule_engine.cpp +++ b/plugins/alarm/rule_engine.cpp @@ -1,12 +1,34 @@ #include "rule_engine.h" #include +#include #include -#include +#include #include +#include "utils/logger.h" + namespace rk3588 { +namespace { + +bool SafeLocalTime(std::time_t t, std::tm& out) { +#if defined(_WIN32) + return localtime_s(&out, &t) == 0; +#elif defined(__unix__) || defined(__APPLE__) + return localtime_r(&t, &out) != nullptr; +#else + static std::mutex mu; + std::lock_guard lock(mu); + std::tm* p = std::localtime(&t); + if (!p) return false; + out = *p; + return true; +#endif +} + +} // namespace + bool RuleEngine::Init(const SimpleJson& rules_config, const std::vector& labels) { labels_ = labels; rules_.clear(); @@ -14,7 +36,7 @@ bool RuleEngine::Init(const SimpleJson& rules_config, const std::vectortm_hour * 60 + local->tm_min; + std::tm local{}; + if (!SafeLocalTime(now, local)) { + return true; + } + 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; diff --git a/plugins/alarm/uploaders/local_uploader.cpp b/plugins/alarm/uploaders/local_uploader.cpp index 5210d63..dd4d453 100644 --- a/plugins/alarm/uploaders/local_uploader.cpp +++ b/plugins/alarm/uploaders/local_uploader.cpp @@ -2,7 +2,8 @@ #include #include -#include + +#include "utils/logger.h" namespace rk3588 { @@ -13,11 +14,11 @@ bool LocalUploader::Init(const SimpleJson& config) { try { std::filesystem::create_directories(base_path_); } catch (const std::exception& e) { - std::cerr << "[LocalUploader] failed to create directory: " << e.what() << "\n"; + LogError(std::string("[LocalUploader] failed to create directory: ") + e.what()); return false; } - std::cout << "[LocalUploader] initialized, path=" << base_path_ << "\n"; + LogInfo("[LocalUploader] initialized, path=" + base_path_); return true; } diff --git a/plugins/alarm/uploaders/minio_uploader.cpp b/plugins/alarm/uploaders/minio_uploader.cpp index 4149521..f21a0bc 100644 --- a/plugins/alarm/uploaders/minio_uploader.cpp +++ b/plugins/alarm/uploaders/minio_uploader.cpp @@ -6,12 +6,12 @@ #include #include #include -#include #include #include #include #include +#include "utils/logger.h" #include "utils/simple_json_writer.h" #if __has_include() @@ -325,15 +325,17 @@ bool MinioUploader::Init(const SimpleJson& config) { #if HAS_CURL static std::once_flag curl_once; - std::call_once(curl_once, [] { curl_global_init(CURL_GLOBAL_DEFAULT); }); + std::call_once(curl_once, [] { + curl_global_init(CURL_GLOBAL_DEFAULT); + std::atexit([]() { curl_global_cleanup(); }); + }); #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"; + LogWarn("[MinioUploader] access_key/secret_key not set; recommended to set presign_endpoint"); } - std::cout << "[MinioUploader] initialized, endpoint=" << endpoint_ - << " bucket=" << bucket_ << "\n"; + LogInfo("[MinioUploader] initialized, endpoint=" + endpoint_ + " bucket=" + bucket_); return true; } @@ -471,8 +473,8 @@ UploadResult MinioUploader::Upload(const std::string& key, return result; #else // Stub for Windows - std::cout << "[MinioUploader] would upload " << size << " bytes as key=" << key - << " (configure presign_endpoint to enable real upload)\n"; + LogInfo("[MinioUploader] would upload " + std::to_string(size) + " bytes as key=" + key + + " (configure presign_endpoint to enable real upload)"); result.success = true; result.url = endpoint_ + "/" + bucket_ + "/" + key; #endif diff --git a/plugins/alarm/uploaders/uploader_factory.cpp b/plugins/alarm/uploaders/uploader_factory.cpp index a776031..a7f8767 100644 --- a/plugins/alarm/uploaders/uploader_factory.cpp +++ b/plugins/alarm/uploaders/uploader_factory.cpp @@ -2,7 +2,7 @@ #include "local_uploader.h" #include "minio_uploader.h" -#include +#include "utils/logger.h" namespace rk3588 { @@ -17,7 +17,7 @@ std::unique_ptr CreateUploader(const SimpleJson& config) { } if (uploader && !uploader->Init(config)) { - std::cerr << "[CreateUploader] failed to init uploader type=" << type << "\n"; + LogError("[CreateUploader] failed to init uploader type=" + type); return nullptr; } diff --git a/plugins/det_post/det_post_node.cpp b/plugins/det_post/det_post_node.cpp index 517ebcf..c68496c 100644 --- a/plugins/det_post/det_post_node.cpp +++ b/plugins/det_post/det_post_node.cpp @@ -4,7 +4,6 @@ #include #include #include -#include #include #include #include @@ -14,6 +13,7 @@ #include #include "node.h" +#include "utils/logger.h" namespace rk3588 { @@ -694,18 +694,18 @@ public: std::shared_ptr snap; std::string err; if (!BuildConfigSnapshot(config, snap, err)) { - std::cerr << "[det_post] invalid config: " << err << "\n"; + LogError("[det_post] invalid config: " + err); return false; } id_ = snap->id; input_queue_ = ctx.input_queue; if (!input_queue_) { - std::cerr << "[det_post] no input queue for node " << id_ << "\n"; + LogError("[det_post] no input queue for node " + id_); return false; } if (ctx.output_queues.empty()) { - std::cerr << "[det_post] no output queue for node " << id_ << "\n"; + LogError("[det_post] no output queue for node " + id_); return false; } output_queues_ = ctx.output_queues; @@ -718,19 +718,19 @@ public: } bool Start() override { - std::cout << "[det_post] started\n"; + LogInfo("[det_post] started"); return true; } void Stop() override { - std::cout << "[det_post] stopped\n"; + LogInfo("[det_post] stopped"); } bool UpdateConfig(const SimpleJson& new_config) override { std::shared_ptr snap; std::string err; if (!BuildConfigSnapshot(new_config, snap, err)) { - std::cerr << "[det_post] UpdateConfig rejected: " << err << "\n"; + LogWarn("[det_post] UpdateConfig rejected: " + err); return false; } if (!id_.empty() && !snap->id.empty() && snap->id != id_) { diff --git a/plugins/input_file/input_file_node.cpp b/plugins/input_file/input_file_node.cpp index 23775c2..6ab5bb5 100644 --- a/plugins/input_file/input_file_node.cpp +++ b/plugins/input_file/input_file_node.cpp @@ -2,7 +2,6 @@ #include #include #include -#include #include #include #include @@ -10,6 +9,7 @@ #include "node.h" #include "utils/thread_affinity.h" +#include "utils/logger.h" #if defined(RK3588_ENABLE_MPP) extern "C" { @@ -50,13 +50,13 @@ public: cpu_affinity_ = ParseCpuAffinity(config); if (ctx.output_queues.empty()) { - std::cerr << "[input_file] no downstream queue configured for node " << id_ << "\n"; + LogError("[input_file] no downstream queue configured for node " + id_); return false; } out_queues_ = ctx.output_queues; if (path_.empty() && !fallback_to_stub_on_fail_) { - std::cerr << "[input_file] path is required\n"; + LogError("[input_file] path is required"); return false; } @@ -83,23 +83,26 @@ public: } #else if (use_ffmpeg_ || use_mpp_) { - std::cerr << "[input_file] requested ffmpeg/mpp but not enabled at build time\n"; + LogError("[input_file] requested ffmpeg/mpp but not enabled at build time"); } worker_ = std::thread(&InputFileNode::LoopStub, this); #endif - std::cout << "[input_file] start path=" << path_ << " loop=" << (loop_ ? "true" : "false") - << " realtime=" << (realtime_ ? "true" : "false") - << " fps=" << fps_; + std::string mode; #if defined(RK3588_ENABLE_MPP) - if (use_ffmpeg_ && use_mpp_) std::cout << " (ffmpeg demux + mpp decode)"; - else if (use_ffmpeg_) std::cout << " (ffmpeg cpu decode)"; - else std::cout << " (stub)"; + if (use_ffmpeg_ && use_mpp_) mode = "ffmpeg demux + mpp decode"; + else if (use_ffmpeg_) mode = "ffmpeg cpu decode"; + else mode = "stub"; #else - if (use_ffmpeg_) std::cout << " (ffmpeg cpu decode)"; - else std::cout << " (stub)"; + if (use_ffmpeg_) mode = "ffmpeg cpu decode"; + else mode = "stub"; #endif - std::cout << "\n"; + + LogInfo("[input_file] start path=" + path_ + + " loop=" + (loop_ ? std::string("true") : std::string("false")) + + " realtime=" + (realtime_ ? std::string("true") : std::string("false")) + + " fps=" + std::to_string(fps_) + + " (" + mode + ")"); return true; } @@ -107,7 +110,7 @@ public: running_.store(false); for (auto& q : out_queues_) q->Stop(); if (worker_.joinable()) worker_.join(); - std::cout << "[input_file] stopped\n"; + LogInfo("[input_file] stopped"); } void Drain() override { @@ -119,7 +122,7 @@ private: if (cpu_affinity_.empty()) return; std::string aerr; if (!SetCurrentThreadAffinity(cpu_affinity_, aerr)) { - std::cerr << "[input_file] SetCurrentThreadAffinity failed: " << aerr << "\n"; + LogWarn("[input_file] SetCurrentThreadAffinity failed: " + aerr); } } @@ -297,7 +300,7 @@ private: std::string oerr; if (!OpenFile(fmt_ctx, video_stream, time_base, fps_out, oerr)) { - std::cerr << "[input_file] open failed: " << oerr << " path=" << path_ << "\n"; + LogError("[input_file] open failed: " + oerr + " path=" + path_); Cleanup(fmt_ctx, codec_ctx, pkt, frm); if (fallback_to_stub_on_fail_) { LoopStub(); @@ -309,7 +312,7 @@ private: const AVCodec* codec = avcodec_find_decoder(fmt_ctx->streams[video_stream]->codecpar->codec_id); if (!codec) { - std::cerr << "[input_file] decoder not found\n"; + LogError("[input_file] decoder not found"); Cleanup(fmt_ctx, codec_ctx, pkt, frm); running_.store(false); break; @@ -317,7 +320,7 @@ private: codec_ctx = avcodec_alloc_context3(codec); avcodec_parameters_to_context(codec_ctx, fmt_ctx->streams[video_stream]->codecpar); if (avcodec_open2(codec_ctx, codec, nullptr) < 0) { - std::cerr << "[input_file] avcodec_open2 failed\n"; + LogError("[input_file] avcodec_open2 failed"); Cleanup(fmt_ctx, codec_ctx, pkt, frm); running_.store(false); break; @@ -546,7 +549,7 @@ private: std::string oerr; if (!OpenFile(fmt_ctx, video_stream, time_base, fps_out, oerr)) { - std::cerr << "[input_file] open failed: " << oerr << " path=" << path_ << "\n"; + LogError("[input_file] open failed: " + oerr + " path=" + path_); Cleanup(fmt_ctx, nullptr, pkt, nullptr); if (fallback_to_stub_on_fail_) { LoopStub(); @@ -561,7 +564,7 @@ private: if (codec_id == AV_CODEC_ID_H264) coding = MPP_VIDEO_CodingAVC; else if (codec_id == AV_CODEC_ID_HEVC) coding = MPP_VIDEO_CodingHEVC; else { - std::cerr << "[input_file] unsupported codec for mpp\n"; + LogError("[input_file] unsupported codec for mpp"); Cleanup(fmt_ctx, nullptr, pkt, nullptr); running_.store(false); break; @@ -569,7 +572,7 @@ private: MppDecoderWrapper dec; if (!dec.Init(coding)) { - std::cerr << "[input_file] mpp init failed\n"; + LogError("[input_file] mpp init failed"); Cleanup(fmt_ctx, nullptr, pkt, nullptr); running_.store(false); break; diff --git a/plugins/input_rtsp/input_rtsp_node.cpp b/plugins/input_rtsp/input_rtsp_node.cpp index 9be6ad7..1002608 100644 --- a/plugins/input_rtsp/input_rtsp_node.cpp +++ b/plugins/input_rtsp/input_rtsp_node.cpp @@ -2,7 +2,6 @@ #include #include #include -#include #include #include #include @@ -63,7 +62,7 @@ public: } cpu_affinity_ = ParseCpuAffinity(config); if (ctx.output_queues.empty()) { - std::cerr << "[input_rtsp] no downstream queue configured for node " << id_ << "\n"; + LogError("[input_rtsp] no downstream queue configured for node " + id_); return false; } out_queues_ = ctx.output_queues; @@ -89,7 +88,7 @@ public: } #else if (use_ffmpeg_ || use_mpp_) { - std::cerr << "[input_rtsp] requested ffmpeg/mpp but not enabled at build time" << "\n"; + LogError("[input_rtsp] requested ffmpeg/mpp but not enabled at build time"); } worker_ = std::thread(&InputRtspNode::LoopStub, this); #endif @@ -172,7 +171,7 @@ private: if (cpu_affinity_.empty()) return; std::string aerr; if (!SetCurrentThreadAffinity(cpu_affinity_, aerr)) { - std::cerr << "[input_rtsp] SetCurrentThreadAffinity failed: " << aerr << "\n"; + LogWarn("[input_rtsp] SetCurrentThreadAffinity failed: " + aerr); } } @@ -232,7 +231,7 @@ private: av_dict_set(&opts, "fflags", "nobuffer", 0); av_dict_set(&opts, "flags", "low_delay", 0); if (avformat_open_input(&fmt_ctx, url_.c_str(), nullptr, &opts) < 0) { - std::cerr << "[input_rtsp] avformat_open_input failed: " << url_ << "\n"; + LogError("[input_rtsp] avformat_open_input failed: " + url_); av_dict_free(&opts); Cleanup(fmt_ctx, codec_ctx, pkt, frm); if (fallback_to_stub_on_fail_) { @@ -246,7 +245,7 @@ private: av_dict_free(&opts); if (avformat_find_stream_info(fmt_ctx, nullptr) < 0) { // For RTSP, codec parameters are often available from SDP already; keep going. - std::cerr << "[input_rtsp] avformat_find_stream_info failed (continuing)\n"; + LogWarn("[input_rtsp] avformat_find_stream_info failed (continuing)"); } for (unsigned i = 0; i < fmt_ctx->nb_streams; ++i) { if (fmt_ctx->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) { @@ -256,7 +255,7 @@ private: } } if (video_stream < 0) { - std::cerr << "[input_rtsp] no video stream\n"; + LogError("[input_rtsp] no video stream"); Cleanup(fmt_ctx, codec_ctx, pkt, frm); std::this_thread::sleep_for(seconds(backoff)); backoff = std::min(backoff_max, backoff * 2); @@ -265,7 +264,7 @@ private: const AVCodec* codec = avcodec_find_decoder(fmt_ctx->streams[video_stream]->codecpar->codec_id); if (!codec) { - std::cerr << "[input_rtsp] decoder not found\n"; + LogError("[input_rtsp] decoder not found"); Cleanup(fmt_ctx, codec_ctx, pkt, frm); std::this_thread::sleep_for(seconds(backoff)); backoff = std::min(backoff_max, backoff * 2); @@ -274,7 +273,7 @@ private: codec_ctx = avcodec_alloc_context3(codec); avcodec_parameters_to_context(codec_ctx, fmt_ctx->streams[video_stream]->codecpar); if (avcodec_open2(codec_ctx, codec, nullptr) < 0) { - std::cerr << "[input_rtsp] avcodec_open2 failed\n"; + LogError("[input_rtsp] avcodec_open2 failed"); Cleanup(fmt_ctx, codec_ctx, pkt, frm); std::this_thread::sleep_for(seconds(backoff)); backoff = std::min(backoff_max, backoff * 2); @@ -477,7 +476,8 @@ private: MPP_RET ret = mpp_packet_init(&src, const_cast(data), size); if (ret != MPP_OK || !src) { if (++fail_copy_init_ <= 3 || fail_copy_init_ % 200 == 0) { - std::cerr << "[input_rtsp] mpp_packet_init failed ret=" << ret << " size=" << size << "\n"; + LogWarn("[input_rtsp] mpp_packet_init failed ret=" + std::to_string(ret) + + " size=" + std::to_string(size)); } return false; } @@ -487,7 +487,8 @@ private: mpp_packet_deinit(&src); if (ret != MPP_OK || !packet) { if (++fail_copy_init_ <= 3 || fail_copy_init_ % 200 == 0) { - std::cerr << "[input_rtsp] mpp_packet_copy_init failed ret=" << ret << " size=" << size << "\n"; + LogWarn("[input_rtsp] mpp_packet_copy_init failed ret=" + std::to_string(ret) + + " size=" + std::to_string(size)); } return false; } @@ -504,7 +505,7 @@ private: if (gr == MPP_ERR_TIMEOUT) return; if (gr != MPP_OK) { if (++fail_get_frame_ <= 3 || fail_get_frame_ % 200 == 0) { - std::cerr << "[input_rtsp] decode_get_frame ret=" << gr << "\n"; + LogWarn("[input_rtsp] decode_get_frame ret=" + std::to_string(gr)); } return; } @@ -543,7 +544,7 @@ private: // Treat as packet drop to keep pipeline progressing; caller can decide to wait for next IDR. if (last_put != MPP_ERR_BUFFER_FULL) { if (++fail_put_packet_ <= 3 || fail_put_packet_ % 200 == 0) { - std::cerr << "[input_rtsp] decode_put_packet failed ret=" << last_put << "\n"; + LogWarn("[input_rtsp] decode_put_packet failed ret=" + std::to_string(last_put)); } } } @@ -553,7 +554,7 @@ private: bool SendExtraData(const uint8_t* data, size_t size) { if (!ctx || !mpi || !data || size == 0) return false; if (!IsAnnexB(data, size)) { - std::cerr << "[input_rtsp] extra_data is not AnnexB; mpp may ignore it\n"; + LogWarn("[input_rtsp] extra_data is not AnnexB; mpp may ignore it"); } MppPacket src = nullptr; @@ -590,7 +591,7 @@ private: usleep(2000); } - std::cerr << "[input_rtsp] extra_data decode_put_packet timeout\n"; + LogWarn("[input_rtsp] extra_data decode_put_packet timeout"); mpp_packet_deinit(&pkt); return false; } @@ -612,7 +613,7 @@ private: !try_group(static_cast(MPP_BUFFER_TYPE_DRM)) && !try_group(static_cast(MPP_BUFFER_TYPE_ION | MPP_BUFFER_FLAGS_CACHABLE)) && !try_group(static_cast(MPP_BUFFER_TYPE_ION))) { - std::cerr << "[input_rtsp] mpp info_change: failed to alloc frame buffer group\n"; + LogError("[input_rtsp] mpp info_change: failed to alloc frame buffer group"); return; } frame_group = group; @@ -658,7 +659,7 @@ private: av_dict_set(&opts, "fflags", "nobuffer", 0); av_dict_set(&opts, "flags", "low_delay", 0); if (avformat_open_input(&fmt_ctx, url_.c_str(), nullptr, &opts) < 0) { - std::cerr << "[input_rtsp] avformat_open_input failed: " << url_ << "\n"; + LogError("[input_rtsp] avformat_open_input failed: " + url_); av_dict_free(&opts); if (bsf_ctx) av_bsf_free(&bsf_ctx); if (filt_pkt) av_packet_free(&filt_pkt); @@ -674,7 +675,7 @@ private: av_dict_free(&opts); if (avformat_find_stream_info(fmt_ctx, nullptr) < 0) { // For RTSP, codec parameters are often available from SDP already; keep going. - std::cerr << "[input_rtsp] avformat_find_stream_info failed (continuing)\n"; + LogWarn("[input_rtsp] avformat_find_stream_info failed (continuing)"); } for (unsigned i = 0; i < fmt_ctx->nb_streams; ++i) { if (fmt_ctx->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) { @@ -684,7 +685,7 @@ private: } } if (video_stream < 0) { - std::cerr << "[input_rtsp] no video stream\n"; + LogError("[input_rtsp] no video stream"); if (bsf_ctx) av_bsf_free(&bsf_ctx); if (filt_pkt) av_packet_free(&filt_pkt); Cleanup(fmt_ctx, nullptr, pkt, nullptr); @@ -698,7 +699,7 @@ private: if (codec_id == AV_CODEC_ID_H264) coding = MPP_VIDEO_CodingAVC; else if (codec_id == AV_CODEC_ID_HEVC) coding = MPP_VIDEO_CodingHEVC; else { - std::cerr << "[input_rtsp] unsupported codec for mpp\n"; + LogError("[input_rtsp] unsupported codec for mpp"); if (bsf_ctx) av_bsf_free(&bsf_ctx); if (filt_pkt) av_packet_free(&filt_pkt); Cleanup(fmt_ctx, nullptr, pkt, nullptr); @@ -709,7 +710,7 @@ private: MppDecoderWrapper dec; if (!dec.Init(coding)) { - std::cerr << "[input_rtsp] mpp init failed\n"; + LogError("[input_rtsp] mpp init failed"); if (bsf_ctx) av_bsf_free(&bsf_ctx); if (filt_pkt) av_packet_free(&filt_pkt); Cleanup(fmt_ctx, nullptr, pkt, nullptr); @@ -761,7 +762,7 @@ private: " annexb=" + std::to_string(IsAnnexB(extra, extra_size) ? 1 : 0)); } if (!dec.SendExtraData(extra, extra_size)) { - std::cerr << "[input_rtsp] send extra_data failed\n"; + LogWarn("[input_rtsp] send extra_data failed"); } } } @@ -800,7 +801,7 @@ private: if (!need_idr && ++decode_fail >= 3) { need_idr = true; announced_wait_idr = false; - std::cerr << "[input_rtsp] decoder desync/backpressure; re-sync on next IDR\n"; + LogWarn("[input_rtsp] decoder desync/backpressure; re-sync on next IDR"); } } }; @@ -866,7 +867,7 @@ private: } } } else { - std::cerr << "[input_rtsp] non-AnnexB bitstream but bsf init failed; mpp decode may produce no frames\n"; + LogWarn("[input_rtsp] non-AnnexB bitstream but bsf init failed; mpp decode may produce no frames"); } av_packet_unref(pkt); diff --git a/plugins/osd/osd_node.cpp b/plugins/osd/osd_node.cpp index 450423b..4d19af5 100644 --- a/plugins/osd/osd_node.cpp +++ b/plugins/osd/osd_node.cpp @@ -3,7 +3,6 @@ #include #include #include -#include #include #include #include @@ -371,11 +370,11 @@ public: input_queue_ = ctx.input_queue; if (!input_queue_) { - std::cerr << "[osd] no input queue for node " << id_ << "\n"; + LogError("[osd] no input queue for node " + id_); return false; } if (ctx.output_queues.empty()) { - std::cerr << "[osd] no output queue for node " << id_ << "\n"; + LogError("[osd] no output queue for node " + id_); return false; } output_queues_ = ctx.output_queues; diff --git a/plugins/preprocess/preprocess_node.cpp b/plugins/preprocess/preprocess_node.cpp index cecfa44..1e2336a 100644 --- a/plugins/preprocess/preprocess_node.cpp +++ b/plugins/preprocess/preprocess_node.cpp @@ -4,7 +4,6 @@ #include #include #include -#include #include #include #include @@ -59,6 +58,7 @@ int GlobalRgaMaxInflight() { int v = std::stoi(s); if (v > 0 && v <= 32) GlobalRgaMaxInflightRef().store(v); } catch (...) { + LogWarn(std::string("[preprocess] invalid RK3588_RGA_MAX_INFLIGHT='") + s + "'"); } }); int v = GlobalRgaMaxInflightRef().load(); @@ -148,7 +148,7 @@ void EnsureRgaInitializedOnce() { std::call_once(once, []() { const IM_STATUS st = imcheckHeader(); if (st != IM_STATUS_NOERROR && st != IM_STATUS_SUCCESS) { - std::cerr << "[preprocess] imcheckHeader failed: " << imStrError(st) << "\n"; + LogWarn(std::string("[preprocess] imcheckHeader failed: ") + imStrError(st)); } }); } @@ -344,25 +344,25 @@ public: input_queue_ = ctx.input_queue; if (!input_queue_) { - std::cerr << "[preprocess] no input queue for node " << id_ << "\n"; + LogError("[preprocess] no input queue for node " + id_); return false; } if (ctx.output_queues.empty()) { - std::cerr << "[preprocess] no output queue for node " << id_ << "\n"; + LogError("[preprocess] no output queue for node " + id_); return false; } output_queues_ = ctx.output_queues; #if !defined(RK3588_ENABLE_RGA) if (requested_use_rga) { - std::cerr << "[preprocess] use_rga=true but RGA not enabled at build time\n"; + LogError("[preprocess] use_rga=true but RGA not enabled at build time"); return false; } use_rga_ = false; #endif #if !defined(RK3588_ENABLE_FFMPEG) if (!use_rga_) { - std::cerr << "[preprocess] neither RGA nor FFmpeg enabled\n"; + LogError("[preprocess] neither RGA nor FFmpeg enabled"); return false; } #endif @@ -507,20 +507,20 @@ private: int dst_hstride = Align16(out_h); if (src_fmt_rga == RK_FORMAT_UNKNOWN || dst_fmt_rga == RK_FORMAT_UNKNOWN) { - std::cerr << "[preprocess] unsupported format for RGA\n"; + LogError("[preprocess] unsupported format for RGA"); return false; } size_t out_size = CalcImageSizeStrided(dst_wstride, dst_hstride, out_fmt); if (out_size == 0) { - std::cerr << "[preprocess] invalid output size for RGA\n"; + LogError("[preprocess] invalid output size for RGA"); return false; } // Use DMA-BUF allocation to avoid >4GB address issue with RGA auto dma_buf = DmaAlloc(out_size); if (!dma_buf || !dma_buf->valid()) { - std::cerr << "[preprocess] DMA alloc failed\n"; + LogError("[preprocess] DMA alloc failed"); return false; } @@ -588,7 +588,7 @@ private: // If there's no DMA fd, we must allocate/copy into a DMA-BUF for RGA. if (frame->dma_fd < 0) { if (!CopySrcToDmaIfPossible()) { - std::cerr << "[preprocess] no dma_fd and src copy failed\n"; + LogError("[preprocess] no dma_fd and src copy failed"); return false; } } @@ -674,7 +674,7 @@ private: const int src_fd = (src_dma_buf && src_dma_buf->valid()) ? src_dma_buf->fd : frame->dma_fd; if (src_fd < 0 || !RunRgaOnce(src_fd, rga_err)) { - std::cerr << "[preprocess] " << (rga_err.empty() ? "RGA failed" : rga_err) << "\n"; + LogError(std::string("[preprocess] ") + (rga_err.empty() ? "RGA failed" : rga_err)); return false; } diff --git a/plugins/publish/publish_node.cpp b/plugins/publish/publish_node.cpp index 1b179fb..ebe6fb9 100644 --- a/plugins/publish/publish_node.cpp +++ b/plugins/publish/publish_node.cpp @@ -6,7 +6,6 @@ #include #include #include -#include #include #include #include @@ -194,7 +193,7 @@ private: if (!warned_) { char errbuf[128]; av_strerror(ret, errbuf, sizeof(errbuf)); - std::cerr << "[publish] write failed for " << url_ << ": " << errbuf << ", resetting...\n"; + LogWarn(std::string("[publish] write failed for ") + url_ + ": " + errbuf + ", resetting..."); warned_ = true; } return false; @@ -216,7 +215,7 @@ private: AVFormatContext* fmt = nullptr; const char* fmt_name = proto_ == "hls" ? "hls" : "rtsp"; if (avformat_alloc_output_context2(&fmt, nullptr, fmt_name, url_.c_str()) < 0 || !fmt) { - std::cerr << "[publish] alloc context failed " << url_ << "\n"; + LogError("[publish] alloc context failed " + url_); return false; } @@ -370,7 +369,7 @@ public: } else if (src.format == PixelFormat::YUV420) { mpp_fmt_ = MPP_FMT_YUV420P; } else { - std::cerr << "[publish] unsupported pixel format for mpp encoder\n"; + LogError("[publish] unsupported pixel format for mpp encoder"); return false; } @@ -381,20 +380,20 @@ public: } if (mpp_create(&ctx_, &mpi_) != MPP_OK) { - std::cerr << "[publish] mpp_create failed\n"; + LogError("[publish] mpp_create failed"); return false; } if (mpp_init(ctx_, MPP_CTX_ENC, coding_) != MPP_OK) { - std::cerr << "[publish] mpp_init enc failed\n"; + LogError("[publish] mpp_init enc failed"); return false; } if (mpp_enc_cfg_init(&cfg_) != MPP_OK) { - std::cerr << "[publish] mpp_enc_cfg_init failed\n"; + LogError("[publish] mpp_enc_cfg_init failed"); return false; } if (mpi_->control(ctx_, MPP_ENC_GET_CFG, cfg_) != MPP_OK) { - std::cerr << "[publish] MPP_ENC_GET_CFG failed\n"; + LogError("[publish] MPP_ENC_GET_CFG failed"); return false; } @@ -416,7 +415,7 @@ public: mpp_enc_cfg_set_s32(cfg_, "codec:type", coding_); if (mpi_->control(ctx_, MPP_ENC_SET_CFG, cfg_) != MPP_OK) { - std::cerr << "[publish] MPP_ENC_SET_CFG failed\n"; + LogError("[publish] MPP_ENC_SET_CFG failed"); return false; } @@ -424,7 +423,7 @@ public: if (coding_ == MPP_VIDEO_CodingAVC || coding_ == MPP_VIDEO_CodingHEVC) { MppEncHeaderMode header_mode = MPP_ENC_HEADER_MODE_EACH_IDR; if (mpi_->control(ctx_, MPP_ENC_SET_HEADER_MODE, &header_mode) != MPP_OK) { - std::cerr << "[publish] MPP_ENC_SET_HEADER_MODE failed\n"; + LogWarn("[publish] MPP_ENC_SET_HEADER_MODE failed"); } } @@ -441,7 +440,7 @@ public: if (mpp_buffer_group_get_internal(&frm_grp_, MPP_BUFFER_TYPE_DRM, NULL) != MPP_OK) { if (mpp_buffer_group_get_internal(&frm_grp_, MPP_BUFFER_TYPE_NORMAL, NULL) != MPP_OK) { - std::cerr << "[publish] mpp_buffer_group_get_internal failed\n"; + LogError("[publish] mpp_buffer_group_get_internal failed"); return false; } } @@ -685,7 +684,7 @@ public: #endif if (!input_queue_) { - std::cerr << "[publish] no input queue for node " << id_ << "\n"; + LogError("[publish] no input queue for node " + id_); return false; } @@ -874,7 +873,7 @@ private: EnsureZlmEnv(); if (!StartRtspServerOnce(port)) { - std::cerr << "[publish] zlm rtsp server start failed on port " << port << "\n"; + LogError("[publish] zlm rtsp server start failed on port " + std::to_string(port)); return false; } @@ -887,13 +886,13 @@ private: media_ = mk_media_create("__defaultVhost__", app_.c_str(), stream_.c_str(), 0, 0, 0); if (!media_) { - std::cerr << "[publish] zlm mk_media_create failed app=" << app_ << " stream=" << stream_ << "\n"; + LogError("[publish] zlm mk_media_create failed app=" + app_ + " stream=" + stream_); return false; } int ok = mk_media_init_video(media_, codec_id, width, height, static_cast(fps), bitrate_kbps * 1000); if (!ok) { - std::cerr << "[publish] zlm mk_media_init_video failed" << "\n"; + LogError("[publish] zlm mk_media_init_video failed"); mk_media_release(media_); media_ = nullptr; return false; @@ -903,7 +902,7 @@ private: is_h265_ = (codec_id == 1); splitter_ = mk_h264_splitter_create(&OnSplitFrame, this); if (!splitter_) { - std::cerr << "[publish] zlm mk_h264_splitter_create failed" << "\n"; + LogError("[publish] zlm mk_h264_splitter_create failed"); mk_media_release(media_); media_ = nullptr; return false; @@ -1232,7 +1231,7 @@ private: if (!encoder_ready_) { if (!mpp_encoder_->InitFromFrame(*frame, codec_, fps_, gop_, bitrate_kbps_)) { - std::cerr << "[publish] encoder init failed, fallback to stub\n"; + LogWarn("[publish] encoder init failed, fallback to stub"); use_mpp_ = false; ProcessStub(frame); return; diff --git a/plugins/storage/storage_node.cpp b/plugins/storage/storage_node.cpp index 9301619..5ad8cf1 100644 --- a/plugins/storage/storage_node.cpp +++ b/plugins/storage/storage_node.cpp @@ -6,12 +6,13 @@ #include #include #include -#include #include #include #include #include +#include "utils/logger.h" + #include "node.h" #include "media/encoded_video_meta.h" @@ -156,22 +157,40 @@ static std::vector AnnexBToLengthPrefixed(const uint8_t* d, size_t n) { } #endif +bool SafeLocalTime(std::time_t t, std::tm& out) { +#if defined(_WIN32) + return localtime_s(&out, &t) == 0; +#elif defined(__unix__) || defined(__APPLE__) + return localtime_r(&t, &out) != nullptr; +#else + static std::mutex mu; + std::lock_guard lock(mu); + std::tm* p = std::localtime(&t); + if (!p) return false; + out = *p; + return true; +#endif +} + 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::tm tm{}; + if (!SafeLocalTime(time_t_now, tm)) { + LogWarn("[storage] localtime failed while formatting filename"); + } 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; + 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 { @@ -201,7 +220,7 @@ public: } else if (fmt == PixelFormat::YUV420) { mpp_fmt_ = MPP_FMT_YUV420P; } else { - std::cerr << "[storage] unsupported pixel format for MPP encoder\n"; + LogError("[storage] unsupported pixel format for MPP encoder"); return false; } @@ -382,31 +401,32 @@ public: input_queue_ = ctx.input_queue; if (!input_queue_) { - std::cerr << "[storage] no input queue\n"; + LogError("[storage] no input queue"); return false; } try { std::filesystem::create_directories(base_path_); } catch (const std::exception& e) { - std::cerr << "[storage] failed to create directory: " << e.what() << "\n"; + LogError(std::string("[storage] failed to create directory: ") + e.what()); return false; } - std::cout << "[storage] initialized, path=" << base_path_ - << " segment=" << segment_sec_ << "s format=" << format_ << "\n"; + LogInfo("[storage] initialized, path=" + base_path_ + + " segment=" + std::to_string(segment_sec_) + + "s format=" + format_); return true; } bool Start() override { - std::cout << "[storage] started\n"; + LogInfo("[storage] started"); return true; } void Stop() override { if (input_queue_) input_queue_->Stop(); CloseCurrentFile(); - std::cout << "[storage] stopped, recorded " << total_frames_ << " frames\n"; + LogInfo("[storage] stopped, recorded " + std::to_string(total_frames_) + " frames"); } void Drain() override { @@ -442,7 +462,11 @@ public: try { std::filesystem::create_directories(base_path_); + } catch (const std::exception& e) { + LogWarn(std::string("[storage] create_directories failed during UpdateConfig: ") + e.what()); + return false; } catch (...) { + LogWarn("[storage] create_directories failed during UpdateConfig: unknown error"); return false; } @@ -501,12 +525,16 @@ private: if (p.has_parent_path()) { std::filesystem::create_directories(p.parent_path()); } - } catch (...) {} + } catch (const std::exception& e) { + LogWarn(std::string("[storage] failed to create directories for segment: ") + e.what()); + } catch (...) { + LogWarn("[storage] failed to create directories for segment: unknown error"); + } 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"; + LogError("[storage] failed to create output context"); return; } @@ -600,9 +628,9 @@ private: file_open_ = true; segment_start_ = std::chrono::steady_clock::now(); segment_frames_ = 0; - std::cout << "[storage] opened: " << current_path_ << "\n"; + LogInfo("[storage] opened: " + current_path_); #else - std::cerr << "[storage] FFmpeg not enabled\n"; + LogError("[storage] FFmpeg not enabled"); #endif } @@ -700,8 +728,7 @@ private: segment_base_pts_ms_ = 0; last_pts90k_ = -1; - std::cout << "[storage] closed: " << current_path_ - << " (" << segment_frames_ << " frames)\n"; + LogInfo("[storage] closed: " + current_path_ + " (" + std::to_string(segment_frames_) + " frames)"); #endif } diff --git a/plugins/zlm_http/zlm_http_node.cpp b/plugins/zlm_http/zlm_http_node.cpp index f68123f..9a8c627 100644 --- a/plugins/zlm_http/zlm_http_node.cpp +++ b/plugins/zlm_http/zlm_http_node.cpp @@ -3,11 +3,11 @@ #include #include #include -#include #include #include #include "node.h" +#include "utils/logger.h" #if defined(RK3588_ENABLE_ZLMEDIAKIT) extern "C" { @@ -189,7 +189,7 @@ public: #if !defined(RK3588_ENABLE_ZLMEDIAKIT) (void)config; - std::cerr << "[zlm_http] RK3588_ENABLE_ZLMEDIAKIT is OFF at build time\n"; + LogError("[zlm_http] RK3588_ENABLE_ZLMEDIAKIT is OFF at build time"); return false; #else root_ = config.ValueOr("root", ""); @@ -199,7 +199,7 @@ public: ssl_ = config.ValueOr("ssl", false); if (root_.empty()) { - std::cerr << "[zlm_http] missing config: root (or path)\n"; + LogError("[zlm_http] missing config: root (or path)"); return false; } if (prefix_.empty()) prefix_ = "/"; @@ -224,11 +224,11 @@ public: auto actual = StartHttpServerOnce(port_, ssl_); if (actual == 0) { - std::cerr << "[zlm_http] failed to start HTTP server on port " << port_ << "\n"; + LogError("[zlm_http] failed to start HTTP server on port " + std::to_string(port_)); return false; } - std::cout << "[zlm_http] http file server started, port=" << actual - << " root=" << root_ << " prefix=" << prefix_ << "\n"; + LogInfo("[zlm_http] http file server started, port=" + std::to_string(actual) + + " root=" + root_ + " prefix=" + prefix_); return true; #endif }