Python 適用的 MediaPipe 架構

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&lt;float&gt; 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&lt;mp::Packet&gt; create_packet_vector(
[packet1, package2])
get_packet_list(p)
對應 [str、封包] std::map<std::string, Pack=""></std::string,> create_string_to_packet_map(
        {&#39;a&#39;: packet1, &#39;b&#39;: 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 以用於 步驟如下:

  1. 寫入 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(...);
    }
    
  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 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 使用者,您可以 就會視為例外狀況計算機圖的生命週期包含 初始化和設定、圖表執行和圖形關閉

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