pipeline.h

Go to the documentation of this file.

 
#ifndef SRC_PIPELINE_H_
#define SRC_PIPELINE_H_
 
#include <CppTimer.h>
 
#include <condition_variable>  //NOLINT [build/c++11]
#include <deque>
#include <mutex>   //NOLINT [build/c++11]
#include <thread>  //NOLINT [build/c++11]
#include <vector>
 
#include "iir.h"
#include "inference_core.h"
#include "opencv2/core.hpp"
#include "opencv2/videoio.hpp"
#include "post_processor.h"
#include "posture_estimator.h"
#include "pre_processor.h"
 
#define FRAME_DELAY_MAX 2000      
#define FRAME_DELAY_MIN 50        
#define FRAME_DELAY_DEFAULT 1000  
 
namespace Buffer {
 
template <typename T>
struct PopResult {
  T value;     
  bool valid;  
};
 
template <typename T>
class Buffer {
 private:
  std::mutex lock_in;    
  std::mutex lock_out;   
  std::vector<T> queue;  
 
  size_t front_index = 0;  
  size_t back_index = 0;   
 
  bool* running;
 
  uint8_t id = -1;
 
  size_t size(void) {
    int size = back_index - front_index;
    if (size < 0) {
      return (queue.size() + size);
    }
    return size;
  }
 
  bool full = false;
 
 public:
  explicit Buffer(bool* running_ptr, size_t max_size)
      : running(running_ptr), queue(max_size) {}
 
  void push(T frame) {
    while (*running) {
      lock_in.lock();
      if ((uint8_t)(id + 1) == frame.id && !full) {
        queue.at(back_index) = frame;
        back_index = (back_index + 1) % queue.size();
        if (back_index == front_index) {
          full = true;
        }
 
        id++;
        lock_in.unlock();
        break;
      }
      lock_in.unlock();
    }
  }
 
  bool try_push(T frame) {
    while (*running) {
      // Immediately return if the queue is full
      lock_in.lock();
      if (full) {
        lock_in.unlock();
        return false;
      }
 
      if ((uint8_t)(id + 1) == frame.id) {
        queue.at(back_index) = frame;
        back_index = (back_index + 1) % queue.size();
        if (front_index == back_index) {
          full = true;
        }
 
        id++;
        lock_in.unlock();
        return true;
      } else {
        lock_in.unlock();
      }
    }
    return false;
  }
 
  PopResult<T> pop() {
    PopResult<T> front = PopResult<T>{T{}, false};
    while (*running) {
      lock_out.lock();
      if (size() != 0 || full) {
        front = PopResult<T>{queue.at(front_index), true};
        front_index = (front_index + 1) % queue.size();
        full = false;
 
        lock_out.unlock();
        break;
      }
      lock_out.unlock();
    }
    return front;
  }
};
}  // namespace Buffer
 
namespace Pipeline {
 
struct FramerateSetting {
  IIR::SmoothingSettings
      smoothing_settings;  
  size_t frame_delay;  
};
 
// Forward declaration of `Pipeline` for use in `FramerateSettings`
class Pipeline;
 
class FramerateSettings {
 private:
  std::vector<FramerateSetting> framerate_settings;
 
  size_t current_setting;
 
  Pipeline* pipeline;
 
  void notify_pipeline(void);
 
 public:
  explicit FramerateSettings(Pipeline* pipeline);
 
  FramerateSetting get_framerate_setting(void);
 
  void increase_framerate(void);
 
  void decrease_framerate(void);
};
 
struct PreprocessedFrame {
  uint8_t id;
  cv::Mat raw_image;
  PreProcessing::PreProcessedImage preprocessed_image;
};
 
struct RawFrame {
  uint8_t id;         
  cv::Mat raw_image;  
};
 
class FrameGenerator : public CppTimer {
 private:
  cv::VideoCapture cap;
 
  uint8_t id = 0;
 
  cv::Mat current_frame;
 
  std::mutex mutex;
 
  std::condition_variable cv;
 
  std::thread thread;
 
  void thread_body(void);
 
  void timerEvent(void);
 
  bool running = true;
 
 public:
  FrameGenerator(void);
 
  ~FrameGenerator();
 
  void updated_framerate(size_t new_frame_delay);
 
  RawFrame next_frame(void);
};
 
struct CoreResults {
  uint8_t id;
  cv::Mat raw_image;
  Inference::InferenceResults image_results;
};
 
class Pipeline {
 private:
  std::vector<std::thread> threads;
 
  bool running;
 
  FramerateSettings framerate_settings;
 
  PreProcessing::PreProcessor preprocessor;
  PostProcessing::PostProcessor post_processor;
  PostureEstimating::PostureEstimator posture_estimator;
 
  FrameGenerator frame_generator;
  Buffer::Buffer<CoreResults> core_results;
 
  void core_thread_body(Inference::InferenceCore);
 
  void post_processing_thread_body(void);
 
  void (*callback)(PostureEstimating::PoseStatus, cv::Mat);
 
 public:
  void updated_framerate(FramerateSetting new_settings);
 
  explicit Pipeline(uint8_t num_inference_core_threads,
                    void (*callback)(PostureEstimating::PoseStatus, cv::Mat));
 
  ~Pipeline();
 
  bool set_confidence_threshold(float threshold);
 
  float get_confidence_threshold();
 
  float increase_framerate(void);
 
  float decrease_framerate(void);
 
  float get_framerate(void);
 
  void set_ideal_posture(PostureEstimating::Pose pose);
 
  bool set_pose_change_threshold(float threshold);
 
  float get_pose_change_threshold();
};
}  // namespace Pipeline
#endif  // SRC_PIPELINE_H_