MediaPipe Python 架構可讓使用者直接存取 MediaPipe C++ 架構,例如 Timestamp、Packet 和 CalculatorGraph 而現成可用的 Python 解決方案 該架構的技術細節,只傳回可讀取的模型 推論結果傳回呼叫端
MediaPipe 架構位於 pybind11 程式庫。 C++ 核心架構會透過 C++/Python 語言繫結在 Python 中公開。 以下內容假設讀者對 MediaPipe C++ 架構如果不是,您可以在 Framework 概念。
封包
封包是 MediaPipe 中的基本資料流單位。封包是由
數值時間戳記,以及指向不可變酬載的共用指標。在 Python 中,
如要建立 MediaPipe 封包,請呼叫以下項目的其中一個封包建立者方法:
這個
mp.packet_creator
敬上
後續課程我們將逐一介紹
預先訓練的 API、AutoML 和自訂訓練相對來說,使用
封包 getter 方法就在
mp.packet_getter
敬上
後續課程我們將逐一介紹
預先訓練的 API、AutoML 和自訂訓練請注意,封包酬載在封包之後「不可變更」
建立。因此,修改擷取的封包內容不會影響
封包中的實際酬載MediaPipe Framework Python API 支援
最常用的 MediaPipe 資料類型 (例如ImageFrame、矩陣、通訊協定
核心繫結中的緩衝區和原始資料類型)。全面
下表顯示 Python 和 C++ 資料類型之間的類型對應
以及每個資料類型的封包建立者和 content getter 方法
MediaPipe Python 架構 API 支援的 3D 繪圖。
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::字串 | create_string('abc') | get_str(packet) |
位元組 | std::字串 | create_string(b'\xd0\xd0\xd0') | get_bytes(packet) |
mp.Packet | mp::Packet | create_packet(p) | get_packet(packet) |
清單 [bool] | std::向量<bool> | create_bool_vector([是, 錯誤]) | get_bool_list(packet) |
List[int] 或 List[np.intc] | 整數 [] | create_int_array([1, 2, 3]) | get_int_list(packet, size=10) |
List[int] 或 List[np.intc] | std::向量<int> | create_int_vector([1, 2, 3]) | get_int_list(packet) |
清單 [浮點值] 或 List [np.float] | float[] | create_float_arrary([0.1, 0.2]) | get_float_list(packet, size=10) |
清單 [浮點值] 或 List [np.float] | std::vector<float> | create_float_vector([0.1, 0.2]) | get_float_list(packet, size=10) |
清單 [str] | std::vector<std::string> | create_string_vector(['a']) | get_str_list(packet) |
清單 [mp.Packet] | std::vector<mp::Packet> | create_packet_vector( [packet1, package2]) |
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::Matrix | create_matrix(data) | get_matrix(packet) |
Google Proto 訊息 | Google Proto 訊息 | create_proto(proto) | get_proto(packet) |
清單 [通訊協定] | std::向量<Proto> | 不適用 | get_proto_list(packet) |
使用者建立自訂 C++ 類別並傳送至 分別是圖表和計算機為了允許在 Python 中使用自訂類別 使用 MediaPipe Framework,您可能要擴充 Packet API 以用於 步驟如下:
寫入 pybind11 類別繫結程式碼 或 自訂類型投放器 專為 cc 檔案中的自訂類型
#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(...); }
在 個別副本檔案。
#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
為自訂類型繫結和新封包新增兩個 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 Epoch 時間轉換為
MediaPipe 時間戳記、
Timestamp API
提供了用於此目的的 mp.Timestamp.from_seconds()
方法。
ImageFrame
ImageFrame 是用來儲存圖片或影片畫面的容器。格式
ImageFrame 支援的
ImageFormat 列舉。
像素是以交錯色彩元件、ImageFrame 編碼的列主要編碼
支援 uint8、uint16 和浮點值做為資料類型。MediaPipe 提供
ImageFrame Python API
存取 ImageFrame C++ 類別。在 Python 中,擷取
像素資料是呼叫 image_frame.numpy_view()
來取得 numpy ndarray。注意事項
您會發現傳回的 numpy ndarray 是內部像素資料的參照,
無法寫入如果呼叫端需要修改 numpy ndarray,則
明確呼叫複製作業來取得副本。當 MediaPipe 接受數字
建立 ImageFrame,它會假設資料是連續儲存。
相對地,ImageFrame 的像素資料也會重新對齊,以便符合
才會連續出現
圖表
在 MediaPipe 架構中,所有處理作業都會進行 計算工具圖表。 計算機圖 Python API 是直接繫結至 C++ CalculatorGraph 類別。主要差異在於 CalculatorGraph Python API 會引發 Python 錯誤,而不是傳回 發生錯誤時的非確定狀態。因此,身為 Python 使用者,您可以 就會視為例外狀況計算機圖的生命週期包含 初始化和設定、圖表執行和圖形關閉
使用 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}'))
開始執行圖形,並將封包傳送至圖表。
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 執行階段執行。