Python의 MediaPipe 프레임워크

MediaPipe Python 프레임워크는 Cloud VPN 게이트웨이의 핵심 구성 요소에 대한 Timestamp, Packet, CalculatorGraph 등 MediaPipe C++ 프레임워크 바로 사용할 수 있는 Python 솔루션은 프레임워크의 기술 세부정보를 가져오고 읽을 수 있는 모델을 반환하기만 하면 됩니다. 호출자에게 반환합니다.

MediaPipe 프레임워크는 pybind11 라이브러리에 있습니다. C++ 핵심 프레임워크는 C++/Python 언어 바인딩을 통해 Python에 노출됩니다. 아래 콘텐츠는 독자가 이미 AI의 기본적인 개념을 이해하고 있다고 가정합니다. MediaPipe C++ 프레임워크입니다. 또는 프레임워크 개념

패킷

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

사용자가 맞춤 C++ 클래스를 만들고 이를 그래프와 계산기가 있습니다. Python에서 커스텀 클래스를 사용하도록 허용하기 위해 MediaPipe 프레임워크로, 새로운 데이터 유형에 대해 패킷 API를 다음 단계를 따르세요.

1. 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. 에서 커스텀 유형의 새 패킷 생성자 및 getter 메서드를 만듭니다. 별도의 cc 파일여야 합니다.

   #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. 커스텀 유형 결합 및 새 패킷에 2개의 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 dirs 중 하나로 이동합니다.

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에서 패킷 API는 편의 메서드 packet.at()를 제공하여 패킷의 타임스탬프입니다. 일반적으로 packet.timestamp는 패킷 클래스입니다. 기본 타임스탬프에 액세스하기 위한 속성입니다. 유닉스 에포크를 MediaPipe 타임스탬프 Timestamp API 는 이를 위해 mp.Timestamp.from_seconds() 메서드를 제공합니다.

ImageFrame

ImageFrame은 이미지 또는 동영상 프레임을 저장하기 위한 컨테이너입니다. 형식 이미지 프레임 지원 목록은 ImageFormat enum 픽셀은 인터리브 처리된 색상 구성요소와 ImageFrame을 사용하여 주요 행 단위로 인코딩됩니다. 는 uint8, uint16 및 float를 데이터 유형으로 지원합니다. MediaPipe는 ImageFrame Python API ImageFrame C++ 클래스에 액세스합니다. Python에서 image_frame.numpy_view()를 호출하여 NumPy ndarray를 얻는 것입니다. 참고 내부 픽셀 데이터에 대한 참조인 반환된 numpy ndarray가 있습니다. 호출자가 numpy ndarray를 수정해야 하는 경우 복사본을 가져오기 위해 복사 작업을 명시적으로 호출합니다. MediaPipe가 NumPy를 받을 때 ndarray를 사용하면 데이터가 연속적으로 저장된다고 가정합니다. 이에 따라 ImageFrame의 픽셀 데이터가 Python 측으로 반환되면 연속됩니다.

그래프

MediaPipe 프레임워크에서 모든 처리는 CalculatorGraph CalculatorGraph Python API 는 C++ CalculatorGraph 클래스에 직접 바인딩됩니다. 가장 큰 차이점은 CalculatorGraph Python API는 오류를 반환하는 대신 Python 오류를 발생시킵니다. non-OK 상태입니다. 따라서 Python 사용자는 예외를 둘 수 있습니다. 계산기 그래프의 수명 주기에는 초기화 및 설정, 그래프 실행, 그래프 종료의 세 단계가 있습니다.

1. CalculatorGraphConfig protobuf 또는 바이너리를 사용하여 계산기 그래프 초기화 protobuf 파일을 열고 출력을 관찰하는 콜백 메서드를 제공합니다. 스트림

   옵션 1. CalculatorGraphConfig protobuf를 사용하여 계산기 그래프 초기화 출력 스트림을 관찰합니다.

   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 런타임에서 실행할 수 있습니다.