videoStreamInfer/src/pipeline/pipeline.hpp

#ifndef PIPELINE_HPP__
#define PIPELINE_HPP__

#include <iostream>
#include <vector>
#include <string>
#include <unordered_map>
#include "nodes/track/trackNode.hpp"
#include "common/json.hpp"
#include "infer/infer.hpp"
#include "nodes/base/base.hpp"
#include "nodes/stream/streamNode.hpp"
#include "nodes/infer/inferNode.hpp"
#include "nodes/analyze/analyzeNode.hpp"
#include "nodes/draw/drawNode.hpp"
#include "nodes/record/recordNode.hpp"
#include "nodes/httpPush/httpPush.hpp"



namespace Pipe
{

struct PipelineInstance {
    std::string pipeline_id;
    std::string description;
    std::vector<std::shared_ptr<GNode::BaseNode>> nodes; // 只包含此 pipeline 的节点，按顺序
};

class PipelineManager
{

public:
    PipelineManager() = default;
    ~PipelineManager() { }


    // 获取配置好的 pipelines (只读)
    const std::vector<PipelineInstance>& getPipelines() const {
        return configured_pipelines_;
    }

    // 获取加载的共享模型 (只读)
    const std::unordered_map<std::string, std::shared_ptr<Infer>>& getSharedModels() const {
        return shared_models_;
    }

    bool set_config(const std::string& pipeline_id, const std::string& node_id, const meta::DrawConfigData& config) 
    {
        auto instance = std::find_if(configured_pipelines_.begin(), configured_pipelines_.end(), 
            [&pipeline_id](const PipelineInstance& instance) { return instance.pipeline_id == pipeline_id; });
        if (instance == configured_pipelines_.end())
        {
            std::cerr << "Pipeline with ID " << pipeline_id << " not found." << std::endl;
            return false;
        }
        auto node = std::find_if(instance->nodes.begin(), instance->nodes.end(), 
            [&node_id](const std::shared_ptr<GNode::BaseNode>& node) { return node->get_name() == node_id; });
        if (node == instance->nodes.end())
        {
            std::cerr << "Node with ID " << node_id << " not found in pipeline " << pipeline_id << "." << std::endl;
            return false;
        }
        (*node)->set_config_data(std::make_shared<meta::DrawConfigData>(config));
        return true;
    }

    bool set_config(const std::string& pipeline_id, const std::string& node_id, const meta::AnalyzeConfigData& config) 
    {
        auto instance = std::find_if(configured_pipelines_.begin(), configured_pipelines_.end(), 
            [&pipeline_id](const PipelineInstance& instance) { return instance.pipeline_id == pipeline_id; });
        if (instance == configured_pipelines_.end())
        {
            std::cerr << "Pipeline with ID " << pipeline_id << " not found." << std::endl;
            return false;
        }
        auto node = std::find_if(instance->nodes.begin(), instance->nodes.end(), 
            [&node_id](const std::shared_ptr<GNode::BaseNode>& node) { return node->get_name() == node_id; });
        if (node == instance->nodes.end())
        {
            std::cerr << "Node with ID " << node_id << " not found in pipeline " << pipeline_id << "." << std::endl;
            return false;
        }
        (*node)->set_config_data(std::make_shared<meta::AnalyzeConfigData>(config));
        return true;
    }

    void start_all_pipelines() 
    {
      for (auto& instance : configured_pipelines_) 
      {
        if (!instance.nodes.empty()) 
        {
            std::cout << "Starting pipeline: " << instance.pipeline_id << std::endl;
            for (auto it = instance.nodes.rbegin(); it != instance.nodes.rend(); ++it) 
            {
              (*it)->start();
            }
        }
      }
    }

    void stop_all_pipelines() 
    {
      for (auto& instance : configured_pipelines_) 
      {
        if (!instance.nodes.empty()) 
        {
            std::cout << "Stopping pipeline: " << instance.pipeline_id << std::endl;
            for (const auto& node : instance.nodes) 
            {
                node->stop();
            }
        }
      }
    }

    void start_pipelines_by_id(std::string pipeline_id) 
    {
      auto instance = std::find_if(configured_pipelines_.begin(), configured_pipelines_.end(), 
        [&pipeline_id](const PipelineInstance& instance) { return instance.pipeline_id == pipeline_id; });
      if (instance == configured_pipelines_.end())
      {
        std::cerr << "Pipeline with ID " << pipeline_id << " not found." << std::endl;
        return;
      }
      std::cout << "Starting pipeline: " << instance->pipeline_id << std::endl;
      // Start the pipeline by starting all nodes in reverse order
      for (auto it = instance->nodes.rbegin(); it != instance->nodes.rend(); ++it) 
      {
        (*it)->start();
      }
    }

    void stop_pipelines_by_id(std::string pipeline_id) 
    {
      auto instance = std::find_if(configured_pipelines_.begin(), configured_pipelines_.end(), 
        [&pipeline_id](const PipelineInstance& instance) { return instance.pipeline_id == pipeline_id; });
      if (instance == configured_pipelines_.end())
      {
        std::cerr << "Pipeline with ID " << pipeline_id << " not found." << std::endl;
        return;
      }
      std::cout << "Stopping pipeline: " << instance->pipeline_id << std::endl;
      // Stop the pipeline by stopping all nodes
      for (const auto& node : instance->nodes) 
      {
        node->stop();
      }
    }

    void show_all_pipeline()
    {
        for (const auto& instance : configured_pipelines_)
        {
            printf("Pipeline ID: %s\nDescription: %s\n", instance.pipeline_id.c_str(), instance.description.c_str());
            printf("start --> ");
            for (const auto& node : instance.nodes)
            {
                printf("%s --> ", node->get_name().c_str());
            }
            printf(" end\n");
        }
    }

    void show_pipeline_by_id(std::string pipeline_id)
    {
        auto instance = std::find_if(configured_pipelines_.begin(), configured_pipelines_.end(), 
            [&pipeline_id](const PipelineInstance& instance) { return instance.pipeline_id == pipeline_id; });
        if (instance == configured_pipelines_.end())
        {
            std::cerr << "Pipeline with ID " << pipeline_id << " not found." << std::endl;
            return;
        }
        printf("Pipeline ID: %s\nDescription: %s\n", instance->pipeline_id.c_str(), instance->description.c_str());
        printf("start --> ");
        for (const auto& node : instance->nodes)
        {
            printf("%s --> ", node->get_name().c_str());
        }
        printf(" end\n");
    }


    void create_from_json(const std::string& json_path);
    void create_from_yaml(const std::string& yaml_path);

private:
    std::unordered_map<std::string, std::shared_ptr<Infer>> shared_models_;
    std::vector<PipelineInstance> configured_pipelines_;
};

}   // namespace Graph

#endif // GRAPH_HPP__