safesight/edge/plugins/event_fusion/event_fusion_node.cpp

#include "event_fusion_node.h"

#include <algorithm>
#include <map>
#include <memory>
#include <set>
#include <sstream>
#include <string>
#include <utility>
#include <vector>

#include "behavior/behavior_event.h"

namespace rk3588 {
namespace {

struct ActiveEvent {
    int event_id = -1;
    BehaviorEventItem last_item{};
    bool seen_in_frame = false;
};

static std::string TypeName(BehaviorEventType type) {
    switch (type) {
        case BehaviorEventType::Intrusion: return "intrusion";
        case BehaviorEventType::Climb: return "climb";
        case BehaviorEventType::Fall: return "fall";
        case BehaviorEventType::Fight: return "fight";
    }
    return "unknown";
}

static std::string MakeKey(const BehaviorEventItem& item) {
    std::vector<int> track_ids = item.track_ids;
    std::sort(track_ids.begin(), track_ids.end());

    std::ostringstream oss;
    oss << TypeName(item.type) << "|" << item.region_id << "|" << item.source;
    for (int track_id : track_ids) {
        oss << "|" << track_id;
    }
    return oss.str();
}

}  // namespace

struct EventFusionNode::Impl {
    int next_event_id = 1;
    std::map<std::string, ActiveEvent> active_events;
};

EventFusionNode::EventFusionNode() : impl_(std::make_unique<Impl>()) {}

EventFusionNode::~EventFusionNode() = default;

std::string EventFusionNode::Id() const {
    return id_;
}

std::string EventFusionNode::Type() const {
    return "event_fusion";
}

bool EventFusionNode::Init(const SimpleJson& config, const NodeContext& ctx) {
    id_ = config.ValueOr<std::string>("id", "event_fusion");
    output_queues_ = ctx.output_queues;
    return true;
}

bool EventFusionNode::Start() {
    return true;
}

void EventFusionNode::Stop() {}

NodeStatus EventFusionNode::Process(FramePtr frame) {
    if (!frame) return NodeStatus::DROP;

    EnsureBehaviorEvents(*frame);
    for (auto& [_, active] : impl_->active_events) {
        active.seen_in_frame = false;
    }

    std::vector<BehaviorEventItem> normalized;
    normalized.reserve(frame->behavior_events->items.size() + impl_->active_events.size());

    for (auto item : frame->behavior_events->items) {
        const std::string key = MakeKey(item);
        auto& active = impl_->active_events[key];
        if (active.event_id < 0) {
            active.event_id = impl_->next_event_id++;
        }
        active.seen_in_frame = true;
        item.event_id = active.event_id;
        active.last_item = item;
        normalized.push_back(std::move(item));
    }

    std::vector<std::string> to_remove;
    for (auto& [key, active] : impl_->active_events) {
        if (active.seen_in_frame) continue;

        BehaviorEventItem ended = active.last_item;
        ended.status = BehaviorEventStatus::Ended;
        ended.event_id = active.event_id;
        ended.last_pts = frame->pts;
        if (frame->pts >= ended.start_pts) {
            ended.duration_ms = frame->pts - ended.start_pts;
        }
        normalized.push_back(std::move(ended));
        to_remove.push_back(key);
    }

    for (const auto& key : to_remove) {
        impl_->active_events.erase(key);
    }

    frame->behavior_events->items = std::move(normalized);
    PushToDownstream(frame);
    return NodeStatus::OK;
}

void EventFusionNode::EnsureBehaviorEvents(Frame& frame) {
    if (!frame.behavior_events) {
        frame.behavior_events = std::make_shared<BehaviorEventResult>();
    }
}

void EventFusionNode::PushToDownstream(const FramePtr& frame) {
    for (auto& q : output_queues_) {
        if (q) q->Push(frame);
    }
}

#ifndef RK3588_TEST_BUILD
REGISTER_NODE(EventFusionNode, "event_fusion");
#endif

}  // namespace rk3588