1162 lines
48 KiB
C++
1162 lines
48 KiB
C++
#include <atomic>
|
|
#include <chrono>
|
|
#include <cctype>
|
|
#include <cstdint>
|
|
#include <cstring>
|
|
#include <algorithm>
|
|
#include <deque>
|
|
#include <iomanip>
|
|
#include <map>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <sstream>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <unordered_map>
|
|
#include <vector>
|
|
|
|
#include "utils/logger.h"
|
|
|
|
#include "face/face_result.h"
|
|
#include "node.h"
|
|
#include "rule_engine.h"
|
|
#include "packet_ring_buffer.h"
|
|
#include "actions/action_base.h"
|
|
#include "actions/log_action.h"
|
|
#include "actions/http_action.h"
|
|
#include "actions/external_api_action.h"
|
|
#include "actions/snapshot_action.h"
|
|
#include "actions/clip_action.h"
|
|
|
|
namespace rk3588 {
|
|
|
|
namespace {
|
|
|
|
struct FaceTrackAggregationConfig {
|
|
int known_min_hits = 3;
|
|
int known_hit_window_ms = 3000;
|
|
int known_reentry_cooldown_ms = 0;
|
|
int unknown_min_track_age_ms = 2000;
|
|
int unknown_min_quality_hits = 4;
|
|
int state_expire_ms = 5000;
|
|
};
|
|
|
|
struct FaceTrackState {
|
|
int track_id = -1;
|
|
uint64_t first_seen_ms = 0;
|
|
uint64_t last_seen_ms = 0;
|
|
int best_known_person_id = -1;
|
|
std::string best_known_name;
|
|
int quality_hits = 0;
|
|
bool reported_known = false;
|
|
bool reported_unknown = false;
|
|
uint64_t known_reentry_suppressed_until_ms = 0;
|
|
std::deque<uint64_t> known_hit_times;
|
|
};
|
|
|
|
struct FaceTrackDecision {
|
|
bool trigger_known = false;
|
|
bool trigger_unknown = false;
|
|
};
|
|
|
|
void PruneKnownHitTimes(
|
|
FaceTrackState& state,
|
|
const FaceTrackAggregationConfig& cfg,
|
|
uint64_t now_ms) {
|
|
const uint64_t known_window_ms = static_cast<uint64_t>(std::max(0, cfg.known_hit_window_ms));
|
|
if (known_window_ms == 0) return;
|
|
while (!state.known_hit_times.empty() &&
|
|
now_ms > state.known_hit_times.front() &&
|
|
(now_ms - state.known_hit_times.front()) > known_window_ms) {
|
|
state.known_hit_times.pop_front();
|
|
}
|
|
}
|
|
|
|
bool IsKnownLeaningTrack(
|
|
const FaceTrackState& state,
|
|
const FaceTrackAggregationConfig& cfg) {
|
|
return !state.reported_known &&
|
|
!state.known_hit_times.empty() &&
|
|
static_cast<int>(state.known_hit_times.size()) < std::max(1, cfg.known_min_hits) &&
|
|
(state.best_known_person_id >= 0 || !state.best_known_name.empty());
|
|
}
|
|
|
|
void ResetFaceTrackState(FaceTrackState& state, int track_id, uint64_t now_ms) {
|
|
state = FaceTrackState{};
|
|
state.track_id = track_id;
|
|
state.first_seen_ms = now_ms;
|
|
state.last_seen_ms = now_ms;
|
|
}
|
|
|
|
FaceTrackAggregationConfig ParseFaceTrackAggregationConfig(const SimpleJson& config) {
|
|
FaceTrackAggregationConfig cfg;
|
|
const SimpleJson* agg = config.Find("face_track_aggregation");
|
|
if (!agg || !agg->IsObject()) return cfg;
|
|
|
|
cfg.known_min_hits = std::max(1, agg->ValueOr<int>("known_min_hits", cfg.known_min_hits));
|
|
cfg.known_hit_window_ms = std::max(0, agg->ValueOr<int>("known_hit_window_ms", cfg.known_hit_window_ms));
|
|
cfg.known_reentry_cooldown_ms = std::max(0, agg->ValueOr<int>("known_reentry_cooldown_ms", cfg.known_reentry_cooldown_ms));
|
|
cfg.unknown_min_track_age_ms = std::max(0, agg->ValueOr<int>("unknown_min_track_age_ms", cfg.unknown_min_track_age_ms));
|
|
cfg.unknown_min_quality_hits = std::max(1, agg->ValueOr<int>("unknown_min_quality_hits", cfg.unknown_min_quality_hits));
|
|
cfg.state_expire_ms = std::max(0, agg->ValueOr<int>("state_expire_ms", cfg.state_expire_ms));
|
|
|
|
if (const SimpleJson* known = agg->Find("known"); known && known->IsObject()) {
|
|
cfg.known_min_hits = std::max(1, known->ValueOr<int>("min_hits", cfg.known_min_hits));
|
|
cfg.known_hit_window_ms = std::max(0, known->ValueOr<int>("hit_window_ms", cfg.known_hit_window_ms));
|
|
cfg.known_reentry_cooldown_ms = std::max(0, known->ValueOr<int>("reentry_cooldown_ms", cfg.known_reentry_cooldown_ms));
|
|
}
|
|
if (const SimpleJson* unknown = agg->Find("unknown"); unknown && unknown->IsObject()) {
|
|
cfg.unknown_min_track_age_ms = std::max(0, unknown->ValueOr<int>("min_track_age_ms", cfg.unknown_min_track_age_ms));
|
|
cfg.unknown_min_quality_hits = std::max(1, unknown->ValueOr<int>("min_quality_hits", cfg.unknown_min_quality_hits));
|
|
}
|
|
|
|
return cfg;
|
|
}
|
|
|
|
FaceTrackDecision UpdateFaceTrackState(
|
|
const FaceTrackAggregationConfig& cfg,
|
|
FaceTrackState& state,
|
|
std::unordered_map<int, uint64_t>& known_identity_last_alarm_ms,
|
|
const FaceRecogItem& item,
|
|
uint64_t now_ms) {
|
|
FaceTrackDecision decision;
|
|
if (item.person_track_id < 0) return decision;
|
|
|
|
const uint64_t expire_ms = static_cast<uint64_t>(std::max(0, cfg.state_expire_ms));
|
|
const bool expired =
|
|
expire_ms > 0 &&
|
|
state.last_seen_ms > 0 &&
|
|
now_ms > state.last_seen_ms &&
|
|
(now_ms - state.last_seen_ms) > expire_ms;
|
|
if (state.track_id != item.person_track_id || expired) {
|
|
ResetFaceTrackState(state, item.person_track_id, now_ms);
|
|
} else if (state.first_seen_ms == 0) {
|
|
state.first_seen_ms = now_ms;
|
|
state.last_seen_ms = now_ms;
|
|
} else {
|
|
state.last_seen_ms = now_ms;
|
|
}
|
|
|
|
PruneKnownHitTimes(state, cfg, now_ms);
|
|
if (FaceRecogStateIsKnown(item.state)) {
|
|
const bool same_identity =
|
|
state.best_known_person_id == item.best_person_id &&
|
|
state.best_known_name == item.best_name;
|
|
if (!same_identity) {
|
|
ResetFaceTrackState(state, item.person_track_id, now_ms);
|
|
state.best_known_person_id = item.best_person_id;
|
|
state.best_known_name = item.best_name;
|
|
}
|
|
|
|
state.known_hit_times.push_back(now_ms);
|
|
state.quality_hits = 0;
|
|
|
|
if (!state.reported_known &&
|
|
static_cast<int>(state.known_hit_times.size()) >= std::max(1, cfg.known_min_hits)) {
|
|
if (cfg.known_reentry_cooldown_ms > 0 && state.best_known_person_id >= 0) {
|
|
const auto it_last = known_identity_last_alarm_ms.find(state.best_known_person_id);
|
|
if (it_last != known_identity_last_alarm_ms.end() &&
|
|
now_ms >= it_last->second &&
|
|
(now_ms - it_last->second) < static_cast<uint64_t>(cfg.known_reentry_cooldown_ms)) {
|
|
state.known_reentry_suppressed_until_ms =
|
|
std::max(state.known_reentry_suppressed_until_ms,
|
|
it_last->second + static_cast<uint64_t>(cfg.known_reentry_cooldown_ms));
|
|
return decision;
|
|
}
|
|
}
|
|
state.reported_known = true;
|
|
state.reported_unknown = false;
|
|
if (state.best_known_person_id >= 0) {
|
|
known_identity_last_alarm_ms[state.best_known_person_id] = now_ms;
|
|
}
|
|
decision.trigger_known = true;
|
|
}
|
|
return decision;
|
|
}
|
|
|
|
if (state.reported_known) return decision;
|
|
if (!state.reported_known &&
|
|
state.best_known_person_id >= 0 &&
|
|
state.known_reentry_suppressed_until_ms > 0) {
|
|
return decision;
|
|
}
|
|
if (state.known_reentry_suppressed_until_ms > now_ms) return decision;
|
|
if (IsKnownLeaningTrack(state, cfg)) return decision;
|
|
|
|
++state.quality_hits;
|
|
if (!state.reported_unknown &&
|
|
now_ms >= state.first_seen_ms &&
|
|
static_cast<int>(now_ms - state.first_seen_ms) >= std::max(0, cfg.unknown_min_track_age_ms) &&
|
|
state.quality_hits >= std::max(1, cfg.unknown_min_quality_hits)) {
|
|
state.reported_unknown = true;
|
|
decision.trigger_unknown = true;
|
|
}
|
|
return decision;
|
|
}
|
|
|
|
uint64_t NowEpochMs() {
|
|
using namespace std::chrono;
|
|
return static_cast<uint64_t>(duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count());
|
|
}
|
|
|
|
std::string SanitizeEventIdPart(const std::string& value) {
|
|
std::string out;
|
|
out.reserve(value.size());
|
|
for (unsigned char ch : value) {
|
|
out.push_back(std::isalnum(ch) ? static_cast<char>(ch) : '_');
|
|
}
|
|
return out.empty() ? "event" : out;
|
|
}
|
|
|
|
class RateLimitedAction final : public IAlarmAction {
|
|
public:
|
|
RateLimitedAction(std::unique_ptr<IAlarmAction> inner, uint64_t min_interval_ms)
|
|
: inner_(std::move(inner)), min_interval_ms_(min_interval_ms) {}
|
|
|
|
bool Init(const SimpleJson& config) override {
|
|
// Inner is expected to be initialized already; keep compatibility if called.
|
|
return inner_ ? inner_->Init(config) : false;
|
|
}
|
|
|
|
void Execute(AlarmEvent& event, std::shared_ptr<Frame> frame) override {
|
|
if (!inner_) return;
|
|
if (min_interval_ms_ == 0) {
|
|
inner_->Execute(event, std::move(frame));
|
|
return;
|
|
}
|
|
|
|
const uint64_t now = NowEpochMs();
|
|
uint64_t last = last_exec_ms_.load(std::memory_order_relaxed);
|
|
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.
|
|
}
|
|
inner_->Execute(event, std::move(frame));
|
|
}
|
|
|
|
void Drain() override {
|
|
if (inner_) inner_->Drain();
|
|
}
|
|
|
|
void Stop() override {
|
|
if (inner_) inner_->Stop();
|
|
}
|
|
|
|
std::string Name() const override {
|
|
return inner_ ? inner_->Name() : std::string("unknown");
|
|
}
|
|
|
|
private:
|
|
std::unique_ptr<IAlarmAction> inner_;
|
|
uint64_t min_interval_ms_ = 0;
|
|
std::atomic<uint64_t> last_exec_ms_{0};
|
|
};
|
|
|
|
std::unique_ptr<IAlarmAction> WrapIfRateLimited(std::unique_ptr<IAlarmAction> action, const SimpleJson& cfg) {
|
|
if (!action) return nullptr;
|
|
const uint64_t min_interval_ms = static_cast<uint64_t>(std::max<int64_t>(
|
|
0, static_cast<int64_t>(cfg.ValueOr<int>("min_interval_ms", 0))));
|
|
if (min_interval_ms == 0) return action;
|
|
return std::make_unique<RateLimitedAction>(std::move(action), min_interval_ms);
|
|
}
|
|
|
|
} // namespace
|
|
|
|
class AlarmNode : public INode {
|
|
private:
|
|
struct AlarmJob {
|
|
AlarmEvent event;
|
|
std::shared_ptr<Frame> frame;
|
|
};
|
|
|
|
static std::shared_ptr<EncodedVideoFrameMeta> TryGetEncodedMeta(const std::shared_ptr<Frame>& frame) {
|
|
if (!frame || !frame->user_meta) return nullptr;
|
|
auto meta = std::static_pointer_cast<EncodedVideoFrameMeta>(frame->user_meta);
|
|
if (!meta || meta->magic != EncodedVideoFrameMeta::kMagic) return nullptr;
|
|
if (!meta->codec || meta->pkt.data.empty() || meta->pkt.pts_ms <= 0) return nullptr;
|
|
return meta;
|
|
}
|
|
|
|
public:
|
|
std::string Id() const override { return id_; }
|
|
std::string Type() const override { return "alarm"; }
|
|
|
|
bool Init(const SimpleJson& config, const NodeContext& ctx) override {
|
|
id_ = config.ValueOr<std::string>("id", "alarm");
|
|
|
|
ParseFaceRules(config);
|
|
track_agg_cfg_ = ParseFaceTrackAggregationConfig(config);
|
|
const FaceDebugConfig face_debug = ParseFaceDebugConfig(config);
|
|
face_debug_log_unknown_candidates_ = face_debug.log_unknown_candidates;
|
|
face_debug_unknown_candidate_interval_ms_ = face_debug.unknown_candidate_interval_ms;
|
|
last_unknown_candidate_log_ms_ = 0;
|
|
face_track_states_.clear();
|
|
known_identity_last_alarm_ms_.clear();
|
|
|
|
// Optional evaluation throttle (for alarm side). 0 = disabled.
|
|
eval_interval_ms_ = std::max<int64_t>(0, static_cast<int64_t>(config.ValueOr<int>("eval_interval_ms", 0)));
|
|
if (eval_interval_ms_ <= 0) {
|
|
const double eval_fps = config.ValueOr<double>("eval_fps", 0.0);
|
|
if (eval_fps > 0.0) {
|
|
eval_interval_ms_ = static_cast<int64_t>(1000.0 / eval_fps);
|
|
if (eval_interval_ms_ < 1) eval_interval_ms_ = 1;
|
|
}
|
|
}
|
|
|
|
// Parse labels for class name mapping
|
|
if (const SimpleJson* labels_cfg = config.Find("labels")) {
|
|
for (const auto& label : labels_cfg->AsArray()) {
|
|
labels_.push_back(label.AsString(""));
|
|
}
|
|
}
|
|
|
|
// Initialize rule engine
|
|
if (const SimpleJson* rules_cfg = config.Find("rules")) {
|
|
if (!rule_engine_.Init(*rules_cfg, labels_)) {
|
|
LogError("[alarm] failed to init rule engine");
|
|
return false;
|
|
}
|
|
}
|
|
|
|
// Derive packet ring window 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);
|
|
}
|
|
}
|
|
const int window_sec = std::max(1, std::max(0, pre_sec) + std::max(0, post_sec) + 2);
|
|
packet_buffer_ = std::make_shared<PacketRingBuffer>(window_sec, fps_hint);
|
|
|
|
// Alarm worker queue (avoid blocking Process())
|
|
size_t alarm_q_size = 32;
|
|
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")) {
|
|
// Log action
|
|
if (const SimpleJson* log_cfg = actions_cfg->Find("log")) {
|
|
if (log_cfg->ValueOr<bool>("enable", true)) {
|
|
auto action = std::make_unique<LogAction>();
|
|
if (action->Init(*log_cfg)) {
|
|
actions_.push_back(WrapIfRateLimited(std::move(action), *log_cfg));
|
|
}
|
|
}
|
|
}
|
|
|
|
// Snapshot action (must be before HTTP to fill snapshot_url)
|
|
if (const SimpleJson* snap_cfg = actions_cfg->Find("snapshot")) {
|
|
if (snap_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<SnapshotAction>();
|
|
if (action->Init(*snap_cfg)) {
|
|
actions_.push_back(WrapIfRateLimited(std::move(action), *snap_cfg));
|
|
}
|
|
}
|
|
}
|
|
|
|
// Clip action (must be before HTTP to fill clip_url)
|
|
if (const SimpleJson* clip_cfg = actions_cfg->Find("clip")) {
|
|
if (clip_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<ClipAction>();
|
|
if (action->Init(*clip_cfg)) {
|
|
auto* clip_ptr = static_cast<ClipAction*>(action.get());
|
|
clip_ptr->SetPacketBuffer(packet_buffer_);
|
|
actions_.push_back(WrapIfRateLimited(std::move(action), *clip_cfg));
|
|
}
|
|
}
|
|
}
|
|
|
|
// HTTP action (should be after snapshot/clip to include media URLs)
|
|
if (const SimpleJson* http_cfg = actions_cfg->Find("http")) {
|
|
if (http_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<HttpAction>();
|
|
if (action->Init(*http_cfg)) {
|
|
actions_.push_back(WrapIfRateLimited(std::move(action), *http_cfg));
|
|
}
|
|
}
|
|
}
|
|
|
|
// External API action (token + message). Should be after snapshot/clip to include media URLs.
|
|
if (const SimpleJson* ext_cfg = actions_cfg->Find("external_api")) {
|
|
if (ext_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<ExternalApiAction>();
|
|
if (action->Init(*ext_cfg)) {
|
|
actions_.push_back(WrapIfRateLimited(std::move(action), *ext_cfg));
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
// Default: just log action
|
|
auto action = std::make_unique<LogAction>();
|
|
SimpleJson empty_cfg;
|
|
action->Init(empty_cfg);
|
|
actions_.push_back(std::move(action));
|
|
}
|
|
|
|
input_queue_ = ctx.input_queue;
|
|
if (!input_queue_) {
|
|
LogError("[alarm] no input queue for node " + id_);
|
|
return false;
|
|
}
|
|
|
|
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);
|
|
LogInfo("[alarm] started");
|
|
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();
|
|
|
|
LogInfo("[alarm] stopped, processed " + std::to_string(processed_frames_) +
|
|
" frames, triggered " + std::to_string(alarm_count_) + " alarms");
|
|
}
|
|
|
|
void Drain() override {
|
|
// 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 {
|
|
const std::string new_id = new_config.ValueOr<std::string>("id", id_);
|
|
if (!new_id.empty() && new_id != id_) {
|
|
return false;
|
|
}
|
|
|
|
// Parse labels
|
|
std::vector<std::string> new_labels;
|
|
if (const SimpleJson* labels_cfg = new_config.Find("labels")) {
|
|
for (const auto& label : labels_cfg->AsArray()) {
|
|
new_labels.push_back(label.AsString(""));
|
|
}
|
|
}
|
|
|
|
// Build new rule engine
|
|
RuleEngine new_engine;
|
|
if (const SimpleJson* rules_cfg = new_config.Find("rules")) {
|
|
if (!new_engine.Init(*rules_cfg, new_labels)) {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
// Parse face rules (independent from detection rules)
|
|
std::vector<FaceRule> new_face_rules = ParseFaceRulesFromConfig(new_config);
|
|
FaceTrackAggregationConfig new_track_agg_cfg = ParseFaceTrackAggregationConfig(new_config);
|
|
const FaceDebugConfig new_face_debug = ParseFaceDebugConfig(new_config);
|
|
|
|
// Packet ring window 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);
|
|
}
|
|
}
|
|
const int window_sec = std::max(1, std::max(0, pre_sec) + std::max(0, post_sec) + 2);
|
|
auto new_packet_buffer = std::make_shared<PacketRingBuffer>(window_sec, fps_hint);
|
|
|
|
// Initialize new actions
|
|
std::vector<std::unique_ptr<IAlarmAction>> new_actions;
|
|
if (const SimpleJson* actions_cfg = new_config.Find("actions")) {
|
|
if (const SimpleJson* log_cfg = actions_cfg->Find("log")) {
|
|
if (log_cfg->ValueOr<bool>("enable", true)) {
|
|
auto action = std::make_unique<LogAction>();
|
|
if (action->Init(*log_cfg)) {
|
|
new_actions.push_back(std::move(action));
|
|
}
|
|
}
|
|
}
|
|
if (const SimpleJson* snap_cfg = actions_cfg->Find("snapshot")) {
|
|
if (snap_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<SnapshotAction>();
|
|
if (action->Init(*snap_cfg)) {
|
|
new_actions.push_back(std::move(action));
|
|
}
|
|
}
|
|
}
|
|
if (const SimpleJson* clip_cfg = actions_cfg->Find("clip")) {
|
|
if (clip_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<ClipAction>();
|
|
if (action->Init(*clip_cfg)) {
|
|
auto* clip_ptr = static_cast<ClipAction*>(action.get());
|
|
clip_ptr->SetPacketBuffer(new_packet_buffer);
|
|
new_actions.push_back(std::move(action));
|
|
}
|
|
}
|
|
}
|
|
if (const SimpleJson* http_cfg = actions_cfg->Find("http")) {
|
|
if (http_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<HttpAction>();
|
|
if (action->Init(*http_cfg)) {
|
|
new_actions.push_back(std::move(action));
|
|
}
|
|
}
|
|
}
|
|
|
|
if (const SimpleJson* ext_cfg = actions_cfg->Find("external_api")) {
|
|
if (ext_cfg->ValueOr<bool>("enable", false)) {
|
|
auto action = std::make_unique<ExternalApiAction>();
|
|
if (action->Init(*ext_cfg)) {
|
|
new_actions.push_back(std::move(action));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (new_actions.empty()) {
|
|
auto action = std::make_unique<LogAction>();
|
|
SimpleJson empty_cfg;
|
|
if (action->Init(empty_cfg)) {
|
|
new_actions.push_back(std::move(action));
|
|
}
|
|
}
|
|
|
|
// 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_);
|
|
labels_ = std::move(new_labels);
|
|
rule_engine_ = std::move(new_engine);
|
|
face_rules_ = std::move(new_face_rules);
|
|
track_agg_cfg_ = new_track_agg_cfg;
|
|
face_debug_log_unknown_candidates_ = new_face_debug.log_unknown_candidates;
|
|
face_debug_unknown_candidate_interval_ms_ = new_face_debug.unknown_candidate_interval_ms;
|
|
last_unknown_candidate_log_ms_ = 0;
|
|
face_last_trigger_.clear();
|
|
face_person_last_trigger_.clear();
|
|
face_track_states_.clear();
|
|
known_identity_last_alarm_ms_.clear();
|
|
packet_buffer_ = std::move(new_packet_buffer);
|
|
old_actions = std::move(actions_);
|
|
actions_ = std::move(new_actions);
|
|
|
|
in_flight_.store(0);
|
|
last_eval_pts_ms_ = 0;
|
|
alarm_queue_ = std::make_shared<SpscQueue<AlarmJob>>(alarm_queue_size_, alarm_queue_strategy_);
|
|
worker_running_.store(true);
|
|
worker_ = std::thread(&AlarmNode::WorkerLoop, this);
|
|
}
|
|
|
|
for (auto& action : old_actions) action->Drain();
|
|
for (auto& action : old_actions) action->Stop();
|
|
return true;
|
|
}
|
|
|
|
bool GetCustomMetrics(SimpleJson& out) const override {
|
|
std::lock_guard<std::mutex> lock(mu_);
|
|
SimpleJson::Object o;
|
|
o["alarm_total"] = SimpleJson(static_cast<double>(alarm_count_));
|
|
o["processed"] = SimpleJson(static_cast<double>(processed_frames_));
|
|
out = SimpleJson(std::move(o));
|
|
return true;
|
|
}
|
|
|
|
NodeStatus Process(FramePtr frame) override {
|
|
if (!frame) return NodeStatus::DROP;
|
|
|
|
if (packet_buffer_) {
|
|
if (auto meta = TryGetEncodedMeta(frame)) {
|
|
packet_buffer_->Push(std::move(meta));
|
|
}
|
|
}
|
|
|
|
{
|
|
std::lock_guard<std::mutex> lock(mu_);
|
|
|
|
// 调试：检查 frame 和 det
|
|
if (frame->det) {
|
|
int no_boots_count = 0;
|
|
for (const auto& d : frame->det->items) {
|
|
if (d.cls_id == 10) no_boots_count++;
|
|
}
|
|
// Log throttled
|
|
}
|
|
|
|
if (eval_interval_ms_ > 0 && frame->pts > 0) {
|
|
const int64_t pts_ms = static_cast<int64_t>(frame->pts / 1000ULL);
|
|
if (last_eval_pts_ms_ > 0 && (pts_ms - last_eval_pts_ms_) < eval_interval_ms_) {
|
|
++processed_frames_;
|
|
return NodeStatus::OK;
|
|
}
|
|
last_eval_pts_ms_ = pts_ms;
|
|
}
|
|
|
|
auto result = rule_engine_.Evaluate(frame);
|
|
if (result.matched) {
|
|
LogInfo("[alarm] RULE MATCHED: " + result.rule_name);
|
|
TriggerAlarm(result, frame);
|
|
}
|
|
|
|
EvaluateFaceRulesLocked(frame);
|
|
++processed_frames_;
|
|
}
|
|
return NodeStatus::OK;
|
|
}
|
|
|
|
private:
|
|
struct FaceRule {
|
|
enum class Kind { Unknown, Person };
|
|
std::string name;
|
|
Kind kind = Kind::Unknown;
|
|
std::vector<std::string> persons; // used when kind==Person; empty => any known person
|
|
int cooldown_ms = 5000;
|
|
int per_person_cooldown_ms = 0;
|
|
float min_face_area_ratio = 0.0f;
|
|
float min_face_aspect = 0.0f;
|
|
float max_face_aspect = 0.0f;
|
|
float min_sim = 0.0f;
|
|
float max_known_sim = 0.0f;
|
|
};
|
|
|
|
struct FaceRuleEval {
|
|
bool matched = false;
|
|
std::string reject_reason;
|
|
std::string detail;
|
|
double face_area_ratio = 0.0;
|
|
double face_aspect = 0.0;
|
|
};
|
|
|
|
struct FaceDebugConfig {
|
|
bool log_unknown_candidates = false;
|
|
uint64_t unknown_candidate_interval_ms = 0;
|
|
};
|
|
|
|
static std::string Fixed3(double value) {
|
|
std::ostringstream oss;
|
|
oss << std::fixed << std::setprecision(3) << value;
|
|
return oss.str();
|
|
}
|
|
|
|
static std::string Fixed6(double value) {
|
|
std::ostringstream oss;
|
|
oss << std::fixed << std::setprecision(6) << value;
|
|
return oss.str();
|
|
}
|
|
|
|
static FaceDebugConfig ParseFaceDebugConfig(const SimpleJson& config) {
|
|
FaceDebugConfig out;
|
|
const SimpleJson* debug = config.Find("face_debug");
|
|
if (!debug || !debug->IsObject()) return out;
|
|
|
|
out.log_unknown_candidates = debug->ValueOr<bool>("log_unknown_candidates", out.log_unknown_candidates);
|
|
out.unknown_candidate_interval_ms = static_cast<uint64_t>(std::max<int64_t>(
|
|
0, static_cast<int64_t>(debug->ValueOr<int>("unknown_candidate_interval_ms", 0))));
|
|
return out;
|
|
}
|
|
|
|
static std::vector<FaceRule> ParseFaceRulesFromConfig(const SimpleJson& config) {
|
|
std::vector<FaceRule> rules;
|
|
const SimpleJson* fr = config.Find("face_rules");
|
|
if (!fr || !fr->IsArray()) return rules;
|
|
|
|
for (const auto& item : fr->AsArray()) {
|
|
if (!item.IsObject()) continue;
|
|
FaceRule r;
|
|
r.name = item.ValueOr<std::string>("name", "face");
|
|
std::string type = item.ValueOr<std::string>("type", "unknown");
|
|
for (auto& c : type) c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
|
|
r.kind = (type == "person") ? FaceRule::Kind::Person : FaceRule::Kind::Unknown;
|
|
r.cooldown_ms = std::max(0, item.ValueOr<int>("cooldown_ms", r.cooldown_ms));
|
|
r.per_person_cooldown_ms = std::max(0, item.ValueOr<int>("per_person_cooldown_ms", r.per_person_cooldown_ms));
|
|
r.min_face_area_ratio = static_cast<float>(
|
|
item.ValueOr<double>("min_face_area_ratio", static_cast<double>(r.min_face_area_ratio)));
|
|
r.min_face_aspect = static_cast<float>(
|
|
item.ValueOr<double>("min_face_aspect", static_cast<double>(r.min_face_aspect)));
|
|
r.max_face_aspect = static_cast<float>(
|
|
item.ValueOr<double>("max_face_aspect", static_cast<double>(r.max_face_aspect)));
|
|
r.min_sim = static_cast<float>(item.ValueOr<double>("min_sim", static_cast<double>(r.min_sim)));
|
|
r.max_known_sim = static_cast<float>(
|
|
item.ValueOr<double>("max_known_sim", static_cast<double>(r.max_known_sim)));
|
|
|
|
if (const SimpleJson* persons = item.Find("persons"); persons && persons->IsArray()) {
|
|
for (const auto& p : persons->AsArray()) {
|
|
const std::string s = p.AsString("");
|
|
if (!s.empty()) r.persons.push_back(s);
|
|
}
|
|
}
|
|
rules.push_back(std::move(r));
|
|
}
|
|
return rules;
|
|
}
|
|
|
|
void ParseFaceRules(const SimpleJson& config) {
|
|
face_rules_ = ParseFaceRulesFromConfig(config);
|
|
face_last_trigger_.clear();
|
|
face_person_last_trigger_.clear();
|
|
face_track_states_.clear();
|
|
known_identity_last_alarm_ms_.clear();
|
|
}
|
|
|
|
static bool FaceItemMatchesRule(
|
|
const FaceRule& rule,
|
|
const FaceRecogItem& it,
|
|
double img_area,
|
|
FaceRuleEval* eval = nullptr) {
|
|
FaceRuleEval local;
|
|
FaceRuleEval& out = eval ? *eval : local;
|
|
out = FaceRuleEval{};
|
|
|
|
if (rule.min_face_area_ratio > 0.0f && img_area > 0.0) {
|
|
const double a = static_cast<double>(it.bbox.w) * static_cast<double>(it.bbox.h);
|
|
const double r = a / img_area;
|
|
out.face_area_ratio = r;
|
|
if (r < static_cast<double>(rule.min_face_area_ratio)) {
|
|
out.reject_reason = "min_face_area_ratio";
|
|
out.detail = "area_ratio=" + Fixed6(r) + " min=" + Fixed6(rule.min_face_area_ratio);
|
|
return false;
|
|
}
|
|
}
|
|
if (rule.min_face_aspect > 0.0f || rule.max_face_aspect > 0.0f) {
|
|
const double h = std::max(1e-6, static_cast<double>(it.bbox.h));
|
|
const double aspect = static_cast<double>(it.bbox.w) / h;
|
|
out.face_aspect = aspect;
|
|
if (rule.min_face_aspect > 0.0f && aspect < static_cast<double>(rule.min_face_aspect)) {
|
|
out.reject_reason = "min_face_aspect";
|
|
out.detail = "aspect=" + Fixed3(aspect) + " min=" + Fixed3(rule.min_face_aspect);
|
|
return false;
|
|
}
|
|
if (rule.max_face_aspect > 0.0f && aspect > static_cast<double>(rule.max_face_aspect)) {
|
|
out.reject_reason = "max_face_aspect";
|
|
out.detail = "aspect=" + Fixed3(aspect) + " max=" + Fixed3(rule.max_face_aspect);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
if (rule.kind == FaceRule::Kind::Unknown) {
|
|
if (FaceRecogStateIsKnown(it.state)) {
|
|
out.reject_reason = "known_state";
|
|
out.detail = "status=known";
|
|
return false;
|
|
}
|
|
if (rule.max_known_sim > 0.0f && it.best_sim >= rule.max_known_sim) {
|
|
out.reject_reason = "max_known_sim";
|
|
out.detail = "best_sim=" + Fixed3(it.best_sim) + " max=" + Fixed3(rule.max_known_sim);
|
|
return false;
|
|
}
|
|
} else {
|
|
if (!FaceRecogStateIsKnown(it.state)) {
|
|
out.reject_reason = "not_known";
|
|
out.detail = std::string("status=") + FaceRecogStateName(it.state);
|
|
return false;
|
|
}
|
|
if (it.best_sim < rule.min_sim) {
|
|
out.reject_reason = "min_sim";
|
|
out.detail = "best_sim=" + Fixed3(it.best_sim) + " min=" + Fixed3(rule.min_sim);
|
|
return false;
|
|
}
|
|
if (!rule.persons.empty()) {
|
|
bool ok = false;
|
|
for (const auto& p : rule.persons) {
|
|
if (p == it.best_name) {
|
|
ok = true;
|
|
break;
|
|
}
|
|
}
|
|
if (!ok) {
|
|
out.reject_reason = "person_filter";
|
|
out.detail = "best_name=" + it.best_name;
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (it.person_track_id < 0) {
|
|
out.reject_reason = "missing_track_id";
|
|
out.detail = "person_track_id=-1";
|
|
return false;
|
|
}
|
|
|
|
out.matched = true;
|
|
return true;
|
|
}
|
|
|
|
void EvaluateFaceRulesLocked(const FramePtr& frame) {
|
|
if (face_rules_.empty()) return;
|
|
if (!frame || !frame->face_recog || frame->face_recog->items.empty()) return;
|
|
|
|
const auto now = std::chrono::steady_clock::now();
|
|
const uint64_t now_epoch_ms = NowEpochMs();
|
|
const int img_w = frame->face_recog->img_w > 0 ? frame->face_recog->img_w : frame->width;
|
|
const int img_h = frame->face_recog->img_h > 0 ? frame->face_recog->img_h : frame->height;
|
|
const double img_area = (img_w > 0 && img_h > 0) ? static_cast<double>(img_w) * static_cast<double>(img_h) : 0.0;
|
|
if (track_agg_cfg_.state_expire_ms > 0) {
|
|
const uint64_t expire_ms = static_cast<uint64_t>(track_agg_cfg_.state_expire_ms);
|
|
for (auto it = face_track_states_.begin(); it != face_track_states_.end();) {
|
|
const auto& state = it->second;
|
|
const bool expired =
|
|
state.last_seen_ms > 0 &&
|
|
now_epoch_ms > state.last_seen_ms &&
|
|
(now_epoch_ms - state.last_seen_ms) > expire_ms;
|
|
if (expired) {
|
|
it = face_track_states_.erase(it);
|
|
} else {
|
|
++it;
|
|
}
|
|
}
|
|
}
|
|
|
|
std::vector<bool> qualifying_observations(frame->face_recog->items.size(), false);
|
|
for (const auto& rule : face_rules_) {
|
|
if (rule.per_person_cooldown_ms <= 0) {
|
|
const auto it_last = face_last_trigger_.find(rule.name);
|
|
if (it_last != face_last_trigger_.end() && rule.cooldown_ms > 0) {
|
|
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(now - it_last->second).count();
|
|
if (elapsed < rule.cooldown_ms) {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
for (size_t i = 0; i < frame->face_recog->items.size(); ++i) {
|
|
const auto& it = frame->face_recog->items[i];
|
|
if (!FaceItemMatchesRule(rule, it, img_area)) continue;
|
|
|
|
if (rule.per_person_cooldown_ms > 0) {
|
|
const std::string key = BuildFaceTrackKey(rule, it.person_track_id);
|
|
auto it_last = face_person_last_trigger_.find(key);
|
|
if (it_last != face_person_last_trigger_.end()) {
|
|
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(now - it_last->second).count();
|
|
if (elapsed < rule.per_person_cooldown_ms) {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
qualifying_observations[i] = true;
|
|
}
|
|
}
|
|
|
|
std::vector<FaceTrackDecision> track_decisions(frame->face_recog->items.size());
|
|
std::vector<bool> have_track_decision(frame->face_recog->items.size(), false);
|
|
for (size_t i = 0; i < frame->face_recog->items.size(); ++i) {
|
|
if (!qualifying_observations[i]) continue;
|
|
const auto& it = frame->face_recog->items[i];
|
|
auto& state = face_track_states_[it.person_track_id];
|
|
track_decisions[i] = UpdateFaceTrackState(
|
|
track_agg_cfg_, state, known_identity_last_alarm_ms_, it, now_epoch_ms);
|
|
have_track_decision[i] = true;
|
|
}
|
|
|
|
if (face_debug_log_unknown_candidates_) {
|
|
for (const auto& rule : face_rules_) {
|
|
if (rule.kind != FaceRule::Kind::Unknown) continue;
|
|
for (size_t i = 0; i < frame->face_recog->items.size(); ++i) {
|
|
FaceRuleEval eval;
|
|
const bool rule_matched = FaceItemMatchesRule(
|
|
rule, frame->face_recog->items[i], img_area, &eval);
|
|
LogUnknownCandidateDebug(
|
|
rule,
|
|
frame,
|
|
i,
|
|
img_area,
|
|
eval,
|
|
rule_matched,
|
|
have_track_decision[i] ? &track_decisions[i] : nullptr,
|
|
now_epoch_ms);
|
|
}
|
|
}
|
|
}
|
|
|
|
for (const auto& rule : face_rules_) {
|
|
if (rule.per_person_cooldown_ms <= 0) {
|
|
const auto it_last = face_last_trigger_.find(rule.name);
|
|
if (it_last != face_last_trigger_.end() && rule.cooldown_ms > 0) {
|
|
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(now - it_last->second).count();
|
|
if (elapsed < rule.cooldown_ms) {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
bool matched = false;
|
|
std::string matched_name;
|
|
std::vector<Detection> dets;
|
|
std::vector<std::string> trigger_keys;
|
|
std::vector<FaceAlarmMatch> face_matches;
|
|
dets.reserve(3);
|
|
trigger_keys.reserve(3);
|
|
face_matches.reserve(3);
|
|
|
|
for (size_t i = 0; i < frame->face_recog->items.size(); ++i) {
|
|
const auto& it = frame->face_recog->items[i];
|
|
if (!FaceItemMatchesRule(rule, it, img_area)) continue;
|
|
|
|
const std::string key = BuildFaceTrackKey(rule, it.person_track_id);
|
|
if (rule.per_person_cooldown_ms > 0) {
|
|
auto it_last = face_person_last_trigger_.find(key);
|
|
if (it_last != face_person_last_trigger_.end()) {
|
|
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(now - it_last->second).count();
|
|
if (elapsed < rule.per_person_cooldown_ms) {
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (!have_track_decision[i]) continue;
|
|
const FaceTrackDecision& decision = track_decisions[i];
|
|
if (rule.kind == FaceRule::Kind::Person && !decision.trigger_known) continue;
|
|
if (rule.kind == FaceRule::Kind::Unknown && !decision.trigger_unknown) continue;
|
|
|
|
matched = true;
|
|
if (rule.kind == FaceRule::Kind::Person) matched_name = it.best_name;
|
|
|
|
Detection d;
|
|
d.cls_id = -1;
|
|
d.score = it.best_sim;
|
|
d.bbox = it.bbox;
|
|
d.track_id = it.person_track_id >= 0 ? it.person_track_id : it.best_person_id;
|
|
dets.push_back(std::move(d));
|
|
trigger_keys.push_back(key);
|
|
FaceAlarmMatch fm;
|
|
fm.status = (rule.kind == FaceRule::Kind::Unknown)
|
|
? FaceAlarmStatus::Unknown
|
|
: (FaceRecogStateIsKnown(it.state) ? FaceAlarmStatus::Known : FaceAlarmStatus::Uncertain);
|
|
fm.candidate_person_id = it.candidate_person_id;
|
|
fm.candidate_name = it.candidate_name;
|
|
fm.best_sim = it.best_sim;
|
|
fm.second_sim = it.second_sim;
|
|
face_matches.push_back(std::move(fm));
|
|
if (dets.size() >= 3) break;
|
|
}
|
|
|
|
if (!matched) continue;
|
|
|
|
if (rule.per_person_cooldown_ms > 0) {
|
|
for (const auto& key : trigger_keys) {
|
|
face_person_last_trigger_[key] = now;
|
|
}
|
|
} else {
|
|
const auto it_last = face_last_trigger_.find(rule.name);
|
|
if (it_last != face_last_trigger_.end() && rule.cooldown_ms > 0) {
|
|
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(now - it_last->second).count();
|
|
if (elapsed < rule.cooldown_ms) {
|
|
continue;
|
|
}
|
|
}
|
|
face_last_trigger_[rule.name] = now;
|
|
}
|
|
|
|
RuleMatchResult fake;
|
|
fake.matched = true;
|
|
fake.rule_name = rule.name;
|
|
if (rule.kind == FaceRule::Kind::Person && !matched_name.empty()) {
|
|
fake.rule_name = rule.name + ":" + matched_name;
|
|
}
|
|
fake.matched_detections = std::move(dets);
|
|
TriggerAlarm(fake, frame, std::move(face_matches));
|
|
}
|
|
}
|
|
|
|
void LogUnknownCandidateDebug(
|
|
const FaceRule& rule,
|
|
const FramePtr& frame,
|
|
size_t item_index,
|
|
double img_area,
|
|
const FaceRuleEval& eval,
|
|
bool rule_matched,
|
|
const FaceTrackDecision* decision,
|
|
uint64_t now_epoch_ms) {
|
|
if (face_debug_unknown_candidate_interval_ms_ > 0 &&
|
|
last_unknown_candidate_log_ms_ > 0 &&
|
|
now_epoch_ms > last_unknown_candidate_log_ms_ &&
|
|
(now_epoch_ms - last_unknown_candidate_log_ms_) < face_debug_unknown_candidate_interval_ms_) {
|
|
return;
|
|
}
|
|
last_unknown_candidate_log_ms_ = now_epoch_ms;
|
|
|
|
const auto& it = frame->face_recog->items[item_index];
|
|
const auto state_it = face_track_states_.find(it.person_track_id);
|
|
const FaceTrackState* state = (state_it != face_track_states_.end()) ? &state_it->second : nullptr;
|
|
const uint64_t track_age_ms = (state && state->first_seen_ms > 0 && now_epoch_ms >= state->first_seen_ms)
|
|
? (now_epoch_ms - state->first_seen_ms)
|
|
: 0;
|
|
const int quality_hits = state ? state->quality_hits : 0;
|
|
const bool known_leaning = state &&
|
|
(state->reported_known || state->known_hit_times.size() >= static_cast<size_t>(track_agg_cfg_.known_min_hits));
|
|
|
|
std::string gate = "rule_rejected";
|
|
if (rule_matched && !decision) {
|
|
gate = "no_track_decision";
|
|
} else if (decision && decision->trigger_unknown) {
|
|
gate = "trigger_unknown";
|
|
} else if (state && state->reported_unknown) {
|
|
gate = "already_reported";
|
|
} else if (known_leaning) {
|
|
gate = "known_leaning";
|
|
} else if (rule_matched && track_age_ms < static_cast<uint64_t>(track_agg_cfg_.unknown_min_track_age_ms)) {
|
|
gate = "waiting_track_age";
|
|
} else if (rule_matched && quality_hits < track_agg_cfg_.unknown_min_quality_hits) {
|
|
gate = "waiting_quality_hits";
|
|
}
|
|
|
|
const double area_ratio = eval.face_area_ratio > 0.0
|
|
? eval.face_area_ratio
|
|
: ((img_area > 0.0) ? (static_cast<double>(it.bbox.w) * static_cast<double>(it.bbox.h) / img_area) : 0.0);
|
|
const double aspect = eval.face_aspect > 0.0
|
|
? eval.face_aspect
|
|
: (static_cast<double>(it.bbox.w) / std::max(1e-6, static_cast<double>(it.bbox.h)));
|
|
|
|
std::ostringstream oss;
|
|
oss << "[alarm] unknown_candidate"
|
|
<< " rule=" << rule.name
|
|
<< " frame=" << frame->frame_id
|
|
<< " item=" << item_index
|
|
<< " status=" << FaceRecogStateName(it.state)
|
|
<< " track_id=" << it.person_track_id
|
|
<< " candidate_id=" << it.candidate_person_id
|
|
<< " candidate=" << (it.candidate_name.empty() ? "-" : it.candidate_name)
|
|
<< " best_sim=" << Fixed3(it.best_sim)
|
|
<< " second_sim=" << Fixed3(it.second_sim)
|
|
<< " margin=" << Fixed3(static_cast<double>(it.best_sim) - static_cast<double>(it.second_sim))
|
|
<< " bbox=(" << Fixed3(it.bbox.x) << "," << Fixed3(it.bbox.y)
|
|
<< "," << Fixed3(it.bbox.w) << "," << Fixed3(it.bbox.h) << ")"
|
|
<< " area_ratio=" << Fixed6(area_ratio)
|
|
<< " aspect=" << Fixed3(aspect)
|
|
<< " rule_matched=" << (rule_matched ? "true" : "false")
|
|
<< " reject_reason=" << (eval.reject_reason.empty() ? "-" : eval.reject_reason)
|
|
<< " reject_detail=" << (eval.detail.empty() ? "-" : eval.detail)
|
|
<< " track_age_ms=" << track_age_ms
|
|
<< " quality_hits=" << quality_hits << "/" << track_agg_cfg_.unknown_min_quality_hits
|
|
<< " min_track_age_ms=" << track_agg_cfg_.unknown_min_track_age_ms
|
|
<< " reported_known=" << (state && state->reported_known ? "true" : "false")
|
|
<< " reported_unknown=" << (state && state->reported_unknown ? "true" : "false")
|
|
<< " gate=" << gate;
|
|
LogInfo(oss.str());
|
|
}
|
|
|
|
static std::string BuildFaceTrackKey(const FaceRule& rule, int track_id) {
|
|
return rule.name + "#track:" + std::to_string(track_id);
|
|
}
|
|
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, std::vector<FaceAlarmMatch> face_matches = {}) {
|
|
++alarm_count_;
|
|
|
|
AlarmEvent event;
|
|
event.node_id = id_;
|
|
event.rule_name = result.rule_name;
|
|
event.timestamp_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
std::chrono::system_clock::now().time_since_epoch()).count();
|
|
event.frame_id = frame->frame_id;
|
|
const uint64_t seq = next_event_seq_.fetch_add(1, std::memory_order_relaxed) + 1;
|
|
event.event_id = id_ + "_" + SanitizeEventIdPart(result.rule_name) + "_" +
|
|
std::to_string(event.frame_id) + "_" + std::to_string(seq);
|
|
event.detections = result.matched_detections;
|
|
event.face_matches = std::move(face_matches);
|
|
event.behavior_events = result.matched_behavior_events;
|
|
|
|
LogInfo("[alarm] trigger event_id=" + event.event_id +
|
|
" rule=" + event.rule_name +
|
|
" frame=" + std::to_string(event.frame_id));
|
|
|
|
AlarmJob job;
|
|
job.event = std::move(event);
|
|
job.frame = std::move(frame);
|
|
|
|
if (alarm_queue_) {
|
|
alarm_queue_->Push(std::move(job));
|
|
}
|
|
}
|
|
|
|
std::string id_;
|
|
std::vector<std::string> labels_;
|
|
RuleEngine rule_engine_;
|
|
std::vector<FaceRule> face_rules_;
|
|
std::map<std::string, std::chrono::steady_clock::time_point> face_last_trigger_;
|
|
std::map<std::string, std::chrono::steady_clock::time_point> face_person_last_trigger_;
|
|
FaceTrackAggregationConfig track_agg_cfg_;
|
|
std::map<int, FaceTrackState> face_track_states_;
|
|
std::unordered_map<int, uint64_t> known_identity_last_alarm_ms_;
|
|
bool face_debug_log_unknown_candidates_ = false;
|
|
uint64_t face_debug_unknown_candidate_interval_ms_ = 0;
|
|
uint64_t last_unknown_candidate_log_ms_ = 0;
|
|
std::shared_ptr<PacketRingBuffer> packet_buffer_;
|
|
std::vector<std::unique_ptr<IAlarmAction>> actions_;
|
|
|
|
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;
|
|
std::atomic<uint64_t> next_event_seq_{0};
|
|
|
|
int64_t eval_interval_ms_ = 0;
|
|
int64_t last_eval_pts_ms_ = 0;
|
|
};
|
|
|
|
REGISTER_NODE(AlarmNode, "alarm");
|
|
|
|
} // namespace rk3588
|