Framework MediaPipe em Python

O framework MediaPipe Python concede acesso direto aos principais componentes do framework do MediaPipe C++, como carimbo de data/hora, pacote e CalculatorGraph, enquanto as soluções Python prontas para uso ocultam os detalhes técnicos do framework e simplesmente retornam os resultados de inferência do modelo legível para os autores da chamada.

O framework MediaPipe fica sobre a biblioteca pybind11 (link em inglês). O framework principal do C++ é exposto em Python por uma vinculação de linguagem C++/Python. O conteúdo abaixo pressupõe que o leitor já tenha uma compreensão básica do framework MediaPipe C++. Também é possível encontrar informações úteis em Conceitos de framework.

Pacote

O pacote é a unidade básica de fluxo de dados no MediaPipe. Um pacote consiste em um carimbo de data/hora numérico e um ponteiro compartilhado para um payload imutável. No Python, um pacote do MediaPipe pode ser criado chamando um dos métodos do criador de pacotes no módulo mp.packet_creator. Da mesma forma, o payload do pacote pode ser recuperado usando um dos métodos getter de pacote no módulo mp.packet_getter. Observe que o payload do pacote se torna imutável após a criação dele. Portanto, a modificação do conteúdo do pacote recuperado não afeta o payload real no pacote. A API Python do framework do MediaPipe oferece suporte aos tipos de dados mais usados do MediaPipe (por exemplo, ImageFrame, Matrix, buffers de protocolo e os tipos de dados primitivos) na vinculação principal. Confira na tabela abaixo os mapeamentos de tipos entre os tipos de dados Python e C++, o criador de pacotes e o método getter de conteúdo para cada tipo de dado aceito pela API MediaPipe Python.

Não é incomum que os usuários criem classes C++ personalizadas e as enviem para gráficos e calculadoras. Para permitir que as classes personalizadas sejam usadas em Python com o MediaPipe Framework, amplie a API Packet para um novo tipo de dados nas seguintes etapas:

1. Escreva o código de vinculação de classe do pybind11 ou um caster de tipo personalizado para o tipo personalizado em um arquivo 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. Crie um novo criador de pacote e um método getter do tipo personalizado em um arquivo cc separado.

   #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. Adicione duas regras de build do Bazel para a vinculação de tipo personalizado e os novos métodos de pacote no arquivo 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. Crie os destinos de extensão pybind (com o sufixo .so) pelo Bazel e mova as bibliotecas dinâmicas geradas para um dos diretórios $LD_LIBRARY_PATH.

5. Usar os módulos de vinculação no 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)
   

Carimbo de data/hora

Cada pacote contém um carimbo de data/hora em unidades de microssegundos. No Python, a API Packet fornece um método de conveniência packet.at() para definir o carimbo de data/hora numérico de um pacote. De modo mais geral, packet.timestamp é a propriedade da classe de pacote para acessar o carimbo de data/hora. Para converter uma época Unix em um carimbo de data/hora do MediaPipe, a API Timestamp oferece um método mp.Timestamp.from_seconds() para essa finalidade.

ImageFrame

ImageFrame é o contêiner para armazenar uma imagem ou um frame de vídeo. Os formatos compatíveis com o ImageFrame estão listados na enumeração ImageFormat. Os pixels são codificados em linha principal com componentes de cores intercaladas, e o ImageFrame oferece suporte a uint8, uint16 e flutuação como tipos de dados. O MediaPipe fornece uma API ImageFrame Python para acessar a classe C++ do ImageFrame. Em Python, a maneira mais fácil de recuperar os dados de pixels é chamar image_frame.numpy_view() para gerar um ndarray numpy. Observe que o numpy ndarray retornado, uma referência aos dados de pixels internos, é não gravável. Se os autores da chamada precisarem modificar a matriz numpy, será necessário chamar explicitamente uma operação de cópia para conseguir uma cópia. Quando o MediaPipe usa um numpy ndarray para criar um ImageFrame, ele pressupõe que os dados são armazenados de forma contígua. Da mesma forma, os dados de pixel de um ImageFrame serão realinhados para serem contíguos quando retornarem para o lado do Python.

Gráfico

No framework do MediaPipe, todo o processamento ocorre no contexto de um CalculatorGraph. A API CalculatorGraph Python é uma vinculação direta com a classe C++ CalculatorGraph. A principal diferença é que a API CalculatorGraph Python gera um erro do Python em vez de retornar um status diferente de OK quando ocorre um erro. Portanto, como usuário de Python, você pode lidar com as exceções normalmente. O ciclo de vida de um CalculatorGraph contém três estágios: inicialização e configuração, execução e encerramento de gráfico.

1. Inicialize um CalculatorGraph com um arquivo protobuf CalculatorGraphConfig ou um arquivo protobuf binário e forneça métodos de callback para observar os streams de saída.

   Opção 1: Inicialize um CalculatorGraph com um protobuf CalculatorGraphConfig ou a representação de texto dele e observe os streams de saída:

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

   Opção 2. Inicialize um CalculatorGraph com um arquivo protobuf binário e observe os streams de saída.

   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. Inicie a execução do gráfico e alimente os pacotes no gráfico.

   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. Fechar o gráfico após o término. É possível reiniciar o gráfico para outra execução após a chamada para close().

   graph.close()
   

O script Python pode ser executado pelo ambiente de execução local do Python.