图表
CalculatorGraphConfig
proto 指定
MediaPipe 图。图表中的每个 node
代表一个特定的计算器或
子图,并指定必要的配置,例如已注册的
计算器/子图类型、输入、输出和可选字段,例如
特定于节点的选项、输入政策和执行器,
同步。
CalculatorGraphConfig
有几个其他用于配置全局图表级的字段
设置,例如执行器配置图、线程数和最大队列大小
输出流。一些图表级别的设置对于调整
图表在不同平台(例如桌面设备和移动设备)上的效果。对于
例如,在移动设备上,将一个大型模型推断计算器连接到一个单独的
Executor 可以提高实时应用的性能,因为这
启用线程局部性。
下面是一个简单的 CalculatorGraphConfig
示例,其中有一系列
直通式计算器 :
# This graph named main_pass_throughcals_nosubgraph.pbtxt contains 4
# passthrough calculators.
input_stream: "in"
output_stream: "out"
node {
calculator: "PassThroughCalculator"
input_stream: "in"
output_stream: "out1"
}
node {
calculator: "PassThroughCalculator"
input_stream: "out1"
output_stream: "out2"
}
node {
calculator: "PassThroughCalculator"
input_stream: "out2"
output_stream: "out3"
}
node {
calculator: "PassThroughCalculator"
input_stream: "out3"
output_stream: "out"
}
MediaPipe 为复杂图(例如机器学习流水线、处理模型元数据、可选节点等)提供替代的 C++
表示法。上图可能如下所示:
CalculatorGraphConfig BuildGraphConfig() {
Graph graph;
// Graph inputs
Stream<AnyType> in = graph.In(0).SetName("in");
auto pass_through_fn = [](Stream<AnyType> in,
Graph& graph) -> Stream<AnyType> {
auto& node = graph.AddNode("PassThroughCalculator");
in.ConnectTo(node.In(0));
return node.Out(0);
};
Stream<AnyType> out1 = pass_through_fn(in, graph);
Stream<AnyType> out2 = pass_through_fn(out1, graph);
Stream<AnyType> out3 = pass_through_fn(out2, graph);
Stream<AnyType> out4 = pass_through_fn(out3, graph);
// Graph outputs
out4.SetName("out").ConnectTo(graph.Out(0));
return graph.GetConfig();
}
如需了解详情,请参阅使用 C++ 构建图。
子图
将 CalculatorGraphConfig
模块化为子模块并帮助重复使用
因此 MediaPipe 图可以定义为 Subgraph
。通过
子图的公共接口,由一组输入和输出流组成
类似于计算器的公共界面然后,子图可以
将 CalculatorGraphConfig
视为计算器。将 MediaPipe 图
会从 CalculatorGraphConfig
加载,则每个子图节点都会替换为
相应的计算器图表因此,语义和性能
与对应的计算器图表相同。
以下示例展示了如何创建一个名为 TwoPassThroughSubgraph
的子图。
定义子图。
# This subgraph is defined in two_pass_through_subgraph.pbtxt # and is registered as "TwoPassThroughSubgraph" type: "TwoPassThroughSubgraph" input_stream: "out1" output_stream: "out3" node { calculator: "PassThroughCalculator" input_stream: "out1" output_stream: "out2" } node { calculator: "PassThroughCalculator" input_stream: "out2" output_stream: "out3" }
子图的公共接口包括:
- 绘制输入流的图形
- 绘制输出流的图像
- 绘制输入端数据包的图形
- 绘制输出端数据包的图形
使用 BUILD 规则
mediapipe_simple_subgraph
注册子图。通过 参数register_as
定义新子图的组件名称。# Small section of BUILD file for registering the "TwoPassThroughSubgraph" # subgraph for use by main graph main_pass_throughcals.pbtxt mediapipe_simple_subgraph( name = "twopassthrough_subgraph", graph = "twopassthrough_subgraph.pbtxt", register_as = "TwoPassThroughSubgraph", deps = [ "//mediapipe/calculators/core:pass_through_calculator", "//mediapipe/framework:calculator_graph", ], )
使用主图中的子图。
# This main graph is defined in main_pass_throughcals.pbtxt # using subgraph called "TwoPassThroughSubgraph" input_stream: "in" node { calculator: "PassThroughCalculator" input_stream: "in" output_stream: "out1" } node { calculator: "TwoPassThroughSubgraph" input_stream: "out1" output_stream: "out3" } node { calculator: "PassThroughCalculator" input_stream: "out3" output_stream: "out4" }
图表选项
可以指定“图表选项”用于 MediaPipe 图的 protobuf
类似于Calculator Options
为 MediaPipe 计算器指定的 protobuf。这些“图表选项”可以是
会指定调用图表的位置,并用于填充计算器选项和
子图选项。
在 CalculatorGraphConfig
中,可以为子图指定图表选项
与计算器选项完全相同,如下所示:
node {
calculator: "FlowLimiterCalculator"
input_stream: "image"
output_stream: "throttled_image"
node_options: {
[type.googleapis.com/mediapipe.FlowLimiterCalculatorOptions] {
max_in_flight: 1
}
}
}
node {
calculator: "FaceDetectionSubgraph"
input_stream: "IMAGE:throttled_image"
node_options: {
[type.googleapis.com/mediapipe.FaceDetectionOptions] {
tensor_width: 192
tensor_height: 192
}
}
}
在 CalculatorGraphConfig
中,可以接受图表选项并用于填充
计算器选项,如下所示:
graph_options: {
[type.googleapis.com/mediapipe.FaceDetectionOptions] {}
}
node: {
calculator: "ImageToTensorCalculator"
input_stream: "IMAGE:image"
node_options: {
[type.googleapis.com/mediapipe.ImageToTensorCalculatorOptions] {
keep_aspect_ratio: true
border_mode: BORDER_ZERO
}
}
option_value: "output_tensor_width:options/tensor_width"
option_value: "output_tensor_height:options/tensor_height"
}
node {
calculator: "InferenceCalculator"
node_options: {
[type.googleapis.com/mediapipe.InferenceCalculatorOptions] {}
}
option_value: "delegate:options/delegate"
option_value: "model_path:options/model_path"
}
在此示例中,FaceDetectionSubgraph
接受图表选项 protobuf
FaceDetectionOptions
。FaceDetectionOptions
用于定义某个字段
计算器选项 ImageToTensorCalculatorOptions
和部分字段中的值
子图选项 InferenceCalculatorOptions
中的值。字段值
均使用 option_value:
语法进行定义。
在 CalculatorGraphConfig::Node
protobuf 中,字段 node_options:
和
option_value:
可同时定义计算器的选项值,例如
ImageToTensorCalculator
。node_options:
字段定义了一组字面量
常量值。每个 option_value:
字段
定义一个 protobuf 字段的值,它使用来自
图表,具体地讲,就是从外围图表选项的字段值
图表。在上面的示例中,option_value:
"output_tensor_width:options/tensor_width"
定义字段
ImageToTensorCalculatorOptions.output_tensor_width
,使用
FaceDetectionOptions.tensor_width
。
option_value:
的语法与 input_stream:
的语法类似。通过
语法为 option_value: "LHS:RHS"
。LHS 标识计算器选项
字段,而 RHS 会标识图表选项字段。具体而言,LHS
和 RHS 均由一系列 protobuf 字段名称组成,
protobuf 消息和字段(以“/”分隔)。这称为“ProtoPath”
语法。LHS 或 RHS 中引用的嵌套消息必须已
其中定义的 protobuf 中的定义,以便使用
option_value:
。
时光
默认情况下,MediaPipe 要求计算器图表是无环的,并且处理循环 在图表中显示为错误如果想要在图表中显示循环次数 在图配置中添加注释。本页将介绍具体操作方法。
注意:当前方法是实验性的,可能会发生变化。我们欢迎 您的反馈。
请使用 CalculatorGraphTest.Cycle
单元测试,
mediapipe/framework/calculator_graph_test.cc
作为示例代码。以下为
测试中的循环图。加法器的 sum
输出是
由整数源计算器生成的整数。
这个简单的图表说明了支持循环图的所有问题。
背面边缘注释
我们要求将每个周期中的一条边缘标注为后边缘。这样, 移除所有背面边缘后,MediaPipe 的拓扑排序即可正常运行。
选择后边缘通常有多种方法。标记了哪些边 哪些节点被视为上游节点,哪些节点 视为下游,这反过来又会影响 MediaPipe 分配的优先级 节点
例如,CalculatorGraphTest.Cycle
测试会将 old_sum
边缘标记为
因此 Delay 节点被视为加法器的下游节点
节点并具有更高的优先级。或者,我们也可以将 sum
标记为
延迟节点的输入作为后边缘,在这种情况下,延迟节点为
被视为添加器节点的上游节点,并被指定较低的优先级。
初始数据包
当整数中的第一个整数为整数时,加法计算器可运行。
我们需要一个初始数据包,其值为 0,
时间戳,位于 old_sum
输入流上。这个初始数据包
应由 Open()
方法中的延迟计算器输出。
循环延迟
每个循环都应产生延迟,以使之前的 sum
输出与下一个循环保持一致
整数输入。延迟节点也会执行此操作。因此延迟节点需要
了解有关整数源计算器的时间戳的以下内容:
第一个输出的时间戳。
连续输出之间的时间戳增量。
我们计划添加只关心数据包的备用调度政策 并忽略数据包时间戳,从而消除这种不便。
一个输入流完成后提前终止计算器
默认情况下,在以下情况下,MediaPipe 会调用非源计算器的 Close()
方法:
所有输入流都已完成。在示例图中,我们想要停止
adder 节点。这是通过
使用备用输入流处理程序配置添加器节点;
EarlyCloseInputStreamHandler
。
相关源代码
延迟计算器
请注意 Open()
中用于输出初始数据包的代码,以及
Process()
,用于为输入数据包添加(单位)延迟。如上所述,
延迟节点假定其输出流与具有
数据包时间戳 0、1、2、3...
class UnitDelayCalculator : public Calculator {
public:
static absl::Status FillExpectations(
const CalculatorOptions& extendable_options, PacketTypeSet* inputs,
PacketTypeSet* outputs, PacketTypeSet* input_side_packets) {
inputs->Index(0)->Set<int>("An integer.");
outputs->Index(0)->Set<int>("The input delayed by one time unit.");
return absl::OkStatus();
}
absl::Status Open() final {
Output()->Add(new int(0), Timestamp(0));
return absl::OkStatus();
}
absl::Status Process() final {
const Packet& packet = Input()->Value();
Output()->AddPacket(packet.At(packet.Timestamp().NextAllowedInStream()));
return absl::OkStatus();
}
};
图配置
请注意 back_edge
注解和备用 input_stream_handler
。
node {
calculator: 'GlobalCountSourceCalculator'
input_side_packet: 'global_counter'
output_stream: 'integers'
}
node {
calculator: 'IntAdderCalculator'
input_stream: 'integers'
input_stream: 'old_sum'
input_stream_info: {
tag_index: ':1' # 'old_sum'
back_edge: true
}
output_stream: 'sum'
input_stream_handler {
input_stream_handler: 'EarlyCloseInputStreamHandler'
}
}
node {
calculator: 'UnitDelayCalculator'
input_stream: 'sum'
output_stream: 'old_sum'
}