Python 適用的 MediaPipe 架構

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 方法。

使用者通常建立自訂 C++ 類別，並將這些類別傳送至圖表和計算機。如要允許在 Python 中搭配 MediaPipe Framework 使用自訂類別，請按照下列步驟為新資料類型擴充 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 時間戳記，時間戳記 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 的生命週期包含三個階段：初始化和設定、執行圖形，以及關閉圖表。

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 執行階段執行。