Python 中的 MediaPipe 框架

MediaPipe Python 框架会授予对 MediaPipe C++ 框架的核心组件（例如时间戳、数据包和 CalculatorGraph）的直接访问权限，而现成的 Python 解决方案会隐藏框架的技术细节，并且直接将可读的模型推断结果返回给调用方。

MediaPipe 框架基于 pybind11 库。C++ 核心框架通过 C++/Python 语言绑定在 Python 中公开。以下内容假定读者已对 MediaPipe C++ 框架有基本的了解。否则，您可以在框架概念中找到有用的信息。

包

数据包是 MediaPipe 中的基本数据流单元。数据包由数字时间戳和指向不可变载荷的共享指针组成。在 Python 中，可以通过调用 mp.packet_creator 模块中的数据包创建者方法之一创建 MediaPipe 数据包。相应地，可以使用 mp.packet_getter 模块中的数据包 getter 方法之一检索数据包载荷。请注意，创建数据包后，数据包载荷会变得不可变。因此，修改检索到的数据包内容不会影响数据包中的实际载荷。MediaPipe 框架 Python API 支持 MediaPipe 最常用的数据类型（例如，ImageFrame、Matrix、Protocol Buffers 和原始数据类型）。下面的综合表显示了 Python 和 C++ 数据类型之间的类型映射，以及 MediaPipe Python 框架 API 支持的每种数据类型的数据包创建者和内容 getter 方法。

用户创建自定义 C++ 类并将其发送到图表和计算器中，这种情况并不少见。若要允许将自定义类在 Python 中与 MediaPipe 框架搭配使用，您可以在以下步骤中针对新数据类型扩展 Packet API：

1. 在 cc 文件中为自定义类型编写 pybind11 类绑定代码或自定义类型转换器。

   #include "path/to/my_type/header/file.h"
   #include "pybind11/pybind11.h"
   
   namespace py = pybind11;
   
   PYBIND11_MODULE(my_type_binding, m) {
     // Write binding code or a custom type caster for MyType.
     py::class_<MyType>(m, "MyType")
         .def(py::init<>())
         .def(...);
   }
   

2. 在单独的 cc 文件中创建新的自定义类型数据包创建者和 getter 方法。

   #include "path/to/my_type/header/file.h"
   #include "mediapipe/framework/packet.h"
   #include "pybind11/pybind11.h"
   
   namespace mediapipe {
   namespace py = pybind11;
   
   PYBIND11_MODULE(my_packet_methods, m) {
     m.def(
         "create_my_type",
         [](const MyType& my_type) { return MakePacket<MyType>(my_type); });
   
     m.def(
         "get_my_type",
         [](const Packet& packet) {
           if(!packet.ValidateAsType<MyType>().ok()) {
             PyErr_SetString(PyExc_ValueError, "Packet data type mismatch.");
             return py::error_already_set();
           }
           return packet.Get<MyType>();
         });
   }
   }  // namespace mediapipe
   

3. 在 BUILD 文件中为自定义类型绑定和新数据包方法添加两条 Bazel 构建规则。

   load("@pybind11_bazel//:build_defs.bzl", "pybind_extension")
   
   pybind_extension(
       name = "my_type_binding",
       srcs = ["my_type_binding.cc"],
       deps = [":my_type"],
   )
   
   pybind_extension(
       name = "my_packet_methods",
       srcs = ["my_packet_methods.cc"],
       deps = [
           ":my_type",
           "//mediapipe/framework:packet"
       ],
   )
   

4. 使用 Bazel 构建 pybind 扩展程序目标（带有 .so 后缀），并将生成的动态库移至某个 $LD_LIBRARY_PATH 目录中。

5. 在 Python 中使用绑定模块。

   import my_type_binding
   import my_packet_methods
   
   packet = my_packet_methods.create_my_type(my_type_binding.MyType())
   my_type = my_packet_methods.get_my_type(packet)
   

时间戳

每个数据包均包含一个以微秒为单位的时间戳。在 Python 中，Packet API 提供了一种便捷方法 packet.at() 来定义数据包的数字时间戳。更笼统地说，packet.timestamp 是用于访问底层时间戳的数据包类属性。为了将 Unix 纪元转换为 MediaPipe 时间戳，Timestamp API 提供了一个 mp.Timestamp.from_seconds() 方法。

ImageFrame

ImageFrame 是用于存储图片或视频帧的容器。ImageFormat 枚举中列出了 ImageFrame 支持的格式。像素以交错颜色组成部分为主编码，而 ImageFrame 支持 uint8、uint16 和 float 作为其数据类型。MediaPipe 提供了 ImageFrame Python API 来访问 ImageFrame C++ 类。在 Python 中，检索像素数据的最简单方法是调用 image_frame.numpy_view() 以获取多维数组 (numpy ndarray)。请注意，返回的 Numpy ndarray（对内部像素数据的引用）不可写入。如果调用方需要修改 Numpy ndarray，则需要显式调用复制操作来获取副本。当 MediaPipe 使用多维数组来生成 ImageFrame 时，它会假定数据连续存储。相应地，当 ImageFrame 的像素数据返回 Python 端时，它会重新校准为连续。

图表

在 MediaPipe 框架中，所有处理都在 CalculatorGraph 上下文中进行。 CalculatorGraph Python API 是对 C++ CalculatorGraph 类的直接绑定，主要区别在于，CalculatorGraph Python API 会在出现错误时引发 Python 错误，而不是返回非正常状态。因此，作为 Python 用户，您可以照常处理异常。CalculatorGraph 的生命周期包括三个阶段：初始化和设置、图表运行和图表关闭。

1. 使用 CalculatorGraphConfig protobuf 或二进制 protobuf 文件初始化 CalculatorGraph，并提供回调方法来观察输出流。

   方法 1. 使用 CalculatorGraphConfig protobuf 或其文本表示形式初始化 CalculatorGraph，并观察输出流：

   import mediapipe as mp
   
   config_text = """
     input_stream: 'in_stream'
     output_stream: 'out_stream'
     node {
       calculator: 'PassThroughCalculator'
       input_stream: 'in_stream'
       output_stream: 'out_stream'
     }
   """
   graph = mp.CalculatorGraph(graph_config=config_text)
   output_packets = []
   graph.observe_output_stream(
       'out_stream',
       lambda stream_name, packet:
           output_packets.append(mp.packet_getter.get_str(packet)))
   

   方法 2. 使用二进制 protobuf 文件初始化 CalculatorGraph，并观察输出流。

   import mediapipe as mp
   # resources dependency
   
   graph = mp.CalculatorGraph(
       binary_graph=os.path.join(
           resources.GetRunfilesDir(), 'path/to/your/graph.binarypb'))
   graph.observe_output_stream(
       'out_stream',
       lambda stream_name, packet: print(f'Get {packet} from {stream_name}'))
   

2. 开始运行图表并将数据包馈送到图表中。

   graph.start_run()
   
   graph.add_packet_to_input_stream(
       'in_stream', mp.packet_creator.create_string('abc').at(0))
   
   rgb_img = cv2.cvtColor(cv2.imread('/path/to/your/image.png'), cv2.COLOR_BGR2RGB)
   graph.add_packet_to_input_stream(
       'in_stream',
       mp.packet_creator.create_image_frame(image_format=mp.ImageFormat.SRGB,
                                            data=rgb_img).at(1))
   

3. 完成后关闭图表。您可以在调用 close() 后重启该图表，以便运行另一个图表。

   graph.close()
   

Python 脚本可以由本地 Python 运行时运行。