Project import generated by Copybara.

GitOrigin-RevId: ea8d45731f5a052f79745e35bfd8240d6ac568d2
This commit is contained in:
MediaPipe Team
2020-12-16 00:05:25 -05:00
committed by chuoling
parent 38be2ec58f
commit 39309bedba
109 changed files with 5803 additions and 1500 deletions
+231
View File
@@ -0,0 +1,231 @@
package(
default_visibility = [":preview_users"],
features = ["-use_header_modules"],
)
# API2 is in preview mode. Internal clients are welcome and encouraged to try
# it out, but be aware that there may be more changes before release. Please
# add your package to this list and reach out to the MediaPipe team (use
# camillol@ as the CL reviewer).
package_group(
name = "preview_users",
packages = [
"//mediapipe/...",
"//video/content_analysis/...",
],
)
licenses(["notice"])
cc_library(
name = "const_str",
hdrs = ["const_str.h"],
)
cc_library(
name = "builder",
hdrs = ["builder.h"],
deps = [
":const_str",
":contract",
":node",
":packet",
":port",
"//mediapipe/framework:calculator_base",
"//mediapipe/framework:calculator_contract",
"@com_google_absl//absl/container:flat_hash_map",
],
)
cc_test(
name = "builder_test",
srcs = ["builder_test.cc"],
deps = [
":builder",
":node",
":packet",
":port",
":tag",
":test_contracts",
"//mediapipe/framework:calculator_framework",
"//mediapipe/framework/deps:message_matchers",
"//mediapipe/framework/port:gtest_main",
"//mediapipe/framework/port:parse_text_proto",
"@com_google_absl//absl/strings",
],
)
cc_library(
name = "contract",
hdrs = ["contract.h"],
deps = [
":const_str",
":packet",
":port",
":tag",
":tuple",
"//mediapipe/framework:calculator_context",
"//mediapipe/framework:calculator_contract",
"//mediapipe/framework:output_side_packet",
"//mediapipe/framework/port:logging",
],
)
cc_test(
name = "contract_test",
srcs = ["contract_test.cc"],
deps = [
":contract",
":port",
":tag",
"//mediapipe/framework:calculator_framework",
"//mediapipe/framework/port:gtest_main",
"//mediapipe/framework/port:parse_text_proto",
],
)
cc_library(
name = "node",
srcs = ["node.cc"],
hdrs = ["node.h"],
deps = [
":const_str",
":contract",
":packet",
":port",
"//mediapipe/framework:calculator_base",
"//mediapipe/framework:calculator_context",
"//mediapipe/framework:calculator_contract",
"//mediapipe/framework:subgraph",
"//mediapipe/framework/deps:no_destructor",
],
)
cc_library(
name = "test_contracts",
testonly = 1,
hdrs = ["test_contracts.h"],
deps = [
":node",
],
)
cc_test(
name = "node_test",
srcs = ["node_test.cc"],
deps = [
":node",
":packet",
":port",
":test_contracts",
":tuple",
"//mediapipe/framework:calculator_framework",
"//mediapipe/framework/port:gtest_main",
"//mediapipe/framework/port:parse_text_proto",
"//mediapipe/framework/port:status",
],
)
cc_library(
name = "packet",
srcs = ["packet.cc"],
hdrs = ["packet.h"],
deps = [
":tuple",
"//mediapipe/framework:packet",
"//mediapipe/framework/port:logging",
],
)
cc_test(
name = "packet_test",
size = "small",
srcs = [
"packet_test.cc",
],
deps = [
":packet",
"//mediapipe/framework/port:gtest_main",
"@com_google_absl//absl/strings",
],
)
cc_library(
name = "port",
hdrs = ["port.h"],
deps = [
":const_str",
":packet",
"//mediapipe/framework:calculator_context",
"//mediapipe/framework:calculator_contract",
"//mediapipe/framework:output_side_packet",
"//mediapipe/framework/port:logging",
"@com_google_absl//absl/strings",
],
)
cc_test(
name = "port_test",
size = "small",
srcs = [
"port_test.cc",
],
deps = [
":port",
"//mediapipe/framework/port:gtest_main",
],
)
cc_test(
name = "subgraph_test",
srcs = ["subgraph_test.cc"],
deps = [
":builder",
":node",
":packet",
":port",
":test_contracts",
"//mediapipe/framework:calculator_framework",
"//mediapipe/framework/deps:message_matchers",
"//mediapipe/framework/port:gtest_main",
"//mediapipe/framework/port:parse_text_proto",
"//mediapipe/framework/tool:subgraph_expansion",
],
)
cc_library(
name = "tag",
hdrs = ["tag.h"],
deps = [":const_str"],
)
cc_test(
name = "tag_test",
size = "small",
srcs = [
"tag_test.cc",
],
deps = [
":tag",
"//mediapipe/framework/port:gtest_main",
],
)
cc_library(
name = "tuple",
hdrs = ["tuple.h"],
deps = ["@com_google_absl//absl/meta:type_traits"],
)
cc_test(
name = "tuple_test",
size = "small",
srcs = [
"tuple_test.cc",
],
deps = [
":tuple",
"//mediapipe/framework/port:gtest_main",
"@com_google_absl//absl/strings",
],
)
+111
View File
@@ -0,0 +1,111 @@
# Experimental new APIs
This directory defines new APIs for MediaPipe:
- Node API, an update to the Calculator API for defining MediaPipe components.
- Builder API, for assembling CalculatorGraphConfigs with C++, as an alternative
to using the proto API directly.
The code is working, and the new APIs interoperate fully with the existing
framework code. They are considered a work in progress, but are being released
now so we can begin adopting them in our calculators.
Developers are welcome to try out these APIs as early adopters, but should
expect breaking changes. The placement of this code under the `mediapipe::api2`
namespace is not final.
## Node API
This API can be used to define calculators. It is designed to be more type-safe
and less verbose than the original API.
Input/output ports (streams and side packets) can now be declared as typed
constants, instead of using plain strings for access.
For example, instead of
```
constexpr char kSelectTag[] = "SELECT";
if (cc->Inputs().HasTag(kSelectTag)) {
cc->Inputs().Tag(kSelectTag).Set<int>();
}
```
you can write
```
static constexpr Input<int>::Optional kSelect{"SELECT"};
```
Instead of setting up the contract procedurally in `GetContract`, add ports to
the contract declaratively, as follows:
```
MEDIAPIPE_NODE_CONTRACT(kInput, kOutput);
```
To access an input in Process, instead of
```
int select = cc->Inputs().Tag(kSelectTag).Get<int>();
```
write
```
int select = kSelectTag(cc).Get(); // alternative: *kSelectTag(cc)
```
Sets of multiple ports can be declared with `::Multiple`. Note, also, that a tag
string must always be provided when declaring a port; use `""` for untagged
ports. For example:
```
for (int i = 0; i < cc->Inputs().NumEntries(); ++i) {
cc->Inputs().Index(i).SetAny();
}
```
becomes
```
static constexpr Input<AnyType>::Multiple kIn{""};
```
For output ports, the payload can be passed directly to the `Send` method. For
example, instead of
```
cc->Outputs().Index(0).Add(
new std::pair<Packet, Packet>(cc->Inputs().Index(0).Value(),
cc->Inputs().Index(1).Value()),
cc->InputTimestamp());
```
you can write
```
kPair(cc).Send({kIn(cc)[0].packet(), kIn(cc)[1].packet()});
```
The input timestamp is propagated to the outputs by default. If your calculator
wants to alter timestamps, it must add a `TimestampChange` entry to its contract
declaration. For example:
```
MEDIAPIPE_NODE_CONTRACT(kMain, kLoop, kPrevLoop,
StreamHandler("ImmediateInputStreamHandler"),
TimestampChange::Arbitrary());
```
Several calculators in
[`calculators/core`](https://github.com/google/mediapipe/tree/master/mediapipe/calculators/core) and
[`calculators/tensor`](https://github.com/google/mediapipe/tree/master/mediapipe/calculators/tensor)
have been updated to use this API. Reference them for more examples.
More complete documentation will be provided in the future.
## Builder API
Documentation will be provided in the future.
+576
View File
@@ -0,0 +1,576 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_BUILDER_H_
#define MEDIAPIPE_FRAMEWORK_API2_BUILDER_H_
#include <string>
#include <type_traits>
#include "absl/container/flat_hash_map.h"
#include "mediapipe/framework/api2/const_str.h"
#include "mediapipe/framework/api2/contract.h"
#include "mediapipe/framework/api2/node.h"
#include "mediapipe/framework/api2/packet.h"
#include "mediapipe/framework/api2/port.h"
#include "mediapipe/framework/calculator_base.h"
#include "mediapipe/framework/calculator_contract.h"
namespace mediapipe {
namespace api2 {
namespace builder {
template <typename T>
T& GetWithAutoGrow(std::vector<std::unique_ptr<T>>* vecp, int index) {
auto& vec = *vecp;
if (vec.size() <= index) {
vec.resize(index + 1);
}
if (vec[index] == nullptr) {
vec[index] = absl::make_unique<T>();
}
return *vec[index];
}
struct TagIndexLocation {
const std::string& tag;
std::size_t index;
std::size_t count;
};
template <typename T>
class TagIndexMap {
public:
std::vector<std::unique_ptr<T>>& operator[](const std::string& tag) {
return map_[tag];
}
void Visit(std::function<void(const TagIndexLocation&, const T&)> fun) const {
for (const auto& tagged : map_) {
TagIndexLocation loc{tagged.first, 0, tagged.second.size()};
for (const auto& item : tagged.second) {
fun(loc, *item);
++loc.index;
}
}
}
void Visit(std::function<void(const TagIndexLocation&, T*)> fun) {
for (auto& tagged : map_) {
TagIndexLocation loc{tagged.first, 0, tagged.second.size()};
for (auto& item : tagged.second) {
fun(loc, item.get());
++loc.index;
}
}
}
// Note: entries are held by a unique_ptr to ensure pointers remain valid.
// Should use absl::flat_hash_map but ordering keys for now.
std::map<std::string, std::vector<std::unique_ptr<T>>> map_;
};
// These structs are used internally to store information about the endpoints
// of a connection.
struct SourceBase;
struct DestinationBase {
SourceBase* source = nullptr;
};
struct SourceBase {
std::vector<DestinationBase*> dests_;
std::string name_;
};
// Following existing GraphConfig usage, we allow using a multiport as a single
// port as well. This is necessary for generic nodes, since we have no
// information about which ports are meant to be multiports or not, but it is
// also convenient with typed nodes.
template <typename Single>
class MultiPort : public Single {
public:
using Base = typename Single::Base;
explicit MultiPort(std::vector<std::unique_ptr<Base>>* vec)
: Single(vec), vec_(*vec) {}
Single operator[](int index) {
CHECK_GE(index, 0);
return Single{&GetWithAutoGrow(&vec_, index)};
}
private:
std::vector<std::unique_ptr<Base>>& vec_;
};
// These classes wrap references to the underlying source/destination
// endpoints, adding type information and the user-visible API.
template <bool AllowMultiple, bool IsSide, typename T = internal::Generic>
class DestinationImpl {
public:
using Base = DestinationBase;
explicit DestinationImpl(std::vector<std::unique_ptr<Base>>* vec)
: DestinationImpl(&GetWithAutoGrow(vec, 0)) {}
explicit DestinationImpl(DestinationBase* base) : base_(*base) {}
DestinationBase& base_;
};
template <bool IsSide, typename T>
class DestinationImpl<true, IsSide, T>
: public MultiPort<DestinationImpl<false, IsSide, T>> {
public:
using MultiPort<DestinationImpl<false, IsSide, T>>::MultiPort;
};
template <bool AllowMultiple, bool IsSide, typename T = internal::Generic>
class SourceImpl {
public:
using Base = SourceBase;
// Src is used as the return type of fluent methods below. Since these are
// single-port methods, it is desirable to always decay to a reference to the
// single-port superclass, even if they are called on a multiport.
using Src = SourceImpl<false, IsSide, T>;
template <typename U>
using Dst = DestinationImpl<false, IsSide, U>;
// clang-format off
template <typename U>
struct AllowConnection : public std::integral_constant<bool,
std::is_same<T, U>{} || std::is_same<T, internal::Generic>{} ||
std::is_same<U, internal::Generic>{}> {};
// clang-format on
explicit SourceImpl(std::vector<std::unique_ptr<Base>>* vec)
: SourceImpl(&GetWithAutoGrow(vec, 0)) {}
explicit SourceImpl(SourceBase* base) : base_(*base) {}
template <typename U,
typename std::enable_if<AllowConnection<U>{}, int>::type = 0>
Src& AddTarget(const Dst<U>& dest) {
CHECK(dest.base_.source == nullptr);
dest.base_.source = &base_;
base_.dests_.emplace_back(&dest.base_);
return *this;
}
Src& SetName(std::string name) {
base_.name_ = std::move(name);
return *this;
}
template <typename U>
Src& operator>>(const Dst<U>& dest) {
return AddTarget(dest);
}
private:
SourceBase& base_;
};
template <bool IsSide, typename T>
class SourceImpl<true, IsSide, T>
: public MultiPort<SourceImpl<false, IsSide, T>> {
public:
using MultiPort<SourceImpl<false, IsSide, T>>::MultiPort;
};
// A source and a destination correspond to an output/input stream on a node,
// and a side source and side destination correspond to an output/input side
// packet.
// For graph inputs/outputs, however, the inputs are sources, and the outputs
// are destinations. This is because graph ports are connected "from inside"
// when building the graph.
template <bool AllowMultiple = false, typename T = internal::Generic>
using Source = SourceImpl<AllowMultiple, false, T>;
template <bool AllowMultiple = false, typename T = internal::Generic>
using SideSource = SourceImpl<AllowMultiple, true, T>;
template <bool AllowMultiple = false, typename T = internal::Generic>
using Destination = DestinationImpl<AllowMultiple, false, T>;
template <bool AllowMultiple = false, typename T = internal::Generic>
using SideDestination = DestinationImpl<AllowMultiple, true, T>;
class NodeBase {
public:
// TODO: right now access to an indexed port is made directly by
// specifying both a tag and an index. It would be better to represent this
// as a two-step lookup, first getting a multi-port, and then accessing one
// of its entries by index. However, for nodes without visible contracts we
// can't know whether a tag is indexable or not, so we would need the
// multi-port to also be usable as a port directly (representing index 0).
Source<true> Out(const std::string& tag) {
return Source<true>(&out_streams_[tag]);
}
Destination<true> In(const std::string& tag) {
return Destination<true>(&in_streams_[tag]);
}
SideSource<true> SideOut(const std::string& tag) {
return SideSource<true>(&out_sides_[tag]);
}
SideDestination<true> SideIn(const std::string& tag) {
return SideDestination<true>(&in_sides_[tag]);
}
// Convenience methods for accessing purely index-based ports.
Source<false> Out(int index) { return Out("")[index]; }
Destination<false> In(int index) { return In("")[index]; }
SideSource<false> SideOut(int index) { return SideOut("")[index]; }
SideDestination<false> SideIn(int index) { return SideIn("")[index]; }
template <typename T>
T& GetOptions() {
options_used_ = true;
return *options_.MutableExtension(T::ext);
}
protected:
NodeBase(std::string type) : type_(std::move(type)) {}
std::string type_;
TagIndexMap<DestinationBase> in_streams_;
TagIndexMap<SourceBase> out_streams_;
TagIndexMap<DestinationBase> in_sides_;
TagIndexMap<SourceBase> out_sides_;
CalculatorOptions options_;
// ideally we'd just check if any extensions are set on options_
bool options_used_ = false;
friend class Graph;
};
template <class Calc = internal::Generic>
class Node;
#if __cplusplus >= 201703L
// Deduction guide to silence -Wctad-maybe-unsupported.
explicit Node()->Node<internal::Generic>;
#endif // C++17
template <>
class Node<internal::Generic> : public NodeBase {
public:
Node(std::string type) : NodeBase(std::move(type)) {}
};
using GenericNode = Node<internal::Generic>;
template <template <bool, class> class BP, class Port, class TagIndexMapT>
auto MakeBuilderPort(const Port& port, TagIndexMapT& streams) {
return BP<Port::kMultiple, typename Port::PayloadT>(&streams[port.Tag()]);
}
template <class Calc>
class Node : public NodeBase {
public:
Node() : NodeBase(Calc::kCalculatorName) {}
// Overrides the built-in calculator type std::string with the provided
// argument. Can be used to create nodes from pure interfaces.
// TODO: only use this for pure interfaces
Node(const std::string& type_override) : NodeBase(type_override) {}
// These methods only allow access to ports declared in the contract.
// The argument must be a tag object created with the MPP_TAG macro.
// These objects encode the tag in their type, which allows us to return
// a result with the appropriate payload type depending on the tag.
template <class Tag>
auto Out(Tag tag) {
constexpr auto& port = Calc::Contract::TaggedOutputs::get(tag);
return MakeBuilderPort<Source>(port, out_streams_);
}
template <class Tag>
auto In(Tag tag) {
constexpr auto& port = Calc::Contract::TaggedInputs::get(tag);
return MakeBuilderPort<Destination>(port, in_streams_);
}
template <class Tag>
auto SideOut(Tag tag) {
constexpr auto& port = Calc::Contract::TaggedSideOutputs::get(tag);
return MakeBuilderPort<SideSource>(port, out_sides_);
}
template <class Tag>
auto SideIn(Tag tag) {
constexpr auto& port = Calc::Contract::TaggedSideInputs::get(tag);
return MakeBuilderPort<SideDestination>(port, in_sides_);
}
// We could allow using the non-checked versions with typed nodes too, but
// we don't.
// using NodeBase::Out;
// using NodeBase::In;
// using NodeBase::SideOut;
// using NodeBase::SideIn;
};
// For legacy PacketGenerators.
class PacketGenerator {
public:
PacketGenerator(std::string type) : type_(std::move(type)) {}
SideSource<true> SideOut(const std::string& tag) {
return SideSource<true>(&out_sides_[tag]);
}
SideDestination<true> SideIn(const std::string& tag) {
return SideDestination<true>(&in_sides_[tag]);
}
// Convenience methods for accessing purely index-based ports.
SideSource<false> SideOut(int index) { return SideOut("")[index]; }
SideDestination<false> SideIn(int index) { return SideIn("")[index]; }
template <typename T>
T& GetOptions() {
options_used_ = true;
return *options_.MutableExtension(T::ext);
}
private:
std::string type_;
TagIndexMap<DestinationBase> in_sides_;
TagIndexMap<SourceBase> out_sides_;
mediapipe::PacketGeneratorOptions options_;
// ideally we'd just check if any extensions are set on options_
bool options_used_ = false;
friend class Graph;
};
class Graph {
public:
void SetType(std::string type) { type_ = std::move(type); }
// Creates a node of a specific type. Should be used for calculators whose
// contract is available.
template <class Calc>
Node<Calc>& AddNode() {
auto node = std::make_unique<Node<Calc>>();
auto node_p = node.get();
nodes_.emplace_back(std::move(node));
return *node_p;
}
// Creates a node of a specific type. Should be used for pure interfaces,
// which do not have a built-in type std::string.
template <class Calc>
Node<Calc>& AddNode(const std::string& type) {
auto node = std::make_unique<Node<Calc>>(type);
auto node_p = node.get();
nodes_.emplace_back(std::move(node));
return *node_p;
}
// Creates a generic node, with no compile-time checking of inputs and
// outputs. This can be used for calculators whose contract is not visible.
GenericNode& AddNode(const std::string& type) {
auto node = std::make_unique<GenericNode>(type);
auto node_p = node.get();
nodes_.emplace_back(std::move(node));
return *node_p;
}
// For legacy PacketGenerators.
PacketGenerator& AddPacketGenerator(const std::string& type) {
auto node = std::make_unique<PacketGenerator>(type);
auto node_p = node.get();
packet_gens_.emplace_back(std::move(node));
return *node_p;
}
// Graph ports, non-typed.
Source<true> In(const std::string& graph_input) {
return graph_boundary_.Out(graph_input);
}
Destination<true> Out(const std::string& graph_output) {
return graph_boundary_.In(graph_output);
}
SideSource<true> SideIn(const std::string& graph_input) {
return graph_boundary_.SideOut(graph_input);
}
SideDestination<true> SideOut(const std::string& graph_output) {
return graph_boundary_.SideIn(graph_output);
}
// Convenience methods for accessing purely index-based ports.
Source<false> In(int index) { return In("")[0]; }
Destination<false> Out(int index) { return Out("")[0]; }
SideSource<false> SideIn(int index) { return SideIn("")[0]; }
SideDestination<false> SideOut(int index) { return SideOut("")[0]; }
// Graph ports, typed.
// TODO: make graph_boundary_ a typed node!
template <class PortT, class Payload = typename PortT::PayloadT,
class Src = Source<PortT::kMultiple, Payload>>
Src In(const PortT& graph_input) {
return Src(&graph_boundary_.out_streams_[graph_input.Tag()]);
}
template <class PortT, class Payload = typename PortT::PayloadT,
class Dst = Destination<PortT::kMultiple, Payload>>
Dst Out(const PortT& graph_output) {
return Dst(&graph_boundary_.in_streams_[graph_output.Tag()]);
}
template <class PortT, class Payload = typename PortT::PayloadT,
class Src = SideSource<PortT::kMultiple, Payload>>
Src SideIn(const PortT& graph_input) {
return Src(&graph_boundary_.out_sides_[graph_input.Tag()]);
}
template <class PortT, class Payload = typename PortT::PayloadT,
class Dst = SideDestination<PortT::kMultiple, Payload>>
Dst SideOut(const PortT& graph_output) {
return Dst(&graph_boundary_.in_sides_[graph_output.Tag()]);
}
// Returns the graph config. This can be used to instantiate and run the
// graph.
CalculatorGraphConfig GetConfig() {
CalculatorGraphConfig config;
if (!type_.empty()) {
config.set_type(type_);
}
FixUnnamedConnections();
CHECK_OK(UpdateBoundaryConfig(&config));
for (const std::unique_ptr<NodeBase>& node : nodes_) {
auto* out_node = config.add_node();
CHECK_OK(UpdateNodeConfig(*node, out_node));
}
for (const std::unique_ptr<PacketGenerator>& node : packet_gens_) {
auto* out_node = config.add_packet_generator();
CHECK_OK(UpdateNodeConfig(*node, out_node));
}
return config;
}
private:
void FixUnnamedConnections(NodeBase* node, int* unnamed_count) {
node->out_streams_.Visit([&](const TagIndexLocation&, SourceBase* source) {
if (source->name_.empty()) {
source->name_ = absl::StrCat("__stream_", (*unnamed_count)++);
}
});
node->out_sides_.Visit([&](const TagIndexLocation&, SourceBase* source) {
if (source->name_.empty()) {
source->name_ = absl::StrCat("__side_packet_", (*unnamed_count)++);
}
});
}
void FixUnnamedConnections() {
int unnamed_count = 0;
FixUnnamedConnections(&graph_boundary_, &unnamed_count);
for (std::unique_ptr<NodeBase>& node : nodes_) {
FixUnnamedConnections(node.get(), &unnamed_count);
}
for (std::unique_ptr<PacketGenerator>& node : packet_gens_) {
node->out_sides_.Visit([&](const TagIndexLocation&, SourceBase* source) {
if (source->name_.empty()) {
source->name_ = absl::StrCat("__side_packet_", unnamed_count++);
}
});
}
}
std::string TaggedName(const TagIndexLocation& loc, const std::string& name) {
if (loc.tag.empty()) {
// ParseTagIndexName does not allow using explicit indices without tags,
// while ParseTagIndex does. There is no explanation for this discrepancy
// in the CLs that introduced them (cl/143209019, cl/156499931).
// TODO: decide whether we should just allow it.
return name;
} else {
if (loc.count <= 1) {
return absl::StrCat(loc.tag, ":", name);
} else {
return absl::StrCat(loc.tag, ":", loc.index, ":", name);
}
}
}
mediapipe::Status UpdateNodeConfig(const NodeBase& node,
CalculatorGraphConfig::Node* config) {
config->set_calculator(node.type_);
node.in_streams_.Visit(
[&](const TagIndexLocation& loc, const DestinationBase& endpoint) {
CHECK(endpoint.source != nullptr);
config->add_input_stream(TaggedName(loc, endpoint.source->name_));
});
node.out_streams_.Visit(
[&](const TagIndexLocation& loc, const SourceBase& endpoint) {
config->add_output_stream(TaggedName(loc, endpoint.name_));
});
node.in_sides_.Visit([&](const TagIndexLocation& loc,
const DestinationBase& endpoint) {
CHECK(endpoint.source != nullptr);
config->add_input_side_packet(TaggedName(loc, endpoint.source->name_));
});
node.out_sides_.Visit(
[&](const TagIndexLocation& loc, const SourceBase& endpoint) {
config->add_output_side_packet(TaggedName(loc, endpoint.name_));
});
if (node.options_used_) {
*config->mutable_options() = node.options_;
}
return {};
}
mediapipe::Status UpdateNodeConfig(const PacketGenerator& node,
PacketGeneratorConfig* config) {
config->set_packet_generator(node.type_);
node.in_sides_.Visit([&](const TagIndexLocation& loc,
const DestinationBase& endpoint) {
CHECK(endpoint.source != nullptr);
config->add_input_side_packet(TaggedName(loc, endpoint.source->name_));
});
node.out_sides_.Visit(
[&](const TagIndexLocation& loc, const SourceBase& endpoint) {
config->add_output_side_packet(TaggedName(loc, endpoint.name_));
});
if (node.options_used_) {
*config->mutable_options() = node.options_;
}
return {};
}
// For special boundary node.
mediapipe::Status UpdateBoundaryConfig(CalculatorGraphConfig* config) {
graph_boundary_.in_streams_.Visit(
[&](const TagIndexLocation& loc, const DestinationBase& endpoint) {
CHECK(endpoint.source != nullptr);
config->add_output_stream(TaggedName(loc, endpoint.source->name_));
});
graph_boundary_.out_streams_.Visit(
[&](const TagIndexLocation& loc, const SourceBase& endpoint) {
config->add_input_stream(TaggedName(loc, endpoint.name_));
});
graph_boundary_.in_sides_.Visit([&](const TagIndexLocation& loc,
const DestinationBase& endpoint) {
CHECK(endpoint.source != nullptr);
config->add_output_side_packet(TaggedName(loc, endpoint.source->name_));
});
graph_boundary_.out_sides_.Visit(
[&](const TagIndexLocation& loc, const SourceBase& endpoint) {
config->add_input_side_packet(TaggedName(loc, endpoint.name_));
});
return {};
}
std::string type_;
std::vector<std::unique_ptr<NodeBase>> nodes_;
std::vector<std::unique_ptr<PacketGenerator>> packet_gens_;
// Special node representing graph inputs and outputs.
NodeBase graph_boundary_{"__GRAPH__"};
};
} // namespace builder
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_BUILDER_H_
+190
View File
@@ -0,0 +1,190 @@
#include "mediapipe/framework/api2/builder.h"
#include "absl/strings/substitute.h"
#include "mediapipe/framework/api2/node.h"
#include "mediapipe/framework/api2/packet.h"
#include "mediapipe/framework/api2/port.h"
#include "mediapipe/framework/api2/tag.h"
#include "mediapipe/framework/api2/test_contracts.h"
#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/deps/message_matchers.h"
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
#include "mediapipe/framework/port/parse_text_proto.h"
#include "mediapipe/framework/port/status_matchers.h"
namespace mediapipe {
namespace api2 {
namespace test {
TEST(BuilderTest, BuildGraph) {
builder::Graph graph;
auto& foo = graph.AddNode("Foo");
auto& bar = graph.AddNode("Bar");
graph.In("IN").SetName("base") >> foo.In("BASE");
graph.SideIn("SIDE").SetName("side") >> foo.SideIn("SIDE");
foo.Out("OUT") >> bar.In("IN");
bar.Out("OUT").SetName("out") >> graph.Out("OUT");
CalculatorGraphConfig expected =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "IN:base"
input_side_packet: "SIDE:side"
output_stream: "OUT:out"
node {
calculator: "Foo"
input_stream: "BASE:base"
input_side_packet: "SIDE:side"
output_stream: "OUT:__stream_0"
}
node {
calculator: "Bar"
input_stream: "IN:__stream_0"
output_stream: "OUT:out"
}
)");
EXPECT_THAT(graph.GetConfig(), EqualsProto(expected));
}
template <class FooT>
void BuildGraphTypedTest() {
builder::Graph graph;
auto& foo = graph.AddNode<FooT>();
auto& bar = graph.AddNode<Bar>();
graph.In("IN").SetName("base") >> foo.In(MPP_TAG("BASE"));
graph.SideIn("SIDE").SetName("side") >> foo.SideIn(MPP_TAG("BIAS"));
foo.Out(MPP_TAG("OUT")) >> bar.In(MPP_TAG("IN"));
bar.Out(MPP_TAG("OUT")).SetName("out") >> graph.Out("OUT");
CalculatorGraphConfig expected =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(
absl::Substitute(R"(
input_stream: "IN:base"
input_side_packet: "SIDE:side"
output_stream: "OUT:out"
node {
calculator: "$0"
input_stream: "BASE:base"
input_side_packet: "BIAS:side"
output_stream: "OUT:__stream_0"
}
node {
calculator: "Bar"
input_stream: "IN:__stream_0"
output_stream: "OUT:out"
}
)",
FooT::kCalculatorName));
EXPECT_THAT(graph.GetConfig(), EqualsProto(expected));
}
TEST(BuilderTest, BuildGraphTyped) { BuildGraphTypedTest<Foo>(); }
TEST(BuilderTest, BuildGraphTyped2) { BuildGraphTypedTest<Foo2>(); }
TEST(BuilderTest, FanOut) {
builder::Graph graph;
auto& foo = graph.AddNode("Foo");
auto& adder = graph.AddNode("FloatAdder");
graph.In("IN").SetName("base") >> foo.In("BASE");
foo.Out("OUT") >> adder.In("IN")[0];
foo.Out("OUT") >> adder.In("IN")[1];
adder.Out("OUT").SetName("out") >> graph.Out("OUT");
CalculatorGraphConfig expected =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "IN:base"
output_stream: "OUT:out"
node {
calculator: "Foo"
input_stream: "BASE:base"
output_stream: "OUT:__stream_0"
}
node {
calculator: "FloatAdder"
input_stream: "IN:0:__stream_0"
input_stream: "IN:1:__stream_0"
output_stream: "OUT:out"
}
)");
EXPECT_THAT(graph.GetConfig(), EqualsProto(expected));
}
TEST(BuilderTest, TypedMultiple) {
builder::Graph graph;
auto& foo = graph.AddNode<Foo>();
auto& adder = graph.AddNode<FloatAdder>();
graph.In("IN").SetName("base") >> foo.In(MPP_TAG("BASE"));
foo.Out(MPP_TAG("OUT")) >> adder.In(MPP_TAG("IN"))[0];
foo.Out(MPP_TAG("OUT")) >> adder.In(MPP_TAG("IN"))[1];
adder.Out(MPP_TAG("OUT")).SetName("out") >> graph.Out("OUT");
CalculatorGraphConfig expected =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "IN:base"
output_stream: "OUT:out"
node {
calculator: "Foo"
input_stream: "BASE:base"
output_stream: "OUT:__stream_0"
}
node {
calculator: "FloatAdder"
input_stream: "IN:0:__stream_0"
input_stream: "IN:1:__stream_0"
output_stream: "OUT:out"
}
)");
EXPECT_THAT(graph.GetConfig(), EqualsProto(expected));
}
TEST(BuilderTest, PacketGenerator) {
builder::Graph graph;
auto& generator = graph.AddPacketGenerator("FloatGenerator");
graph.SideIn("IN") >> generator.SideIn("IN");
generator.SideOut("OUT") >> graph.SideOut("OUT");
CalculatorGraphConfig expected =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_side_packet: "IN:__side_packet_0"
output_side_packet: "OUT:__side_packet_1"
packet_generator {
packet_generator: "FloatGenerator"
input_side_packet: "IN:__side_packet_0"
output_side_packet: "OUT:__side_packet_1"
}
)");
EXPECT_THAT(graph.GetConfig(), EqualsProto(expected));
}
TEST(BuilderTest, EmptyTag) {
builder::Graph graph;
auto& foo = graph.AddNode("Foo");
graph.In("A").SetName("a") >> foo.In("")[0];
graph.In("C").SetName("c") >> foo.In("")[2];
graph.In("B").SetName("b") >> foo.In("")[1];
foo.Out("")[0].SetName("x") >> graph.Out("ONE");
foo.Out("")[1].SetName("y") >> graph.Out("TWO");
CalculatorGraphConfig expected =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "A:a"
input_stream: "B:b"
input_stream: "C:c"
output_stream: "ONE:x"
output_stream: "TWO:y"
node {
calculator: "Foo"
input_stream: "a"
input_stream: "b"
input_stream: "c"
output_stream: "x"
output_stream: "y"
}
)");
EXPECT_THAT(graph.GetConfig(), EqualsProto(expected));
}
} // namespace test
} // namespace api2
} // namespace mediapipe
+43
View File
@@ -0,0 +1,43 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_CONST_STR_H_
#define MEDIAPIPE_FRAMEWORK_API2_CONST_STR_H_
#include <string>
namespace mediapipe {
namespace api2 {
// This class stores a constant std::string that can be inspected at compile
// time in constexpr code.
class const_str {
public:
constexpr const_str(std::size_t size, const char* data)
: len_(size - 1), data_(data) {}
template <std::size_t N>
explicit constexpr const_str(const char (&str)[N]) : const_str(N, str) {}
constexpr std::size_t len() const { return len_; }
constexpr const char* data() const { return data_; }
constexpr bool operator==(const const_str& other) const {
return len_ == other.len_ && equal(len_, data_, other.data_);
}
constexpr char operator[](const std::size_t idx) const {
return idx <= len_ ? data_[idx] : '\0';
}
private:
static constexpr bool equal(std::size_t len, const char* const p,
const char* const q) {
return len == 0 || (*p == *q && equal(len - 1, p + 1, q + 1));
}
const std::size_t len_;
const char* const data_;
};
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_CONST_STR_H_
+387
View File
@@ -0,0 +1,387 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_CONTRACT_H_
#define MEDIAPIPE_FRAMEWORK_API2_CONTRACT_H_
#include <tuple>
#include <type_traits>
#include <utility>
#include <vector>
#include "mediapipe/framework/api2/const_str.h"
#include "mediapipe/framework/api2/packet.h"
#include "mediapipe/framework/api2/port.h"
#include "mediapipe/framework/api2/tag.h"
#include "mediapipe/framework/api2/tuple.h"
#include "mediapipe/framework/calculator_context.h"
#include "mediapipe/framework/calculator_contract.h"
#include "mediapipe/framework/output_side_packet.h"
#include "mediapipe/framework/port/logging.h"
namespace mediapipe {
namespace api2 {
class StreamHandler {
public:
template <std::size_t N>
explicit constexpr StreamHandler(const char (&name)[N]) : name_(N, name) {}
const const_str& name() { return name_; }
mediapipe::Status AddToContract(CalculatorContract* cc) const {
cc->SetInputStreamHandler(name_.data());
return {};
}
private:
const const_str name_;
};
class TimestampChange {
public:
// Note: we don't use TimestampDiff as an argument because it's not constexpr.
static constexpr TimestampChange Offset(int64_t offset) {
return TimestampChange(offset);
}
static constexpr TimestampChange Arbitrary() {
// Same value as used for Timestamp::Unset.
return TimestampChange(kUnset);
}
mediapipe::Status AddToContract(CalculatorContract* cc) const {
if (offset_ != kUnset) cc->SetTimestampOffset(offset_);
return {};
}
private:
constexpr TimestampChange(int64_t offset) : offset_(offset) {}
static constexpr int64_t kUnset = std::numeric_limits<int64_t>::min();
int64_t offset_;
};
namespace internal {
template <class Base>
struct IsSubclass {
template <class T>
using pred = std::is_base_of<Base, std::decay_t<T>>;
};
template <class T, class = void>
struct HasProcessMethod : std::false_type {};
template <class T>
struct HasProcessMethod<
T, std::void_t<decltype(mediapipe::Status(
std::declval<std::decay_t<T>>().Process(
std::declval<mediapipe::CalculatorContext*>())))>>
: std::true_type {};
template <class T, class = void>
struct HasNestedItems : std::false_type {};
template <class T>
struct HasNestedItems<
T, std::void_t<decltype(std::declval<std::decay_t<T>>().nested_items())>>
: std::true_type {};
// Helper to construct a tuple of Tag types (see tag.h) from a tuple of ports.
template <class TupleRef>
struct TagTuple {
template <std::size_t J>
struct S {
const const_str tag{std::get<J>(TupleRef::get()).tag_};
};
template <std::size_t... I>
static constexpr auto Make(std::index_sequence<I...> indices) {
return std::make_tuple(mediapipe::api2::internal::tag_build(S<I>{})...);
}
static constexpr auto Make() {
using TupleT = decltype(TupleRef::get());
return Make(internal::tuple_index_sequence<TupleT>());
}
};
// Helper to access a tuple of ports by static tag. Attempts to look up a
// missing tag will not compile.
template <class TupleRef>
struct TaggedAccess {
// This is not functionally necessary (we could do the tag search directly
// on the port tuple), but it gives a more readable error message when the
// static_assert below fails.
static constexpr auto kTagTuple = TagTuple<TupleRef>::Make();
template <class Tag>
static constexpr auto& get(Tag tag) {
constexpr auto i =
internal::tuple_find([tag](auto x) { return x == tag; }, kTagTuple);
static_assert(i < std::tuple_size_v<decltype(kTagTuple)>, "tag not found");
return std::get<i>(TupleRef::get());
}
};
template <class... T>
constexpr auto ExtractNestedItems(std::tuple<T...> tuple) {
return internal::flatten_tuple(internal::map_tuple(
[](auto&& item) {
if constexpr (HasNestedItems<decltype(item)>{}) {
return std::tuple_cat(std::make_tuple(item), item.nested_items());
} else {
return std::make_tuple(item);
}
},
tuple));
}
// Internal contract type. Takes a list of ports or other contract items.
template <typename... T>
class Contract {
public:
constexpr Contract(std::tuple<T...> tuple) : items(tuple) {}
constexpr Contract(T&&... args)
: Contract(std::tuple<T...>{std::move(args)...}) {}
mediapipe::Status GetContract(mediapipe::CalculatorContract* cc) const {
std::vector<mediapipe::Status> statuses;
auto store_status = [&statuses](mediapipe::Status status) {
if (!status.ok()) statuses.push_back(std::move(status));
};
internal::tuple_for_each(
[cc, &store_status](auto&& item) {
store_status(item.AddToContract(cc));
},
all_items);
if (timestamp_change_count() == 0) {
// Default to SetOffset(0);
store_status(TimestampChange::Offset(0).AddToContract(cc));
}
if (statuses.empty()) return {};
if (statuses.size() == 1) return statuses[0];
return tool::CombinedStatus("Multiple errors", statuses);
}
std::tuple<T...> items;
// TODO: when forwarding nested items (e.g. ports), check for conflicts.
decltype(ExtractNestedItems(items)) all_items{ExtractNestedItems(items)};
constexpr auto inputs() const {
return internal::filter_tuple<IsSubclass<InputBase>::pred>(all_items);
}
constexpr auto outputs() const {
return internal::filter_tuple<IsSubclass<OutputBase>::pred>(all_items);
}
constexpr auto side_inputs() const {
return internal::filter_tuple<IsSubclass<SideInputBase>::pred>(all_items);
}
constexpr auto side_outputs() const {
return internal::filter_tuple<IsSubclass<SideOutputBase>::pred>(all_items);
}
constexpr auto timestamp_change_count() const {
return internal::filtered_tuple_indices<IsSubclass<TimestampChange>::pred>(
all_items)
.size();
}
constexpr auto process_items() const {
return internal::filter_tuple<HasProcessMethod>(all_items);
}
};
// Helpers to construct a Contract.
template <typename... T>
constexpr auto MakeContract(T&&... args) {
return Contract<T...>(std::forward<T>(args)...);
}
template <typename... T>
constexpr auto MakeContract(const std::tuple<T...>& tuple) {
return Contract<T...>(tuple);
}
// Helper for accessing the ports of a Contract by static tags.
template <typename C2T, const C2T& c2>
class TaggedContract {
public:
constexpr TaggedContract() = default;
static mediapipe::Status GetContract(mediapipe::CalculatorContract* cc) {
return c2.GetContract(cc);
}
template <class Tuple, Tuple (C2T::*member)() const>
struct GetMember {
static constexpr const auto get() { return (c2.*member)(); }
};
using TaggedInputs =
TaggedAccess<GetMember<decltype(c2.inputs()), &C2T::inputs>>;
using TaggedOutputs =
TaggedAccess<GetMember<decltype(c2.outputs()), &C2T::outputs>>;
using TaggedSideInputs =
TaggedAccess<GetMember<decltype(c2.side_inputs()), &C2T::side_inputs>>;
using TaggedSideOutputs =
TaggedAccess<GetMember<decltype(c2.side_outputs()), &C2T::side_outputs>>;
};
// Support for function-based Process.
template <class T>
struct IsInputPort
: std::bool_constant<std::is_base_of<InputBase, std::decay_t<T>>{} ||
std::is_base_of<SideInputBase, std::decay_t<T>>{}> {};
template <class T>
struct IsOutputPort
: std::bool_constant<std::is_base_of<OutputBase, std::decay_t<T>>{} ||
std::is_base_of<SideOutputBase, std::decay_t<T>>{}> {};
// Helper class that converts a port specification into a function argument.
template <class P>
class PortArg {
public:
PortArg(CalculatorContext* cc, const P& port) : cc_(cc), port_(port) {}
using PayloadT = typename P::PayloadT;
operator const PayloadT&() { return port_(cc_).Get(); }
operator Packet<typename P::value_t>() { return port_(cc_); }
operator PacketBase() { return port_(cc_).packet(); }
private:
CalculatorContext* cc_;
const P& port_;
};
template <class P>
auto MakePortArg(CalculatorContext* cc, const P& port) {
return PortArg<P>(cc, port);
}
// Helper class that takes a function result and sends it into outputs.
template <class... P>
class OutputSender {
public:
OutputSender(P&&... args) : outputs_(args...) {}
OutputSender(std::tuple<P...>&& args) : outputs_(args) {}
template <class R, std::enable_if_t<sizeof...(P) == 1, int> = 0>
mediapipe::Status operator()(CalculatorContext* cc,
mediapipe::StatusOr<R>&& result) {
if (result.ok()) {
return this(cc, result.ValueOrDie());
} else {
return result.status();
}
}
template <class R, std::enable_if_t<sizeof...(P) == 1, int> = 0>
mediapipe::Status operator()(CalculatorContext* cc, R&& result) {
std::get<0>(outputs_)(cc).Send(std::forward<R>(result));
return {};
}
template <class... R>
mediapipe::Status operator()(CalculatorContext* cc,
mediapipe::StatusOr<std::tuple<R...>>&& result) {
if (result.ok()) {
return this(cc, result.ValueOrDie());
} else {
return result.status();
}
}
template <class... R>
mediapipe::Status operator()(CalculatorContext* cc,
std::tuple<R...>&& result) {
static_assert(sizeof...(P) == sizeof...(R), "");
internal::tuple_for_each(
[cc, &result](const auto& port, auto i_const) {
constexpr std::size_t i = decltype(i_const)::value;
port(cc).Send(std::get<i>(result));
},
outputs_);
return {};
}
std::tuple<P...> outputs_;
};
template <class... P>
auto MakeOutputSender(P&&... args) {
return OutputSender<P...>(std::forward<P>(args)...);
}
template <class... P>
auto MakeOutputSender(std::tuple<P...>&& args) {
return OutputSender<P...>(std::forward<std::tuple<P...>>(args));
}
// Contract item that specifies that certain I/O ports are handled by invoking
// a specific function.
template <class F, class... P>
class FunCaller {
public:
constexpr FunCaller(F&& f, P&&... args) : f_(f), args_(args...) {}
auto operator()(CalculatorContext* cc) const {
auto output_sender = MakeOutputSender(outputs());
// tuple_apply gives better error messages than std::apply if the argument
// types don't match.
return output_sender(
cc, internal::tuple_apply(f_, internal::map_tuple(
[cc](const auto& port) {
return MakePortArg(cc, port);
},
inputs())));
}
auto inputs() const { return internal::filter_tuple<IsInputPort>(args_); }
auto outputs() const { return internal::filter_tuple<IsOutputPort>(args_); }
mediapipe::Status AddToContract(CalculatorContract* cc) const { return {}; }
mediapipe::Status Process(CalculatorContext* cc) const { return (*this)(cc); }
constexpr std::tuple<P...> nested_items() const { return args_; }
F f_;
std::tuple<P...> args_;
};
// Helper function to invoke function callers in Process.
// TODO: implement multiple callers for syncsets.
template <class... T>
mediapipe::Status ProcessFnCallers(CalculatorContext* cc,
std::tuple<T...> callers);
inline mediapipe::Status ProcessFnCallers(CalculatorContext* cc, std::tuple<>) {
return mediapipe::InternalError("Process unimplemented");
}
template <class T>
mediapipe::Status ProcessFnCallers(CalculatorContext* cc,
std::tuple<T> callers) {
return std::get<0>(callers).Process(cc);
}
} // namespace internal
// Function used to add a process function to a calculator contract.
template <class F, class... P>
constexpr auto ProcessFn(F&& f, P&&... args) {
return internal::FunCaller<F, P...>(std::forward<F>(f),
std::forward<P>(args)...);
}
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_CONTRACT_H_
+73
View File
@@ -0,0 +1,73 @@
#include "mediapipe/framework/api2/contract.h"
#include <tuple>
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
namespace mediapipe {
namespace api2 {
namespace {
struct ProcessItem {
mediapipe::Status Process(CalculatorContext* cc) { return {}; }
};
struct ItemWithNested {
constexpr auto nested_items() { return std::make_tuple(Input<char>{"FWD"}); }
};
static constexpr auto kTestContract = internal::MakeContract(
Input<int>{"BASE"}, Input<float>::Optional{"SCALE"}, Output<float>{"OUT"},
SideInput<float>::Optional{"BIAS"}, SideOutput<char>{"SIDE"},
ProcessItem{});
static_assert(std::tuple_size_v<decltype(kTestContract.inputs())> == 2, "");
static_assert(std::tuple_size_v<decltype(kTestContract.outputs())> == 1, "");
static_assert(std::tuple_size_v<decltype(kTestContract.side_inputs())> == 1,
"");
static_assert(std::tuple_size_v<decltype(kTestContract.side_outputs())> == 1,
"");
static_assert(internal::HasProcessMethod<ProcessItem>{}, "");
static_assert(!internal::HasProcessMethod<Input<int>>{}, "");
static_assert(std::tuple_size_v<decltype(kTestContract.process_items())> == 1,
"");
static constexpr auto kExtractNested1 = internal::ExtractNestedItems(
std::make_tuple(Input<int>{"BASE"}, Input<float>::Optional{"SCALE"},
Output<float>{"OUT"}));
static_assert(std::tuple_size_v<decltype(kExtractNested1)> == 3, "");
static constexpr auto kExtractNested2 = internal::ExtractNestedItems(
std::make_tuple(Input<int>{"BASE"}, Input<float>::Optional{"SCALE"},
Output<float>{"OUT"}, ItemWithNested{}));
static_assert(std::tuple_size_v<decltype(kExtractNested2)> == 5, "");
using TaggedTestContract =
internal::TaggedContract<decltype(kTestContract), kTestContract>;
static constexpr auto kBASE = MPP_TAG("BASE");
static constexpr auto kSCALE = MPP_TAG("SCALE");
static constexpr auto kBIAS = MPP_TAG("BIAS");
static constexpr auto kOUT = MPP_TAG("OUT");
static constexpr auto kSIDE = MPP_TAG("SIDE");
static_assert(TaggedTestContract::TaggedInputs::get(kBASE).tag_ == kBASE.kStr,
"");
static_assert(TaggedTestContract::TaggedInputs::get(kSCALE).tag_ == kSCALE.kStr,
"");
static_assert(TaggedTestContract::TaggedOutputs::get(kOUT).tag_ == kOUT.kStr,
"");
static_assert(TaggedTestContract::TaggedSideInputs::get(kBIAS).tag_ ==
kBIAS.kStr,
"");
static_assert(TaggedTestContract::TaggedSideOutputs::get(kSIDE).tag_ ==
kSIDE.kStr,
"");
} // namespace
} // namespace api2
} // namespace mediapipe
+9
View File
@@ -0,0 +1,9 @@
#include "mediapipe/framework/api2/node.h"
namespace mediapipe {
namespace api2 {
Node::~Node() {}
} // namespace api2
} // namespace mediapipe
+248
View File
@@ -0,0 +1,248 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_NODE_H_
#define MEDIAPIPE_FRAMEWORK_API2_NODE_H_
#include <functional>
#include <string>
#include "mediapipe/framework/api2/const_str.h"
#include "mediapipe/framework/api2/contract.h"
#include "mediapipe/framework/api2/packet.h"
#include "mediapipe/framework/api2/port.h"
#include "mediapipe/framework/calculator_base.h"
#include "mediapipe/framework/calculator_context.h"
#include "mediapipe/framework/calculator_contract.h"
#include "mediapipe/framework/deps/no_destructor.h"
#include "mediapipe/framework/subgraph.h"
namespace mediapipe {
namespace api2 {
class NodeIntf {};
class Node : public CalculatorBase {
public:
virtual ~Node();
};
} // namespace api2
namespace internal {
template <class T>
class CalculatorBaseFactoryFor<
T,
typename std::enable_if<std::is_base_of<mediapipe::api2::Node, T>{}>::type>
: public CalculatorBaseFactory {
public:
mediapipe::Status GetContract(CalculatorContract* cc) final {
auto status = T::Contract::GetContract(cc);
if (status.ok()) {
status = UpdateContract<T>(cc);
}
return status;
}
std::unique_ptr<CalculatorBase> CreateCalculator(
CalculatorContext* calculator_context) final {
return absl::make_unique<T>();
}
private:
template <typename U>
auto UpdateContract(CalculatorContract* cc)
-> decltype(U::UpdateContract(cc)) {
return U::UpdateContract(cc);
}
template <typename U>
mediapipe::Status UpdateContract(...) {
return {};
}
};
} // namespace internal
namespace api2 {
namespace internal {
// Defining a member of this type causes P to be ODR-used, which forces its
// instantiation if it's a static member of a template.
// Previously we depended on the pointer's value to determine whether the size
// of a character array is 0 or 1, forcing it to be instantiated so the
// compiler can determine the object's layout. But using it as a template
// argument is more compact.
template <auto* P>
struct ForceStaticInstantiation {
#ifdef _MSC_VER
// Just having it as the template argument does not count as a use for
// MSVC.
static constexpr bool Use() { return P != nullptr; }
char force_static[Use()];
#endif // _MSC_VER
};
// Helper template for forcing the definition of a static registration token.
template <typename T>
struct NodeRegistrationStatic {
static NoDestructor<mediapipe::RegistrationToken> registration;
static mediapipe::RegistrationToken Make() {
return mediapipe::CalculatorBaseRegistry::Register(
T::kCalculatorName,
absl::make_unique<mediapipe::internal::CalculatorBaseFactoryFor<T>>);
}
using RequireStatics = ForceStaticInstantiation<&registration>;
};
// Static members of template classes can be defined in the header.
template <typename T>
NoDestructor<mediapipe::RegistrationToken>
NodeRegistrationStatic<T>::registration(NodeRegistrationStatic<T>::Make());
template <typename T>
struct SubgraphRegistrationImpl {
static NoDestructor<mediapipe::RegistrationToken> registration;
static mediapipe::RegistrationToken Make() {
return mediapipe::SubgraphRegistry::Register(T::kCalculatorName,
absl::make_unique<T>);
}
using RequireStatics = ForceStaticInstantiation<&registration>;
};
template <typename T>
NoDestructor<mediapipe::RegistrationToken>
SubgraphRegistrationImpl<T>::registration(
SubgraphRegistrationImpl<T>::Make());
} // namespace internal
// By passing the Impl parameter, registration is done automatically. No need
// to use MEDIAPIPE_NODE_IMPLEMENTATION.
// For backward compatibility, Impl can be omitted; use
// MEDIAPIPE_NODE_IMPLEMENTATION with this.
// TODO: migrate and remove.
template <class Impl = void>
class RegisteredNode;
template <class Impl>
class RegisteredNode : public Node {
private:
// The member below triggers instantiation of the registration static.
// Note that the constructor of calculator subclasses is only invoked through
// the registration token, and so we cannot simply use the static in the
// constructor.
typename internal::NodeRegistrationStatic<Impl>::RequireStatics register_;
};
// No-op version for backwards compatibility.
template <>
class RegisteredNode<void> : public Node {};
template <class Impl>
struct FunctionNode : public RegisteredNode<Impl> {
mediapipe::Status Process(CalculatorContext* cc) override {
return internal::ProcessFnCallers(cc, Impl::kContract.process_items());
}
};
template <class Intf, class Impl = void>
class NodeImpl : public RegisteredNode<Impl>, public Intf {
protected:
// These methods allow accessing a node's ports by tag. This can be useful in
// a few cases, e.g. if the port is not available as a named constant.
// They parallel the corresponding methods on builder nodes.
template <class Tag>
static constexpr auto Out(Tag t) {
return Intf::Contract::TaggedOutputs::get(t);
}
template <class Tag>
static constexpr auto In(Tag t) {
return Intf::Contract::TaggedInputs::get(t);
}
template <class Tag>
static constexpr auto SideOut(Tag t) {
return Intf::Contract::TaggedSideOutputs::get(t);
}
template <class Tag>
static constexpr auto SideIn(Tag t) {
return Intf::Contract::TaggedSideInputs::get(t);
}
// Convenience.
template <class Tag, class CC>
static auto Out(Tag t, CC cc) {
return Out(t)(cc);
}
template <class Tag, class CC>
static auto In(Tag t, CC cc) {
return In(t)(cc);
}
template <class Tag, class CC>
static auto SideOut(Tag t, CC cc) {
return SideOut(t)(cc);
}
template <class Tag, class CC>
static auto SideIn(Tag t, CC cc) {
return SideIn(t)(cc);
}
};
// This macro is used to define the contract, without also giving the
// node a type name. It can be used directly in pure interfaces.
#define MEDIAPIPE_NODE_CONTRACT(...) \
static constexpr auto kContract = \
mediapipe::api2::internal::MakeContract(__VA_ARGS__); \
using Contract = \
typename mediapipe::api2::internal::TaggedContract<decltype(kContract), \
kContract>;
// This macro is used to define the contract and the type name of a node.
// This saves the name of the calculator, making it available to the
// implementation too, and to the registration macro for it. The reason is
// that the name must be available with the contract (so that it can be used
// to build a graph config, for instance); however, it is the implementation
// that needs to be registered.
// TODO: rename to MEDIAPIPE_NODE_DECLARATION?
// TODO: more detailed explanation.
#define MEDIAPIPE_NODE_INTERFACE(name, ...) \
static constexpr char kCalculatorName[] = #name; \
MEDIAPIPE_NODE_CONTRACT(__VA_ARGS__)
// TODO: verify that the subgraph config fully implements the
// declared interface.
template <class Intf, class Impl>
class SubgraphImpl : public Subgraph, public Intf {
private:
typename internal::SubgraphRegistrationImpl<Impl>::RequireStatics register_;
};
// This macro is used to register a calculator that does not use automatic
// registration. Deprecated.
#define MEDIAPIPE_NODE_IMPLEMENTATION(Impl) \
static mediapipe::NoDestructor<mediapipe::RegistrationToken> \
REGISTRY_STATIC_VAR(calculator_registration, __LINE__)( \
mediapipe::CalculatorBaseRegistry::Register( \
Impl::kCalculatorName, \
absl::make_unique< \
mediapipe::internal::CalculatorBaseFactoryFor<Impl>>))
// This macro is used to register a non-split-contract calculator. Deprecated.
#define MEDIAPIPE_REGISTER_NODE(name) REGISTER_CALCULATOR(name)
// This macro is used to define a subgraph that does not use automatic
// registration. Deprecated.
#define MEDIAPIPE_SUBGRAPH_IMPLEMENTATION(Impl) \
static mediapipe::NoDestructor<mediapipe::RegistrationToken> \
REGISTRY_STATIC_VAR(subgraph_registration, \
__LINE__)(mediapipe::SubgraphRegistry::Register( \
Impl::kCalculatorName, absl::make_unique<Impl>))
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_NODE_H_
+527
View File
@@ -0,0 +1,527 @@
#include "mediapipe/framework/api2/node.h"
#include <tuple>
#include <utility>
#include "mediapipe/framework/api2/packet.h"
#include "mediapipe/framework/api2/port.h"
#include "mediapipe/framework/api2/test_contracts.h"
#include "mediapipe/framework/api2/tuple.h"
#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/port/canonical_errors.h"
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
#include "mediapipe/framework/port/parse_text_proto.h"
#include "mediapipe/framework/port/status_matchers.h"
namespace mediapipe {
namespace api2 {
namespace test {
using testing::ElementsAre;
// Returns the packet values for a vector of Packets.
template <typename T>
std::vector<T> PacketValues(const std::vector<mediapipe::Packet>& packets) {
std::vector<T> result;
for (const auto& packet : packets) {
result.push_back(packet.Get<T>());
}
return result;
}
class FooImpl : public NodeImpl<Foo, FooImpl> {
public:
mediapipe::Status Process(CalculatorContext* cc) override {
float bias = kBias(cc).GetOr(0.0);
float scale = kScale(cc).GetOr(1.0);
kOut(cc).Send(*kBase(cc) * scale + bias);
return {};
}
};
class Foo3 : public FunctionNode<Foo3> {
public:
static constexpr Input<int> kBase{"BASE"};
static constexpr Input<float>::Optional kScale{"SCALE"};
static constexpr Output<float> kOut{"OUT"};
static constexpr SideInput<float>::Optional kBias{"BIAS"};
static float foo(int base, Packet<float> bias, Packet<float> scale) {
return base * scale.GetOr(1.0) + bias.GetOr(0.0);
}
// TODO: add support for methods.
MEDIAPIPE_NODE_INTERFACE(Foo3, ProcessFn(&foo, kBase, kBias, kScale, kOut));
};
class Foo4 : public FunctionNode<Foo4> {
public:
static float foo(int base, Packet<float> bias, Packet<float> scale) {
return base * scale.GetOr(1.0) + bias.GetOr(0.0);
}
MEDIAPIPE_NODE_INTERFACE(Foo4, ProcessFn(&foo, Input<int>{"BASE"},
SideInput<float>::Optional{"BIAS"},
Input<float>::Optional{"SCALE"},
Output<float>{"OUT"}));
};
class Foo5 : public FunctionNode<Foo5> {
public:
MEDIAPIPE_NODE_INTERFACE(
Foo5, ProcessFn(
[](int base, Packet<float> bias, Packet<float> scale) {
return base * scale.GetOr(1.0) + bias.GetOr(0.0);
},
Input<int>{"BASE"}, SideInput<float>::Optional{"BIAS"},
Input<float>::Optional{"SCALE"}, Output<float>{"OUT"}));
};
class Foo2Impl : public NodeImpl<Foo2, Foo2Impl> {
public:
mediapipe::Status Process(CalculatorContext* cc) override {
float bias = SideIn(MPP_TAG("BIAS"), cc).GetOr(0.0);
float scale = In(MPP_TAG("SCALE"), cc).GetOr(1.0);
Out(MPP_TAG("OUT"), cc).Send(*In(MPP_TAG("BASE"), cc) * scale + bias);
return {};
}
};
class BarImpl : public NodeImpl<Bar, BarImpl> {
public:
mediapipe::Status Process(CalculatorContext* cc) override {
Packet p = kIn(cc);
kOut(cc).Send(p);
return {};
}
};
class BazImpl : public NodeImpl<Baz> {
public:
static mediapipe::Status UpdateContract(CalculatorContract* cc) { return {}; }
mediapipe::Status Process(CalculatorContext* cc) override {
for (int i = 0; i < kData(cc).Count(); ++i) {
kDataOut(cc)[i].Send(kData(cc)[i]);
}
return {};
}
};
MEDIAPIPE_NODE_IMPLEMENTATION(BazImpl);
class IntForwarderImpl : public NodeImpl<IntForwarder, IntForwarderImpl> {
public:
mediapipe::Status Process(CalculatorContext* cc) override {
kOut(cc).Send(*kIn(cc));
return {};
}
};
class ToFloatImpl : public NodeImpl<ToFloat, ToFloatImpl> {
public:
mediapipe::Status Process(CalculatorContext* cc) override {
kIn(cc).Visit([cc](auto x) { kOut(cc).Send(x); });
return {};
}
};
TEST(NodeTest, GetContract) {
// In the old API, contracts are defined "backwards"; first you fill it in
// with what you have in the graph, then you let the calculator fill it in
// with what it expects, and then you see if they match.
const CalculatorGraphConfig::Node node_config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig::Node>(R"(
calculator: "Foo"
input_stream: "BASE:base"
input_stream: "SCALE:scale"
output_stream: "OUT:out"
)");
mediapipe::CalculatorContract contract;
MP_EXPECT_OK(contract.Initialize(node_config));
MP_EXPECT_OK(Foo::Contract::GetContract(&contract));
MP_EXPECT_OK(ValidatePacketTypeSet(contract.Inputs()));
MP_EXPECT_OK(ValidatePacketTypeSet(contract.Outputs()));
}
TEST(NodeTest, GetContractMulti) {
const CalculatorGraphConfig::Node node_config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig::Node>(R"(
calculator: "Baz"
input_stream: "DATA:0:b"
input_stream: "DATA:1:c"
output_stream: "DATA:0:d"
output_stream: "DATA:1:e"
)");
mediapipe::CalculatorContract contract;
MP_EXPECT_OK(contract.Initialize(node_config));
MP_EXPECT_OK(Baz::Contract::GetContract(&contract));
MP_EXPECT_OK(ValidatePacketTypeSet(contract.Inputs()));
MP_EXPECT_OK(ValidatePacketTypeSet(contract.Outputs()));
}
TEST(NodeTest, CreateByName) {
MP_EXPECT_OK(CalculatorBaseRegistry::CreateByName("Foo"));
}
void RunFooCalculatorInGraph(const std::string& foo_name) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(
absl::Substitute(R"(
input_stream: "base"
input_stream: "scale"
output_stream: "out"
node {
calculator: "$0"
input_stream: "BASE:base"
input_stream: "SCALE:scale"
output_stream: "OUT:out"
}
)",
foo_name));
std::vector<mediapipe::Packet> out_packets;
tool::AddVectorSink("out", &config, &out_packets);
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(graph.StartRun({}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"base", mediapipe::MakePacket<int>(10).At(Timestamp(1))));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"scale", mediapipe::MakePacket<float>(2.0).At(Timestamp(1))));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
EXPECT_THAT(PacketValues<float>(out_packets), testing::ElementsAre(20.0));
}
TEST(NodeTest, RunInGraph) { RunFooCalculatorInGraph("Foo"); }
TEST(NodeTest, RunInGraph3) { RunFooCalculatorInGraph("Foo3"); }
TEST(NodeTest, RunInGraph4) { RunFooCalculatorInGraph("Foo4"); }
TEST(NodeTest, RunInGraph5) { RunFooCalculatorInGraph("Foo5"); }
TEST(NodeTest, OptionalStream) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "base"
input_side_packet: "bias"
output_stream: "out"
node {
calculator: "Foo"
input_stream: "BASE:base"
input_side_packet: "BIAS:bias"
output_stream: "OUT:out"
}
)");
std::vector<mediapipe::Packet> out_packets;
tool::AddVectorSink("out", &config, &out_packets);
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(graph.StartRun({{"bias", mediapipe::MakePacket<float>(30.0)}}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"base", mediapipe::MakePacket<int>(10).At(Timestamp(1))));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
EXPECT_THAT(PacketValues<float>(out_packets), testing::ElementsAre(40.0));
}
TEST(NodeTest, DynamicTypes) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "in"
output_stream: "out"
node {
calculator: "Bar"
input_stream: "IN:in"
output_stream: "OUT:bar"
}
node {
calculator: "IntForwarder"
input_stream: "IN:bar"
output_stream: "OUT:out"
}
)");
std::vector<mediapipe::Packet> out_packets;
tool::AddVectorSink("out", &config, &out_packets);
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(graph.StartRun({}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in", mediapipe::MakePacket<int>(10).At(Timestamp(1))));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
EXPECT_THAT(PacketValues<int>(out_packets), testing::ElementsAre(10));
}
TEST(NodeTest, MultiPort) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "in0"
input_stream: "in1"
output_stream: "out0"
output_stream: "out1"
node {
calculator: "Baz"
input_stream: "DATA:0:in0"
input_stream: "DATA:1:in1"
output_stream: "DATA:0:baz0"
output_stream: "DATA:1:baz1"
}
node {
calculator: "IntForwarder"
input_stream: "IN:baz0"
output_stream: "OUT:out0"
}
node {
calculator: "IntForwarder"
input_stream: "IN:baz1"
output_stream: "OUT:out1"
}
)");
std::vector<mediapipe::Packet> out0_packets;
std::vector<mediapipe::Packet> out1_packets;
tool::AddVectorSink("out0", &config, &out0_packets);
tool::AddVectorSink("out1", &config, &out1_packets);
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(graph.StartRun({}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in0", mediapipe::MakePacket<int>(10).At(Timestamp(1))));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in1", mediapipe::MakePacket<int>(5).At(Timestamp(1))));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in0", mediapipe::MakePacket<int>(15).At(Timestamp(2))));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in1", mediapipe::MakePacket<int>(7).At(Timestamp(2))));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
std::vector<int> out0_values;
std::vector<int> out1_values;
for (auto& packet : out0_packets) {
out0_values.push_back(packet.Get<int>());
}
for (auto& packet : out1_packets) {
out1_values.push_back(packet.Get<int>());
}
EXPECT_EQ(out0_values, (std::vector<int>{10, 15}));
EXPECT_EQ(out1_values, (std::vector<int>{5, 7}));
}
struct SideFallback : public Node {
static constexpr Input<int> kIn{"IN"};
static constexpr Input<int>::SideFallback kFactor{"FACTOR"};
static constexpr Output<int> kOut{"OUT"};
MEDIAPIPE_NODE_CONTRACT(kIn, kFactor, kOut);
mediapipe::Status Process(CalculatorContext* cc) override {
kOut(cc).Send(kIn(cc).Get() * kFactor(cc).Get());
return {};
}
};
MEDIAPIPE_REGISTER_NODE(SideFallback);
TEST(NodeTest, SideFallbackWithStream) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "in"
input_stream: "factor"
output_stream: "out"
node {
calculator: "SideFallback"
input_stream: "IN:in"
input_stream: "FACTOR:factor"
output_stream: "OUT:out"
}
)");
std::vector<int> outputs;
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(
graph.ObserveOutputStream("out", [&outputs](const mediapipe::Packet& p) {
outputs.push_back(p.Get<int>());
return mediapipe::OkStatus();
}));
MP_EXPECT_OK(graph.StartRun({}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in", mediapipe::MakePacket<int>(10).At(Timestamp(0))));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"factor", mediapipe::MakePacket<int>(2).At(Timestamp(0))));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
EXPECT_EQ(outputs, std::vector<int>{20});
}
TEST(NodeTest, SideFallbackWithSide) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "in"
input_side_packet: "factor"
output_stream: "out"
node {
calculator: "SideFallback"
input_stream: "IN:in"
input_side_packet: "FACTOR:factor"
output_stream: "OUT:out"
}
)");
std::vector<int> outputs;
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(
graph.ObserveOutputStream("out", [&outputs](const mediapipe::Packet& p) {
outputs.push_back(p.Get<int>());
return mediapipe::OkStatus();
}));
MP_EXPECT_OK(graph.StartRun({{"factor", mediapipe::MakePacket<int>(2)}}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in", mediapipe::MakePacket<int>(10).At(Timestamp(0))));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
EXPECT_EQ(outputs, std::vector<int>{20});
}
TEST(NodeTest, SideFallbackWithNone) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "in"
output_stream: "out"
node {
calculator: "SideFallback"
input_stream: "IN:in"
output_stream: "OUT:out"
}
)");
std::vector<int> outputs;
mediapipe::CalculatorGraph graph;
auto status = graph.Initialize(config, {});
EXPECT_THAT(status.message(), testing::HasSubstr("must be connected"));
}
TEST(NodeTest, SideFallbackWithBoth) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "in"
input_stream: "factor"
input_side_packet: "factor_side"
output_stream: "out"
node {
calculator: "SideFallback"
input_stream: "IN:in"
input_stream: "FACTOR:factor"
input_side_packet: "FACTOR:factor_side"
output_stream: "OUT:out"
}
)");
std::vector<int> outputs;
mediapipe::CalculatorGraph graph;
auto status = graph.Initialize(config, {});
EXPECT_THAT(status.message(), testing::HasSubstr("not both"));
}
TEST(NodeTest, OneOf) {
CalculatorGraphConfig config =
::mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "in"
output_stream: "out"
node {
calculator: "ToFloat"
input_stream: "IN:in"
output_stream: "OUT:out"
}
)");
std::vector<mediapipe::Packet> out_packets;
tool::AddVectorSink("out", &config, &out_packets);
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(graph.StartRun({}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in", mediapipe::MakePacket<int>(10).At(Timestamp(1))));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"in", mediapipe::MakePacket<float>(5.0).At(Timestamp(2))));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
EXPECT_THAT(PacketValues<float>(out_packets), testing::ElementsAre(10, 5.0));
}
struct DropEvenTimestamps : public Node {
static constexpr Input<AnyType> kIn{"IN"};
static constexpr Output<SameType<kIn>> kOut{"OUT"};
MEDIAPIPE_NODE_CONTRACT(kIn, kOut);
mediapipe::Status Process(CalculatorContext* cc) override {
if (cc->InputTimestamp().Value() % 2) {
kOut(cc).Send(kIn(cc));
}
return {};
}
};
MEDIAPIPE_REGISTER_NODE(DropEvenTimestamps);
struct ListIntPackets : public Node {
static constexpr Input<int>::Multiple kIn{"INT"};
static constexpr Output<std::string> kOut{"STR"};
MEDIAPIPE_NODE_CONTRACT(kIn, kOut);
mediapipe::Status Process(CalculatorContext* cc) override {
std::string result = absl::StrCat(cc->InputTimestamp().DebugString(), ":");
for (int i = 0; i < kIn(cc).Count(); ++i) {
if (kIn(cc)[i].IsEmpty()) {
absl::StrAppend(&result, " empty");
} else {
absl::StrAppend(&result, " ", *kIn(cc)[i]);
}
}
kOut(cc).Send(std::move(result));
return {};
}
};
MEDIAPIPE_REGISTER_NODE(ListIntPackets);
TEST(NodeTest, DefaultTimestampChange0) {
CalculatorGraphConfig config =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "a"
input_stream: "b"
output_stream: "out"
node {
calculator: "DropEvenTimestamps"
input_stream: "IN:a"
output_stream: "OUT:a2"
}
node {
calculator: "IntForwarder"
input_stream: "IN:a2"
output_stream: "OUT:a3"
}
node {
calculator: "ListIntPackets"
input_stream: "INT:0:a3"
input_stream: "INT:1:b"
output_stream: "STR:out"
}
)");
std::vector<mediapipe::Packet> out_packets;
tool::AddVectorSink("out", &config, &out_packets);
mediapipe::CalculatorGraph graph;
MP_EXPECT_OK(graph.Initialize(config, {}));
MP_EXPECT_OK(graph.StartRun({}));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"a", mediapipe::MakePacket<int>(10).At(Timestamp(2))));
MP_EXPECT_OK(graph.AddPacketToInputStream(
"b", mediapipe::MakePacket<int>(10).At(Timestamp(2))));
MP_EXPECT_OK(graph.WaitUntilIdle());
// The packet sent to a should have been dropped, but the timestamp bound
// should be forwarded by IntForwarder, and ListIntPackets should have run.
EXPECT_THAT(PacketValues<std::string>(out_packets),
testing::ElementsAre("2: empty 10"));
MP_EXPECT_OK(graph.CloseAllPacketSources());
MP_EXPECT_OK(graph.WaitUntilDone());
}
} // namespace test
} // namespace api2
} // namespace mediapipe
+15
View File
@@ -0,0 +1,15 @@
#include "mediapipe/framework/api2/packet.h"
namespace mediapipe {
namespace api2 {
PacketBase FromOldPacket(const mediapipe::Packet& op) {
return PacketBase(packet_internal::GetHolderShared(op)).At(op.Timestamp());
}
mediapipe::Packet ToOldPacket(const PacketBase& p) {
return mediapipe::packet_internal::Create(p.payload_, p.timestamp_);
}
} // namespace api2
} // namespace mediapipe
+353
View File
@@ -0,0 +1,353 @@
// This file defines a typed Packet type. It fully interoperates with the older
// mediapipe::Packet; creating an api::Packet<T> that refers to an existing
// Packet (or vice versa) is cheap, just like copying a Packet. Ownership of
// the payload is shared. Consider this as a typed view into the same data.
//
// Conversion is currently done explicitly with the FromOldPacket and
// ToOldPacket functions, but calculator code does not need to concern itself
// with it.
#ifndef MEDIAPIPE_FRAMEWORK_API2_PACKET_H_
#define MEDIAPIPE_FRAMEWORK_API2_PACKET_H_
#include <functional>
#include <type_traits>
#include "mediapipe/framework/api2/tuple.h"
#include "mediapipe/framework/packet.h"
#include "mediapipe/framework/port/logging.h"
namespace mediapipe {
namespace api2 {
using Timestamp = mediapipe::Timestamp;
using HolderBase = mediapipe::packet_internal::HolderBase;
template <typename T>
class Packet;
// Type-erased packet.
class PacketBase {
public:
// Empty.
PacketBase() = default;
// Copy.
PacketBase(const PacketBase&) = default;
PacketBase& operator=(const PacketBase&) = default;
// Move.
PacketBase(PacketBase&&) = default;
PacketBase& operator=(PacketBase&&) = default;
// Get timestamp.
Timestamp timestamp() const { return timestamp_; }
// The original API has a Timestamp method, but it shadows the Timestamp
// type within this class, which is annoying.
// Timestamp Timestamp() const { return timestamp_; }
PacketBase At(Timestamp timestamp) const&;
PacketBase At(Timestamp timestamp) &&;
bool IsEmpty() const { return payload_ == nullptr; }
template <typename T>
Packet<T> As() const;
// Returns the reference to the object of type T if it contains
// one, crashes otherwise.
template <typename T>
const T& Get() const;
// Conversion to old Packet type.
operator mediapipe::Packet() const { return ToOldPacket(*this); }
protected:
explicit PacketBase(std::shared_ptr<HolderBase> payload)
: payload_(std::move(payload)) {}
std::shared_ptr<HolderBase> payload_;
Timestamp timestamp_;
template <typename T>
friend PacketBase PacketBaseAdopting(const T* ptr);
friend PacketBase FromOldPacket(const mediapipe::Packet& op);
friend mediapipe::Packet ToOldPacket(const PacketBase& p);
};
PacketBase FromOldPacket(const mediapipe::Packet& op);
mediapipe::Packet ToOldPacket(const PacketBase& p);
template <typename T>
inline const T& PacketBase::Get() const {
CHECK(payload_);
packet_internal::Holder<T>* typed_payload = payload_->As<T>();
CHECK(typed_payload) << absl::StrCat(
"The Packet stores \"", payload_->DebugTypeName(), "\", but \"",
MediaPipeTypeStringOrDemangled<T>(), "\" was requested.");
return typed_payload->data();
}
// This is used to indicate that the packet could be holding one of a set of
// types, e.g. Packet<OneOf<A, B>>.
//
// A Packet<OneOf<T...>> has an interface similar to std::variant<T...>.
// However, we cannot use std::variant directly, since it requires that the
// contained object be stored in place within the variant.
// Suppose we have a stream that accepts an Image or an ImageFrame, and it
// receives a Packet<ImageFrame>. To present it as a
// std::variant<Image, ImageFrame> we would have to move the ImageFrame into
// the variant (or copy it), but that is not compatible with Packet's existing
// ownership model.
// We could have Get() return a std::variant<std::reference_wrapper<Image>,
// std::reference_wrapper<ImageFrame>>, but that would just make user code more
// convoluted.
//
// TODO: should we just use Packet<T...>?
template <class... T>
struct OneOf {};
namespace internal {
template <class T>
inline void CheckCompatibleType(const HolderBase& holder, internal::Wrap<T>) {
const packet_internal::Holder<T>* typed_payload = holder.As<T>();
CHECK(typed_payload) << absl::StrCat(
"The Packet stores \"", holder.DebugTypeName(), "\", but \"",
MediaPipeTypeStringOrDemangled<T>(), "\" was requested.");
// CHECK(payload_->has_type<T>());
}
template <class... T>
inline void CheckCompatibleType(const HolderBase& holder,
internal::Wrap<OneOf<T...>>) {
bool compatible = (holder.As<T>() || ...);
CHECK(compatible)
<< "The Packet stores \"" << holder.DebugTypeName() << "\", but one of "
<< absl::StrJoin(
{absl::StrCat("\"", MediaPipeTypeStringOrDemangled<T>(), "\"")...},
", ")
<< " was requested.";
}
struct Generic {
Generic() = delete;
};
}; // namespace internal
template <typename T>
inline Packet<T> PacketBase::As() const {
if (!payload_) return Packet<T>().At(timestamp_);
packet_internal::Holder<T>* typed_payload = payload_->As<T>();
internal::CheckCompatibleType(*payload_, internal::Wrap<T>{});
return Packet<T>(payload_).At(timestamp_);
}
template <>
inline Packet<internal::Generic> PacketBase::As<internal::Generic>() const;
template <typename T = internal::Generic>
class Packet;
#if __cplusplus >= 201703L
// Deduction guide to silence -Wctad-maybe-unsupported.
explicit Packet()->Packet<internal::Generic>;
#endif // C++17
template <>
class Packet<internal::Generic> : public PacketBase {
public:
Packet() = default;
Packet<internal::Generic> At(Timestamp timestamp) const&;
Packet<internal::Generic> At(Timestamp timestamp) &&;
protected:
explicit Packet(std::shared_ptr<HolderBase> payload)
: PacketBase(std::move(payload)) {}
friend PacketBase;
};
// Having Packet<T> subclass Packet<Generic> will require hiding some methods
// like As. May be better not to subclass, and allow implicit conversion
// instead.
template <typename T>
class Packet : public Packet<internal::Generic> {
public:
Packet() = default;
Packet<T> At(Timestamp timestamp) const&;
Packet<T> At(Timestamp timestamp) &&;
const T& Get() const {
CHECK(payload_);
packet_internal::Holder<T>* typed_payload = payload_->As<T>();
CHECK(typed_payload);
return typed_payload->data();
}
const T& operator*() const { return Get(); }
template <typename U>
T GetOr(U&& v) const {
return IsEmpty() ? static_cast<T>(absl::forward<U>(v)) : **this;
}
private:
explicit Packet(std::shared_ptr<HolderBase> payload)
: Packet<internal::Generic>(std::move(payload)) {}
friend PacketBase;
template <typename U, typename... Args>
friend Packet<U> MakePacket(Args&&... args);
template <typename U>
friend Packet<U> PacketAdopting(const U* ptr);
template <typename U>
friend Packet<U> PacketAdopting(std::unique_ptr<U> ptr);
};
namespace internal {
template <class... F>
struct Overload : F... {
using F::operator()...;
};
template <class... F>
explicit Overload(F...) -> Overload<F...>;
template <class T, class... U>
struct First {
using type = T;
};
} // namespace internal
template <class... T>
class Packet<OneOf<T...>> : public PacketBase {
public:
Packet() = default;
template <class U>
using AllowedType = std::enable_if_t<(std::is_same_v<U, T> || ...)>;
template <class U, class = AllowedType<U>>
Packet(const Packet<U>& p) : PacketBase(p) {}
template <class U, class = AllowedType<U>>
Packet<OneOf<T...>>& operator=(const Packet<U>& p) {
PacketBase::operator=(p);
return *this;
}
template <class U, class = AllowedType<U>>
Packet(Packet<U>&& p) : PacketBase(std::move(p)) {}
template <class U, class = AllowedType<U>>
Packet<OneOf<T...>>& operator=(Packet<U>&& p) {
PacketBase::operator=(std::move(p));
return *this;
}
Packet<OneOf<T...>> At(Timestamp timestamp) const& {
return Packet<OneOf<T...>>(*this).At(timestamp);
}
Packet<OneOf<T...>> At(Timestamp timestamp) && {
timestamp_ = timestamp;
return std::move(*this);
}
template <class U, class = AllowedType<U>>
const U& Get() const {
CHECK(payload_);
packet_internal::Holder<U>* typed_payload = payload_->As<U>();
CHECK(typed_payload);
return typed_payload->data();
}
template <class U, class = AllowedType<U>>
bool Has() const {
return payload_ && payload_->As<U>();
}
template <class... F>
auto Visit(const F&... args) const {
CHECK(payload_);
auto f = internal::Overload{args...};
using FirstT = typename internal::First<T...>::type;
using ResultType = absl::result_of_t<decltype(f)(const FirstT&)>;
static_assert(
(std::is_same_v<ResultType, absl::result_of_t<decltype(f)(const T&)>> &&
...),
"All visitor overloads must have the same return type");
return Invoke<decltype(f), T...>(f);
}
protected:
explicit Packet(std::shared_ptr<HolderBase> payload)
: PacketBase(std::move(payload)) {}
friend PacketBase;
private:
template <class F, class U>
auto Invoke(const F& f) const {
return f(Get<U>());
}
template <class F, class U, class V, class... W>
auto Invoke(const F& f) const {
return Has<U>() ? f(Get<U>()) : Invoke<F, V, W...>(f);
}
};
template <>
inline Packet<internal::Generic> PacketBase::As<internal::Generic>() const {
if (!payload_) return Packet<internal::Generic>().At(timestamp_);
return Packet<internal::Generic>(payload_).At(timestamp_);
}
inline PacketBase PacketBase::At(Timestamp timestamp) const& {
return PacketBase(*this).At(timestamp);
}
inline PacketBase PacketBase::At(Timestamp timestamp) && {
timestamp_ = timestamp;
return std::move(*this);
}
template <typename T>
inline Packet<T> Packet<T>::At(Timestamp timestamp) const& {
return Packet<T>(*this).At(timestamp);
}
template <typename T>
inline Packet<T> Packet<T>::At(Timestamp timestamp) && {
timestamp_ = timestamp;
return std::move(*this);
}
inline Packet<internal::Generic> Packet<internal::Generic>::At(
Timestamp timestamp) const& {
return Packet<internal::Generic>(*this).At(timestamp);
}
inline Packet<internal::Generic> Packet<internal::Generic>::At(
Timestamp timestamp) && {
timestamp_ = timestamp;
return std::move(*this);
}
template <typename T, typename... Args>
Packet<T> MakePacket(Args&&... args) {
return Packet<T>(std::make_shared<packet_internal::Holder<T>>(
new T(std::forward<Args>(args)...)));
}
template <typename T>
Packet<T> PacketAdopting(const T* ptr) {
return Packet<T>(std::make_shared<packet_internal::Holder<T>>(ptr));
}
template <typename T>
Packet<T> PacketAdopting(std::unique_ptr<T> ptr) {
return Packet<T>(std::make_shared<packet_internal::Holder<T>>(ptr.release()));
}
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_PACKET_H_
+16
View File
@@ -0,0 +1,16 @@
#include "mediapipe/framework/api2/packet.h"
namespace api2 {
namespace {
#if defined(TEST_NO_ASSIGN_WRONG_PACKET_TYPE)
void AssignWrongPacketType() { Packet<int> p = MakePacket<float>(1.0); }
#elif defined(TEST_NO_ASSIGN_GENERIC_TO_SPECIFIC)
void AssignWrongPacketType() {
Packet<> p = MakePacket<float>(1.0);
Packet<int> p2 = p;
}
#endif
} // namespace
}; // namespace api2
+195
View File
@@ -0,0 +1,195 @@
#include "mediapipe/framework/api2/packet.h"
#include "absl/strings/str_cat.h"
#include "mediapipe/framework/port/gtest.h"
namespace mediapipe {
namespace api2 {
namespace {
class LiveCheck {
public:
explicit LiveCheck(bool* alive) : alive_(*alive) { alive_ = true; }
~LiveCheck() { alive_ = false; }
private:
bool& alive_;
};
TEST(PacketTest, PacketBaseDefault) {
PacketBase p;
EXPECT_TRUE(p.IsEmpty());
}
TEST(PacketTest, PacketBaseNonEmpty) {
PacketBase p = PacketAdopting(new int(5));
EXPECT_FALSE(p.IsEmpty());
}
TEST(PacketTest, PacketBaseRefCount) {
bool alive = false;
PacketBase p = PacketAdopting(new LiveCheck(&alive));
EXPECT_TRUE(alive);
PacketBase p2 = p;
p = {};
EXPECT_TRUE(alive);
p2 = {};
EXPECT_FALSE(alive);
}
TEST(PacketTest, PacketBaseSame) {
int* ip = new int(5);
PacketBase p = PacketAdopting(ip);
PacketBase p2 = p;
EXPECT_EQ(&p2.Get<int>(), ip);
}
TEST(PacketTest, PacketNonEmpty) {
Packet<int> p = MakePacket<int>(5);
EXPECT_FALSE(p.IsEmpty());
}
TEST(PacketTest, Get) {
Packet<int> p = MakePacket<int>(5);
EXPECT_EQ(*p, 5);
EXPECT_EQ(p.Get(), 5);
}
TEST(PacketTest, GetOr) {
Packet<int> p_0 = MakePacket<int>(0);
Packet<int> p_5 = MakePacket<int>(5);
Packet<int> p_empty;
EXPECT_EQ(p_0.GetOr(1), 0);
EXPECT_EQ(p_5.GetOr(1), 5);
EXPECT_EQ(p_empty.GetOr(1), 1);
}
// This show how GetOr can be used with a lambda that is only called if the "or"
// case is needed. Can be useful when generating the fallback value is
// expensive.
// We could also add an overload to GetOr for types which are not convertible to
// T, but are callable and return T.
// TODO: consider adding it to make things easier.
template <typename F>
struct Lazy {
F f;
using ValueT = decltype(f());
Lazy(F fun) : f(fun) {}
operator ValueT() const { return f(); }
};
template <typename F>
Lazy(F f) -> Lazy<F>;
TEST(PacketTest, GetOrLazy) {
int expensive_call_count = 0;
auto expensive_string_generation = [&expensive_call_count] {
++expensive_call_count;
return "an expensive fallback";
};
auto p_hello = MakePacket<std::string>("hello");
Packet<std::string> p_empty;
EXPECT_EQ(p_hello.GetOr(Lazy(expensive_string_generation)), "hello");
EXPECT_EQ(expensive_call_count, 0);
EXPECT_EQ(p_empty.GetOr(Lazy(expensive_string_generation)),
"an expensive fallback");
EXPECT_EQ(expensive_call_count, 1);
}
TEST(PacketTest, OneOf) {
Packet<OneOf<std::string, int>> p = MakePacket<std::string>("hi");
EXPECT_TRUE(p.Has<std::string>());
EXPECT_FALSE(p.Has<int>());
EXPECT_EQ(p.Get<std::string>(), "hi");
std::string out =
p.Visit([](std::string s) { return absl::StrCat("string: ", s); },
[](int i) { return absl::StrCat("int: ", i); });
EXPECT_EQ(out, "string: hi");
p = MakePacket<int>(2);
EXPECT_FALSE(p.Has<std::string>());
EXPECT_TRUE(p.Has<int>());
EXPECT_EQ(p.Get<int>(), 2);
out = p.Visit([](std::string s) { return absl::StrCat("string: ", s); },
[](int i) { return absl::StrCat("int: ", i); });
EXPECT_EQ(out, "int: 2");
}
TEST(PacketTest, PacketRefCount) {
bool alive = false;
auto p = MakePacket<LiveCheck>(&alive);
EXPECT_TRUE(alive);
auto p2 = p;
p = {};
EXPECT_TRUE(alive);
p2 = {};
EXPECT_FALSE(alive);
}
TEST(PacketTest, PacketTimestamp) {
auto p = MakePacket<int>(5);
EXPECT_EQ(p.timestamp(), Timestamp::Unset());
auto p2 = p.At(Timestamp(1));
EXPECT_EQ(p.timestamp(), Timestamp::Unset());
EXPECT_EQ(p2.timestamp(), Timestamp(1));
auto p3 = std::move(p2).At(Timestamp(3));
EXPECT_EQ(p3.timestamp(), Timestamp(3));
}
TEST(PacketTest, PacketFromGeneric) {
Packet<> pb = PacketAdopting(new int(5));
Packet<int> p = pb.As<int>();
EXPECT_EQ(p.Get(), 5);
}
TEST(PacketTest, PacketAdopting) {
Packet<float> p = PacketAdopting(new float(1.0));
EXPECT_FALSE(p.IsEmpty());
}
TEST(PacketTest, PacketGeneric) {
// With C++17, Packet<> could be written simply as Packet.
Packet<> p = PacketAdopting(new float(1.0));
EXPECT_FALSE(p.IsEmpty());
}
TEST(PacketTest, PacketGenericTimestamp) {
Packet<> p = MakePacket<int>(5);
EXPECT_EQ(p.timestamp(), mediapipe::Timestamp::Unset());
auto p2 = p.At(Timestamp(1));
EXPECT_EQ(p.timestamp(), mediapipe::Timestamp::Unset());
EXPECT_EQ(p2.timestamp(), Timestamp(1));
auto p3 = std::move(p2).At(Timestamp(3));
EXPECT_EQ(p3.timestamp(), Timestamp(3));
}
TEST(PacketTest, FromOldPacket) {
mediapipe::Packet op = mediapipe::MakePacket<int>(7);
Packet<int> p = FromOldPacket(op).As<int>();
EXPECT_EQ(p.Get(), 7);
}
TEST(PacketTest, ToOldPacket) {
auto p = MakePacket<int>(7);
mediapipe::Packet op = ToOldPacket(p);
EXPECT_EQ(op.Get<int>(), 7);
}
TEST(PacketTest, OldRefCounting) {
bool alive = false;
PacketBase p = PacketAdopting(new LiveCheck(&alive));
EXPECT_TRUE(alive);
mediapipe::Packet op = ToOldPacket(p);
p = {};
EXPECT_TRUE(alive);
PacketBase p2 = FromOldPacket(op);
op = {};
EXPECT_TRUE(alive);
p2 = {};
EXPECT_FALSE(alive);
}
} // namespace
} // namespace api2
} // namespace mediapipe
+628
View File
@@ -0,0 +1,628 @@
// This file defines an API to define a node's ports in a concise, type-safe
// way. Example usage in a node:
//
// static constexpr Input<int> kBase("IN");
// static constexpr Output<float> kOut("OUT");
// static constexpr SideInput<float>::Optional kDelta("DELTA");
// static constexpr SideOutput<float> kForward("FORWARD");
//
// Pass a CalculatorContext to a port to access the inputs or outputs in the
// context. For example:
//
// kBase(cc) yields an InputShardAccess<int>
// kOut(cc) yields an OutputShardAccess<float>
// kDelta(cc) yields an InputSidePacketAccess<float>
// kForward(cc) yields an OutputSidePacketAccess<float>
#ifndef MEDIAPIPE_FRAMEWORK_API2_PORT_H_
#define MEDIAPIPE_FRAMEWORK_API2_PORT_H_
#include <type_traits>
#include <utility>
#include "absl/strings/str_cat.h"
#include "mediapipe/framework/api2/const_str.h"
#include "mediapipe/framework/api2/packet.h"
#include "mediapipe/framework/calculator_context.h"
#include "mediapipe/framework/calculator_contract.h"
#include "mediapipe/framework/output_side_packet.h"
#include "mediapipe/framework/port/logging.h"
namespace mediapipe {
namespace api2 {
// typeid is not constexpr, but a pointer to this is.
template <typename T>
size_t get_type_hash() {
return typeid(T).hash_code();
}
using type_id_fptr = size_t (*)();
// This is a base class for various types of port. It is not meant to be used
// directly by node code.
class PortBase {
public:
constexpr PortBase(std::size_t tag_size, const char* tag,
type_id_fptr get_type_id, bool optional, bool multiple)
: tag_(tag_size, tag),
optional_(optional),
multiple_(multiple),
type_id_getter_(get_type_id) {}
bool IsOptional() const { return optional_; }
bool IsMultiple() const { return multiple_; }
const char* Tag() const { return tag_.data(); }
size_t type_id() const { return type_id_getter_(); }
const const_str tag_;
const bool optional_;
const bool multiple_;
protected:
type_id_fptr type_id_getter_;
};
// These four base classes are used to distinguish between ports of different
// kinds. They are not meant to be used directly by node code.
class InputBase : public PortBase {
using PortBase::PortBase;
};
class OutputBase : public PortBase {
using PortBase::PortBase;
};
class SideInputBase : public PortBase {
using PortBase::PortBase;
};
class SideOutputBase : public PortBase {
using PortBase::PortBase;
};
struct NoneType {
private:
NoneType() = delete;
};
struct DynamicType {};
struct AnyType : public DynamicType {};
template <auto& P>
class SameType : public DynamicType {
public:
static constexpr const decltype(P)& kPort = P;
};
class PacketTypeAccess;
class PacketTypeAccessFallback;
template <typename T>
class InputShardAccess;
template <typename T>
class OutputShardAccess;
template <typename T>
class InputSidePacketAccess;
template <typename T>
class OutputSidePacketAccess;
template <typename T>
class InputShardOrSideAccess;
namespace internal {
// Forward declaration for AddToContract friend.
template <typename...>
class Contract;
template <class CC>
auto GetCollection(CC* cc, const InputBase& port) -> decltype(cc->Inputs()) {
return cc->Inputs();
}
template <class CC>
auto GetCollection(CC* cc, const SideInputBase& port)
-> decltype(cc->InputSidePackets()) {
return cc->InputSidePackets();
}
template <class CC>
auto GetCollection(CC* cc, const OutputBase& port) -> decltype(cc->Outputs()) {
return cc->Outputs();
}
template <class CC>
auto GetCollection(CC* cc, const SideOutputBase& port)
-> decltype(cc->OutputSidePackets()) {
return cc->OutputSidePackets();
}
template <class Collection>
auto GetOrNull(Collection& collection, const std::string& tag, int index)
-> decltype(&collection.Get(std::declval<CollectionItemId>())) {
CollectionItemId id = collection.GetId(tag, index);
return id.IsValid() ? &collection.Get(id) : nullptr;
}
template <class T>
struct IsOneOf : std::false_type {};
template <class... T>
struct IsOneOf<OneOf<T...>> : std::true_type {};
template <typename T, typename std::enable_if<
!std::is_base_of<DynamicType, T>{} && !IsOneOf<T>{},
int>::type = 0>
inline void SetType(CalculatorContract* cc, PacketType& pt) {
pt.Set<T>();
}
template <typename T, typename std::enable_if<std::is_base_of<DynamicType, T>{},
int>::type = 0>
inline void SetType(CalculatorContract* cc, PacketType& pt) {
pt.SetSameAs(&internal::GetCollection(cc, T::kPort).Tag(T::kPort.Tag()));
}
template <>
inline void SetType<AnyType>(CalculatorContract* cc, PacketType& pt) {
pt.SetAny();
}
template <>
inline void SetType<NoneType>(CalculatorContract* cc, PacketType& pt) {
// This is used for header-only streams. Should it be removed?
pt.SetNone();
}
template <typename T, typename std::enable_if<IsOneOf<T>{}, int>::type = 0>
inline void SetType(CalculatorContract* cc, PacketType& pt) {
pt.SetAny();
}
template <typename ValueT>
InputShardAccess<ValueT> SinglePortAccess(mediapipe::CalculatorContext* cc,
const InputStreamShard* stream) {
return InputShardAccess<ValueT>(*cc, stream);
}
template <typename ValueT>
OutputShardAccess<ValueT> SinglePortAccess(mediapipe::CalculatorContext* cc,
OutputStreamShard* stream) {
return OutputShardAccess<ValueT>(*cc, stream);
}
template <typename ValueT>
InputSidePacketAccess<ValueT> SinglePortAccess(
mediapipe::CalculatorContext* cc, const mediapipe::Packet* packet) {
return InputSidePacketAccess<ValueT>(packet);
}
template <typename ValueT>
OutputSidePacketAccess<ValueT> SinglePortAccess(
mediapipe::CalculatorContext* cc, OutputSidePacket* osp) {
return OutputSidePacketAccess<ValueT>(osp);
}
template <typename ValueT>
InputShardOrSideAccess<ValueT> SinglePortAccess(
mediapipe::CalculatorContext* cc, const InputStreamShard* stream,
const mediapipe::Packet* packet) {
return InputShardOrSideAccess<ValueT>(*cc, stream, packet);
}
template <typename ValueT>
PacketTypeAccess SinglePortAccess(mediapipe::CalculatorContract* cc,
PacketType* pt);
template <typename ValueT>
PacketTypeAccessFallback SinglePortAccess(mediapipe::CalculatorContract* cc,
PacketType* pt, bool is_stream);
template <typename ValueT, typename PortT, class CC>
auto AccessPort(std::false_type, const PortT& port, CC* cc) {
auto& collection = GetCollection(cc, port);
return SinglePortAccess<ValueT>(
cc, internal::GetOrNull(collection, port.Tag(), 0));
}
template <typename ValueT, typename X, class CC>
class MultiplePortAccess {
public:
MultiplePortAccess(CC* cc, X* first, int count)
: cc_(cc), first_(first), count_(count) {}
// TODO: maybe this should be size(), like in a standard C++
// container?
int Count() { return count_; }
auto operator[](int pos) {
CHECK_GE(pos, 0);
CHECK_LT(pos, count_);
return SinglePortAccess<ValueT>(cc_, &first_[pos]);
}
// TODO: add begin/end.
private:
CC* cc_;
X* first_;
int count_;
};
template <typename ValueT, typename PortT, class CC>
auto AccessPort(std::true_type, const PortT& port, CC* cc) {
auto& collection = GetCollection(cc, port);
auto* first = internal::GetOrNull(collection, port.Tag(), 0);
using EntryT = typename std::remove_pointer<decltype(first)>::type;
return MultiplePortAccess<ValueT, EntryT, CC>(
cc, first, collection.NumEntries(port.Tag()));
}
template <class Base>
struct SideBase;
template <>
struct SideBase<InputBase> {
using type = SideInputBase;
};
} // namespace internal
// TODO: maybe return a PacketBase instead of a Packet<internal::Generic>?
template <typename T, typename std::enable_if<
!std::is_base_of<DynamicType, T>{}, int>::type = 0>
auto ActualValueT(T) -> T;
auto ActualValueT(DynamicType) -> internal::Generic;
template <typename Base, typename ValueT, bool IsOptional = false,
bool IsMultiple = false>
class SideFallbackT;
// This template is used to define a port. Nodes should use it through one
// of the aliases below (Input, Output, SideInput, SideOutput).
template <typename Base, typename ValueT, bool IsOptionalV = false,
bool IsMultipleV = false>
class PortCommon : public Base {
public:
using value_t = ValueT;
static constexpr bool kOptional = IsOptionalV;
static constexpr bool kMultiple = IsMultipleV;
using Optional = PortCommon<Base, ValueT, true, IsMultipleV>;
using Multiple = PortCommon<Base, ValueT, IsOptionalV, true>;
using SideFallback = SideFallbackT<Base, ValueT, IsOptionalV, IsMultipleV>;
template <std::size_t N>
explicit constexpr PortCommon(const char (&tag)[N])
: Base(N, tag, &get_type_hash<ValueT>, IsOptionalV, IsMultipleV) {}
using PayloadT = decltype(ActualValueT(std::declval<ValueT>()));
auto operator()(CalculatorContext* cc) const {
return internal::AccessPort<PayloadT>(
std::integral_constant<bool, IsMultipleV>{}, *this, cc);
}
auto operator()(CalculatorContract* cc) const {
return internal::AccessPort<PayloadT>(
std::integral_constant<bool, IsMultipleV>{}, *this, cc);
}
private:
mediapipe::Status AddToContract(CalculatorContract* cc) const {
if (kMultiple) {
AddMultiple(cc);
} else {
auto& pt = internal::GetCollection(cc, *this).Tag(this->Tag());
internal::SetType<value_t>(cc, pt);
if (kOptional) {
pt.Optional();
}
}
return {};
}
void AddMultiple(CalculatorContract* cc) const {
auto& collection = internal::GetCollection(cc, *this);
int count = collection.NumEntries(this->Tag());
for (int i = 0; i < count; ++i) {
internal::SetType<value_t>(cc, collection.Get(this->Tag(), i));
}
}
template <typename...>
friend class internal::Contract;
template <typename B, typename VT, bool, bool>
friend class mediapipe::api2::SideFallbackT;
};
// Use one of these templates to define a port in node code.
template <typename T = internal::Generic>
using Input = PortCommon<InputBase, T>;
template <typename T = internal::Generic>
using Output = PortCommon<OutputBase, T>;
template <typename T = internal::Generic>
using SideInput = PortCommon<SideInputBase, T>;
template <typename T = internal::Generic>
using SideOutput = PortCommon<SideOutputBase, T>;
template <typename Base, typename ValueT, bool IsOptionalV, bool IsMultipleV>
class SideFallbackT : public Base {
public:
using value_t = ValueT;
static constexpr bool kOptional = IsOptionalV;
static constexpr bool kMultiple = IsMultipleV;
using Optional = SideFallbackT<Base, ValueT, true, IsMultipleV>;
using PayloadT = decltype(ActualValueT(std::declval<ValueT>()));
const char* Tag() const { return stream_port.Tag(); }
auto operator()(CalculatorContract* cc) const {
bool is_stream = true;
auto& stream_collection = internal::GetCollection(cc, stream_port);
auto* packet_type = internal::GetOrNull(stream_collection, Tag(), 0);
if (packet_type == nullptr) {
auto& side_collection = internal::GetCollection(cc, side_port);
packet_type = internal::GetOrNull(side_collection, Tag(), 0);
is_stream = false;
}
return internal::SinglePortAccess<PayloadT>(cc, packet_type, is_stream);
}
auto operator()(CalculatorContext* cc) const {
auto& stream_collection = internal::GetCollection(cc, stream_port);
auto& side_collection = internal::GetCollection(cc, side_port);
return internal::SinglePortAccess<PayloadT>(
cc, internal::GetOrNull(stream_collection, Tag(), 0),
internal::GetOrNull(side_collection, Tag(), 0));
}
template <std::size_t N>
explicit constexpr SideFallbackT(const char (&tag)[N])
: Base(N, tag, &get_type_hash<ValueT>, IsOptionalV, IsMultipleV),
stream_port(tag),
side_port(tag) {}
protected:
mediapipe::Status AddToContract(CalculatorContract* cc) const {
stream_port.AddToContract(cc);
side_port.AddToContract(cc);
int connected_count =
stream_port(cc).IsConnected() + side_port(cc).IsConnected();
if (connected_count > 1)
return mediapipe::InvalidArgumentError(absl::StrCat(
Tag(),
" can be connected as a stream or as a side packet, but not both"));
if (!IsOptionalV && connected_count == 0)
return mediapipe::InvalidArgumentError(
absl::StrCat(Tag(), " must be connected"));
return {};
}
using StreamPort = PortCommon<Base, ValueT, true, IsMultipleV>;
using SidePort = PortCommon<typename internal::SideBase<Base>::type, ValueT,
true, IsMultipleV>;
StreamPort stream_port;
SidePort side_port;
template <typename...>
friend class internal::Contract;
};
// An OutputShardAccess is returned when accessing an output stream within a
// CalculatorContext (e.g. kOut(cc)), and provides a type-safe interface to
// OutputStreamShard. Like that class, this class will not be usually named in
// calculator code, but used as a temporary object (e.g. kOut(cc).Send(...)).
class OutputShardAccessBase {
public:
OutputShardAccessBase(const CalculatorContext& cc, OutputStreamShard* output)
: context_(cc), output_(output) {}
void SetNextTimestampBound(Timestamp timestamp) {
if (output_) output_->SetNextTimestampBound(timestamp);
}
bool IsClosed() { return output_ ? output_->IsClosed() : true; }
void Close() {
if (output_) output_->Close();
}
bool IsConnected() { return output_ != nullptr; }
protected:
const CalculatorContext& context_;
OutputStreamShard* output_;
};
template <typename T>
class OutputShardAccess : public OutputShardAccessBase {
public:
void Send(Packet<T>&& packet) {
if (output_) output_->AddPacket(ToOldPacket(std::move(packet)));
}
void Send(const Packet<T>& packet) {
if (output_) output_->AddPacket(ToOldPacket(packet));
}
void Send(const T& payload, Timestamp time) {
Send(api2::MakePacket<T>(payload).At(time));
}
void Send(const T& payload) { Send(payload, context_.InputTimestamp()); }
void Send(std::unique_ptr<T> payload, Timestamp time) {
Send(api2::PacketAdopting(std::move(payload)).At(time));
}
void Send(std::unique_ptr<T> payload) {
Send(std::move(payload), context_.InputTimestamp());
}
private:
OutputShardAccess(const CalculatorContext& cc, OutputStreamShard* output)
: OutputShardAccessBase(cc, output) {}
friend OutputShardAccess<T> internal::SinglePortAccess<T>(
mediapipe::CalculatorContext*, OutputStreamShard*);
};
template <>
class OutputShardAccess<internal::Generic> : public OutputShardAccessBase {
public:
void Send(PacketBase&& packet) {
if (output_) output_->AddPacket(ToOldPacket(std::move(packet)));
}
void Send(const PacketBase& packet) {
if (output_) output_->AddPacket(ToOldPacket(packet));
}
void SetHeader(const PacketBase& header) {
if (output_) output_->SetHeader(ToOldPacket(header));
}
private:
OutputShardAccess(const CalculatorContext& cc, OutputStreamShard* output)
: OutputShardAccessBase(cc, output) {}
friend OutputShardAccess<internal::Generic>
internal::SinglePortAccess<internal::Generic>(mediapipe::CalculatorContext*,
OutputStreamShard*);
};
// Equivalent of OutputShardAccess, but for side packets.
template <typename T>
class OutputSidePacketAccess {
public:
void Set(Packet<T> packet) {
if (output_) output_->Set(ToOldPacket(std::move(packet)));
}
void Set(const T& payload) { Set(MakePacket<T>(payload)); }
private:
OutputSidePacketAccess(OutputSidePacket* output) : output_(output) {}
OutputSidePacket* output_;
friend OutputSidePacketAccess<T> internal::SinglePortAccess<T>(
mediapipe::CalculatorContext*, OutputSidePacket*);
};
template <typename T>
class InputShardAccess : public Packet<T> {
public:
const PacketBase& packet() const& { return *this; }
// Since InputShardAccess is currently created as a temporary, this avoids
// easy mistakes with dangling references.
PacketBase packet() const&& { return *this; }
bool IsDone() const { return stream_->IsDone(); }
bool IsConnected() { return stream_ != nullptr; }
PacketBase Header() const { return FromOldPacket(stream_->Header()); }
private:
InputShardAccess(const CalculatorContext&, const InputStreamShard* stream)
: Packet<T>(stream ? FromOldPacket(stream->Value()).template As<T>()
: Packet<T>()),
stream_(stream) {}
const InputStreamShard* stream_;
friend InputShardAccess<T> internal::SinglePortAccess<T>(
mediapipe::CalculatorContext*, const InputStreamShard*);
};
template <typename T>
class InputSidePacketAccess : public Packet<T> {
public:
const PacketBase& packet() const& { return *this; }
PacketBase packet() const&& { return *this; }
bool IsConnected() { return connected_; }
private:
InputSidePacketAccess(const mediapipe::Packet* packet)
: Packet<T>(packet ? FromOldPacket(*packet).template As<T>()
: Packet<T>()),
connected_(packet != nullptr) {}
bool connected_;
friend InputSidePacketAccess<T> internal::SinglePortAccess<T>(
mediapipe::CalculatorContext*, const mediapipe::Packet*);
};
template <typename T>
class InputShardOrSideAccess : public Packet<T> {
public:
const PacketBase& packet() const& { return *this; }
PacketBase packet() const&& { return *this; }
bool IsDone() const { return stream_->IsDone(); }
bool IsConnected() { return connected_; }
bool IsStream() { return stream_ != nullptr; }
PacketBase Header() const { return FromOldPacket(stream_->Header()); }
private:
InputShardOrSideAccess(const CalculatorContext&,
const InputStreamShard* stream,
const mediapipe::Packet* packet)
: Packet<T>(stream ? FromOldPacket(stream->Value()).template As<T>()
: packet ? FromOldPacket(*packet).template As<T>()
: Packet<T>()),
stream_(stream),
connected_(stream_ != nullptr || packet != nullptr) {}
const InputStreamShard* stream_;
bool connected_;
friend InputShardOrSideAccess<T> internal::SinglePortAccess<T>(
mediapipe::CalculatorContext*, const InputStreamShard*,
const mediapipe::Packet*);
};
class PacketTypeAccess {
public:
bool IsConnected() { return packet_type_ != nullptr; }
protected:
PacketTypeAccess(PacketType* pt) : packet_type_(pt) {}
PacketType* packet_type_;
template <typename T>
friend PacketTypeAccess internal::SinglePortAccess(
mediapipe::CalculatorContract*, PacketType*);
};
class PacketTypeAccessFallback : public PacketTypeAccess {
public:
bool IsStream() { return is_stream_; }
private:
PacketTypeAccessFallback(PacketType* pt, bool is_stream)
: PacketTypeAccess(pt), is_stream_(is_stream) {}
bool is_stream_;
template <typename T>
friend PacketTypeAccessFallback internal::SinglePortAccess(
mediapipe::CalculatorContract*, PacketType*, bool);
};
namespace internal {
template <typename ValueT>
PacketTypeAccess SinglePortAccess(mediapipe::CalculatorContract* cc,
PacketType* pt) {
return PacketTypeAccess(pt);
}
template <typename ValueT>
PacketTypeAccessFallback SinglePortAccess(mediapipe::CalculatorContract* cc,
PacketType* pt, bool is_stream) {
return PacketTypeAccessFallback(pt, is_stream);
}
} // namespace internal
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_PORT_H_
+26
View File
@@ -0,0 +1,26 @@
#include "mediapipe/framework/api2/port.h"
#include "mediapipe/framework/port/gtest.h"
namespace mediapipe {
namespace api2 {
namespace {
TEST(PortTest, IntInput) {
static constexpr auto port = Input<int>("FOO");
EXPECT_EQ(port.type_id(), typeid(int).hash_code());
}
TEST(PortTest, OptionalInput) {
static constexpr auto port = Input<float>::Optional("BAR");
EXPECT_TRUE(port.IsOptional());
}
TEST(PortTest, Tag) {
static constexpr auto port = Input<int>("FOO");
EXPECT_EQ(std::string(port.Tag()), "FOO");
}
} // namespace
} // namespace api2
} // namespace mediapipe
+157
View File
@@ -0,0 +1,157 @@
#include "mediapipe/framework/api2/builder.h"
#include "mediapipe/framework/api2/node.h"
#include "mediapipe/framework/api2/packet.h"
#include "mediapipe/framework/api2/port.h"
#include "mediapipe/framework/api2/test_contracts.h"
#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/deps/message_matchers.h"
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
#include "mediapipe/framework/port/parse_text_proto.h"
#include "mediapipe/framework/port/status_matchers.h"
#include "mediapipe/framework/tool/subgraph_expansion.h"
namespace mediapipe {
namespace api2 {
namespace test {
class FooBarImpl1 : public SubgraphImpl<FooBar1, FooBarImpl1> {
public:
mediapipe::StatusOr<CalculatorGraphConfig> GetConfig(
const SubgraphOptions& /*options*/) {
builder::Graph graph;
auto& foo = graph.AddNode("Foo");
auto& bar = graph.AddNode("Bar");
graph.In(kIn) >> foo.In("BASE");
foo.Out("OUT") >> bar.In("IN");
bar.Out("OUT") >> graph.Out(kOut);
return graph.GetConfig();
}
};
class FooBarImpl2 : public SubgraphImpl<FooBar2, FooBarImpl2> {
public:
mediapipe::StatusOr<CalculatorGraphConfig> GetConfig(
const SubgraphOptions& /*options*/) {
builder::Graph graph;
auto& foo = graph.AddNode<Foo>();
auto& bar = graph.AddNode<Bar>();
graph.In(kIn) >> foo.In(MPP_TAG("BASE"));
foo.Out(MPP_TAG("OUT")) >> bar.In(MPP_TAG("IN"));
bar.Out(MPP_TAG("OUT")) >> graph.Out(kOut);
return graph.GetConfig();
}
};
TEST(SubgraphTest, SubgraphConfig) {
CalculatorGraphConfig subgraph = FooBarImpl1().GetConfig({}).ValueOrDie();
const CalculatorGraphConfig expected_graph =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "IN:__stream_0"
output_stream: "OUT:__stream_2"
node {
calculator: "Foo"
input_stream: "BASE:__stream_0"
output_stream: "OUT:__stream_1"
}
node {
calculator: "Bar"
input_stream: "IN:__stream_1"
output_stream: "OUT:__stream_2"
}
)");
EXPECT_THAT(subgraph, EqualsProto(expected_graph));
}
TEST(SubgraphTest, TypedSubgraphConfig) {
CalculatorGraphConfig subgraph = FooBarImpl2().GetConfig({}).ValueOrDie();
const CalculatorGraphConfig expected_graph =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "IN:__stream_0"
output_stream: "OUT:__stream_2"
node {
calculator: "Foo"
input_stream: "BASE:__stream_0"
output_stream: "OUT:__stream_1"
}
node {
calculator: "Bar"
input_stream: "IN:__stream_1"
output_stream: "OUT:__stream_2"
}
)");
EXPECT_THAT(subgraph, EqualsProto(expected_graph));
}
TEST(SubgraphTest, ProtoApiConfig) {
CalculatorGraphConfig graph;
graph.add_input_stream("IN:__stream_0");
graph.add_output_stream("OUT:__stream_2");
auto* foo = graph.add_node();
foo->set_calculator("Foo");
foo->add_input_stream("BASE:__stream_0");
foo->add_output_stream("OUT:__stream_1");
auto* bar = graph.add_node();
bar->set_calculator("Bar");
bar->add_input_stream("IN:__stream_1");
bar->add_output_stream("OUT:__stream_2");
const CalculatorGraphConfig expected_graph =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
input_stream: "IN:__stream_0"
output_stream: "OUT:__stream_2"
node {
calculator: "Foo"
input_stream: "BASE:__stream_0"
output_stream: "OUT:__stream_1"
}
node {
calculator: "Bar"
input_stream: "IN:__stream_1"
output_stream: "OUT:__stream_2"
}
)");
EXPECT_THAT(graph, EqualsProto(expected_graph));
}
TEST(SubgraphTest, ExpandSubgraphs) {
CalculatorGraphConfig supergraph =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
node {
name: "simple_source"
calculator: "SomeSourceCalculator"
output_stream: "foo"
}
node {
calculator: "FooBar"
input_stream: "IN:foo"
output_stream: "OUT:output"
}
)");
const CalculatorGraphConfig expected_graph =
mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
node {
name: "simple_source"
calculator: "SomeSourceCalculator"
output_stream: "foo"
}
node {
name: "foobar__Foo"
calculator: "Foo"
input_stream: "BASE:foo"
output_stream: "OUT:foobar____stream_1"
}
node {
name: "foobar__Bar"
calculator: "Bar"
input_stream: "IN:foobar____stream_1"
output_stream: "OUT:output"
}
)");
MP_EXPECT_OK(tool::ExpandSubgraphs(&supergraph));
EXPECT_THAT(supergraph, EqualsProto(expected_graph));
}
} // namespace test
} // namespace api2
} // namespace mediapipe
+72
View File
@@ -0,0 +1,72 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_TAG_H_
#define MEDIAPIPE_FRAMEWORK_API2_TAG_H_
#include <utility>
#include "mediapipe/framework/api2/const_str.h"
namespace mediapipe {
namespace api2 {
// This template is used to define a separate type for each tag.
// This makes it possible to obtain results of different types depending on
// the tag. See MPP_TAG below for usage examples.
template <char... C>
struct Tag {
static constexpr char const kChars[sizeof...(C) + 1] = {C..., '\0'};
static constexpr const_str const kStr{kChars};
static const std::string str() {
return std::string(kStr.data(), kStr.len());
}
template <char... Q>
constexpr bool operator==(const Tag<Q...>& other) const {
return kStr == other.kStr;
}
template <char... Q>
constexpr bool operator!=(const Tag<Q...>& other) const {
return !(*this == other);
}
};
template <char... C>
constexpr bool is_tag(Tag<C...>) {
return true;
}
template <typename A>
constexpr bool is_tag(A) {
return false;
}
namespace internal {
template <typename S, std::size_t... I>
constexpr auto tag_build_impl(S, std::index_sequence<I...>)
-> Tag<S().tag[I]...> {
return {};
}
template <typename S>
constexpr auto tag_build(S) {
return tag_build_impl(S(), std::make_index_sequence<S().tag.len()>{});
}
} // namespace internal
// Use this to create typed tag objects.
// For example:
// auto kFOO = MPP_TAG(FOO);
// auto kBAR = MPP_TAG(BAR);
#define MPP_TAG(s) \
([] { \
struct S { \
const const_str tag{s}; \
}; \
return ::mediapipe::api2::internal::tag_build(S()); \
}())
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_TAG_H_
+48
View File
@@ -0,0 +1,48 @@
#include "mediapipe/framework/api2/tag.h"
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
namespace mediapipe {
namespace api2 {
namespace {
template <typename A, typename B>
constexpr bool same_type(A, B) {
return false;
}
template <typename A>
constexpr bool same_type(A, A) {
return true;
}
auto kFOO = MPP_TAG("FOO");
auto kFOO2 = MPP_TAG("FOO");
auto kBAR = MPP_TAG("BAR");
TEST(TagTest, String) {
EXPECT_EQ(kFOO.str(), "FOO");
EXPECT_EQ(kBAR.str(), "BAR");
}
// Separate invocations of MPP_TAG with the same std::string produce objects of
// the same type.
TEST(TagTest, SameType) { EXPECT_TRUE(same_type(kFOO, kFOO2)); }
// Different tags have different types.
TEST(TagTest, DifferentType) { EXPECT_FALSE(same_type(kFOO, kBAR)); }
TEST(TagTest, Equal) {
EXPECT_EQ(kFOO, kFOO2);
EXPECT_NE(kFOO, kBAR);
}
TEST(TagTest, IsTag) {
EXPECT_TRUE(is_tag(kFOO));
EXPECT_FALSE(is_tag("FOO"));
}
} // namespace
} // namespace api2
} // namespace mediapipe
+87
View File
@@ -0,0 +1,87 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_TEST_CONTRACTS_H_
#define MEDIAPIPE_FRAMEWORK_API2_TEST_CONTRACTS_H_
#include "mediapipe/framework/api2/node.h"
namespace mediapipe {
namespace api2 {
namespace test {
struct Foo : public NodeIntf {
static constexpr Input<int> kBase{"BASE"};
static constexpr Input<float>::Optional kScale{"SCALE"};
static constexpr Output<float> kOut{"OUT"};
static constexpr SideInput<float>::Optional kBias{"BIAS"};
MEDIAPIPE_NODE_INTERFACE(Foo, kBase, kScale, kOut, kBias);
};
struct Foo2 : public NodeIntf {
// clang-format off
static constexpr auto kPorts = std::make_tuple(
Input<int>{"BASE"},
Input<float>::Optional{"SCALE"},
Output<float>{"OUT"},
SideInput<float>::Optional{"BIAS"}
);
// clang-format on
MEDIAPIPE_NODE_INTERFACE(Foo2, kPorts);
};
struct Bar : public NodeIntf {
static constexpr Input<AnyType> kIn{"IN"};
// Should all outputs be treated as optional by default?
static constexpr Output<SameType<kIn>>::Optional kOut{"OUT"};
MEDIAPIPE_NODE_INTERFACE(Bar, kIn, kOut);
};
struct Baz : public NodeIntf {
static constexpr Input<AnyType>::Multiple kData{"DATA"};
// Should all outputs be treated as optional by default?
static constexpr Output<SameType<kData>>::Multiple kDataOut{"DATA"};
MEDIAPIPE_NODE_INTERFACE(Baz, kData, kDataOut);
};
struct IntForwarder : public NodeIntf {
static constexpr Input<int> kIn{"IN"};
static constexpr Output<int> kOut{"OUT"};
MEDIAPIPE_NODE_INTERFACE(IntForwarder, kIn, kOut);
};
struct FloatAdder : public NodeIntf {
static constexpr Input<float>::Multiple kIn{"IN"};
static constexpr Output<float> kOut{"OUT"};
MEDIAPIPE_NODE_INTERFACE(FloatAdder, kIn, kOut);
};
struct ToFloat : public NodeIntf {
static constexpr Input<OneOf<float, int>> kIn{"IN"};
static constexpr Output<float> kOut{"OUT"};
MEDIAPIPE_NODE_INTERFACE(ToFloat, kIn, kOut);
};
struct FooBar : public NodeIntf {
static constexpr Input<int> kIn{"IN"};
static constexpr Output<float> kOut{"OUT"};
MEDIAPIPE_NODE_CONTRACT(kIn, kOut);
};
struct FooBar1 : public FooBar {
static constexpr char kCalculatorName[] = "FooBar";
};
struct FooBar2 : public FooBar {
static constexpr char kCalculatorName[] = "FooBar2";
};
} // namespace test
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_TEST_CONTRACTS_H_
+187
View File
@@ -0,0 +1,187 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_TUPLE_H_
#define MEDIAPIPE_FRAMEWORK_API2_TUPLE_H_
#include <tuple>
#include <type_traits>
#include <utility>
#include "absl/meta/type_traits.h"
// This file contains utilities for working with constexpr tuples.
namespace mediapipe {
namespace api2 {
namespace internal {
// Defines a std::index_sequence with indices for each item of the tuple.
template <class Tuple>
using tuple_index_sequence =
std::make_index_sequence<std::tuple_size_v<std::decay_t<Tuple>>>;
// Concatenates two std::index_sequences.
template <std::size_t... I, std::size_t... J>
constexpr auto index_sequence_cat(std::index_sequence<I...>,
std::index_sequence<J...>)
-> std::index_sequence<I..., J...> {
return {};
}
template <std::size_t... I, std::size_t... J, class... Tail>
constexpr auto index_sequence_cat(std::index_sequence<I...>,
std::index_sequence<J...>, Tail... tail) {
return index_sequence_cat(std::index_sequence<I..., J...>(), tail...);
}
template <template <typename...> class Pred, typename Tuple, std::size_t... I>
constexpr auto filtered_tuple_indices_impl(Tuple&& t,
std::index_sequence<I...>) {
return index_sequence_cat(
std::conditional_t<
Pred<std::tuple_element_t<I, std::decay_t<Tuple>>>::value,
std::index_sequence<I>, std::index_sequence<>>{}...);
}
// Returns a std::index_sequence with the indices of the tuple items whose
// type satisfied Pred.
template <template <typename...> class Pred, typename Tuple>
constexpr auto filtered_tuple_indices(Tuple&& tuple) {
return filtered_tuple_indices_impl<Pred>(tuple,
tuple_index_sequence<Tuple>());
}
// Convenience type to pass any type as a value.
template <typename T>
struct Wrap {
using type = T;
};
template <class F, typename Tuple, std::size_t... I>
constexpr auto filtered_tuple_indices_impl(Tuple&& t,
std::index_sequence<I...>) {
return index_sequence_cat(
std::conditional_t<
F{}(Wrap<std::tuple_element_t<I, std::decay_t<Tuple>>>{}),
std::index_sequence<I>, std::index_sequence<>>{}...);
}
// Returns a std::index_sequence with the indices of the tuple items for which
// F{}(Wrap<item_type>) returns true.
template <class F, typename Tuple>
constexpr auto filtered_tuple_indices(Tuple&& tuple) {
return filtered_tuple_indices_impl<F>(std::forward<Tuple>(tuple),
tuple_index_sequence<Tuple>());
}
// Returns a tuple of references to the tuple items with the specified indices.
template <typename Tuple, std::size_t... I>
constexpr auto select_tuple_indices(Tuple&& tuple, std::index_sequence<I...>) {
return std::forward_as_tuple(std::get<I>(std::forward<Tuple>(tuple))...);
}
// Returns a tuple of references to the tuple items whose types satisfy Pred.
template <template <typename...> class Pred, typename Tuple>
constexpr auto filter_tuple(Tuple&& t) {
return select_tuple_indices(std::forward<Tuple>(t),
filtered_tuple_indices<Pred>(t));
}
// Returns a tuple of references to the tuple items for which
// F{}(Wrap<item_type>) returns true.
template <typename F, typename Tuple>
constexpr auto filter_tuple(Tuple&& t) {
return select_tuple_indices(
std::forward<Tuple>(t),
filtered_tuple_indices<F>(std::forward<Tuple>(t)));
}
// TODO: ensure only one of these is enabled?
template <class F, class T, class I>
constexpr auto call_with_optional_index(F&& f, T&& t, I i)
-> absl::void_t<decltype(f(std::forward<T>(t), i))> {
return f(std::forward<T>(t), i);
}
template <class F, class T, class I>
constexpr auto call_with_optional_index(F&& f, T&& t, I i)
-> absl::void_t<decltype(f(std::forward<T>(t)))> {
return f(std::forward<T>(t));
}
template <class F, class Tuple, std::size_t... I>
constexpr void tuple_for_each_impl(F&& f, Tuple&& tuple,
std::index_sequence<I...>) {
int unpack[] = {
(call_with_optional_index(std::forward<F>(f),
std::get<I>(std::forward<Tuple>(tuple)),
std::integral_constant<std::size_t, I>{}),
0)...};
(void)unpack;
}
// Invokes f for each item in tuple.
// If f takes one argument, it will be called as f(item).
// If f takes two arguments, it will be called as
// f(item, std::integral_constant<std::size_t, index>{}).
template <class F, class Tuple>
constexpr void tuple_for_each(F&& f, Tuple&& tuple) {
return tuple_for_each_impl(std::forward<F>(f), std::forward<Tuple>(tuple),
tuple_index_sequence<Tuple>());
}
template <class F, class Tuple, std::size_t... I>
constexpr auto map_tuple_impl(F&& f, Tuple&& tuple, std::index_sequence<I...>) {
return std::make_tuple(f(std::get<I>(std::forward<Tuple>(tuple)))...);
}
// Returns a tuple where each item is the result of calling f on the
// corresponding item of the provided tuple.
template <class F, class Tuple>
constexpr auto map_tuple(F&& f, Tuple&& tuple) {
return map_tuple_impl(std::forward<F>(f), std::forward<Tuple>(tuple),
tuple_index_sequence<Tuple>());
}
template <class F, class Tuple, std::size_t... I>
constexpr auto tuple_apply_impl(F&& f, Tuple&& tuple,
std::index_sequence<I...>) {
return f(std::get<I>(std::forward<Tuple>(tuple))...);
}
// Invokes f passing the tuple's items as arguments.
template <class F, class Tuple>
constexpr auto tuple_apply(F&& f, Tuple&& tuple) {
return tuple_apply_impl(std::forward<F>(f), std::forward<Tuple>(tuple),
tuple_index_sequence<Tuple>());
}
// Returns the index [0, tuple_size) of the first item for which f returns true,
// or tuple_size if no such item is found.
template <class F, class Tuple, std::size_t i = 0>
constexpr std::enable_if_t<i == std::tuple_size_v<std::decay_t<Tuple>>,
std::size_t>
tuple_find(F&& f, Tuple&& tuple) {
return i;
}
template <class F, class Tuple, std::size_t i = 0>
constexpr std::enable_if_t<i != std::tuple_size_v<std::decay_t<Tuple>>,
std::size_t>
tuple_find(F&& f, Tuple&& tuple) {
if (f(std::get<i>(std::forward<Tuple>(tuple)))) {
return i;
}
return tuple_find<F, Tuple, i + 1>(std::forward<F>(f),
std::forward<Tuple>(tuple));
}
template <class Tuple>
constexpr auto flatten_tuple(Tuple&& tuple) {
return tuple_apply([](auto&&... args) { return std::tuple_cat(args...); },
tuple);
}
} // namespace internal
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_TUPLE_H_
+147
View File
@@ -0,0 +1,147 @@
#include "mediapipe/framework/api2/tuple.h"
#include <tuple>
#include <type_traits>
#include <utility>
#include "absl/strings/str_cat.h"
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
namespace mediapipe {
namespace api2 {
namespace internal {
namespace {
template <typename A, typename B>
constexpr bool same_type(A, B) {
return false;
}
template <typename A>
constexpr bool same_type(A, A) {
return true;
}
template <std::size_t... I>
using iseq = std::index_sequence<I...>;
TEST(TupleTest, IndexSeq) {
EXPECT_TRUE(
same_type(iseq<0, 1, 2>(), index_sequence_cat(iseq<0, 1>(), iseq<2>())));
EXPECT_TRUE(same_type(iseq<0, 1, 2>(),
index_sequence_cat(iseq<0, 1>(), iseq<>(), iseq<2>())));
}
TEST(TupleTest, FilteredIndices) {
EXPECT_TRUE(same_type(
filtered_tuple_indices<std::is_integral>(std::tuple<int, float, char>()),
iseq<0, 2>()));
}
TEST(TupleTest, SelectIndices) {
auto t = std::make_tuple(5.0, 10, "hi");
EXPECT_EQ((select_tuple_indices(t, iseq<0, 2>())),
(std::make_tuple(5.0, "hi")));
}
TEST(TupleTest, FilterTuple) {
auto t = std::make_tuple(5.0, 10, "hi");
EXPECT_EQ((filter_tuple<std::is_integral>(t)), (std::make_tuple(10)));
}
TEST(TupleTest, FilterTupleRefs) {
auto t = std::make_tuple(5.0, 10, "hi");
auto tr = filter_tuple<std::is_integral>(t);
int x;
EXPECT_TRUE(same_type(tr, std::tuple<int&>{x}));
EXPECT_FALSE(same_type(tr, std::tuple<int>{x}));
auto tr_copy =
std::apply([](auto&&... item) { return std::make_tuple(item...); },
filter_tuple<std::is_integral>(t));
EXPECT_TRUE(same_type(tr_copy, std::tuple<int>{x}));
}
struct is_integral {
template <class W>
constexpr bool operator()(W&&) {
return std::is_integral<typename W::type>{};
}
};
TEST(TupleTest, FilteredIndices2) {
EXPECT_TRUE(same_type(
filtered_tuple_indices<is_integral>(std::tuple<int, float, char>()),
iseq<0, 2>()));
}
// TEST(TupleTest, FilterTuple2) {
// auto t = std::make_tuple(5.0, 10, "hi");
// auto is_int = [](auto&& x) {
// return std::is_integral_v<decltype(x)>;
// };
// EXPECT_EQ((filter_tuple(is_int, t)), (std::make_tuple(10)));
// }
TEST(TupleTest, ForEach) {
auto t = std::make_tuple(5.0, 10, "hi");
std::vector<std::string> s;
tuple_for_each([&s](auto&& item) { s.push_back(absl::StrCat(item)); }, t);
EXPECT_EQ(s, (std::vector<std::string>{"5", "10", "hi"}));
}
TEST(TupleTest, ForEachWithIndex) {
auto t = std::make_tuple(5.0, 10, "hi");
std::vector<std::string> s;
tuple_for_each(
[&s](auto&& item, std::size_t i) {
s.push_back(absl::StrCat(i, ":", item));
},
t);
EXPECT_EQ(s, (std::vector<std::string>{"0:5", "1:10", "2:hi"}));
}
TEST(TupleTest, ForEachZip) {
auto t = std::make_tuple(5.0, 10, "hi");
auto u = std::make_tuple(2.0, 3, "lo");
std::vector<std::string> s;
tuple_for_each(
[&s, &u](auto&& item, auto i_const) {
constexpr std::size_t i = decltype(i_const)::value;
s.push_back(absl::StrCat(i, ":", item, ",", std::get<i>(u)));
},
t);
EXPECT_EQ(s, (std::vector<std::string>{"0:5,2", "1:10,3", "2:hi,lo"}));
}
TEST(TupleTest, Apply) {
auto t = std::make_tuple(5.0, 10, "hi");
std::string s = tuple_apply(
[](float f, int i, const char* s) { return absl::StrCat(f, i, s); }, t);
EXPECT_EQ(s, "510hi");
}
TEST(TupleTest, Map) {
auto t = std::make_tuple(5.0, 10, 2L);
auto t2 = map_tuple([](auto x) { return x * 2; }, t);
EXPECT_EQ(t2, std::make_tuple(10.0, 20, 4L));
}
TEST(TupleFind, Find) {
auto t = std::make_tuple(5.0, 10, 2L);
auto i = tuple_find([](auto x) { return x > 3; }, t);
EXPECT_EQ(i, 0);
}
TEST(TupleFind, Flatten) {
auto t1 = std::make_tuple(5.0, 10);
auto t2 = std::make_tuple(2L);
auto t = std::make_tuple(t1, t2);
auto tf = flatten_tuple(t);
EXPECT_EQ(tf, std::make_tuple(5.0, 10, 2L));
}
} // namespace
} // namespace internal
} // namespace api2
} // namespace mediapipe
+136
View File
@@ -0,0 +1,136 @@
#ifndef MEDIAPIPE_FRAMEWORK_API2_TYPE_LIST_H_
#define MEDIAPIPE_FRAMEWORK_API2_TYPE_LIST_H_
#include <string>
#include <type_traits>
#include <utility>
namespace mediapipe {
namespace api2 {
namespace types {
// A list of types. This allows us to store a template parameter pack.
template <typename... Args>
struct List {};
// Concatenate two lists.
template <typename... As, typename... Bs>
auto concat(List<As...>, List<Bs...>) -> List<As..., Bs...> {
return {};
}
// Filter a list using a predicate.
template <template <typename> class Pred, typename... Args>
auto filter(List<Args...>) -> List<Args...> {
return {};
}
template <template <typename> class Pred, typename Head, typename... Tail>
auto filter(List<Head, Tail...>) -> decltype(concat(
typename std::conditional<Pred<Head>::value, List<Head>, List<>>::type{},
filter<Pred>(List<Tail...>{}))) {
return {};
}
template <typename Pred>
auto filter(Pred, List<>) -> List<> {
return {};
}
template <typename Pred, typename Head, typename... Tail>
auto filter(Pred pred, List<Head, Tail...>) -> decltype(concat(
typename std::conditional<pred(Head{}), List<Head>, List<>>::type{},
filter(pred, List<Tail...>{}))) {
return {};
}
// Invoke a template using a list's types as parameters.
template <template <typename...> class T, typename... Args>
auto apply(List<Args...>) -> T<Args...> {
return {};
}
// Wraps a single type. The wrapper can always be instantiated as a value,
// even if T cannot.
template <typename T>
struct Wrap {
using type = T;
};
// Find first match for a predicate.
template <template <typename> class Pred, typename... Args>
auto find(List<Args...>) -> Wrap<void> {
return {};
}
template <template <typename> class Pred, typename Head, typename... Tail>
auto find(List<Head, Tail...>) ->
typename std::conditional<Pred<Head>::value, Wrap<Head>,
decltype(find<Pred>(List<Tail...>{}))>::type {
return {};
}
template <class Pred, typename... Args>
auto find(Pred, List<Args...>) -> Wrap<void> {
return {};
}
template <class Pred, typename Head, typename... Tail>
auto find(Pred pred, List<Head, Tail...>) ->
typename std::conditional<pred(Head{}), Wrap<Head>,
decltype(find(pred, List<Tail...>{}))>::type {
return {};
}
// Apply a function to each item in a list.
template <template <typename> class Fun, typename... Items>
auto map(List<Items...>) -> List<typename Fun<Items>::type...> {
return {};
}
// Get the list's head.
template <typename... Args>
constexpr auto head(List<Args...>) -> Wrap<void> {
return {};
}
template <typename H, typename... T>
constexpr auto head(List<H, T...>) -> Wrap<H> {
return {};
}
// Get the list's length.
template <typename... Args>
constexpr std::size_t length(List<Args...>) {
return 0;
}
template <typename H, typename... T>
constexpr std::size_t length(List<H, T...>) {
return length(List<T...>{}) + 1;
}
// Add indices.
template <std::size_t I, typename T>
struct IndexedType {
static constexpr std::size_t kIndex = I;
using type = T;
};
template <typename... Args, std::size_t... Is>
auto enumerate_impl(List<Args...>, std::index_sequence<Is...>)
-> List<IndexedType<Is, Args>...> {
return {};
}
template <typename... Args>
auto enumerate(List<Args...> a)
-> decltype(enumerate_impl(a, std::index_sequence_for<Args...>{})) {
return {};
}
} // namespace types
} // namespace api2
} // namespace mediapipe
#endif // MEDIAPIPE_FRAMEWORK_API2_TYPE_LIST_H_
+101
View File
@@ -0,0 +1,101 @@
#include "mediapipe/framework/api2/type_list.h"
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
namespace mediapipe {
namespace api2 {
namespace types {
namespace {
template <typename A, typename B>
constexpr bool same_type(A, B) {
return false;
}
template <typename A>
constexpr bool same_type(A, A) {
return true;
}
struct Foo {};
struct Bar {};
struct Baz {};
TEST(TypeListFTest, SameType) {
EXPECT_FALSE(same_type(List<Foo>{}, List<>{}));
EXPECT_TRUE(same_type(List<Foo>{}, List<Foo>{}));
}
TEST(TypeListFTest, Length) {
EXPECT_EQ(length(List<float, int>{}), 2);
EXPECT_EQ(length(List<>{}), 0);
}
TEST(TypeListFTest, Head) {
using Empty = List<>;
using ListA = List<Foo, Bar>;
EXPECT_TRUE(same_type(Wrap<Foo>{}, head(ListA{})));
EXPECT_TRUE(same_type(Wrap<void>{}, head(Empty{})));
}
TEST(TypeListFTest, Concat) {
using Empty = List<>;
using ListA = List<Foo>;
EXPECT_TRUE(same_type(ListA{}, concat(ListA{}, Empty{})));
EXPECT_TRUE(same_type(concat(ListA{}, Empty{}), ListA{}));
using ListB = List<Bar, Baz>;
EXPECT_TRUE(same_type(concat(ListA{}, ListB{}), List<Foo, Bar, Baz>{}));
}
TEST(TypeListFTest, Filter) {
EXPECT_TRUE(same_type(filter<std::is_integral>(List<>{}), List<>{}));
EXPECT_TRUE(same_type(filter<std::is_integral>(List<int, float, char>{}),
List<int, char>{}));
}
TEST(TypeListFTest, Filter2) {
constexpr auto is_integral = [](auto x) {
return std::is_integral<decltype(x)>{};
};
auto x = filter(is_integral, List<>{});
EXPECT_TRUE(same_type(x, List<>{}));
auto y = filter(is_integral, List<int, float, char>{});
EXPECT_TRUE(same_type(y, List<int, char>{}));
auto z = filter([](auto x) { return std::is_integral<decltype(x)>{}; },
List<int, double>{});
EXPECT_TRUE(same_type(z, List<int>{}));
}
TEST(TypeListFTest, Find) {
EXPECT_TRUE(same_type(find<std::is_integral>(List<>{}), Wrap<void>{}));
EXPECT_TRUE(
same_type(find<std::is_integral>(List<float, int>{}), Wrap<int>()));
}
TEST(TypeListFTest, Find2) {
constexpr auto is_integral = [](auto x) {
return std::is_integral<decltype(x)>{};
};
EXPECT_TRUE(same_type(find(is_integral, List<>{}), Wrap<void>{}));
EXPECT_TRUE(same_type(find(is_integral, List<float, int>{}), Wrap<int>()));
}
TEST(TypeListFTest, Map) {
EXPECT_TRUE(
same_type(map<std::remove_cv>(List<const int, const float, const char>{}),
List<int, float, char>{}));
}
TEST(TypeListFTest, Enumerate) {
EXPECT_TRUE(same_type(enumerate(List<int, float, char>{}),
List<IndexedType<0, int>, IndexedType<1, float>,
IndexedType<2, char>>{}));
}
} // namespace
} // namespace types
} // namespace api2
} // namespace mediapipe