Python の MediaPipe フレームワーク

MediaPipe Python フレームワークは、タイムスタンプ、パケット、電卓グラフなどの MediaPipe C++ フレームワークのコア コンポーネントに直接アクセスします。一方、すぐに使用できる Python ソリューションはフレームワークの技術的な詳細を隠し、読み取り可能なモデル推論の結果を呼び出し元に返します。

MediaPipe フレームワークは、pybind11 ライブラリの上にあります。C++ コア フレームワークは、C++/Python 言語バインディングを介して Python で公開されます。以下のコンテンツは、MediaPipe C++ フレームワークに関する基本的な知識をお持ちであることを前提としています。それ以外の場合は、フレームワークのコンセプトに有用な情報があります。

パケット

パケットは MediaPipe の基本的なデータフロー単位です。パケットは、数値タイムスタンプと、不変ペイロードへの共有ポインタで構成されます。Python では、mp.packet_creator モジュールのパケット作成者メソッドのいずれかを呼び出すことで、MediaPipe パケットを作成できます。また、パケット ペイロードは、mp.packet_getter モジュールのパケット取得メソッドのいずれかを使用して取得できます。パケットの作成後は、パケット ペイロードは不変になります。したがって、取得したパケットの内容を変更しても、パケット内の実際のペイロードには影響しません。MediaPipe フレームワークの Python API は 最もよく使用される MediaPipe のデータ型（ImageFrame、Matrix、プロトコル バッファ、プリミティブ データ型など）をコア バインディングに含めます。以下の表に、Python と C++ のデータ型の型マッピングと、MediaPipe Python Framework 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. BUILD ファイルに、カスタムタイプ バインディング用の 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 は、画像または動画フレームを保存するためのコンテナです。ImageFrame でサポートされている形式については、ImageFormat 列挙型をご覧ください。ピクセルはインターリーブされた色コンポーネントで行優先でエンコードされ、ImageFrame はそのデータ型として uint8、uint16、float をサポートしています。MediaPipe には、ImageFrame C++ クラスにアクセスするための ImageFrame Python API が用意されています。Python でピクセルデータを取得する最も簡単な方法は、image_frame.numpy_view() を呼び出して numpy ndarray を取得することです。返される numpy ndarray（内部ピクセルデータへの参照）は書き込みできないことに注意してください。呼び出し元が numpy ndarray を変更する必要がある場合は、コピーオペレーションを明示的に呼び出してコピーを取得する必要があります。MediaPipe は numpy ndarray を使用して ImageFrame を作成する場合、データが連続して保存されていることを前提としています。これに対応して、ImageFrame のピクセルデータは、Python 側に返されると、連続するように再配置されます。

グラフ

MediaPipe フレームワークでは、すべての処理が CalculatorGraph のコンテキスト内で行われます。CalculatorGraph Python API は、C++ CalculatorGraph クラスへの直接バインディングです。主な違いは、CalculatorGraph Python API が、エラーが発生したときに OK 以外のステータスを返すのではなく、Python エラーを生成することです。そのため、Python ユーザーは通常どおり例外を処理できます。CalculatorGraph のライフサイクルには、初期化と設定、グラフの実行、グラフのシャットダウンの 3 つのステージがあります。

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