MediaPipe Python 架構可讓您直接存取 MediaPipe C++ 架構的核心元件,例如時間戳記、封包和計算機圖表,而現成的 Python 解決方案會隱藏架構的技術詳細資料,直接將可讀取的模型推論結果傳回呼叫端。
MediaPipe 架構位於 pybind11 程式庫頂端。C++ 核心架構會透過 C++/Python 語言繫結在 Python 中公開。以下內容假設讀者已對 MediaPipe C++ 架構有基本瞭解。否則,您可以在架構概念中找到實用資訊。
封包
封包是 MediaPipe 中的基本資料流程單位。封包是由數字時間戳記和不可變更酬載的共用指標組成。在 Python 中,您可以在 mp.packet_creator
模組中呼叫其中一個封包建立者方法,藉此建立 MediaPipe 封包。相對地,您可以使用 mp.packet_getter
模組中的任一封包 getter 方法擷取封包酬載。請注意,封包酬載在建立封包之後就「無法變更」。因此,修改已擷取的封包內容,不會影響封包中的實際酬載。MediaPipe 架構 Python API 支援最常用的 MediaPipe 資料類型 (例如在核心繫結中的 ImageFrame、矩陣、通訊協定緩衝區和原始資料類型)。下表提供完整的表格,列出 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 | 建立 16(2**15-1) | get_int(packet) |
int 或 np.int32 | int32_t | 建立 32(2**31-1) | get_int(packet) |
int 或 np.int64 | int64_t | 建立 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 | 建立 uint16(2**16-1) | get_uint(packet) |
int 或 np.uint32 | uint32_t | 建立 uint32(2**32-1) | get_uint(packet) |
int 或 np.uint64 | uint64_t | 建立 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 | 建立字串('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([是, 錯誤]) | get_bool_list(packet) |
List[int] 或 List[np.intc] | int[] | create_int_array([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(封包, size=10) |
List[float] 或 List[np.float] | std::vector<float> | create_float_vector([0.1, 0.2]) | get_float_list(封包, size=10) |
清單 [str] | std::vector<std::string> | 建立字串向量(['a']) | get_str_list(packet) |
清單 [mp.Packet] | std::vector<mp::Packet> | create_packet_vector( [packet1, pack2]) |
get_packet_list(p) |
媒合 [str, 封包] | std::map<std::string, pack=""></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> | n/a | get_proto_list(packet) |
使用者通常建立自訂 C++ 類別,並將這些類別傳送至圖表和計算機。如要允許在 Python 中搭配 MediaPipe Framework 使用自訂類別,請按照下列步驟為新資料類型擴充 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 時間戳記,時間戳記 API 提供 mp.Timestamp.from_seconds()
方法。
ImageFrame
ImageFrame 是一種儲存圖片或影片影格的容器。ImageFrame 支援的格式列於 ImageFormat 列舉中。像素是以交錯色彩元件為列 major 編碼,而 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 架構中,所有處理作業都會在計算機圖表的情境中進行。計算機圖表 Python API 是與 C++ 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 執行階段執行。