Platforma MediaPipe w Pythonie

Platforma MediaPipe Python zapewnia bezpośredni dostęp do podstawowych komponentów platformy C++ MediaPipe, takich jak sygnatura czasowa, pakiet i kalkulator wykresu, podczas gdy gotowe do użycia rozwiązania w języku Python ukrywają szczegóły techniczne platformy i po prostu zwracają czytelne wyniki wnioskowania modelu do elementów wywołujących.

Platforma MediaPipe znajduje się nad biblioteką pybind11. Podstawowa platforma C++ jest udostępniana w Pythonie przez powiązanie języka C++/Python. W poniższej treści założono, że czytelnik zna już podstawową platformę MediaPipe C++. Więcej przydatnych informacji znajdziesz w sekcji Framework Concepts.

Pakiet

Pakiet jest podstawową jednostką przepływu danych w MediaPipe. Pakiet składa się z liczbowej sygnatury czasowej i wspólnego wskaźnika do ładunku stałego. W Pythonie pakiet MediaPipe można utworzyć przez wywołanie jednej z metod ich tworzenia w module mp.packet_creator. Ponadto ładunek pakietu można pobrać przy użyciu jednej z metod pobierania pakietów w module mp.packet_getter. Pamiętaj, że po utworzeniu pakietu ładunek pakietu staje się niezmienny. Dlatego zmiana zawartości pobranego pakietu nie ma wpływu na rzeczywisty ładunek w pakiecie. Interfejs MediaPipe Framework Python API obsługuje najczęściej używane typy danych MediaPipe (np. ramki obrazu, macierzy, buforów protokołu i typów danych podstawowych) w podstawowym powiązaniu. W kompleksowej tabeli poniżej znajdziesz mapowania typów danych między Pythonem a C++ razem z kreatorem pakietów i metodą pobierania treści w przypadku poszczególnych typów danych obsługiwanych przez interfejs MediaPipe Python Framework API.

Często użytkownicy tworzą własne klasy w C++ i przesyłają je do wykresów oraz kalkulatorów. Aby umożliwić korzystanie z klas niestandardowych w Pythonie z MediaPipe Framework, możesz rozszerzyć interfejs Packet API dla nowego typu danych, wykonując te czynności:

1. W pliku DW wpisz kod powiązania klasy pybind11 lub element rzutujący typ niestandardowy na potrzeby typu niestandardowego.

   #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. Utwórz nowy kreator pakietów i metodę getter niestandardowego typu w osobnym pliku DW.

   #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. Dodaj 2 reguły kompilacji bazel dla wiązania typu niestandardowego i nowe metody pakietów w pliku 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. Utwórz miejsca docelowe rozszerzeń pybind (z przyrostkiem .so) przez Bazel i przenieś wygenerowane biblioteki dynamiczne do jednego z katalogów $LD_LIBRARY_PATH.

5. Użyj modułów wiązania w Pythonie.

   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)
   

Sygnatura czasowa

Każdy pakiet zawiera sygnaturę czasową wyrażoną w mikrosekundach. W Pythonie interfejs Packet API udostępnia wygodną metodę packet.at() do definiowania liczbowej sygnatury czasowej pakietu. Ogólnie rzecz biorąc, packet.timestamp to właściwość klasy pakietu umożliwiająca dostęp do bazowej sygnatury czasowej. Do przekonwertowania epoki uniksowej na sygnaturę czasową MediaPipe z użyciem sygnatury czasowej MediaPipe dostępna jest metoda mp.Timestamp.from_seconds() dostępna w interfejsie Timetamp API.

ImageFrame

ImageFrame to kontener do przechowywania obrazu lub klatki wideo. Formaty obsługiwane przez ImageFrame są wymienione w wyliczeniu ImageFormat. Piksele zakodowane są jako główne wiersze z przeplatanymi komponentami kolorów. Ramka ImageFrame obsługuje typy danych uint8, uint16 i float. MediaPipe udostępnia interfejs API ImageFrame w języku Python umożliwiający dostęp do klasy ImageFrame w C++. W Pythonie najprostszym sposobem pobrania danych pikseli jest wywołanie metody image_frame.numpy_view() w celu uzyskania numpy ndarray. Pamiętaj, że zwrócona tablica numpy, czyli odwołanie do wewnętrznych danych piksela, jest niemożliwa do zapisu. Jeśli elementy wywołujące muszą zmodyfikować numpy ndarray, w celu uzyskania kopii wymagane jest wyraźne wywołanie operacji kopiowania. Gdy MediaPipe tworzy ramkę obrazu z użyciem numpy ndarray, zakłada, że dane są przechowywane w sąsiedztwie. Odpowiednio dane pikselowe elementu ImageFrame zostaną ponownie wyrównane po zwróceniu na stronę Pythona.

Wykres

W ramach MediaPipe Framework wszystkie procesy przebiegają w kontekście kalkulatora Interfejs CalculatorGraph Python API to bezpośrednie powiązanie z klasą CalculatorGraph w C++. Główna różnica polega na tym, że interfejs CalculatorGraph Python API w przypadku błędu zgłasza błąd Pythona, zamiast zwracać stan nieprawidłowy. Dlatego jako użytkownik Pythona możesz obsługiwać wyjątki w zwykły sposób. Cykl życia kalkulatora zawiera 3 etapy: inicjowanie i konfiguracja, uruchomienie wykresu i wyłączenie wykresu.

1. Zainicjuj narzędzie CalculatorGraph za pomocą protobufa CalculatorGraphConfig lub pliku binarnego protokołuprotobuf i podaj metody wywołania zwrotnego do obserwowania strumieni wyjściowych.

   Opcja 1. Zainicjuj program CalculatorGraph za pomocą protobufa CalculatorGraphConfig lub jego tekstowej reprezentacji, i obserwuj strumienie wyjściowe:

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

   Opcja 2. Zainicjuj narzędzie CalculatorGraph za pomocą binarnego pliku protobufa i obserwuj strumienie wyjściowe.

   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. Rozpocznij uruchomienie wykresu i przekaż pakiety do wykresu.

   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. Po zakończeniu zamknij wykres. Możesz ponownie uruchomić wykres, aby utworzyć kolejny wykres uruchomiony po wywołaniu funkcji close().

   graph.close()
   

Skrypt Pythona może zostać uruchomiony przez lokalne środowisko wykonawcze języka Python.