MediaPipe 系列 11：自定义 Calculator 第一步——从零到运行

一、学习目标

完成本篇后，你将能够：

  1. 创建自定义 Proto Options
  2. 实现完整的 Calculator 类
  3. 编写 Graph 配置文件
  4. 配置 Bazel BUILD
  5. 编译运行测试程序
  6. 编写单元测试

最终效果：

1
2
3
4
5
$ ./hello_world_demo --name=Mars
Hello, Mars!

$ ./hello_world_demo --name=IMS
Hello, IMS!

二、Calculator 开发流程

2.1 开发步骤概览

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
┌─────────────────────────────────────────────────────────────┐
│                  Calculator 开发流程                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Step 1: 定义 Proto Options                                 │
│  ┌─────────────────────────────────────────────┐           │
│  │ message MyCalculatorOptions {               │           │
│  │   optional string param = 1;                │           │
│  │ }                                           │           │
│  └─────────────────────────────────────────────┘           │
│                         │                                   │
│                         ▼                                   │
│  Step 2: 实现 Calculator 类                                 │
│  ┌─────────────────────────────────────────────┐           │
│  │ class MyCalculator : public CalculatorBase {│           │
│  │   GetContract()  // 定义输入输出            │           │
│  │   Open()         // 初始化                  │           │
│  │   Process()      // 处理数据                │           │
│  │   Close()        // 清理资源                │           │
│  │ };                                          │           │
│  └─────────────────────────────────────────────┘           │
│                         │                                   │
│                         ▼                                   │
│  Step 3: 编写 Graph 配置                                    │
│  ┌─────────────────────────────────────────────┐           │
│  │ node {                                      │           │
│  │   calculator: "MyCalculator"                │           │
│  │   input_stream: "input"                     │           │
│  │   output_stream: "output"                   │           │
│  │ }                                           │           │
│  └─────────────────────────────────────────────┘           │
│                         │                                   │
│                         ▼                                   │
│  Step 4: 配置 Bazel BUILD                                   │
│  ┌─────────────────────────────────────────────┐           │
│  │ cc_library(                                 │           │
│  │   name = "my_calculator",                   │           │
│  │   srcs = ["my_calculator.cc"],              │           │
│  │   alwayslink = 1,  // 重要！                │           │
│  │ )                                           │           │
│  └─────────────────────────────────────────────┘           │
│                         │                                   │
│                         ▼                                   │
│  Step 5: 编写测试程序                                       │
│  ┌─────────────────────────────────────────────┐           │
│  │ int main() {                                │           │
│  │   CalculatorGraph graph;                    │           │
│  │   graph.Initialize(config);                 │           │
│  │   graph.StartRun({});                       │           │
│  │   graph.AddPacketToInputStream(...);        │           │
│  │ }                                           │           │
│  └─────────────────────────────────────────────┘           │
│                         │                                   │
│                         ▼                                   │
│  Step 6: 编译运行测试                                       │
│  ┌─────────────────────────────────────────────┐           │
│  │ $ bazel build //path:target                 │           │
│  │ $ ./bazel-bin/path/target                   │           │
│  └─────────────────────────────────────────────┘           │
│                                                             │
└─────────────────────────────────────────────────────────────┘

三、Step 1: 定义 Proto Options

3.1 为什么需要 Proto Options？

Options 的作用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
┌─────────────────────────────────────────────────────────────┐
│                    Options 的作用                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Calculator 代码 = 固定逻辑                                 │
│  Options = 可配置参数                                       │
│                                                             │
│  ┌─────────────────┐        ┌─────────────────┐            │
│  │ Calculator Code │ ◀────▶ │ Options (Proto) │            │
│  │ (编译后固定)    │        │ (运行时可改)    │            │
│  └─────────────────┘        └─────────────────┘            │
│                                                             │
│  好处：                                                     │
│  1. 不重新编译就能调参                                      │
│  2. Graph 配置文件中设置参数                                │
│  3. 不同场景使用不同参数                                    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 同一个 Calculator，不同 Options
node {
  calculator: "ThresholdCalculator"
  input_stream: "data"
  output_stream: "result"
  options {
    [mediapipe.ThresholdCalculatorOptions.ext] {
      threshold: 0.5  # 配置 A
    }
  }
}

node {
  calculator: "ThresholdCalculator"
  input_stream: "data"
  output_stream: "result"
  options {
    [mediapipe.ThresholdCalculatorOptions.ext] {
      threshold: 0.8  # 配置 B（不同阈值）
    }
  }
}

3.2 创建 Proto 文件

文件结构：

1
2
3
4
5
6
7
mediapipe/
└── calculators/
    └── my/
        ├── BUILD
        ├── hello_world_calculator_options.proto
        ├── hello_world_calculator.h
        └── hello_world_calculator.cc

Proto 文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// mediapipe/calculators/my/hello_world_calculator_options.proto
syntax = "proto3";

package mediapipe;

import "mediapipe/framework/calculator_options.proto";

message HelloWorldCalculatorOptions {
  // 扩展 CalculatorOptions
  extend CalculatorOptions {
    optional HelloWorldCalculatorOptions ext = 1000000;
  }
  
  // 自定义参数
  optional string greeting = 1 [default = "Hello"];
  
  // 可以添加更多参数
  optional bool uppercase = 2 [default = false];
  optional string suffix = 3 [default = "!"];
}

Proto 语法说明：

1
2
3
4
5
6
7
8
9
10
11
12
13
syntax = "proto3";        // 使用 proto3 语法
package mediapipe;        // 包名（影响 C++ 命名空间）

message HelloWorldCalculatorOptions {
  // 字段定义
  optional string greeting = 1 [default = "Hello"];
  //           │          │
  //           │          └── 字段编号（唯一）
  //           └── 字段类型和名称
  
  // 默认值：如果不设置，使用 default 指定的值
  // [default = "Hello"]
}

3.3 Proto 编译过程

1
2
3
4
5
6
7
8
.proto 文件
     │
     ▼ (protoc 编译器)
.pb.h 文件 (C++ 头文件)
.pb.cc 文件 (C++ 实现文件)
     │
     ▼ (C++ 编译器)
.o 文件 → .so/.a 库文件

四、Step 2: 实现 Calculator 类

4.1 Calculator 生命周期

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
┌─────────────────────────────────────────────────────────────┐
│                  Calculator 生命周期                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Graph 启动                                                 │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐                                           │
│  │ GetContract │ ← 只调用一次（验证 Graph 配置）            │
│  │             │   定义输入输出端口类型                     │
│  └─────────────┘                                           │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐                                           │
│  │    Open     │ ← 每个实例调用一次                         │
│  │             │   初始化资源、读取 Options                 │
│  └─────────────┘                                           │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐                                           │
│  │   Process   │ ← 每个输入 Packet 调用一次                 │
│  │             │   处理数据、输出结果                       │
│  │             │   (可能调用多次)                           │
│  └─────────────┘                                           │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐                                           │
│  │    Close    │ ← Graph 关闭时调用一次                     │
│  │             │   释放资源                                 │
│  └─────────────┘                                           │
│       │                                                     │
│       ▼                                                     │
│  Graph 关闭                                                 │
│                                                             │
└─────────────────────────────────────────────────────────────┘

4.2 头文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
// mediapipe/calculators/my/hello_world_calculator.h
#ifndef MEDIAPIPE_CALCULATORS_MY_HELLO_WORLD_CALCULATOR_H_
#define MEDIAPIPE_CALCULATORS_MY_HELLO_WORLD_CALCULATOR_H_

#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/calculators/my/hello_world_calculator_options.pb.h"

namespace mediapipe {

class HelloWorldCalculator : public CalculatorBase {
 public:
  // GetContract：定义输入输出端口
  // 在 Graph 初始化时调用，验证配置正确性
  static absl::Status GetContract(CalculatorContract* cc);
  
  // Open：初始化
  // 在 Graph 启动时调用一次
  absl::Status Open(CalculatorContext* cc) override;
  
  // Process：处理数据
  // 每个输入 Packet 调用一次
  absl::Status Process(CalculatorContext* cc) override;
  
  // Close：清理资源（可选）
  // 在 Graph 关闭时调用
  absl::Status Close(CalculatorContext* cc) override;

 private:
  // 从 Options 读取的配置
  std::string greeting_;
  bool uppercase_ = false;
  std::string suffix_;
  
  // 统计信息
  int process_count_ = 0;
};

}  // namespace mediapipe

#endif  // MEDIAPIPE_CALCULATORS_MY_HELLO_WORLD_CALCULATOR_H_

4.3 实现文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
// mediapipe/calculators/my/hello_world_calculator.cc
#include "mediapipe/calculators/my/hello_world_calculator.h"
#include "mediapipe/framework/port/logging.h"
#include "mediapipe/framework/port/ret_check.h"
#include "mediapipe/framework/port/status.h"

namespace mediapipe {

using mediapipe::CalculatorContract;
using mediapipe::CalculatorContext;

absl::Status HelloWorldCalculator::GetContract(CalculatorContract* cc) {
  // ========== 定义输入端口 ==========
  // Index(0)：第一个输入端口
  // Set<std::string>()：输入类型为 string
  cc->Inputs().Index(0).Set<std::string>();
  
  // 也可以使用 Tag（标签）命名端口
  // cc->Inputs().Tag("NAME").Set<std::string>();
  
  // ========== 定义输出端口 ==========
  cc->Outputs().Index(0).Set<std::string>();
  
  // ========== 声明使用 Options ==========
  // 这会验证 Graph 配置中的 options 是否正确
  cc->Options<HelloWorldCalculatorOptions>();
  
  // ========== 定义 Side Packet（可选）==========
  // Side Packet 是静态配置数据，不随时间变化
  // cc->InputSidePackets().Tag("MODEL").Set<TfLiteModelPtr>();
  
  return absl::OkStatus();
}

absl::Status HelloWorldCalculator::Open(CalculatorContext* cc) {
  // ========== 读取 Options ==========
  const auto& options = cc->Options<HelloWorldCalculatorOptions>();
  
  greeting_ = options.greeting();
  uppercase_ = options.uppercase();
  suffix_ = options.suffix();
  
  // ========== 初始化资源 ==========
  // 例如：加载模型、分配内存等
  
  // ========== 日志输出 ==========
  LOG(INFO) << "HelloWorldCalculator initialized";
  LOG(INFO) << "  greeting: " << greeting_;
  LOG(INFO) << "  uppercase: " << uppercase_;
  LOG(INFO) << "  suffix: " << suffix_;
  
  return absl::OkStatus();
}

absl::Status HelloWorldCalculator::Process(CalculatorContext* cc) {
  // ========== 检查输入是否为空 ==========
  // 重要：输入可能为空（某些情况下）
  if (cc->Inputs().Index(0).IsEmpty()) {
    LOG(WARNING) << "Empty input at timestamp: " << cc->InputTimestamp();
    return absl::OkStatus();  // 返回 OK，跳过处理
  }
  
  // ========== 获取输入数据 ==========
  const std::string& input = cc->Inputs().Index(0).Get<std::string>();
  
  // ========== 处理数据 ==========
  std::string output;
  
  // 组合问候语
  output = greeting_ + ", " + input + suffix_;
  
  // 可选：转换为大写
  if (uppercase_) {
    std::transform(output.begin(), output.end(), output.begin(), ::toupper);
  }
  
  // ========== 日志输出 ==========
  LOG(INFO) << "Process #" << process_count_ 
            << ": Input=\"" << input 
            << "\" -> Output=\"" << output << "\"";
  
  // ========== 输出结果 ==========
  // MakePacket：创建 Packet
  // .At(Timestamp)：设置时间戳
  cc->Outputs().Index(0).AddPacket(
      MakePacket<std::string>(output).At(cc->InputTimestamp()));
  
  process_count_++;
  
  return absl::OkStatus();
}

absl::Status HelloWorldCalculator::Close(CalculatorContext* cc) {
  // ========== 清理资源 ==========
  LOG(INFO) << "HelloWorldCalculator closed";
  LOG(INFO) << "  Total process count: " << process_count_;
  
  return absl::OkStatus();
}

// ========== 注册 Calculator ==========
// 重要：必须注册，否则 Graph 找不到这个 Calculator
REGISTER_CALCULATOR(HelloWorldCalculator);

}  // namespace mediapipe

4.4 关键代码详解

GetContract 的作用：

1
2
3
4
5
6
7
8
9
10
11
12
// GetContract 在 Graph 初始化时调用
// 作用：告诉框架这个 Calculator 需要什么输入、产生什么输出
// 
// 例如：
// cc->Inputs().Index(0).Set<std::string>();
// 表示：第 0 个输入端口，类型为 string
//
// 框架会检查：
// 1. 上游 Calculator 的输出类型是否匹配
// 2. Graph 配置中的端口连接是否正确
//
// 如果不匹配，Graph 初始化会失败

Packet 的概念：

1
2
3
4
5
6
7
8
9
10
// Packet = 数据 + 时间戳
// 
// Packet<std::string> packet = MakePacket<std::string>("hello").At(Timestamp(0));
// 
// packet.Get<std::string>()  // 获取数据 -> "hello"
// packet.Timestamp()          // 获取时间戳 -> 0
//
// 重要：
// 1. 时间戳必须递增
// 2. 不同流的相同时间戳会被同步处理

REGISTER_CALCULATOR 宏：

1
2
3
4
5
6
7
8
// 这个宏做了什么？
// 1. 创建一个工厂函数
// 2. 注册到 CalculatorRegistry
// 3. Graph 配置中的 calculator: "HelloWorldCalculator" 
//    通过 Registry 找到对应的工厂函数，创建实例
//
// 如果忘记注册：
// 错误信息：Calculator "HelloWorldCalculator" not found

五、Step 3: 编写 Graph 配置

5.1 Graph 配置语法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# mediapipe/graphs/my/hello_world_graph.pbtxt

# ============== 输入输出流定义 ==============
# 声明 Graph 对外的接口
input_stream: "input_name"      # 输入：名字
output_stream: "output_greeting"  # 输出：问候语

# ============== 节点定义 ==============
node {
  # Calculator 类型（与 REGISTER_CALCULATOR 的名字对应）
  calculator: "HelloWorldCalculator"
  
  # 输入流：Graph 输入 -> Calculator 输入
  input_stream: "input_name"
  
  # 输出流：Calculator 输出 -> Graph 输出
  output_stream: "output_greeting"
  
  # Options：Calculator 配置
  options {
    [mediapipe.HelloWorldCalculatorOptions.ext] {
      greeting: "Hello"
      uppercase: false
      suffix: "!"
    }
  }
}

5.2 多节点 Graph 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 多个 Calculator 组成的 Graph
input_stream: "input_name"
output_stream: "final_output"

# 节点 1：HelloWorld
node {
  calculator: "HelloWorldCalculator"
  input_stream: "input_name"
  output_stream: "greeting"
  options {
    [mediapipe.HelloWorldCalculatorOptions.ext] {
      greeting: "Hello"
    }
  }
}

# 节点 2：添加时间戳
node {
  calculator: "TimestampAppenderCalculator"
  input_stream: "greeting"
  input_stream: "timestamp"  # 也可以有多个输入
  output_stream: "final_output"
}

六、Step 4: 配置 Bazel BUILD

6.1 BUILD 文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# mediapipe/calculators/my/BUILD

# 加载 MediaPipe 的构建规则
load("//mediapipe/framework/port:build_config.bzl", "mediapipe_proto_library")

# ============== Proto 库 ==============
mediapipe_proto_library(
    name = "hello_world_calculator_options_proto",
    srcs = ["hello_world_calculator_options.proto"],
    deps = [
        "//mediapipe/framework:calculator_options_proto",
    ],
    visibility = ["//visibility:public"],
)

# ============== Calculator 库 ==============
cc_library(
    name = "hello_world_calculator",
    srcs = ["hello_world_calculator.cc"],
    hdrs = ["hello_world_calculator.h"],
    deps = [
        ":hello_world_calculator_options_proto",
        "//mediapipe/framework:calculator_framework",
        "//mediapipe/framework/port:logging",
        "//mediapipe/framework/port:ret_check",
        "//mediapipe/framework/port:status",
        "@com_google_absl//absl/strings",
    ],
    # 重要：必须设置 alwayslink = 1
    # 否则 REGISTER_CALCULATOR 可能被优化掉
    alwayslink = 1,
    visibility = ["//visibility:public"],
)

# ============== Demo 程序 ==============
cc_binary(
    name = "hello_world_demo",
    srcs = ["hello_world_demo.cc"],
    deps = [
        ":hello_world_calculator",
        "//mediapipe/framework:calculator_framework",
        "//mediapipe/framework/port:parse_text_proto",
        "//mediapipe/framework/port:status",
        "@com_google_absl//absl/flags:flag",
        "@com_google_absl//absl/strings",
    ],
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
问题背景：
┌─────────────────────────────────────────────────────────────┐
│                    链接器优化问题                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Calculator 代码：                                          │
│  REGISTER_CALCULATOR(HelloWorldCalculator);                 │
│                                                             │
│  这是一个"副作用"代码：                                      │
│  - 不被任何其他代码调用                                      │
│  - 只是注册自己到 Registry                                  │
│                                                             │
│  链接器的行为：                                              │
│  - 如果发现某个符号没有被引用                                │
│  - 会把它优化掉（不链接进最终程序）                          │
│                                                             │
│  结果：                                                      │
│  - REGISTER_CALCULATOR 被优化掉                             │
│  - Registry 里找不到 HelloWorldCalculator                   │
│  - 报错：Calculator "HelloWorldCalculator" not found        │
│                                                             │
│  解决方案：                                                  │
│  alwayslink = 1                                             │
│  - 强制链接整个库                                            │
│  - 即使看起来没有被引用                                      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

七、Step 5: 编写测试程序

7.1 完整 Demo 程序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
// mediapipe/calculators/my/hello_world_demo.cc
#include <iostream>
#include <string>

#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/port/parse_text_proto.h"
#include "mediapipe/framework/port/status.h"
#include "absl/flags/flag.h"
#include "absl/flags/parse.h"
#include "absl/strings/string_view.h"

// 定义命令行参数
ABSL_FLAG(std::string, name, "World", "Name to greet");
ABSL_FLAG(bool, verbose, false, "Enable verbose output");

namespace mediapipe {

// 运行 Graph 的函数
absl::Status RunGraph(const std::string& name, bool verbose) {
  // ========== 1. 定义 Graph 配置 ==========
  // 可以从文件读取，也可以内嵌定义
  CalculatorGraphConfig config = ParseTextProtoOrDie<CalculatorGraphConfig>(R"(
    input_stream: "input_name"
    output_stream: "output_greeting"
    
    node {
      calculator: "HelloWorldCalculator"
      input_stream: "input_name"
      output_stream: "output_greeting"
      options {
        [mediapipe.HelloWorldCalculatorOptions.ext] {
          greeting: "Hello"
          uppercase: false
          suffix: "!"
        }
      }
    }
  )");

  if (verbose) {
    LOG(INFO) << "Graph config:\n" << config.DebugString();
  }

  // ========== 2. 创建 Graph ==========
  CalculatorGraph graph;
  
  // 初始化 Graph
  MP_RETURN_IF_ERROR(graph.Initialize(config));
  
  if (verbose) {
    LOG(INFO) << "Graph initialized";
  }

  // ========== 3. 设置输出回调 ==========
  // 当有输出时，这个函数会被调用
  MP_RETURN_IF_ERROR(graph.ObserveOutputStream(
      "output_greeting",
      [&name, verbose](const Packet& packet) {
        // 获取输出数据
        const std::string& greeting = packet.Get<std::string>();
        
        // 打印结果
        std::cout << greeting << std::endl;
        
        if (verbose) {
          LOG(INFO) << "Received output at timestamp: " << packet.Timestamp();
        }
        
        return absl::OkStatus();
      }));

  // ========== 4. 启动 Graph ==========
  MP_RETURN_IF_ERROR(graph.StartRun({}));
  
  if (verbose) {
    LOG(INFO) << "Graph started";
  }

  // ========== 5. 发送输入数据 ==========
  // 创建 Packet 并发送到输入流
  // 注意：时间戳必须递增
  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
      "input_name",
      MakePacket<std::string>(name).At(Timestamp(0))));

  if (verbose) {
    LOG(INFO) << "Input sent: " << name;
  }

  // ========== 6. 等待处理完成 ==========
  // 关闭输入流，等待 Graph 处理完所有数据
  MP_RETURN_IF_ERROR(graph.CloseInputStream("input_name"));
  MP_RETURN_IF_ERROR(graph.WaitUntilDone());
  
  if (verbose) {
    LOG(INFO) << "Graph finished";
  }

  return absl::OkStatus();
}

}  // namespace mediapipe

// ========== 主函数 ==========
int main(int argc, char** argv) {
  // 初始化 Google Logging
  google::InitGoogleLogging(argv[0]);
  
  // 解析命令行参数
  absl::ParseCommandLine(argc, argv);
  
  // 获取参数值
  std::string name = absl::GetFlag(FLAGS_name);
  bool verbose = absl::GetFlag(FLAGS_verbose);
  
  // 运行 Graph
  absl::Status status = mediapipe::RunGraph(name, verbose);
  
  // 检查结果
  if (!status.ok()) {
    LOG(ERROR) << "Error: " << status;
    return 1;
  }
  
  return 0;
}

7.2 代码详解

Graph 运行流程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// 1. 创建配置
CalculatorGraphConfig config = ParseTextProtoOrDie<...>(R"(
  ...
)");

// 2. 创建并初始化 Graph
CalculatorGraph graph;
graph.Initialize(config);  // 验证配置、创建节点

// 3. 设置输出回调
graph.ObserveOutputStream("output", [](const Packet& p) {
  // 处理输出
  return absl::OkStatus();
});

// 4. 启动 Graph
graph.StartRun({});  // 调用所有 Calculator 的 Open()

// 5. 发送输入
graph.AddPacketToInputStream("input", MakePacket<std::string>("data").At(Timestamp(0)));

// 6. 关闭并等待
graph.CloseInputStream("input");  // 关闭输入
graph.WaitUntilDone();            // 等待所有 Calculator 的 Close() 完成

八、Step 6: 编译运行测试

8.1 编译

1
2
3
4
5
6
7
8
9
10
11
12
# 进入 MediaPipe 目录
cd mediapipe

# 编译
bazel build -c opt //mediapipe/calculators/my:hello_world_demo

# 输出：
# INFO: Analyzed target //mediapipe/calculators/my:hello_world_demo (123 packages loaded, 4567 targets configured).
# INFO: Found 1 target...
# Target //mediapipe/calculators/my:hello_world_demo up-to-date:
#   bazel-bin/mediapipe/calculators/my/hello_world_demo
# INFO: Elapsed time: 12.345s, Critical Path: 10.12s

8.2 运行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 基本运行
GLOG_logtostderr=1 bazel-bin/mediapipe/calculators/my/hello_world_demo

# 输出：
# Hello, World!

# 自定义名字
GLOG_logtostderr=1 bazel-bin/mediapipe/calculators/my/hello_world_demo --name=Mars

# 输出：
# Hello, Mars!

# 详细日志
GLOG_logtostderr=1 GLOG_v=1 bazel-bin/mediapipe/calculators/my/hello_world_demo --name=IMS --verbose

# 输出：
# I20260312 09:00:00.000000 12345 hello_world_calculator.cc:45] HelloWorldCalculator initialized
# I20260312 09:00:00.000100 12345 hello_world_calculator.cc:46]   greeting: Hello
# I20260312 09:00:00.000100 12345 hello_world_calculator.cc:47]   uppercase: false
# I20260312 09:00:00.000100 12345 hello_world_calculator.cc:48]   suffix: !
# I20260312 09:00:00.000200 12345 hello_world_demo.cc:78] Graph config: ...
# Hello, IMS!

8.3 常见编译错误

错误 1：找不到 Proto 文件

1
2
3
4
5
6
error: 'mediapipe/calculators/my/hello_world_calculator_options.pb.h' file not found

解决：
1. 检查 Proto 文件路径是否正确
2. 检查 BUILD 文件中的 mediapipe_proto_library 是否正确配置
3. 检查 deps 是否包含 Proto 库

错误 2：Calculator not found

1
2
3
4
5
6
Calculator "HelloWorldCalculator" not found

解决：
1. 检查是否添加了 alwayslink = 1
2. 检查 REGISTER_CALCULATOR 是否正确
3. 检查 Calculator 库是否被正确链接

错误 3：链接错误

1
2
3
4
5
6
undefined reference to `mediapipe::HelloWorldCalculator::GetContract(...)'

解决：
1. 检查 cc_library 的 deps 是否完整
2. 检查 Calculator 是否正确继承 CalculatorBase
3. 检查命名空间是否正确

九、单元测试

9.1 编写测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
// mediapipe/calculators/my/hello_world_calculator_test.cc
#include "mediapipe/framework/calculator_framework.h"
#include "mediapipe/framework/calculator_runner.h"
#include "mediapipe/framework/port/gmock.h"
#include "mediapipe/framework/port/gtest.h"
#include "mediapipe/framework/port/status_matchers.h"

namespace mediapipe {
namespace {

using ::testing::HasSubstr;

TEST(HelloWorldCalculatorTest, ProducesCorrectOutput) {
  // 创建 CalculatorRunner
  CalculatorRunner runner(R"(
    calculator: "HelloWorldCalculator"
    input_stream: "input_name"
    output_stream: "output_greeting"
    options {
      [mediapipe.HelloWorldCalculatorOptions.ext] {
        greeting: "Hello"
        suffix: "!"
      }
    }
  )");

  // 准备输入
  const std::string input = "MediaPipe";
  runner.MutableInputs()
      ->Index(0)
      .packets.push_back(MakePacket<std::string>(input).At(Timestamp(0)));

  // 运行
  MP_ASSERT_OK(runner.Run());

  // 验证输出
  const auto& outputs = runner.Outputs().Index(0).packets;
  ASSERT_EQ(outputs.size(), 1);
  
  const std::string& output = outputs[0].Get<std::string>();
  EXPECT_EQ(output, "Hello, MediaPipe!");
}

TEST(HelloWorldCalculatorTest, HandlesEmptyInput) {
  CalculatorRunner runner(R"(
    calculator: "HelloWorldCalculator"
    input_stream: "input_name"
    output_stream: "output_greeting"
    options {
      [mediapipe.HelloWorldCalculatorOptions.ext] {
        greeting: "Hello"
      }
    }
  )");

  // 不添加输入（模拟空输入）
  
  // 运行
  MP_ASSERT_OK(runner.Run());

  // 验证：空输入应该不产生输出
  const auto& outputs = runner.Outputs().Index(0).packets;
  EXPECT_EQ(outputs.size(), 0);
}

TEST(HelloWorldCalculatorTest, UppercaseOption) {
  CalculatorRunner runner(R"(
    calculator: "HelloWorldCalculator"
    input_stream: "input_name"
    output_stream: "output_greeting"
    options {
      [mediapipe.HelloWorldCalculatorOptions.ext] {
        greeting: "Hello"
        uppercase: true
        suffix: "!"
      }
    }
  )");

  runner.MutableInputs()
      ->Index(0)
      .packets.push_back(MakePacket<std::string>("test").At(Timestamp(0)));

  MP_ASSERT_OK(runner.Run());

  const auto& outputs = runner.Outputs().Index(0).packets;
  ASSERT_EQ(outputs.size(), 1);
  
  const std::string& output = outputs[0].Get<std::string>();
  EXPECT_EQ(output, "HELLO, TEST!");
}

}  // namespace
}  // namespace mediapipe

9.2 运行测试

1
2
3
4
5
6
7
8
9
10
11
12
# 编译测试
bazel build -c opt //mediapipe/calculators/my:hello_world_calculator_test

# 运行测试
bazel test //mediapipe/calculators/my:hello_world_calculator_test

# 输出：
# INFO: Elapsed time: 5.432s, Critical Path: 3.21s
//mediapipe/calculators/my:hello_world_calculator_test   PASSED in 1.2s

# 详细输出
bazel test //mediapipe/calculators/my:hello_world_calculator_test --test_output=all

十、常见问题与解决方案

10.1 编译问题

问题 原因 解决方案
Proto not found BUILD 配置错误 检查 mediapipe_proto_library
Calculator not found 链接优化 添加 alwayslink = 1
Undefined reference deps 不完整 添加缺失的依赖

10.2 运行时问题

问题 原因 解决方案
空输入崩溃 未检查 IsEmpty() 添加空检查
时间戳错误 时间戳不递增 确保时间戳递增
内存泄漏 未在 Close 清理 实现 Close 方法

10.3 调试技巧

1
2
3
4
5
6
7
8
9
10
11
12
// 1. 使用 VLOG 输出调试信息
VLOG(1) << "Debug info: " << value;
// 运行时：GLOG_v=1 ./program

// 2. 使用 LOG(INFO) 输出重要信息
LOG(INFO) << "Important: " << value;

// 3. 使用 CHECK 断言
CHECK(condition) << "Condition failed";

// 4. 使用 DCHECK 调试断言（仅 debug 模式）
DCHECK(condition) << "Debug assertion failed";

十一、总结

11.1 开发 Checklist

  • 定义 Proto Options
  • 实现 Calculator 类（GetContract、Open、Process、Close）
  • 注册 Calculator（REGISTER_CALCULATOR）
  • 编写 Graph 配置
  • 配置 Bazel BUILD（alwayslink = 1）
  • 编写测试程序
  • 编写单元测试

11.2 最佳实践

  1. 输入检查：始终检查输入是否为空
  2. 错误处理：使用 absl::Status 返回错误
  3. 资源管理：在 Open 分配，在 Close 释放
  4. 日志分级：使用 VLOG 调试，LOG(INFO) 重要信息
  5. 单元测试：覆盖正常和边界情况

系列进度： 11/55
更新时间： 2026-03-12

MediaPipe 系列 > Calculator 开发
#MediaPipe #C++ #Calculator #Bazel #Hello World
MediaPipe 系列 11：自定义 Calculator 第一步——从零到运行
https://dapalm.com/2026/03/12/MediaPipe系列11-自定义Calculator第一步：Hello-World/
作者
Mars
发布于
2026年3月12日
许可协议