Платформа MediaPipe на Python

Платформа MediaPipe Python предоставляет прямой доступ к основным компонентам платформы MediaPipe C++, таким как Timestamp, Packet и CalculatorGraph, тогда как готовые к использованию решения Python скрывают технические детали платформы и просто возвращают обратно читаемые результаты вывода модели. звонящим.

Фреймворк MediaPipe находится поверх библиотеки pybind11 . Базовая платформа C++ предоставляется в Python через привязку языков C++/Python. Содержание ниже предполагает, что читатель уже имеет базовое представление о платформе MediaPipe C++. В противном случае вы можете найти полезную информацию в Framework Concepts .

Пакет

Пакет — это базовая единица потока данных в MediaPipe. Пакет состоит из числовой отметки времени и общего указателя на неизменяемую полезную нагрузку. В Python пакет MediaPipe можно создать, вызвав один из методов создания пакета в модуле mp.packet_creator . Соответственно, полезную нагрузку пакета можно получить с помощью одного из методов получения пакетов в модуле mp.packet_getter . Обратите внимание, что полезная нагрузка пакета становится неизменной после создания пакета. Таким образом, изменение содержимого полученного пакета не влияет на фактическую полезную нагрузку в пакете. API Python платформы MediaPipe поддерживает наиболее часто используемые типы данных MediaPipe (например, ImageFrame, Matrix, Protocol Buffers и примитивные типы данных) в базовой привязке. В подробной таблице ниже показаны сопоставления типов между типами данных Python и C++, а также создатель пакета и метод получения контента для каждого типа данных, поддерживаемого API платформы MediaPipe Python.

Нередко пользователи создают собственные классы 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 два правила сборки 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. Создайте целевые объекты расширения pybind (с суффиксом .so) от Bazel и переместите сгенерированные динамические библиотеки в один из каталогов $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 Timestamp предлагает для этой цели метод mp.Timestamp.from_seconds() .

ИзображениеКадр

ImageFrame — это контейнер для хранения изображения или видеокадра. Форматы, поддерживаемые ImageFrame, перечислены в перечислении ImageFormat . Пиксели кодируются по строкам с чередующимися цветовыми компонентами, а ImageFrame поддерживает uint8, uint16 и float в качестве типов данных. MediaPipe предоставляет API ImageFrame Python для доступа к классу ImageFrame C++. В Python самый простой способ получить данные пикселей — вызвать image_frame.numpy_view() чтобы получить numpy ndarray. Обратите внимание, что возвращаемый numpy ndarray, ссылка на внутренние данные пикселей, недоступен для записи. Если вызывающим сторонам необходимо изменить numpy ndarray, необходимо явно вызвать операцию копирования для получения копии. Когда MediaPipe использует numpy ndarray для создания ImageFrame, он предполагает, что данные хранятся последовательно. Соответственно, пиксельные данные ImageFrame будут перестроены так, чтобы быть непрерывными, когда они вернутся на сторону Python.

График

В MediaPipe Framework вся обработка происходит в контексте CalculatorGraph. API Python CalculatorGraph — это прямая привязка к классу C++ CalculatorGraph. Основное отличие заключается в том, что API Python CalculatorGraph выдает ошибку Python вместо возврата статуса «Не ОК» при возникновении ошибки. Таким образом, как пользователь Python, вы можете обрабатывать исключения, как обычно. Жизненный цикл CalculatorGraph состоит из трех этапов: инициализация и настройка, запуск графика и закрытие графика.

1. Инициализируйте CalculatorGraph с помощью файла protobuf CalculatorGraphConfig или двоичного файла protobuf и предоставьте методы обратного вызова для наблюдения за выходным потоком.

   Вариант 1. Инициализируйте CalculatorGraph с помощью protobuf CalculatorGraphConfig или его текстового представления и наблюдайте за выходным потоком(ами):

   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. Инициализируйте CalculatorGraph с помощью двоичного файла protobuf и наблюдайте за выходными потоками.

   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.