Python 中的 MediaPipe 框架

MediaPipe Python 框架可以直接访问 MediaPipe C++ 框架，例如 Timestamp、Packet 和 CalculatorGraph， 而现成的 Python 解决方案则隐藏了 框架的技术细节，然后仅返回可读的模型 将推理结果返回给调用方。

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

包

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

用户创建自定义 C++ 类并将其发送到 以及图形和计算器。允许在 Python 中使用自定义类 使用 MediaPipe Framework，您可以扩展 Packet API，以处理 操作步骤：

1. 编写 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. 在 单独的抄送文件。

   #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. 为自定义类型绑定和新数据包添加两个 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，以行主格式编码 支持 uint8、uint16 和 float 作为其数据类型。MediaPipe 提供 一个 ImageFrame Python API 访问 ImageFrame C++ 类。在 Python 中，检索 像素数据是要调用 image_frame.numpy_view() 来获取多维数组 (numpy ndarray)。注意事项 返回的 Numpy ndarray（对内部像素数据的引用）为 不可写入。如果调用方需要修改 numpy ndarray，则需要 显式调用复制操作来获取副本。当 MediaPipe 处理 Numpy 时， ndarray 作为 ImageFrame，假定数据是连续存储的。 相应地，系统会将 ImageFrame 的像素数据重新对齐 与 Python 端保持一致。

图表

在 MediaPipe 框架中，所有处理都是在 CalculatorGraph 中。 CalculatorGraph Python API 是对 C++ CalculatorGraph 类的直接绑定。主要区别在于 CalculatorGraph Python API 将引发 Python 错误，而不是返回 异常状态。因此，作为一名 Python 用户 像往常一样处理异常CalculatorGraph 的生命周期包含 三个阶段：初始化和设置、图运行和图关闭。

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

   方法 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 运行时运行。