diff --git a/CMakeLists.txt b/CMakeLists.txt index 7d1c639..af66d39 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -60,6 +60,21 @@ set(SRC_FILES src/utils/config_expand.cpp ) +# Shared cross-plugin state (used by gate/tracker and potentially other nodes). +add_library(rk_shared_state SHARED src/utils/shared_state.cpp) +target_compile_definitions(rk_shared_state PRIVATE RK3588_BUILD_SHARED_STATE) +target_include_directories(rk_shared_state PRIVATE ${CMAKE_SOURCE_DIR}/include) +target_link_libraries(rk_shared_state PRIVATE project_options Threads::Threads) +set_target_properties(rk_shared_state PROPERTIES + OUTPUT_NAME "rk_shared_state" + RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR} + LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR} +) +install(TARGETS rk_shared_state + RUNTIME DESTINATION bin + LIBRARY DESTINATION bin +) + add_executable(media-server ${SRC_FILES}) set_target_properties(media-server PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}) target_include_directories(media-server diff --git a/configs/sample_cam1.json b/configs/sample_cam1.json index 15851ba..b5ef274 100644 --- a/configs/sample_cam1.json +++ b/configs/sample_cam1.json @@ -84,7 +84,8 @@ "enable": true, "mode": "bytetrack_lite", "per_class": true, - "track_classes": [], + "state_key": "cam1_sample_full_pipeline", + "track_classes": [0], "ignore_classes": [], "allowed_models": ["yolov5", "yolov8"], "high_th": 0.5, @@ -95,6 +96,32 @@ "max_tracks": 128, "debug": { "stats": false, "stats_interval": 200 } }, + { + "id": "gate_face_cam1", + "type": "gate", + "role": "filter", + "enable": true, + "state_key": "cam1_sample_full_pipeline", + "class_ids": [0], + "require_confirmed": true, + "min_count": 1, + "max_age_ms": 1000, + "min_interval_ms": 300, + "min_box_area_ratio": 0.01 + }, + { + "id": "pre_face_cam1", + "type": "preprocess", + "role": "filter", + "enable": true, + "dst_w": 1280, + "dst_h": 720, + "dst_format": "rgb", + "dst_packed": true, + "keep_ratio": false, + "rga_gate": "cam1_sample_full_pipeline", + "use_rga": true + }, { "id": "osd_cam1", "type": "osd", @@ -151,10 +178,7 @@ "cooldown_ms": 3000 } ], - "face_rules": [ - { "name": "unknown_face", "type": "unknown", "cooldown_ms": 5000, "min_sim": 0.0 }, - { "name": "known_person", "type": "person", "cooldown_ms": 5000, "min_sim": 0.5 } - ], + "face_rules": [], "actions": { "log": { "enable": true, "level": "info" }, "snapshot": { @@ -193,18 +217,59 @@ "method": "POST" } } + }, + { + "id": "alarm_face_cam1", + "type": "alarm", + "role": "sink", + "enable": true, + "eval_fps": 5, + "labels": [], + "rules": [], + "face_rules": [ + { "name": "unknown_face", "type": "unknown", "cooldown_ms": 5000, "min_sim": 0.0 }, + { "name": "known_person", "type": "person", "cooldown_ms": 5000, "min_sim": 0.5 } + ], + "actions": { + "log": { "enable": true, "level": "info" }, + "snapshot": { + "enable": true, + "format": "jpg", + "quality": 85, + "upload": { + "type": "minio", + "endpoint": "http://10.0.0.5:9000", + "bucket": "test", + "region": "us-east-1", + "access_key": "minioadmin", + "secret_key": "minioadmin" + } + }, + "clip": { "enable": false }, + "http": { + "enable": false, + "url": "http://127.0.0.1:8080/api/alarm", + "timeout_ms": 3000, + "include_media_url": true, + "method": "POST" + } + } } ], "edges": [ ["in_cam1", "pre_cam1"], ["pre_cam1", "yolo_cam1"], - ["yolo_cam1", "face_det_cam1"], - ["face_det_cam1", "face_recog_cam1"], - ["face_recog_cam1", "trk_cam1"], + ["yolo_cam1", "trk_cam1"], ["trk_cam1", "osd_cam1"], ["osd_cam1", "post_cam1"], ["post_cam1", "pub_cam1"], - ["pub_cam1", "alarm_cam1"] + ["pub_cam1", "alarm_cam1"], + + ["in_cam1", "gate_face_cam1", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["gate_face_cam1", "pre_face_cam1", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["pre_face_cam1", "face_det_cam1", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["face_det_cam1", "face_recog_cam1", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["face_recog_cam1", "alarm_face_cam1"] ] }, { @@ -290,7 +355,8 @@ "enable": true, "mode": "bytetrack_lite", "per_class": true, - "track_classes": [], + "state_key": "cam2_sample_full_pipeline", + "track_classes": [0], "ignore_classes": [], "allowed_models": ["yolov5", "yolov8"], "high_th": 0.5, @@ -301,6 +367,32 @@ "max_tracks": 128, "debug": { "stats": false, "stats_interval": 200 } }, + { + "id": "gate_face_cam2", + "type": "gate", + "role": "filter", + "enable": true, + "state_key": "cam2_sample_full_pipeline", + "class_ids": [0], + "require_confirmed": true, + "min_count": 1, + "max_age_ms": 1000, + "min_interval_ms": 300, + "min_box_area_ratio": 0.01 + }, + { + "id": "pre_face_cam2", + "type": "preprocess", + "role": "filter", + "enable": true, + "dst_w": 1280, + "dst_h": 720, + "dst_format": "rgb", + "dst_packed": true, + "keep_ratio": false, + "rga_gate": "cam2_sample_full_pipeline", + "use_rga": true + }, { "id": "osd_cam2", "type": "osd", @@ -357,10 +449,7 @@ "cooldown_ms": 3000 } ], - "face_rules": [ - { "name": "unknown_face", "type": "unknown", "cooldown_ms": 5000, "min_sim": 0.0 }, - { "name": "known_person", "type": "person", "cooldown_ms": 5000, "min_sim": 0.5 } - ], + "face_rules": [], "actions": { "log": { "enable": true, "level": "info" }, "snapshot": { @@ -399,18 +488,59 @@ "method": "POST" } } + }, + { + "id": "alarm_face_cam2", + "type": "alarm", + "role": "sink", + "enable": true, + "eval_fps": 5, + "labels": [], + "rules": [], + "face_rules": [ + { "name": "unknown_face", "type": "unknown", "cooldown_ms": 5000, "min_sim": 0.0 }, + { "name": "known_person", "type": "person", "cooldown_ms": 5000, "min_sim": 0.5 } + ], + "actions": { + "log": { "enable": true, "level": "info" }, + "snapshot": { + "enable": true, + "format": "jpg", + "quality": 85, + "upload": { + "type": "minio", + "endpoint": "http://10.0.0.5:9000", + "bucket": "test", + "region": "us-east-1", + "access_key": "minioadmin", + "secret_key": "minioadmin" + } + }, + "clip": { "enable": false }, + "http": { + "enable": false, + "url": "http://127.0.0.1:8080/api/alarm", + "timeout_ms": 3000, + "include_media_url": true, + "method": "POST" + } + } } ], "edges": [ ["in_cam2", "pre_cam2"], ["pre_cam2", "yolo_cam2"], - ["yolo_cam2", "face_det_cam2"], - ["face_det_cam2", "face_recog_cam2"], - ["face_recog_cam2", "trk_cam2"], + ["yolo_cam2", "trk_cam2"], ["trk_cam2", "osd_cam2"], ["osd_cam2", "post_cam2"], ["post_cam2", "pub_cam2"], - ["pub_cam2", "alarm_cam2"] + ["pub_cam2", "alarm_cam2"], + + ["in_cam2", "gate_face_cam2", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["gate_face_cam2", "pre_face_cam2", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["pre_face_cam2", "face_det_cam2", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["face_det_cam2", "face_recog_cam2", {"queue": {"size": 2, "strategy": "drop_oldest"}}], + ["face_recog_cam2", "alarm_face_cam2"] ] } ] diff --git a/include/utils/shared_state.h b/include/utils/shared_state.h new file mode 100644 index 0000000..970bbb1 --- /dev/null +++ b/include/utils/shared_state.h @@ -0,0 +1,63 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "frame/frame.h" + +// SharedState is a process-wide, cross-plugin state store. +// It is implemented in a shared library so all dynamically loaded node plugins +// see the same singleton instance. + +namespace rk3588 { + +#if defined(_WIN32) +#if defined(RK3588_BUILD_SHARED_STATE) +#define RK3588_SHARED_STATE_API __declspec(dllexport) +#else +#define RK3588_SHARED_STATE_API __declspec(dllimport) +#endif +#else +#define RK3588_SHARED_STATE_API +#endif + +struct TrackedObject { + int cls_id = -1; + int track_id = -1; + float score = 0.0f; + Rect bbox{}; // coordinates in snapshot image space (img_w/img_h) + bool confirmed = false; +}; + +struct TargetsSnapshot { + uint64_t update_steady_us = 0; + int img_w = 0; + int img_h = 0; + std::string model_name; + std::vector objects; +}; + +RK3588_SHARED_STATE_API uint64_t NowSteadyUs(); + +class RK3588_SHARED_STATE_API SharedState { +public: + static SharedState& Instance(); + + void SetTargets(std::string key, TargetsSnapshot snap); + std::shared_ptr GetTargets(const std::string& key) const; + void ClearTargets(const std::string& key); + +private: + SharedState() = default; + SharedState(const SharedState&) = delete; + SharedState& operator=(const SharedState&) = delete; + + mutable std::mutex mu_; + std::unordered_map> targets_; +}; + +} // namespace rk3588 diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index a9a80cb..a2e7c77 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -282,12 +282,28 @@ set_target_properties(det_post PROPERTIES # tracker plugin (ByteTrack-lite style IOU association) add_library(tracker SHARED tracker/tracker_node.cpp) target_include_directories(tracker PRIVATE ${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/third_party) -target_link_libraries(tracker PRIVATE project_options Threads::Threads) +target_link_libraries(tracker PRIVATE project_options Threads::Threads rk_shared_state) set_target_properties(tracker PROPERTIES OUTPUT_NAME "tracker" LIBRARY_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} RUNTIME_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} ) +if(UNIX AND NOT APPLE) + set_target_properties(tracker PROPERTIES BUILD_RPATH "\$ORIGIN/.." INSTALL_RPATH "\$ORIGIN/..") +endif() + +# gate plugin (conditional pass/drop based on shared state) +add_library(gate SHARED gate/gate_node.cpp) +target_include_directories(gate PRIVATE ${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/third_party) +target_link_libraries(gate PRIVATE project_options Threads::Threads rk_shared_state) +set_target_properties(gate PROPERTIES + OUTPUT_NAME "gate" + LIBRARY_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} + RUNTIME_OUTPUT_DIRECTORY ${RK_PLUGIN_OUTPUT_DIR} +) +if(UNIX AND NOT APPLE) + set_target_properties(gate PROPERTIES BUILD_RPATH "\$ORIGIN/.." INSTALL_RPATH "\$ORIGIN/..") +endif() # osd plugin (on-screen display for detection results) add_library(osd SHARED @@ -406,7 +422,7 @@ if(RK3588_ENABLE_ZLMEDIAKIT AND RK_ZLMK_API_LIB) ) endif() -install(TARGETS input_rtsp input_file publish preprocess ai_yolo ai_face_det ai_face_recog tracker osd alarm storage ai_scheduler +install(TARGETS input_rtsp input_file publish preprocess ai_yolo ai_face_det ai_face_recog tracker gate osd alarm storage ai_scheduler LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}/rk3588-media-server/plugins RUNTIME DESTINATION ${CMAKE_INSTALL_LIBDIR}/rk3588-media-server/plugins ) diff --git a/plugins/gate/gate_node.cpp b/plugins/gate/gate_node.cpp new file mode 100644 index 0000000..5082be7 --- /dev/null +++ b/plugins/gate/gate_node.cpp @@ -0,0 +1,208 @@ +#include +#include +#include +#include +#include +#include +#include + +#include "node.h" +#include "utils/logger.h" +#include "utils/shared_state.h" + +namespace rk3588 { + +namespace { + +struct ConfigSnapshot { + std::string id; + + std::string state_key; + + std::set class_ids; // empty => any + bool require_confirmed = true; + int min_count = 1; + + int64_t max_age_ms = 800; // state freshness + int64_t min_interval_ms = 300; // throttle output + float min_box_area_ratio = 0.0f; // optional filter + + bool pass_through_if_no_state = false; +}; + +static bool ParseIntSet(const SimpleJson& arr, std::set& out, std::string& err) { + if (!arr.IsArray()) { + err = "expected array"; + return false; + } + out.clear(); + for (const auto& it : arr.AsArray()) { + const int v = it.AsInt(-1); + if (v < 0) continue; + out.insert(v); + } + return true; +} + +static bool BuildConfigSnapshot(const SimpleJson& config, std::shared_ptr& out, + std::string& err) { + auto snap = std::make_shared(); + snap->id = config.ValueOr("id", "gate"); + snap->state_key = config.ValueOr("state_key", ""); + snap->require_confirmed = config.ValueOr("require_confirmed", true); + snap->min_count = config.ValueOr("min_count", 1); + snap->max_age_ms = static_cast(config.ValueOr("max_age_ms", 800)); + snap->min_interval_ms = static_cast(config.ValueOr("min_interval_ms", 300)); + snap->min_box_area_ratio = config.ValueOr("min_box_area_ratio", 0.0f); + snap->pass_through_if_no_state = config.ValueOr("pass_through_if_no_state", false); + + if (const SimpleJson* ids = config.Find("class_ids")) { + if (!ParseIntSet(*ids, snap->class_ids, err)) { + err = "class_ids: " + err; + return false; + } + } + + if (snap->min_count < 1) snap->min_count = 1; + if (snap->max_age_ms < 0) snap->max_age_ms = 0; + if (snap->min_interval_ms < 0) snap->min_interval_ms = 0; + if (snap->min_box_area_ratio < 0.0f) snap->min_box_area_ratio = 0.0f; + if (snap->min_box_area_ratio > 1.0f) snap->min_box_area_ratio = 1.0f; + + out = std::move(snap); + return true; +} + +} // namespace + +class GateNode final : public INode { +public: + std::string Id() const override { return id_; } + std::string Type() const override { return "gate"; } + + bool Init(const SimpleJson& config, const NodeContext& ctx) override { + std::string err; + std::shared_ptr snap; + if (!BuildConfigSnapshot(config, snap, err)) { + LogError("[GateNode] invalid config: " + err); + return false; + } + id_ = snap->id; + ctx_ = ctx; + { + std::lock_guard lk(mu_); + cfg_ = std::move(snap); + } + return true; + } + + bool Start() override { return true; } + void Stop() override {} + + bool UpdateConfig(const SimpleJson& new_config) override { + std::string err; + std::shared_ptr snap; + if (!BuildConfigSnapshot(new_config, snap, err)) { + LogWarn("[GateNode] UpdateConfig ignored: " + err); + return false; + } + if (snap->id != id_) { + LogWarn("[GateNode] UpdateConfig ignored: id mismatch"); + return false; + } + std::lock_guard lk(mu_); + cfg_ = std::move(snap); + return true; + } + + NodeStatus Process(FramePtr frame) override { + if (!frame) return NodeStatus::DROP; + + std::shared_ptr cfg; + { + std::lock_guard lk(mu_); + cfg = cfg_; + } + if (!cfg) return NodeStatus::DROP; + + if (cfg->state_key.empty()) { + // Without a state source, this node cannot decide; choose conservative behavior. + if (cfg->pass_through_if_no_state) { + PushToDownstream(std::move(frame)); + return NodeStatus::OK; + } + return NodeStatus::DROP; + } + + const uint64_t now_us = NowSteadyUs(); + auto snap = SharedState::Instance().GetTargets(cfg->state_key); + if (!snap) { + if (cfg->pass_through_if_no_state) { + PushToDownstream(std::move(frame)); + return NodeStatus::OK; + } + return NodeStatus::DROP; + } + + if (cfg->max_age_ms > 0) { + const uint64_t age_us = (now_us > snap->update_steady_us) ? (now_us - snap->update_steady_us) : 0; + if (age_us > static_cast(cfg->max_age_ms) * 1000ULL) { + return NodeStatus::DROP; + } + } + + const double img_area = (snap->img_w > 0 && snap->img_h > 0) ? + (static_cast(snap->img_w) * static_cast(snap->img_h)) : + 0.0; + + int match_count = 0; + for (const auto& obj : snap->objects) { + if (!cfg->class_ids.empty() && cfg->class_ids.count(obj.cls_id) == 0) continue; + if (cfg->require_confirmed && !obj.confirmed) continue; + + if (cfg->min_box_area_ratio > 0.0f && img_area > 0.0) { + const double a = static_cast(obj.bbox.w) * static_cast(obj.bbox.h); + const double r = a / img_area; + if (r < static_cast(cfg->min_box_area_ratio)) continue; + } + + ++match_count; + if (match_count >= cfg->min_count) break; + } + + if (match_count < cfg->min_count) { + return NodeStatus::DROP; + } + + if (cfg->min_interval_ms > 0) { + const uint64_t last = last_emit_us_.load(std::memory_order_relaxed); + if (last != 0 && now_us < last + static_cast(cfg->min_interval_ms) * 1000ULL) { + return NodeStatus::DROP; + } + last_emit_us_.store(now_us, std::memory_order_relaxed); + } + + PushToDownstream(std::move(frame)); + return NodeStatus::OK; + } + +private: + void PushToDownstream(FramePtr frame) { + for (auto& q : ctx_.output_queues) { + if (!q) continue; + q->push(frame); + } + } + + std::string id_; + NodeContext ctx_{}; + + mutable std::mutex mu_; + std::shared_ptr cfg_; + + std::atomic last_emit_us_{0}; +}; + +REGISTER_NODE(GateNode, "gate"); + +} // namespace rk3588 diff --git a/plugins/tracker/tracker_node.cpp b/plugins/tracker/tracker_node.cpp index 54dce22..0b0b0dc 100644 --- a/plugins/tracker/tracker_node.cpp +++ b/plugins/tracker/tracker_node.cpp @@ -12,6 +12,7 @@ #include "node.h" #include "utils/logger.h" +#include "utils/shared_state.h" namespace rk3588 { namespace { @@ -62,6 +63,9 @@ struct ConfigSnapshot { std::string mode; // off | bytetrack_lite bool per_class = true; + // If set, write current tracked targets into SharedState under this key. + std::string state_key; + std::set track_classes; // whitelist std::set ignore_classes; // blacklist std::unordered_set allowed_models; @@ -111,6 +115,7 @@ static bool BuildConfigSnapshot(const SimpleJson& config, std::shared_ptrid = config.ValueOr("id", "tracker"); snap->mode = config.ValueOr("mode", "bytetrack_lite"); snap->per_class = config.ValueOr("per_class", true); + snap->state_key = config.ValueOr("state_key", ""); if (const SimpleJson* tc = config.Find("track_classes")) { if (!ParseIntSet(*tc, snap->track_classes, err)) { @@ -369,6 +374,7 @@ public: if (!frame->det || frame->det->items.empty()) { MaybeLogStats(now_us, *cfg); + MaybeUpdateSharedState(*cfg, frame->det.get()); PushToDownstream(frame); total_process_us_.fetch_add(NowUsSteady() - t0); processed_frames_.fetch_add(1); @@ -379,6 +385,7 @@ public: if (cfg->allowed_models.find(frame->det->model_name) == cfg->allowed_models.end()) { // Not an enabled model: pass-through (do not rewrite track_id). MaybeLogStats(now_us, *cfg); + MaybeUpdateSharedState(*cfg, frame->det.get()); PushToDownstream(frame); total_process_us_.fetch_add(NowUsSteady() - t0); processed_frames_.fetch_add(1); @@ -410,6 +417,7 @@ public: if (high_dets.empty() && low_dets.empty()) { MaybeLogStats(now_us, *cfg); + MaybeUpdateSharedState(*cfg, frame->det.get()); PushToDownstream(frame); total_process_us_.fetch_add(NowUsSteady() - t0); processed_frames_.fetch_add(1); @@ -546,6 +554,7 @@ public: if (unmatched_dets_local) unmatched_dets_total_.fetch_add(unmatched_dets_local); MaybeLogStats(now_us, *cfg); + MaybeUpdateSharedState(*cfg, frame->det.get()); PushToDownstream(frame); total_process_us_.fetch_add(NowUsSteady() - t0); @@ -554,6 +563,34 @@ public: } private: + void MaybeUpdateSharedState(const ConfigSnapshot& cfg, const DetectionResult* det_hint) { + if (cfg.state_key.empty()) return; + + TargetsSnapshot snap; + snap.update_steady_us = NowSteadyUs(); + if (det_hint) { + snap.img_w = det_hint->img_w; + snap.img_h = det_hint->img_h; + snap.model_name = det_hint->model_name; + } + + { + std::lock_guard lk(mu_); + snap.objects.reserve(tracks_.size()); + for (const auto& t : tracks_) { + TrackedObject obj; + obj.cls_id = t.cls_id; + obj.track_id = t.id; + obj.score = 0.0f; + obj.bbox = t.bbox; + obj.confirmed = t.confirmed; + snap.objects.push_back(std::move(obj)); + } + } + + SharedState::Instance().SetTargets(cfg.state_key, std::move(snap)); + } + uint64_t ResolveNowUsLocked(const Frame& frame) { uint64_t t = frame.pts; if (t == 0) { diff --git a/src/utils/shared_state.cpp b/src/utils/shared_state.cpp new file mode 100644 index 0000000..889845d --- /dev/null +++ b/src/utils/shared_state.cpp @@ -0,0 +1,41 @@ +#include "utils/shared_state.h" + +#include + +namespace rk3588 { + +uint64_t NowSteadyUs() { + using clock = std::chrono::steady_clock; + return static_cast( + std::chrono::duration_cast(clock::now().time_since_epoch()) + .count()); +} + +SharedState& SharedState::Instance() { + static SharedState inst; + return inst; +} + +void SharedState::SetTargets(std::string key, TargetsSnapshot snap) { + if (snap.update_steady_us == 0) { + snap.update_steady_us = NowSteadyUs(); + } + + auto ptr = std::make_shared(std::move(snap)); + std::lock_guard lk(mu_); + targets_[std::move(key)] = std::move(ptr); +} + +std::shared_ptr SharedState::GetTargets(const std::string& key) const { + std::lock_guard lk(mu_); + auto it = targets_.find(key); + if (it == targets_.end()) return {}; + return it->second; +} + +void SharedState::ClearTargets(const std::string& key) { + std::lock_guard lk(mu_); + targets_.erase(key); +} + +} // namespace rk3588