MediaPipe Framework ב-Python

ה-framework של Python ב-MediaPipe מעניק גישה ישירה לרכיבי הליבה של MediaPipe C++, כמו Timestamp, Packet ו-CalculatorGraph, בעוד שפתרונות Python המוכנים לשימוש מסתירים את הפרטים הטכניים של המסגרת ופשוט מחזירים למתקשרים את תוצאות ההסקה הקריאות של המודל.

ה-framework של MediaPipe נמצא מעל ספריית pybind11. ה-framework של הליבה C++ נחשפת ב-Python באמצעות קישור השפה C++/Python. התוכן שבהמשך מבוסס על ההנחה שלקורא יש כבר הבנה בסיסית של מסגרת MediaPipe C++. אם לא, תוכלו למצוא מידע שימושי בנושא Framework Concepts.

מנה

ה-Package הוא היחידה הבסיסית לזרימת נתונים ב-MediaPipe. חבילה מורכבת מחותמת זמן מספרית ומסמן משותף למטען ייעודי (payload) שלא ניתן לשינוי. ב-Python, אפשר ליצור חבילת MediaPipe על ידי קריאה לאחת מהשיטות של יוצר החבילות במודול mp.packet_creator. בהתאם לכך, אפשר לאחזר את המטען הייעודי (payload) של החבילות באמצעות אחת מהשיטות של מקבל החבילות במודול mp.packet_getter. שימו לב שהמטען הייעודי (payload) של החבילות הופך ללא ניתן לשינוי אחרי יצירת החבילות. לכן, השינוי בתוכן של החבילה שאוחזר לא משפיע על המטען הייעודי (payload) בפועל בחבילה. MediaPipe framework Python API תומך בסוגי הנתונים הנפוצים ביותר של MediaPipe (למשל, ImageFrame , Matrix, מאגרי פרוטוקולים וסוגי נתונים פרימיטיביים) בקישור הליבה. בטבלה המקיפה שבהמשך מוצגים מיפויי הסוגים בין Python לבין סוג הנתונים C++ , יחד עם יוצר החבילות והשיטה של קבלת התוכן לכל סוג נתונים שנתמך על ידי MediaPipe Python framework API.

לעיתים קרובות המשתמשים יוצרים מחלקות C++ מותאמות אישית ושולחים אותן לגרפים ולמחשבונים. כדי לאפשר את השימוש במחלקות בהתאמה אישית ב-Python עם MediaPipe Framework, אפשר להרחיב את ה-Packet API לסוג נתונים חדש, באמצעות השלבים הבאים:

1. כתבו את קוד קישור המחלקה pybind11 או ממיר סוג מותאם אישית עבור הסוג המותאם אישית בקובץ 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. יוצרים חבילות חדשות ושיטת קבלה חדשה מהסוג המותאם אישית, בקובץ נפרד בשדה 'עותק'.

   #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 עבור קישור הסוג המותאם אישית ושיטות המנה החדשות בקובץ ה-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. צור את יעדי התוספים של pybind (עם הסיומת .so) על ידי Bazel, והעבר את הספריות הדינמיות שנוצרו לאחת מה-dirs של $LD_LIBRARY_PATH.

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 ובצף כסוגי הנתונים שלו. MediaPipe מספק ImageFrame Python API כדי לגשת למחלקה C++ של ImageFrame. ב-Python, הדרך הקלה ביותר לאחזר את נתוני הפיקסלים היא לקרוא ל-image_frame.numpy_view() כדי לקבל ndarray numpy. שימו לב שה-Narray numpy שמוחזר, הפניה לנתוני הפיקסלים הפנימיים, לא ניתן לכתיבה. אם המתקשרים צריכים לשנות את המערך ה-numpy, הם צריכים לקרוא לפעולת העתקה באופן מפורש כדי לקבל עותק. כאשר MediaPipe לוקח מערך numpy כדי ליצור ImageFrame, הוא מניח שהנתונים מאוחסנים באופן רציף. בהתאם, נתוני הפיקסלים של ImageFrame ישופרו מחדש כך שיהיו רציפים כשהיא מוחזרת לצד של Python.

תרשים

ב-MediaPipe Framework, כל העיבודים מתבצעים בהקשר של תרשים המחשבון. ממשק ה-API של CalculatorGraph Python הוא קישור ישיר למחלקה C++ CalculatorGraph. ההבדל העיקרי הוא שממשק ה-API של אליהו Python ב-Python מעלה שגיאה ב-Python במקום להחזיר סטטוס לא תקין כשמתרחשת שגיאה. לכן, כמשתמשי Python תוכלו לטפל בחריגים כרגיל. מחזור החיים של תרשים כולל שלושה שלבים: אתחול והגדרה, הרצת גרף וסגירת גרף.

1. בצעו אתחול של CalculatorGraph עם קובץ Protobuf [מחשבון] או עם קובץ protobuf בינארי, וספקו שיטות של קריאה חוזרת (callback) כדי לצפות בזרמי הפלט.

   דרך 1. בצעו אתחול של CalculatorGraph עם a 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. מפעילים מחשבון Chart עם קובץ protobuf בינארי, ובוחנים את מקורות הפלט.

   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.