9.1/service-maker/includes/pipeline.hpp

Go to the documentation of this file.

/*
 * SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#ifndef PIPELINE_HPP
#define PIPELINE_HPP
 
#include <string>
#include <memory>
#include <thread>
#include <unordered_map>
 
#include "object.hpp"
#include "element.hpp"
#include "buffer_probe.hpp"
#include "signal_handler.hpp"
#include "metadata.hpp"
 
namespace deepstream
{
 
 
struct RecordingInfo {
    RecordingInfo(void* data=nullptr);
    virtual ~RecordingInfo();
  
    unsigned int getSessionId() const;
    std::string getFileName() const;
    std::string getFileDirectory() const;
    unsigned int getDuration() const;
    std::string getContainerType() const;
    unsigned int getWidth() const;
    unsigned int getHeight() const;
    bool containsVideo() const;
    bool containsAudio() const;
    unsigned int getChannels() const;
    unsigned int getSamplingRate() const;
 
    operator bool() const {
      return data_ != nullptr;
    }
  
   protected:
    void * data_;
    
};
 
class Pipeline : public Object {
 public:
 
  typedef enum {
    INVALID,
    EMPTY,
    READY,
    PAUSED,
    PLAYING
  } State;
 
  class Message {
    public:
      virtual ~Message() {}
      uint32_t type() { return type_; }
    protected:
      uint32_t type_;
 
      Message(uint32_t type): type_(type) {}
  };
 
  class EOSMessage : public Message {
    public:
      EOSMessage(void*);
  };
 
  class DynamicSourceMessage : public Message {
    public:
      DynamicSourceMessage(void*);
      DynamicSourceMessage(uint32_t source_id, bool source_added);
 
      inline bool isSourceAdded() const { return source_added_; }
      inline uint32_t getSourceId() const { return source_id_; }
      inline const std::string& getSensorId() const { return sensor_id_; }
      inline const std::string& getSensorName() const { return sensor_name_; }
      inline const std::string& getUri() const { return uri_; }
 
    protected:
      bool source_added_ = false;
      uint32_t source_id_ = 0;
      std::string sensor_id_;
      std::string sensor_name_;
      std::string uri_;
  };
 
  class StateTransitionMessage : public Message {
    public:
      StateTransitionMessage(void*);
      inline const std::string& getName() const { return name_; };
      inline void getState(State& old_state, State& new_state) const {
        old_state = old_state_;
        new_state = new_state_;
      }
    protected:
      State old_state_;
      State new_state_;
      std::string name_;
  };
 
  Pipeline(const char* name);
 
  Pipeline(const char* name, const std::string& config_file);
 
  virtual ~Pipeline();
 
  template<typename... Args>
  Pipeline& add(const std::string& element_type, const std::string& element_name, const Args&... args) {
    Element element = Element(element_type, element_name);
    if constexpr (sizeof...(args) > 0) {
      element.set(args...);
    }
    return this->add(element);
  }
 
  Pipeline& add(Element element);
 
  Element* find(const::std::string& name);
 
  Element& operator[](const std::string& name) {
    Element* e = this->find(name);
    if (e == nullptr) {
      throw std::runtime_error(name + " Not found");
    }
    return *e;
  }
 
  Pipeline& link(
    std::pair<std::string, std::string> route,
    std::pair<std::string, std::string> tips
  );
 
  template <typename... Args>
  Pipeline &link(const std::string &arg1, const std::string arg2, const Args &...args)
  {
    link(std::make_pair(arg1, arg2), std::make_pair("", ""));
    if constexpr (sizeof...(args) > 0) {
        return this->link(arg2, args...);
    } else {
        return *this;
    }
  }
 
  Pipeline& attach(const std::string& elmenent_name, CustomObject* object, const std::string tip="");
 
  Pipeline& attach(
    const std::string& element_name,
    const std::string& plugin_name,
    const std::string& object_name,
    const std::string tip="");
 
  template<typename... Args>
  Pipeline& attach(
    const std::string& element_name,
    const std::string& plugin_name,
    const std::string& object_name,
    const std::string tip,
    const Args&... args) {
    attach(element_name, plugin_name, object_name, tip);
    if (sizeof...(args) > 0) {
      auto& element = (*this)[element_name];
      if (auto handler = element.getSignalHandler(object_name)) {
        handler->set(args...);
      } else if (auto probe = element.getProbe(object_name)) {
        probe->set(args...);
      }
    }
    return *this;
  }
 
  Pipeline& install(std::function<void(Pipeline&, int key)> keyboard_listener) {
    keyboard_listener_ = keyboard_listener;
    return *this;
  }
 
  int prepare();
  int prepare(std::function<void(Pipeline &, const Message &)> listener);
 
  Pipeline& start();
  Pipeline& start(std::function<void(Pipeline&, const Message&)> listener);
  Pipeline &activate();
  Pipeline& wait();
 
  Pipeline& stop();
 
  bool isRunning() const;
 
  Pipeline& pause();
  Pipeline& resume();
  Pipeline& seek(uint64_t timestamp);
 
  uint32_t startRTSP(uint16_t rtsp_port, uint16_t udp_port, uint32_t buffer_size=0);
 
  uint32_t startRecording(const std::string& element_name,
                      uint32_t startTime = 0,
                      uint32_t duration = 0,
                      std::function<void(const RecordingInfo &)> callback = [](const RecordingInfo &info) {
                        printf("\nSource recorded at %s/%s. Duration %.2f sec\n",
                          info.getFileDirectory().c_str(), info.getFileName().c_str(),
                          info.getDuration() / 1000.0);
                      },
                      void* userData = nullptr);
 
  bool stopRecording(const std::string& element_name);
 
  bool stopRecording(uint32_t sessionId);
 
  void handleKey(int key);
 
protected:
  int run();
  int run_after_prepare();
 
  // This is GMainLoop*
  void *loop_ = NULL;
  // gstreamer bus watch id
  uint bus_watch_id_ = 0;
  // gstreamer bus data
  void *bus_data_ = NULL;
  std::function<void(Pipeline&, int key)> keyboard_listener_;
 
  // Thread
  std::thread thread_;
  // Hold all the element references here
  std::map<std::string, Element> elements_;
  // Hold the signal emitters for action control
  std::map<std::string, std::unique_ptr<SignalEmitter>> action_owners_;
  // start timestamp for the whole pipeline
  uint64_t start_pts_ = 0;
  // Smart Record session id per element name
  std::unordered_map<std::string, uint32_t> element_to_session_id;
  std::unordered_map<uint32_t, std::string> session_id_to_element;
};
 
} // namespace deepstream
 
#endif