Framework de MediaPipe en Python

El framework de MediaPipe para Python otorga acceso directo a los componentes principales del framework de MediaPipe C++, como Timestamp, Packet y CalculatorGraph, mientras que las soluciones de Python listas para usar ocultan los detalles técnicos del framework y simplemente muestran los resultados de la inferencia del modelo legible a los emisores.

El framework de MediaPipe se encuentra en la parte superior de la biblioteca pybind11. El framework principal de C++ se expone en Python a través de una vinculación de lenguaje C++/Python. En el siguiente contenido, se da por sentado que el lector ya tiene conocimientos básicos sobre el framework de C++ de MediaPipe. De lo contrario, puedes encontrar información útil en Conceptos del framework.

Paquete

El paquete es la unidad básica de flujo de datos en MediaPipe. Un paquete consta de una marca de tiempo numérica y un puntero compartido para una carga útil inmutable. En Python, se puede crear un paquete MediaPipe mediante una llamada a uno de los métodos del creador de paquetes en el módulo mp.packet_creator. En consecuencia, la carga útil del paquete se puede recuperar mediante uno de los métodos get de paquetes en el módulo mp.packet_getter. Ten en cuenta que la carga útil del paquete se vuelve inmutable después de la creación del paquete. Por lo tanto, la modificación del contenido del paquete recuperado no afecta la carga útil real en el paquete. La API de Python del framework de MediaPipe admite los tipos de datos de MediaPipe más usados (p.ej., ImageFrame, Matrix, búferes de protocolo y tipos de datos primitivos) en la vinculación de núcleo. En la siguiente tabla completa, se muestran las asignaciones de tipos entre los tipos de datos de Python y C++, junto con el creador de paquetes y el método get de contenido para cada tipo de datos compatible con la API del framework de MediaPipe Python.

Es común que los usuarios creen clases de C++ personalizadas y las envíen a los gráficos y calculadoras. Para permitir que las clases personalizadas se usen en Python con el framework de MediaPipe, puedes extender la API de paquetes para un nuevo tipo de datos en los siguientes pasos:

1. Escribe el código de vinculación de clase pybind11 o una conversión de tipo personalizado para el tipo personalizado en un archivo 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. Crea un nuevo creador de paquetes y un método get del tipo personalizado en un archivo cc independiente.

   #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. Agrega dos reglas de compilación de Bazel para la vinculación de tipos personalizados y los métodos de paquetes nuevos en el archivo COMPILACIÓN.

   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. Compila las orientaciones de extensión de pybind (con el sufijo .so) de Bazel y mueve las bibliotecas dinámicas generadas a uno de los dir de $LD_LIBRARY_PATH.

5. Usar los módulos de vinculación en 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)
   

Marca de tiempo

Cada paquete contiene una marca de tiempo en unidades de microsegundos. En Python, la API de paquetes proporciona un método de conveniencia packet.at() para definir la marca de tiempo numérica de un paquete. En términos más generales, packet.timestamp es la propiedad de la clase del paquete para acceder a la marca de tiempo subyacente. Para convertir un tiempo Unix en una marca de tiempo de MediaPipe, la API de Timestamp ofrece un método mp.Timestamp.from_seconds() para este propósito.

ImageFrame

ImageFrame es el contenedor para almacenar una imagen o un fotograma de video. Los formatos compatibles con ImageFrame se enumeran en la enumeración de ImageFormat. Los píxeles están codificados como fila principal con componentes de color intercalados, y ImageFrame admite uint8, uint16 y float como sus tipos de datos. MediaPipe proporciona una API de Python de ImageFrame para acceder a la clase de C++ de ImageFrame. En Python, la forma más fácil de recuperar los datos de píxeles es llamar a image_frame.numpy_view() para obtener un ndarray de NumPy. Ten en cuenta que no se puede escribir en el ndarray de NumPy que se muestra, una referencia a los datos internos de píxeles. Si los emisores necesitan modificar el ndarray de NumPy, se debe llamar de forma explícita a una operación de copia para obtener una copia. Cuando MediaPipe toma un ndarray de NumPy para crear un ImageFrame, supone que los datos se almacenan de forma contiguo. En consecuencia, los datos de píxeles de un ImageFrame se volverán a alinear para que sean contiguos cuando se muestren al lado de Python.

Gráfico

En el framework de MediaPipe, todo el procesamiento se lleva a cabo dentro del contexto de CalculatorGraph. La API de CalculatorGraph para Python es una vinculación directa a la clase CalculatorGraph de C++. La diferencia principal es que la API de Python para CalculatorGraph genera un error de Python en lugar de mostrar un estado incorrecto cuando se produce un error. Por lo tanto, como usuario de Python, puedes controlar las excepciones como lo haces normalmente. El ciclo de vida de un CalculatorGraph contiene tres etapas: inicialización y configuración, ejecución del grafo y cierre del grafo.

1. Inicializa un objeto CalculatorGraph con un archivo protobuf de CalculatorGraphConfig o protobuf binario y proporciona métodos de devolución de llamada para observar las transmisiones de salida.

   Opción 1: Inicializa un objeto CalculatorGraph con un protobuf de CalculatorGraphConfig o su representación de texto y observa las transmisiones de salida:

   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)))
   

   Opción 2. Inicializa un CalculatorGraph con un archivo protobuf binario y observa los flujos de salida.

   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. Iniciar la ejecución del grafo y enviar paquetes al grafo

   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. Cierra el gráfico después de finalizar. Puedes reiniciar el gráfico para otra ejecución de gráfico después de la llamada a close().

   graph.close()
   

El entorno de ejecución local de Python puede ejecutar la secuencia de comandos de Python.