Cvb/CppCompositeStreamHandler

#include <iostream>
#include <string>
#include <vector>
#include <condition_variable>
 
#include <cvb/device_factory.hpp>
#include <cvb/global.hpp>
#include <cvb/driver/composite_stream.hpp>
#include <cvb/genapi/node_map_enumerator.hpp>
#include <cvb/async/async.hpp>
#include <cvb/async/stream_handler_base.hpp>
 
static const constexpr auto TIMEOUT                = std::chrono::milliseconds(3000);
static const constexpr int NUM_ELEMENTS_TO_ACQUIRE = 10;  // number of elements to be acquired
static std::map<Cvb::WaitStatus, const char *> WAIT_ERROR_STATES{{Cvb::WaitStatus::Timeout, "timeout"},
                                                                 {Cvb::WaitStatus::Abort, "abort"}};
 
// derives the CompositeStreamHandler class so that we can implement our own customized tasks
// for the demonstration.
class CustomStreamHandler final : public Cvb::Async::CompositeStreamHandler
{
  struct PrivateTag
  {
  };
 
public:
  static std::unique_ptr<CustomStreamHandler>
  Create(const Cvb::Async::CompositeStreamHandler::StreamVectorType &streams)
  {
    return std::make_unique<CustomStreamHandler>(streams, PrivateTag{});
  }
 
  explicit CustomStreamHandler(const Cvb::Async::CompositeStreamHandler::StreamVectorType &streams, PrivateTag)
      : Cvb::Async::CompositeStreamHandler(streams)
      , numDeliverables_(0)
  {
  }
 
  std::cv_status WaitUntil(std::unique_lock<std::mutex> &lock, const std::chrono::system_clock::time_point &absTime)
  {
    return observer_.wait_until(lock, absTime);
  }
 
private:
  // asynchronously called for all registered streams.
  void HandleAsyncStream(const typename Cvb::Async::CompositeStreamHandler::StreamVectorType &streams) override
  {
    // the following code is the same as the default implementation;
    // this is just to demonstrate that you can freely override depending on your demand.
    std::vector<Cvb::WaitResultTuple<Cvb::Composite>> waitResultList;
    waitResultList.reserve(streams.size());
    for (auto n = 0; n < streams.size(); ++n)
      waitResultList.emplace_back(streams[n]->WaitFor(TIMEOUT));
 
    // let it take care of the acquired deliverables.
    HandleAsyncWaitResult(waitResultList);
  }
 
  // asynchronously called for all acquired deliverables.
  void HandleAsyncWaitResult(const std::vector<Cvb::WaitResultTuple<Cvb::Composite>> &waitResultList) override
  {
    // the default implementation does nothing; you may add tasks by yourself.
    std::cout << "round: #" << numDeliverables_ << "\n";
 
    // iterate over the streams.
    for (auto m = 0; m < waitResultList.size(); ++m)
    {
      // check the wait status.
      auto waitStatus = std::get<Cvb::WaitStatus>(waitResultList[m]);
      std::cout << "stream #" << m << "\n";
      std::cout << "wait status: ";
      if (waitStatus != Cvb::WaitStatus::Ok)
      {
        switch (waitStatus)
        {
          case Cvb::WaitStatus::Abort:
          case Cvb::WaitStatus::Timeout:
            std::cout << WAIT_ERROR_STATES[waitStatus];
            break;
          default:
            std::cout << "unknown";
        }
        std::cout << "; no deliverable is available.\n";
        continue;  // work on the next stream.
      }
      else
        std::cout << "ok\n";
 
      // pick up the delivered composite.
      auto composite = std::get<Cvb::CompositePtr>(waitResultList[m]);
 
      // identify the deliverable type of each item.
      const auto numItems = composite->ItemCount();
      std::cout << "number of items: " << numItems << "\n";
      for (auto n = 0; n < numItems; ++n)
      {
        auto item      = composite->ItemAt(n);
        auto item_type = std::string("unknown");
        if (Cvb::holds_alternative<Cvb::ImagePtr>(item))
          item_type = "image";
        else if (Cvb::holds_alternative<Cvb::PlanePtr>(item))
          item_type = "plane";
        else if (Cvb::holds_alternative<Cvb::PlaneEnumeratorPtr>(item))
          item_type = "plane enumerator";
        else if (Cvb::holds_alternative<Cvb::BufferPtr>(item))
          item_type = "buffer";
        else if (Cvb::holds_alternative<Cvb::PFNCBufferPtr>(item))
          item_type = "pfnc buffer";
        std::cout << "composite item #" << n << ": " << item_type << "\n";
      }
    }
    std::cout << "\n";
 
    ++numDeliverables_;
    if (numDeliverables_ == NUM_ELEMENTS_TO_ACQUIRE)
      observer_.notify_all();
  }
 
private:
  std::condition_variable observer_;
  int numDeliverables_;
};
 
int main()
{
  try
  {
    // enumerate devices.
    auto infoList = Cvb::DeviceFactory::Discover(Cvb::DiscoverFlags::IgnoreVins);
 
    // cannot continue the demonstration if CVMockTL is not present
    Cvb::String accessToken;
    for (const auto &info : infoList)
      if (info.AccessToken().find("MockTL") != Cvb::String::npos)
        accessToken = info.AccessToken();
 
    if (accessToken.empty())
      throw std::runtime_error("there is no available device for this demonstration.");
 
    // instantiate the first device on the list; however, it is worth knowing that
    // you can bind streams of other devices to a stream handle if needed.
    auto device =
        Cvb::DeviceFactory::Open<Cvb::GenICamDevice>(infoList.front().AccessToken(), Cvb::AcquisitionStack::GenTL);
 
    // create a stream handler.
    // IMPORTANT: this tutorial assumes that each stream is synchronized, i.e., it
    // is running at the same acquisition frame rate; if any of them is out of sync
    // the application will lose images. if any of the streams are not synchronized,
    // please consider preparing a dedicated stream handler object for every single
    // stream.
    std::vector<Cvb::CompositeStreamPtr> streams;
    streams.reserve(device->StreamCount());
    for (auto n = 0; n < device->StreamCount(); ++n)
      streams.emplace_back(device->Stream<Cvb::CompositeStream>(n));
    auto handler = CustomStreamHandler::Create(streams);
 
    // start the data acquisition.
    handler->Run();
 
    // wait for the job is done.
    {
      std::mutex mutex;
      std::unique_lock<std::mutex> lock(mutex);
      if (std::cv_status::no_timeout != handler->WaitUntil(lock, std::chrono::system_clock::now() + TIMEOUT))
        throw std::runtime_error("could not complete the job");
    }
 
    // stop the data acquisition, ignore errors.
    handler->TryFinish();
  }
  catch (const std::exception &e)
  {
    std::cout << e.what() << std::endl;
  }
  return 0;
}