Raspberry Pi 5 Hailo-8L AIアクセラレータのサンプルコードを解説

import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
import os
import numpy as np
import cv2
import hailo
from hailo_rpi_common import (
    get_caps_from_pad,
    get_numpy_from_buffer,
    app_callback_class,
)
from detection_pipeline import GStreamerDetectionApp

class user_app_callback_class(app_callback_class):
    def __init__(self):
        super().__init__()
        self.new_variable = 42  # 新しい変数（例：42を初期値として設定）

    def new_function(self):  # 新しい関数の定義（例）
        return "The meaning of life is: "

def app_callback(pad, info, user_data):
    # GstBuffer（メディアデータのバッファ）をプローブ情報から取得
    buffer = info.get_buffer()
    # バッファが有効かどうかを確認
    if buffer is None:
        return Gst.PadProbeReturn.OK  # 無効な場合は処理を継続

    # user_dataを使ってフレーム数をカウント
    user_data.increment()
    string_to_print = f"Frame count: {user_data.get_count()}\n"  # フレーム数を表示する文字列を作成

    # パッドからキャプス（フォーマット、幅、高さ）を取得
    format, width, height = get_caps_from_pad(pad)

    # user_data.use_frameがTrueの場合、ビデオフレームをバッファから取得可能
    frame = None
    if user_data.use_frame and format is not None and width is not None and height is not None:
        # ビデオフレームを取得
        frame = get_numpy_from_buffer(buffer, format, width, height)

    # バッファから検出結果を取得
    roi = hailo.get_roi_from_buffer(buffer)
    detections = roi.get_objects_typed(hailo.HAILO_DETECTION)

    # 検出結果を解析
    detection_count = 0
    for detection in detections:
        label = detection.get_label()  # 検出されたオブジェクトのラベルを取得
        bbox = detection.get_bbox()  # バウンディングボックスを取得
        confidence = detection.get_confidence()  # 信頼度を取得
        if label == "person":  # ラベルが「person」の場合
            string_to_print += f"Detection: {label} {confidence:.2f}\n"  # 検出結果を表示
            detection_count += 1

    if user_data.use_frame:
        # 注意：imshowを使うとメインスレッドでないため正しく表示できない
        # フレームに検出数を表示
        cv2.putText(frame, f"Detections: {detection_count}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
        # 新しい変数と新しい関数の結果をフレームに表示
        cv2.putText(frame, f"{user_data.new_function()} {user_data.new_variable}", (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
        # フレームをBGR形式に変換
        frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
        user_data.set_frame(frame)  # 更新したフレームを設定

if __name__ == "__main__":
    # ユーザー定義コールバッククラスのインスタンスを作成
    user_data = user_app_callback_class()
    # GStreamerDetectionAppを初期化し、パイプラインを実行
    app = GStreamerDetectionApp(app_callback, user_data)
    app.run()  # アプリケーションを実行

import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
import os
import numpy as np
import cv2
import hailo
from hailo_rpi_common import (
    get_caps_from_pad,
    get_numpy_from_buffer,
    app_callback_class,
)
from detection_pipeline import GStreamerDetectionApp

# -----------------------------------------------------------------------------------------------
# ユーザー定義のコールバッククラス
# -----------------------------------------------------------------------------------------------
# app_callback_classを継承してカスタマイズしたクラスを作成
class user_app_callback_class(app_callback_class):
    def __init__(self):
        super().__init__()
        self.new_variable = 42  # 新しい変数（例：42を初期値として設定）

    def new_function(self):  # 新しい関数の定義（例）
        return "The meaning of life is: "  # 返り値として意味を持つ文字列を返す

# -----------------------------------------------------------------------------------------------
# ユーザー定義のコールバック関数
# -----------------------------------------------------------------------------------------------

# この関数は、パイプラインからデータが利用可能になったときに呼び出されるコールバック関数
def app_callback(pad, info, user_data):
    # GstBuffer（メディアデータのバッファ）をプローブ情報から取得
    buffer = info.get_buffer()
    # バッファが有効かどうかを確認
    if buffer is None:
        return Gst.PadProbeReturn.OK  # 無効な場合は処理を継続

    # user_dataを使ってフレーム数をカウント
    user_data.increment()
    string_to_print = f"Frame count: {user_data.get_count()}\n"  # フレーム数を表示する文字列を作成

    # パッドからキャプス（フォーマット、幅、高さ）を取得
    format, width, height = get_caps_from_pad(pad)

    # user_data.use_frameがTrueの場合、ビデオフレームをバッファから取得可能
    frame = None
    if user_data.use_frame and format is not None and width is not None and height is not None:
        # ビデオフレームを取得
        frame = get_numpy_from_buffer(buffer, format, width, height)

    # バッファから検出結果を取得
    roi = hailo.get_roi_from_buffer(buffer)
    detections = roi.get_objects_typed(hailo.HAILO_DETECTION)

    # 検出結果を解析
    detection_count = 0
    for detection in detections:
        label = detection.get_label()  # 検出されたオブジェクトのラベルを取得
        bbox = detection.get_bbox()  # バウンディングボックスを取得
        confidence = detection.get_confidence()  # 信頼度を取得
        if label == "person":  # ラベルが「person」の場合
            string_to_print += f"Detection: {label} {confidence:.2f}\n"  # 検出結果を表示
            detection_count += 1
    if user_data.use_frame:
        # 注意：imshowを使うとメインスレッドでないため正しく表示できない
        # フレームに検出数を表示
        cv2.putText(frame, f"Detections: {detection_count}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
        # 新しい変数と新しい関数の結果をフレームに表示
        cv2.putText(frame, f"{user_data.new_function()} {user_data.new_variable}", (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
        # フレームをBGR形式に変換
        frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
        user_data.set_frame(frame)  # 更新したフレームを設定

    print(string_to_print)  # 検出結果をコンソールに表示
    return Gst.PadProbeReturn.OK  # 処理を続行

if __name__ == "__main__":
    # ユーザー定義コールバッククラスのインスタンスを作成
    user_data = user_app_callback_class()
    # GStreamerDetectionAppを初期化し、パイプラインを実行
    app = GStreamerDetectionApp(app_callback, user_data)
    app.run()  # アプリケーションを実行