Khung MediaPipe trong Python

Khung MediaPipe Python cấp quyền truy cập trực tiếp vào các thành phần cốt lõi của khung MediaPipe C++ như Timestamp, Packet và CalculatorGraph, trong khi các giải pháp Python sẵn sàng sử dụng sẽ ẩn thông tin kỹ thuật của khung và chỉ cần trả về kết quả dự đoán mô hình có thể đọc được cho phương thức gọi.

Khung MediaPipe nằm ở đầu thư viện pybind11. Khung cốt lõi C++ được hiển thị bằng Python thông qua liên kết ngôn ngữ C++/Python. Nội dung bên dưới giả định rằng người đọc đã có kiến thức cơ bản về khung MediaPipe C++. Nếu không, bạn có thể tìm thấy thông tin hữu ích trong phần Framework Concepts (Khái niệm về khung).

Gói

Gói này là đơn vị luồng dữ liệu cơ bản trong MediaPipe. Một gói bao gồm một dấu thời gian dạng số và một con trỏ dùng chung đến một tải trọng không thể thay đổi. Trong Python, bạn có thể tạo gói MediaPipe bằng cách gọi một trong các phương thức tạo gói trong mô-đun mp.packet_creator. Tương ứng, bạn có thể truy xuất tải trọng gói dữ liệu bằng một trong các phương thức getter của gói dữ liệu trong mô-đun mp.packet_getter. Lưu ý rằng tải trọng gói trở nên không thể thay đổi sau khi tạo gói. Do đó, việc sửa đổi nội dung gói đã truy xuất không ảnh hưởng đến tải trọng thực tế trong gói đó. API Python hỗ trợ khung MediaPipe hỗ trợ các loại dữ liệu thường dùng nhất của MediaPipe (ví dụ: ImageFrame, Matrix, Protocol Buffers và các kiểu dữ liệu gốc) trong liên kết lõi. Bảng toàn diện dưới đây cho thấy các liên kết kiểu giữa kiểu dữ liệu Python và C++, cùng với trình tạo gói và phương thức getter nội dung cho từng loại dữ liệu được API khung MediaPipe Python hỗ trợ.

Không có gì lạ khi người dùng tạo các lớp C++ tuỳ chỉnh và gửi các lớp đó vào biểu đồ và máy tính. Để cho phép sử dụng các lớp tuỳ chỉnh trong Python bằng Khung MediaPipe, bạn có thể mở rộng API Gói cho một loại dữ liệu mới theo các bước sau:

1. Viết mã liên kết lớp pybind11 hoặc một Caster loại tuỳ chỉnh cho loại tuỳ chỉnh trong tệp 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. Tạo một trình tạo gói mới và phương thức getter của loại tuỳ chỉnh trong một tệp cc riêng.

   #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. Thêm 2 quy tắc bản dựng cơ sở cho liên kết kiểu tuỳ chỉnh và các phương thức gói mới trong tệp 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. Tạo các mục tiêu tiện ích pybind (có hậu tố .so) của Bazel và di chuyển các thư viện động đã tạo vào một trong các thư mục $LD_Library_PATH.

5. Sử dụng các mô-đun liên kết trong 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)
   

Dấu thời gian

Mỗi gói chứa một dấu thời gian tính bằng đơn vị micrô giây. Trong Python, API Gói cung cấp một phương thức packet.at() tiện lợi để xác định dấu thời gian bằng số của một gói. Nói chung, packet.timestamp là thuộc tính lớp gói để truy cập vào dấu thời gian cơ bản. Để chuyển đổi thời gian bắt đầu của hệ thống Unix thành dấu thời gian MediaPipe, API dấu thời gian sẽ cung cấp một phương thức mp.Timestamp.from_seconds() cho mục đích này.

ImageFrame

ImageFrame là vùng chứa để lưu trữ hình ảnh hoặc khung video. Các định dạng mà ImageFrame hỗ trợ được liệt kê trong enum ImageFormat. Pixel là hàng chính được mã hoá với các thành phần màu xen kẽ, và ImageFrame hỗ trợ uint8, uint16 và float làm các kiểu dữ liệu. MediaPipe cung cấp API ImageFrame Python để truy cập vào lớp ImageFrame C++. Trong Python, cách dễ nhất để truy xuất dữ liệu pixel là gọi image_frame.numpy_view() để lấy một ndarray numpy. Xin lưu ý rằng numpy ndarray được trả về, một tham chiếu đến dữ liệu pixel nội bộ, là không thể ghi. Nếu phương thức gọi cần sửa đổi ndarray numpy, bạn phải gọi một cách rõ ràng một thao tác sao chép để lấy bản sao. Khi dùng một ndarray để tạo một ImageFrame, MediaPipe sẽ giả định rằng dữ liệu được lưu trữ liền nhau. Tương ứng, dữ liệu pixel của ImageFrame sẽ được căn chỉnh lại để liền kề khi dữ liệu này được trả về phía Python.

Biểu đồ

Trong Khung MediaPipe, mọi quá trình xử lý đều diễn ra trong bối cảnh của CalculatorGraph. API CalculatorGraph Python là một mối liên kết trực tiếp với lớp CalculatorGraph C++. Điểm khác biệt chính là API CalculatorGraph Python tăng lỗi Python thay vì trả về Trạng thái không OK khi có lỗi. Do đó, là người dùng Python, bạn có thể xử lý các trường hợp ngoại lệ như bình thường. Vòng đời của CalculatorGraph bao gồm 3 giai đoạn: khởi chạy và thiết lập, chạy biểu đồ và tắt biểu đồ.

1. Khởi chạy CalculatorGraph với một tệp protobuf CalculatorGraphConfig hoặc protobuf nhị phân và cung cấp(các) phương thức gọi lại để quan sát(các) luồng đầu ra.

   Tùy chọn 1. Khởi động CalculatorGraph bằng một protobuf CalculatorGraphConfig hoặc bản trình bày văn bản của CalculatorGraph và quan sát(các) luồng đầu ra:

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

   Tùy chọn 2. Khởi động CalculatorGraph với tệp protobuf nhị phân và quan sát(các) luồng đầu ra.

   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. Bắt đầu chạy biểu đồ và đưa các gói nguồn cấp dữ liệu vào biểu đồ.

   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. Đóng biểu đồ sau khi hoàn tất. Bạn có thể khởi động lại biểu đồ để chạy một biểu đồ khác sau khi gọi close().

   graph.close()
   

Bạn có thể chạy tập lệnh Python bằng môi trường thời gian chạy Python cục bộ.