create.hpp

/****************************************************************************
 *  hipipe library
 *  Copyright (c) 2017, Cognexa Solutions s.r.o.
 *  Copyright (c) 2018, Iterait a.s.
 *  Author(s) Filip Matzner
 *
 *  This file is distributed under the MIT License.
 *  See the accompanying file LICENSE.txt for the complete license agreement.
 ****************************************************************************/
 
#pragma once
 
#include <hipipe/core/stream/stream_t.hpp>
#include <hipipe/core/utility/tuple.hpp>
 
#include <range/v3/view/transform.hpp>
#include <range/v3/view/chunk.hpp>
 
#include <typeinfo>
#include <unordered_map>
 
namespace hipipe::stream {
 
namespace rgv = ranges::views;
 
namespace detail {
 
    template<typename... Columns>
    struct create_impl {
 
        template<typename Source>
        batch_t operator()(Source&& source) const
        {
            batch_t batch;
 
            if constexpr(sizeof...(Columns) == 0) {
                static_assert("hipipe::stream::create: At least one column has to be provided.");
            } else if constexpr(sizeof...(Columns) == 1) {
                static_assert(std::is_constructible_v<Columns..., Source&&>,
                  "hipipe::stream::create: "
                  "Cannot convert the given data range to the selected column type.");
                batch.insert_or_assign<Columns...>(ranges::to_vector(std::forward<Source>(source)));
            } else {
                using SourceValue = ranges::range_value_t<Source>;
                static_assert(std::is_constructible_v<
                  std::tuple<typename Columns::example_type...>, SourceValue&&>,
                  "hipipe::stream::create: "
                  "Cannot convert the given data range to the selected column types.");
                std::tuple<Columns...> data = utility::unzip(std::forward<Source>(source));
                utility::tuple_for_each(data, [&batch](auto& column){
                    batch.insert_or_assign<std::decay_t<decltype(column)>>(std::move(column));
                });
            }
 
            return batch;
        }
    };
 
    template<typename... Columns>
    class create_fn {
    private:
        friend rgv::view_access;
 
        static auto bind(create_fn<Columns...> fun, std::size_t batch_size = 1)
        {
            return ranges::make_pipeable(std::bind(fun, std::placeholders::_1, batch_size));
        }
    public:
        CPP_template(class Rng)(requires ranges::forward_range<Rng>)
        forward_stream_t operator()(Rng&& rng, std::size_t batch_size = 1) const
        {
            return rgv::transform(
              rgv::chunk(std::forward<Rng>(rng), batch_size),
              create_impl<Columns...>{});
        }
 
        CPP_template(class Rng)(requires !ranges::forward_range<Rng>)
        void operator()(Rng&&, std::size_t batch_size = 1) const
        {
            CONCEPT_ASSERT_MSG(ranges::forward_range<Rng>(),
              "stream::create only works on ranges satisfying the forward_range concept.");
        }
    };
 
}  // namespace detail
 
template<typename... Columns>
rgv::view<detail::create_fn<Columns...>> create{};
 
} // end namespace hipipe::stream