Combine enable/disable stream with processors

src/api/api.cc

@@ -152,6 +152,10 @@ void API::EnableStreamData(const Stream &stream) {
   synthetic_->EnableStreamData(stream);
 }
 
+void API::DisableStreamData(const Stream &stream) {
+  synthetic_->DisableStreamData(stream);
+}
+
 api::StreamData API::GetStreamData(const Stream &stream) {
   return synthetic_->GetStreamData(stream);
 }

src/api/api.h

@@ -79,6 +79,8 @@ class MYNTEYE_API API {
   void WaitForStreams();
 
   void EnableStreamData(const Stream &stream);
+  void DisableStreamData(const Stream &stream);
+
   api::StreamData GetStreamData(const Stream &stream);
   std::vector<api::StreamData> GetStreamDatas(const Stream &stream);
 

src/api/processor/processor.cc

@@ -59,10 +59,10 @@ void Processor::SetProcessCallback(ProcessCallback callback) {
   callback_ = std::move(callback);
 }
 
-void Processor::Activate(bool tree) {
+void Processor::Activate(bool parents) {
   if (activated_)
     return;
-  if (tree) {
+  if (parents) {
     // Activate all parents
     Processor *parent = parent_;
     while (parent != nullptr) {
@@ -75,10 +75,10 @@ void Processor::Activate(bool tree) {
   // thread_.detach();
 }
 
-void Processor::Deactivate(bool tree) {
+void Processor::Deactivate(bool childs) {
   if (!activated_)
     return;
-  if (tree) {
+  if (childs) {
     // Deactivate all childs
     iterate_processors(GetChilds(), [](std::shared_ptr<Processor> proc) {
       proc->Deactivate();

src/api/processor/processor.h

@@ -39,8 +39,8 @@ class Processor /*: public std::enable_shared_from_this<Processor>*/ {
   void SetPostProcessCallback(PostProcessCallback callback);
   void SetProcessCallback(ProcessCallback callback);
 
-  void Activate(bool tree = false);
+  void Activate(bool parents = false);
-  void Deactivate(bool tree = false);
+  void Deactivate(bool childs = false);
   bool IsActivated();
 
   bool IsIdle();

src/api/synthetic.cc

@@ -2,11 +2,14 @@
 
 #include <glog/logging.h>
 
+#include <algorithm>
+#include <functional>
 #include <stdexcept>
 
 #include "api/processor/depth_processor.h"
 #include "api/processor/disparity_normalized_processor.h"
 #include "api/processor/disparity_processor.h"
+#include "api/processor/object.h"
 #include "api/processor/points_processor.h"
 #include "api/processor/processor.h"
 #include "api/processor/rectify_processor.h"
@@ -18,10 +21,15 @@ Synthetic::Synthetic(API *api) : api_(api) {
   VLOG(2) << __func__;
   CHECK_NOTNULL(api_);
   InitStreamSupports();
+  InitProcessors();
 }
 
 Synthetic::~Synthetic() {
   VLOG(2) << __func__;
+  if (processor_) {
+    processor_->Deactivate(true);
+    processor_ = nullptr;
+  }
 }
 
 Synthetic::mode_t Synthetic::GetMode(const Stream &stream) const {
@@ -37,67 +45,28 @@ bool Synthetic::Supports(const Stream &stream) const {
 }
 
 void Synthetic::EnableStreamData(const Stream &stream) {
-  if (Supports(stream))
+  EnableStreamData(stream, 0);
-    return;
-  switch (stream) {
-    case Stream::LEFT_RECTIFIED: {
-      if (Supports(Stream::LEFT)) {
-        stream_supports_mode_[stream] = MODE_SYNTHETIC;
-        return;
-      }
-    } break;
-    case Stream::RIGHT_RECTIFIED: {
-      if (Supports(Stream::RIGHT)) {
-        stream_supports_mode_[stream] = MODE_SYNTHETIC;
-        return;
-      }
-    } break;
-    case Stream::DISPARITY: {
-      if (Supports(Stream::LEFT) && Supports(Stream::RIGHT)) {
-        stream_supports_mode_[stream] = MODE_SYNTHETIC;
-        return;
-      }
-    } break;
-    case Stream::DISPARITY_NORMALIZED: {
-      if ((Supports(Stream::LEFT) && Supports(Stream::RIGHT)) ||
-          Supports(Stream::DISPARITY)) {
-        stream_supports_mode_[stream] = MODE_SYNTHETIC;
-        return;
-      }
-    } break;
-    case Stream::DEPTH: {
-      if ((Supports(Stream::LEFT) && Supports(Stream::RIGHT)) ||
-          Supports(Stream::DISPARITY) || Supports(Stream::POINTS)) {
-        stream_supports_mode_[stream] = MODE_SYNTHETIC;
-        return;
-      }
-    } break;
-    case Stream::POINTS: {
-      if ((Supports(Stream::LEFT) && Supports(Stream::RIGHT)) ||
-          Supports(Stream::DISPARITY)) {
-        stream_supports_mode_[stream] = MODE_SYNTHETIC;
-        return;
-      }
-    } break;
-    default:
-      break;
-  }
-  LOG(WARNING) << "Enable stream data of " << stream << " failed";
 }
 
 void Synthetic::DisableStreamData(const Stream &stream) {
-  UNUSED(stream)
+  DisableStreamData(stream, 0);
+}
+
+bool Synthetic::IsStreamDataEnabled(const Stream &stream) const {
+  return stream_supports_mode_.find(stream) != stream_supports_mode_.end();
 }
 
 void Synthetic::SetStreamCallback(
     const Stream &stream, stream_callback_t callback) {
-  UNUSED(stream)
+  if (callback == nullptr) {
-  UNUSED(callback)
+    stream_callbacks_.erase(stream);
+  } else {
+    stream_callbacks_[stream] = callback;
+  }
 }
 
 bool Synthetic::HasStreamCallback(const Stream &stream) const {
-  UNUSED(stream)
+  return stream_callbacks_.find(stream) != stream_callbacks_.end();
-  return false;
 }
 
 void Synthetic::StartVideoStreaming() {}
@@ -126,4 +95,181 @@ void Synthetic::InitStreamSupports() {
   }
 }
 
+void Synthetic::EnableStreamData(const Stream &stream, std::uint32_t depth) {
+  if (Supports(stream))
+    return;
+  // Activate processors of synthetic stream
+  switch (stream) {
+    case Stream::LEFT_RECTIFIED: {
+      if (!Supports(Stream::LEFT))
+        break;
+      stream_supports_mode_[stream] = MODE_SYNTHETIC;
+      CHECK(ActivateProcessor<RectifyProcessor>());
+    }
+      return;
+    case Stream::RIGHT_RECTIFIED: {
+      if (!Supports(Stream::RIGHT))
+        break;
+      stream_supports_mode_[stream] = MODE_SYNTHETIC;
+      CHECK(ActivateProcessor<RectifyProcessor>());
+    }
+      return;
+    case Stream::DISPARITY: {
+      stream_supports_mode_[stream] = MODE_SYNTHETIC;
+      EnableStreamData(Stream::LEFT_RECTIFIED, depth + 1);
+      EnableStreamData(Stream::RIGHT_RECTIFIED, depth + 1);
+      CHECK(ActivateProcessor<DisparityProcessor>());
+    }
+      return;
+    case Stream::DISPARITY_NORMALIZED: {
+      stream_supports_mode_[stream] = MODE_SYNTHETIC;
+      EnableStreamData(Stream::DISPARITY, depth + 1);
+      CHECK(ActivateProcessor<DisparityNormalizedProcessor>());
+    }
+      return;
+    case Stream::POINTS: {
+      stream_supports_mode_[stream] = MODE_SYNTHETIC;
+      EnableStreamData(Stream::DISPARITY, depth + 1);
+      CHECK(ActivateProcessor<PointsProcessor>());
+    }
+      return;
+    case Stream::DEPTH: {
+      stream_supports_mode_[stream] = MODE_SYNTHETIC;
+      EnableStreamData(Stream::POINTS, depth + 1);
+      CHECK(ActivateProcessor<DepthProcessor>());
+    }
+      return;
+    default:
+      break;
+  }
+  if (depth == 0) {
+    LOG(WARNING) << "Enable stream data of " << stream << " failed";
+  }
+}
+
+void Synthetic::DisableStreamData(const Stream &stream, std::uint32_t depth) {
+  if (!IsStreamDataEnabled(stream))
+    return;
+  // Deactivate processors of synthetic stream
+  if (stream_supports_mode_[stream] != MODE_NATIVE) {
+    stream_supports_mode_.erase(stream);
+    switch (stream) {
+      case Stream::LEFT_RECTIFIED: {
+        if (GetMode(Stream::RIGHT_RECTIFIED) == MODE_SYNTHETIC) {
+          DisableStreamData(Stream::RIGHT_RECTIFIED, depth + 1);
+        }
+        if (GetMode(Stream::DISPARITY) == MODE_SYNTHETIC) {
+          DisableStreamData(Stream::DISPARITY, depth + 1);
+        }
+        DeactivateProcessor<RectifyProcessor>();
+      } break;
+      case Stream::RIGHT_RECTIFIED: {
+        if (GetMode(Stream::LEFT_RECTIFIED) == MODE_SYNTHETIC) {
+          DisableStreamData(Stream::LEFT_RECTIFIED, depth + 1);
+        }
+        if (GetMode(Stream::DISPARITY) == MODE_SYNTHETIC) {
+          DisableStreamData(Stream::DISPARITY, depth + 1);
+        }
+        DeactivateProcessor<RectifyProcessor>();
+      } break;
+      case Stream::DISPARITY: {
+        if (GetMode(Stream::DISPARITY_NORMALIZED) == MODE_SYNTHETIC) {
+          DisableStreamData(Stream::DISPARITY_NORMALIZED, depth + 1);
+        }
+        if (GetMode(Stream::POINTS) == MODE_SYNTHETIC) {
+          DisableStreamData(Stream::POINTS, depth + 1);
+        }
+        DeactivateProcessor<DisparityProcessor>();
+      } break;
+      case Stream::DISPARITY_NORMALIZED: {
+        DeactivateProcessor<DisparityNormalizedProcessor>();
+      } break;
+      case Stream::POINTS: {
+        if (GetMode(Stream::DEPTH) == MODE_SYNTHETIC) {
+          DisableStreamData(Stream::DEPTH, depth + 1);
+        }
+        DeactivateProcessor<PointsProcessor>();
+      } break;
+      case Stream::DEPTH: {
+        DeactivateProcessor<DepthProcessor>();
+      } break;
+      default:
+        return;
+    }
+    if (depth > 0) {
+      LOG(WARNING) << "Disable synthetic stream data of " << stream << " too";
+    }
+  } else if (depth == 0) {
+    LOG(WARNING) << "Disable native stream data of " << stream << " failed";
+  }
+}
+
+void Synthetic::InitProcessors() {
+  auto &&rectify_processor = std::make_shared<RectifyProcessor>();
+  auto &&disparity_processor = std::make_shared<DisparityProcessor>();
+  auto &&disparitynormalized_processor =
+      std::make_shared<DisparityNormalizedProcessor>();
+  auto &&points_processor = std::make_shared<PointsProcessor>();
+  auto &&depth_processor = std::make_shared<DepthProcessor>();
+
+  using namespace std::placeholders;  // NOLINT
+  rectify_processor->SetProcessCallback(
+      std::bind(&Synthetic::OnRectifyProcess, this, _1, _2, _3));
+  disparity_processor->SetProcessCallback(
+      std::bind(&Synthetic::OnDisparityProcess, this, _1, _2, _3));
+  disparitynormalized_processor->SetProcessCallback(
+      std::bind(&Synthetic::OnDisparityNormalizedProcess, this, _1, _2, _3));
+  points_processor->SetProcessCallback(
+      std::bind(&Synthetic::OnPointsProcess, this, _1, _2, _3));
+  depth_processor->SetProcessCallback(
+      std::bind(&Synthetic::OnDepthProcess, this, _1, _2, _3));
+
+  rectify_processor->AddChild(disparity_processor);
+  disparity_processor->AddChild(disparitynormalized_processor);
+  disparity_processor->AddChild(points_processor);
+  points_processor->AddChild(depth_processor);
+
+  processor_ = rectify_processor;
+}
+
+bool Synthetic::OnRectifyProcess(
+    Object *const in, Object *const out, Processor *const parent) {
+  UNUSED(in)
+  UNUSED(out)
+  UNUSED(parent)
+  return false;
+}
+
+bool Synthetic::OnDisparityProcess(
+    Object *const in, Object *const out, Processor *const parent) {
+  UNUSED(in)
+  UNUSED(out)
+  UNUSED(parent)
+  return false;
+}
+
+bool Synthetic::OnDisparityNormalizedProcess(
+    Object *const in, Object *const out, Processor *const parent) {
+  UNUSED(in)
+  UNUSED(out)
+  UNUSED(parent)
+  return false;
+}
+
+bool Synthetic::OnPointsProcess(
+    Object *const in, Object *const out, Processor *const parent) {
+  UNUSED(in)
+  UNUSED(out)
+  UNUSED(parent)
+  return false;
+}
+
+bool Synthetic::OnDepthProcess(
+    Object *const in, Object *const out, Processor *const parent) {
+  UNUSED(in)
+  UNUSED(out)
+  UNUSED(parent)
+  return false;
+}
+
 MYNTEYE_END_NAMESPACE

src/api/synthetic.h

@@ -4,6 +4,7 @@
 
 #include <map>
 #include <memory>
+#include <string>
 #include <vector>
 
 #include "api/api.h"
@@ -11,6 +12,7 @@
 MYNTEYE_BEGIN_NAMESPACE
 
 class API;
+class Object;
 class Processor;
 
 class Synthetic {
@@ -31,6 +33,7 @@ class Synthetic {
 
   void EnableStreamData(const Stream &stream);
   void DisableStreamData(const Stream &stream);
+  bool IsStreamDataEnabled(const Stream &stream) const;
 
   void SetStreamCallback(const Stream &stream, stream_callback_t callback);
   bool HasStreamCallback(const Stream &stream) const;
@@ -46,13 +49,86 @@ class Synthetic {
  private:
   void InitStreamSupports();
 
+  void EnableStreamData(const Stream &stream, std::uint32_t depth);
+  void DisableStreamData(const Stream &stream, std::uint32_t depth);
+
+  void InitProcessors();
+
+  template <class T>
+  bool ActivateProcessor(bool tree = false);
+  template <class T>
+  bool DeactivateProcessor(bool tree = false);
+
+  bool OnRectifyProcess(
+      Object *const in, Object *const out, Processor *const parent);
+  bool OnDisparityProcess(
+      Object *const in, Object *const out, Processor *const parent);
+  bool OnDisparityNormalizedProcess(
+      Object *const in, Object *const out, Processor *const parent);
+  bool OnPointsProcess(
+      Object *const in, Object *const out, Processor *const parent);
+  bool OnDepthProcess(
+      Object *const in, Object *const out, Processor *const parent);
+
   API *api_;
 
   std::map<Stream, mode_t> stream_supports_mode_;
 
-  std::vector<std::shared_ptr<Processor>> processors_;
+  std::map<Stream, stream_callback_t> stream_callbacks_;
+
+  std::shared_ptr<Processor> processor_;
 };
 
+template <class T, class P>
+std::shared_ptr<T> find_processor(const P &processor) {
+  if (processor->Name() == T::NAME) {
+    return std::dynamic_pointer_cast<T>(processor);
+  }
+  auto &&childs = processor->GetChilds();
+  return find_processor<T>(std::begin(childs), std::end(childs), T::NAME);
+}
+
+template <class T, class InputIt>
+std::shared_ptr<T> find_processor(
+    InputIt first, InputIt last, const std::string &name) {
+  if (first == last)
+    return nullptr;
+  for (auto it = first; it != last; ++it) {
+    if ((*it)->Name() == name) {
+      return std::dynamic_pointer_cast<T>(*it);
+    }
+  }
+  for (auto it = first; it != last; ++it) {
+    auto &&childs = (*it)->GetChilds();
+    if (childs.empty())
+      continue;
+    auto &&result =
+        find_processor<T>(std::begin(childs), std::end(childs), name);
+    if (result == nullptr)
+      continue;
+    return result;
+  }
+  return nullptr;
+}
+
+template <class T>
+bool Synthetic::ActivateProcessor(bool parents) {
+  auto &&processor = find_processor<T>(processor_);
+  if (processor == nullptr)
+    return false;
+  processor->Activate(parents);
+  return true;
+}
+
+template <class T>
+bool Synthetic::DeactivateProcessor(bool childs) {
+  auto &&processor = find_processor<T>(processor_);
+  if (processor == nullptr)
+    return false;
+  processor->Deactivate(childs);
+  return true;
+}
+
 MYNTEYE_END_NAMESPACE
 
 #endif  // MYNTEYE_SYNTHETIC_H_ NOLINT