Framework MediaPipe in Python

Il framework Python di MediaPipe consente l'accesso diretto ai componenti principali del framework C++ di MediaPipe, come Timestamp, Packet e CalculatorGraph, mentre le soluzioni Python pronte all'uso nascondono i dettagli tecnici del framework e restituiscono semplicemente i risultati leggibili di inferenza del modello ai chiamanti.

Il framework MediaPipe si trova sopra la libreria pybind11. Il framework principale di C++ viene esposto in Python tramite un'associazione di linguaggio C++/Python. I contenuti di seguito presupponeno che il lettore abbia già una conoscenza di base del framework C++ di MediaPipe. In caso contrario, puoi trovare informazioni utili in Concetti del framework.

Pacchetto

Il pacchetto è l'unità di flusso di dati di base in MediaPipe. Un pacchetto è costituito da un timestamp numerico e da un puntatore condiviso a un payload immutabile. In Python, è possibile creare un pacchetto MediaPipe chiamando uno dei metodi del creatore di pacchetti nel modulo mp.packet_creator. Di conseguenza, il payload dei pacchetti può essere recuperato utilizzando uno dei metodi getter dei pacchetti nel modulo mp.packet_getter. Tieni presente che il payload del pacchetto diventa immutabile dopo la creazione del pacchetto. Pertanto, la modifica del contenuto del pacchetto recuperato non influisce sul payload effettivo nel pacchetto. L'API Python framework MediaPipe supporta i tipi di dati più utilizzati di MediaPipe (ad es. ImageFrame, Matrix, buffer di protocollo e tipi di dati primitivi) nell'associazione del core. La tabella completa di seguito mostra le mappature dei tipi tra il tipo di dati Python e C++, il creatore del pacchetto e il metodo getter del contenuto per ogni tipo di dati supportato dall'API del framework Python MediaPipe.

Non è raro che gli utenti creino classi C++ personalizzate e le inviino nei grafici e nelle calcolatrici. Per consentire l'utilizzo delle classi personalizzate in Python con il framework MediaPipe, puoi estendere l'API Packet per un nuovo tipo di dati nei passaggi seguenti:

1. Scrivi il codice di associazione di classe pybind11 o un caster di tipo personalizzato per il tipo personalizzato in un file 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 nuovo metodo di creazione dei pacchetti e getter di tipo personalizzato in un file Cc separato.

   #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. Aggiungi due regole di build bazel per l'associazione di tipi personalizzati e i nuovi metodi di pacchetto nel file COSTRUIRE.

   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. Crea i target dell'estensione pybind (con il suffisso .so) da parte di Bazel e sposta le librerie dinamiche generate in una delle directory $LD_LIBRARY_PATH.

5. Utilizza i moduli di associazione in 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)
   

Timestamp

Ogni pacchetto contiene un timestamp espresso in unità di microsecondi. In Python, l'API Packet fornisce un metodo packet.at() per definire il timestamp numerico di un pacchetto. Più in generale, packet.timestamp è la proprietà della classe del pacchetto per accedere al timestamp sottostante. Per convertire un'epoca Unix in un timestamp di MediaPipe, l'API Timestamp offre un metodo mp.Timestamp.from_seconds() a questo scopo.

ImageFrame

ImageFrame è il contenitore per l'archiviazione di un'immagine o di un frame video. I formati supportati da ImageFrame sono elencati nell'enumerazione ImageFormat. I pixel sono codificati nella riga principale con componenti di colori con interleaving e ImageFrame supporta i tipi di dati uint8, uint16 e float. MediaPipe fornisce un'API Python ImageFrame per accedere alla classe ImageFrame C++. In Python, il modo più semplice per recuperare i dati del pixel è chiamare image_frame.numpy_view() per ottenere un ndarray numpy. Tieni presente che il ndarray numpy restituito, un riferimento ai dati dei pixel interni, non è scrivibile. Se i chiamanti devono modificare il ndarray numpy, è necessario chiamare esplicitamente un'operazione di copia per ottenerne una copia. Quando MediaPipe prende un ndarray numpy per creare un ImageFrame, presuppone che i dati siano archiviati in modo contiguo. Di conseguenza, i dati dei pixel di un ImageFrame verranno riallineati per essere contigui quando vengono restituiti al lato Python.

Grafico

Nel framework MediaPipe, tutte le elaborazioni vengono effettuate nel contesto di un Calculator Graph. L'API Python CalculatorGraph è un'associazione diretta alla classe C++ CalculatorGraph. La differenza principale è che l'API Python CalculatorGraph genera un errore Python anziché restituire uno stato non OK quando si verifica un errore. Pertanto, in qualità di utente Python, puoi gestire le eccezioni come faresti normalmente. Il ciclo di vita di una CalculatorGraph contiene tre fasi: inizializzazione e configurazione, esecuzione del grafico e chiusura del grafico.

1. Inizializza un oggetto CalculatorGraph con un protobuf o un file protobuf binario CalculatorGraphConfig e fornisci metodi di callback per osservare i flussi di output.

   Opzione 1. Inizializza un CalculatorGraph con un protobuf CalculatorGraphConfig o una sua rappresentazione testuale e osserva i flussi di output:

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

   Opzione 2. Inizializza un oggetto CalculatorGraph con un file protobuf binario e osserva i flussi di output.

   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. Avvia l'esecuzione del grafico e inserisci i pacchetti nel grafico.

   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. Chiudi il grafico al termine. Puoi riavviare il grafico per eseguire un altro grafico dopo la chiamata a close().

   graph.close()
   

Lo script Python può essere eseguito dal tuo runtime Python locale.