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 方法。
Python 数据类型 | C++ 数据类型 | 数据包创建者 | 内容 getter |
---|---|---|---|
bool | bool | create_bool(True) | get_bool(packet) |
int 或 np.intc | int_t | create_int(1) | get_int(packet) |
int 或 np.int8 | int8_t | create_int8(2**7-1) | get_int(packet) |
int 或 np.int16 | int16_t | create_int16(2**15-1) | get_int(packet) |
int 或 np.int32 | int32_t | create_int32(2**31-1) | get_int(packet) |
int 或 np.int64 | int64_t | create_int64(2**63-1) | get_int(packet) |
int 或 np.uint8 | uint8_t | create_uint8(2**8-1) | get_uint(packet) |
int 或 np.uint16 | uint16_t | create_uint16(2**16-1) | get_uint(packet) |
int 或 np.uint32 | uint32_t | create_uint32(2**32-1) | get_uint(packet) |
int 或 np.uint64 | uint64_t | create_uint64(2**64-1) | get_uint(packet) |
float 或 np.float32 | float | create_float(1.1) | get_float(packet) |
float 或 np.double | 双精度 | create_double(1.1) | get_float(packet) |
str (UTF-8) | std::string | create_string('abc') | get_str(packet) |
字节 | std::string | create_string(b'\xd0\xd0\xd0') | get_bytes(packet) |
mp.Packet | mp::数据包 | create_packet(p) | get_packet(packet) |
列表 [bool] | std::vector<bool> | create_bool_vector([True, False]) | get_bool_list(packet) |
List[int] 或 List[np.intc] | int[] | 创建整数数组([1, 2, 3]) | get_int_list(数据包, size=10) |
List[int] 或 List[np.intc] | std::vector<int> | create_int_vector([1, 2, 3]) | get_int_list(packet) |
List[float] 或 List[np.float] | float[] | create_float_arrary([0.1, 0.2]) | get_float_list(packet, size=10) |
List[float] 或 List[np.float] | std::vector<float> | create_float_vector([0.1, 0.2]) | get_float_list(packet, size=10) |
List[str] | std::vector<std::string> | 创建字符串矢量(['a']) | get_str_list(packet) |
列表 [mp.Packet] | std::vector<mp::Packet> | create_packet_vector( [packet1, packet2]) |
get_packet_list(p) |
映射 [str, Packet] | std::map<std::string, packet=""></std::string,> | create_string_to_packet_map( {'a': packet1, 'b': packet2}) |
get_str_to_packet_dict(packet) |
np.ndarray (cv.mat 和 PIL.Image) |
mp::ImageFrame | create_image_frame( format=ImageFormat.SRGB, data=mat) |
get_image_frame(packet) |
np.ndarray | mp::矩阵 | create_matrix(data) | get_matrix(packet) |
Google Proto 消息 | Google Proto 消息 | create_proto(proto) | get_proto(packet) |
列表 [Proto] | std::vector<Proto> | 不适用 | get_proto_list(packet) |
用户创建自定义 C++ 类并将其发送到图表和计算器中,这种情况并不少见。若要允许将自定义类在 Python 中与 MediaPipe 框架搭配使用,您可以在以下步骤中针对新数据类型扩展 Packet API:
在 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(...); }
在单独的 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
在 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" ], )
使用 Bazel 构建 pybind 扩展程序目标(带有 .so 后缀),并将生成的动态库移至某个 $LD_LIBRARY_PATH 目录中。
在 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 的生命周期包括三个阶段:初始化和设置、图表运行和图表关闭。
使用 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}'))
开始运行图表并将数据包馈送到图表中。
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))
完成后关闭图表。您可以在调用
close()
后重启该图表,以便运行另一个图表。graph.close()
Python 脚本可以由本地 Python 运行时运行。