Framework MediaPipe en Python

Le framework Python de MediaPipe offre un accès direct aux composants principaux du framework C++ MediaPipe, tels que Timestamp, Packet et CalculatorGraph, tandis que les solutions Python prêtes à l'emploi masquent les détails techniques du framework et renvoient simplement les résultats d'inférence lisibles aux appelants.

Le framework MediaPipe se trouve au-dessus de la bibliothèque pybind11. Le framework principal C++ est exposé en Python via une liaison en langage C++/Python. Le contenu ci-dessous suppose que le lecteur possède déjà des connaissances de base sur le framework C++ MediaPipe. Sinon, vous trouverez des informations utiles dans la section Concepts liés aux frameworks.

Paquet

Le paquet est l'unité de flux de données de base dans MediaPipe. Un paquet se compose d'un horodatage numérique et d'un pointeur partagé vers une charge utile immuable. En Python, vous pouvez créer un paquet MediaPipe en appelant l'une des méthodes de création de paquets dans le module mp.packet_creator. En conséquence, la charge utile du paquet peut être récupérée en utilisant l'une des méthodes du getter de paquets du module mp.packet_getter. Notez que la charge utile du paquet devient immuable après la création du paquet. Ainsi, la modification du contenu du paquet récupéré n'affecte pas la charge utile réelle dans le paquet. L'API Python du framework MediaPipe est compatible avec les types de données MediaPipe les plus couramment utilisés (par exemple, ImageFrame, Matrix, Protocol Buffers et les types de données primitifs) dans la liaison principale. Le tableau complet ci-dessous présente les mappages de types entre les types de données Python et C++, ainsi que le créateur du paquet et la méthode getter de contenu pour chaque type de données compatible avec l'API du framework Python MediaPipe.

Il n'est pas rare que les utilisateurs créent des classes C++ personnalisées et les envoient dans les graphiques et les calculatrices. Pour permettre l'utilisation des classes personnalisées en Python avec MediaPipe Framework, vous pouvez étendre l'API Packet à un nouveau type de données en procédant comme suit:

1. Écrivez le code de liaison de classe pybind11 ou un lanceur de type personnalisé pour le type personnalisé dans un fichier 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. Créez un créateur de paquets et une méthode getter du type personnalisé dans un fichier Cc distinct.

   #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. Ajoutez deux règles de compilation Bazel pour la liaison de type personnalisé et les nouvelles méthodes de paquets dans le fichier BUILD.

   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. Créez les cibles d'extension pybind (avec le suffixe .so) par Bazel, puis déplacez les bibliothèques dynamiques générées vers l'un des répertoires $LD_LIBRARY_PATH.

5. Utilisez les modules de liaison 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)
   

Code temporel

Chaque paquet contient un horodatage exprimé en microsecondes. En Python, l'API Packet fournit une méthode pratique packet.at() pour définir l'horodatage numérique d'un paquet. Plus généralement, packet.timestamp est la propriété de classe de paquet permettant d'accéder à l'horodatage sous-jacent. Pour convertir une epoch Unix en horodatage MediaPipe, l'API Timestamp propose une méthode mp.Timestamp.from_seconds() à cet effet.

ImageFrame

"ImageFrame" est le conteneur permettant de stocker une image ou une image vidéo. Les formats compatibles avec ImageFrame sont répertoriés dans l'énumération ImageFormat. Les pixels sont encodés par ligne principale avec des composants de couleur entrelacés, et ImageFrame prend en charge les types de données uint8, uint16 et float. MediaPipe fournit une API ImageFrame Python pour accéder à la classe C++ ImageFrame. En Python, le moyen le plus simple de récupérer les données de pixels consiste à appeler image_frame.numpy_view() pour obtenir un tableau de données Numpy. Notez que le ndarray Numpy renvoyé, une référence aux données de pixels internes, n'est pas accessible en écriture. Si les appelants doivent modifier le ndarray Numpy, il est nécessaire d'appeler explicitement une opération de copie pour obtenir une copie. Lorsque MediaPipe utilise un ndarray Numpy pour créer un ImageFrame, il suppose que les données sont stockées de manière contiguë. En conséquence, les données de pixels d'un ImageFrame sont réalignées pour être contiguës lorsqu'elles sont renvoyées du côté Python.

Graphique

Dans le framework MediaPipe, l'intégralité du traitement s'effectue dans le contexte d'un CalculatorGraph. L'API Python CalculatorGraph est une liaison directe à la classe CalculatorGraph C++. La principale différence est que l'API Python CalculatorGraph génère une erreur Python au lieu de renvoyer un état non OK en cas d'erreur. Par conséquent, en tant qu'utilisateur Python, vous pouvez gérer les exceptions comme vous le faites habituellement. Le cycle de vie d'un CalculatorGraph comporte trois étapes: l'initialisation et la configuration, l'exécution du graphique et son arrêt.

1. Initialisez un CalculatorGraph avec un fichier protobuf ou binaire CalculatorGraphConfig, puis fournissez une ou plusieurs méthodes de rappel pour observer les flux de sortie.

   Option 1. Initialisez un CalculatorGraph avec un tampon de protocole CalculatorGraphConfig ou sa représentation textuelle, puis observez les flux de sortie :

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

   Option 2. Initialisez un CalculatorGraph avec un fichier protobuf binaire et observez le ou les flux de sortie.

   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. Démarrez l'exécution du graphe et insérez des paquets dans le graphe.

   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. Une fois l'opération terminée, fermez le graphique. Vous pouvez redémarrer le graphique pour une autre exécution de graphe après l'appel de close().

   graph.close()
   

Le script Python peut être exécuté par votre environnement d'exécution Python local.