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 でサポートされている各データ型のパケット作成者とコンテンツ ゲッター メソッドを示します。
Python のデータ型 | C++ データ型 | パケット作成者 | コンテンツ ゲッター |
---|---|---|---|
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::string | create_string('abc') | get_str(packet) |
バイト | std::string | create_string(b'\xd0\xd0\xd0') | get_bytes(packet) |
mp.Packet | mp::パケット | create_packet(p) | get_packet(packet) |
リスト [ブール値] | std::vector<bool> | create_bool_vector([True, False]) | get_bool_list(packet) |
List[int] または List[np.intc] | int[] | create_int_array([1, 2, 3]) | get_int_list(packet, size=10) |
List[int] または List[np.intc] | std::vector<int> | create_int_vector([1, 2, 3]) | get_int_list(packet) |
List[float] または List[np.float] | float[] | create_float_arrary([0.1, 0.2]) | get_float_list(packet, size=10) |
List[float] または List[np.float] | std::vector<float> | create_float_vector([0.1, 0.2]) | get_float_list(packet, size=10) |
List[str] | std::vector<std::string> | create_string_vector(['a']) | get_str_list(packet) |
リスト [mp.Packet] | std::vector<mp::Packet> | create_packet_vector( [パケット 1, パケット 2]) |
get_packet_list(p) |
マッピング [str, Packet] | std::マップ<std::string, packets=""></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::行列 | create_matrix(data) | get_matrix(packet) |
Google Proto メッセージ | Google Proto メッセージ | create_proto(proto) | get_proto(packet) |
リスト [Proto] | std::vector<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(...); }
カスタムタイプの新しいパケット作成者とゲッター メソッドを、別の 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
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" ], )
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 エポックを 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 つのステージがあります。
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}'))
グラフの実行を開始し、グラフにパケットをフィードします。
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 ランタイムで実行できます。