Python의 MediaPipe 프레임워크

MediaPipe Python 프레임워크는 Timestamp, Packet, CalculatorGraph와 같은 MediaPipe C++ 프레임워크의 핵심 구성요소에 직접 액세스할 수 있는 권한을 부여하는 반면, 즉시 사용할 수 있는 Python 솔루션은 프레임워크의 기술적 세부정보를 숨기고 읽을 수 있는 모델 추론 결과를 호출자에게 반환합니다.

MediaPipe 프레임워크는 pybind11 라이브러리 위에 있습니다. C++ 핵심 프레임워크는 C++/Python 언어 바인딩을 통해 Python에서 노출됩니다. 아래 콘텐츠는 독자가 이미 MediaPipe C++ 프레임워크에 관한 기본 사항을 이해하고 있다고 가정합니다. 또는 프레임워크 개념에서 유용한 정보를 찾을 수 있습니다.

패킷

패킷은 MediaPipe의 기본 데이터 흐름 단위입니다. 패킷은 숫자 타임스탬프와 변경 불가능한 페이로드의 공유 포인터로 구성됩니다. Python에서는 mp.packet_creator 모듈에서 패킷 생성자 메서드 중 하나를 호출하여 MediaPipe 패킷을 만들 수 있습니다. 따라서 패킷 페이로드는 mp.packet_getter 모듈의 패킷 getter 메서드 중 하나를 사용하여 검색할 수 있습니다. 패킷 페이로드는 패킷 생성 후 변경 불가능합니다. 따라서 검색된 패킷 콘텐츠의 수정은 패킷의 실제 페이로드에 영향을 주지 않습니다. MediaPipe 프레임워크 Python API는 MediaPipe에서 가장 일반적으로 사용되는 데이터 유형 (예: ImageFrame, 매트릭스, 프로토콜 버퍼, 원시 데이터 유형)을 포함합니다. 아래의 포괄적인 표는 MediaPipe Python 프레임워크 API에서 지원하는 각 데이터 유형의 콘텐츠 getter 메서드와 함께 Python 및 C++ 데이터 유형 간의 유형 매핑을 보여줍니다.

사용자가 맞춤 C++ 클래스를 만들어 그래프와 계산기로 보내는 것은 드문 일이 아닙니다. MediaPipe 프레임워크를 Python에서 맞춤 클래스를 사용할 수 있도록 하려면 다음 단계에서 새 데이터 유형에 맞게 Packet API를 확장하면 됩니다.

1. cc 파일에 맞춤 유형의 pybind11 클래스 결합 코드 또는 맞춤 유형 캐스터를 작성합니다.

   #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. 별도의 cc 파일에 맞춤 유형의 새 패킷 생성자 및 getter 메서드를 만듭니다.

   #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. BUILD 파일에 커스텀 유형 결합과 새 패킷 메서드에 관한 두 개의 bazel 빌드 규칙을 추가합니다.

   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. Bazel로 pybind 확장 프로그램 타겟 (접미사 .so 포함)을 빌드하고 생성된 동적 라이브러리를 $LD_LIBRARY_PATH dir 중 하나로 이동합니다.

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

타임스탬프

각 패킷에는 마이크로초 단위의 타임스탬프가 포함됩니다. Python에서 Packet API는 패킷의 숫자 타임스탬프를 정의하는 편의 메서드 packet.at()를 제공합니다. 보다 일반적으로 packet.timestamp은 기본 타임스탬프에 액세스하기 위한 패킷 클래스 속성입니다. Unix 에포크를 MediaPipe 타임스탬프로 변환하기 위해 Timestamp API는 이러한 용도로 mp.Timestamp.from_seconds() 메서드를 제공합니다.

ImageFrame

ImageFrame은 이미지 또는 동영상 프레임을 저장하는 컨테이너입니다. ImageFrame에서 지원하는 형식은 ImageFormat enum에 나열되어 있습니다. 픽셀은 인터리브 처리된 색상 구성요소로 행 단위로 인코딩되며 ImageFrame은 uint8, uint16, float를 데이터 유형으로 지원합니다. MediaPipe는 ImageFrame C++ 클래스에 액세스할 수 있는 ImageFrame Python API를 제공합니다. Python에서 픽셀 데이터를 검색하는 가장 쉬운 방법은 image_frame.numpy_view()를 호출하여 Numpy ndarray를 가져오는 것입니다. 반환된 Numpy ndarray라는 내부 픽셀 데이터의 참조는 쓸 수 없습니다. 호출자가 Numpy ndarray를 수정해야 하는 경우 복사 작업을 명시적으로 호출하여 사본을 가져와야 합니다. MediaPipe는 Numpy ndarray를 사용하여 ImageFrame을 만들 때 데이터가 연속적으로 저장된다고 가정합니다. 따라서 ImageFrame의 픽셀 데이터가 Python 측에 반환될 때 인접하도록 재정렬됩니다.

그래프

MediaPipe 프레임워크에서는 모든 처리가 CalculatorGraph의 컨텍스트 내에서 이루어집니다. CalculatorGraph Python API는 C++ CalculatorGraph 클래스에 직접 바인딩됩니다. 주요 차이점은 CalculatorGraph Python API는 오류가 발생할 때 non-OK 상태를 반환하는 대신 Python 오류를 발생시킨다는 것입니다. 따라서 Python 사용자는 평소와 같이 예외를 처리할 수 있습니다. CalculatorGraph의 수명 주기는 초기화 및 설정, 그래프 실행, 그래프 종료라는 세 단계로 구성됩니다.

1. CalculatorGraphConfig protobuf 또는 바이너리 protobuf 파일로 CalculatorGraph를 초기화하고 출력 스트림을 관찰하는 콜백 메서드를 제공합니다.

   옵션 1. CalculatorGraphConfig protobuf 또는 텍스트 표현을 사용하여 CalculatorGraph를 초기화하고 출력 스트림을 관찰합니다.

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

   옵션 2. 바이너리 protobuf 파일로 CalculatorGraph를 초기화하고 출력 스트림을 관찰합니다.

   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. 그래프 실행을 시작하고 그래프에 패킷을 제공합니다.

   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. 완료 후 그래프를 닫습니다. close() 호출 후 다른 그래프 실행을 위해 그래프를 다시 시작할 수 있습니다.

   graph.close()
   

Python 스크립트는 로컬 Python 런타임에서 실행할 수 있습니다.