Python の MediaPipe フレームワーク

MediaPipe Python フレームワークは、MediaPipe のコア コンポーネントへの MediaPipe C++ フレームワーク（Timestamp、Packet、CalculatorGraph など）です。 一方、すぐに使える Python ソリューションでは、 フレームワークの技術的な詳細を表現し、読み取り可能なモデルを返すだけです。 呼び出し元に返すことができます

MediaPipe フレームワークは Google Cloud の pybind11 ライブラリを使用します。 C++ コア フレームワークは、C++/Python 言語バインディングを介して Python で公開されます。 以下の内容は、読者が MediaPipe C++ フレームワークです。そうでない場合は、 フレームワークのコンセプト。

パケット

パケットは、MediaPipe の基本的なデータフロー ユニットです。パケットは 数値タイムスタンプと不変ペイロードへの共有ポインタです。Python では、 MediaPipe パケットを作成するには、 mp.packet_creator 説明します。これに対応して、パケット ペイロードは、 パケット ゲッター メソッドを mp.packet_getter 説明します。パケットのペイロードは、パケットの後、不変になります。 あります。したがって、取得したパケット コンテンツを変更しても、 パケット内の実際のペイロードを返します。MediaPipe フレームワークの Python API は、 MediaPipe の最もよく使用されるデータ型（例:ImageFrame、Matrix、プロトコル バッファ、プリミティブ データ型など）をコア バインディングに含めます。包括的な 次の表は、Python データ型と C++ データ型の型マッピングを示しています 各データ型のパケット作成者とコンテンツ ゲッター メソッドも示されています。 MediaPipe Python フレームワーク API で サポートされています

ユーザーがカスタム 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. カスタムタイプの新しいパケット作成ツールとゲッター メソッドを 別の cc ファイルに分割することもできます。

   #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. カスタム型バインディングと新しいパケットの 2 つの 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 タイムスタンプ、 Timestamp API この目的のために mp.Timestamp.from_seconds() メソッドが用意されています。

ImageFrame

ImageFrame は、画像または動画フレームを保存するためのコンテナです。形式 主な機能の一覧は、 ImageFormat 列挙型。 ピクセルは、インターリーブされた色コンポーネントと ImageFrame を使用して行メジャーにエンコードされます。 データ型として uint8、uint16、float がサポートされています。MediaPipe は、 ImageFrame Python API ImageFrame C++ クラスにアクセスします。Python でデータを取得する最も簡単な方法は、 ピクセルデータは、image_frame.numpy_view() を呼び出して numpy 配列を取得することです。備考 内部ピクセルデータへの参照である numpy ndarray が 書き込み不可にします。呼び出し元が numpy ndarray を変更する必要がある場合は、 コピー操作を明示的に呼び出してコピーを取得できます。MediaPipe が numpy 関数を ndarray を使用して ImageFrame に変換する場合、データが連続して格納されていることを前提としています。 それに応じて、ImageFrame のピクセルデータは、 連続した値を返します。

グラフ

MediaPipe フレームワークでは、すべての処理は、 CalculatorGraph。 CalculatorGraph Python API C++ CalculatorGraph クラスへの直接バインディングです。主な違いは CalculatorGraph Python API は、メッセージを返す代わりに Python エラーを OK 以外のステータスです。そのため、Python ユーザーは、 例外をプロビジョニングしますCalculatorGraph のライフサイクルには 初期化と設定、グラフ実行、グラフシャットダウンの 3 つのステージがあります。

1. CalculatorGraphConfig 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 ランタイムで実行できます。