Raspberry Pi 5 Hailo-8L AIアクセラレータでカメラ映像をリアルタイムにオブジェクト検出する方法

import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib
import os
import argparse
import multiprocessing
import numpy as np
import setproctitle
import cv2
import time
import hailo
from hailo_rpi_common import (
    get_default_parser,
    QUEUE,
    get_caps_from_pad,
    get_numpy_from_buffer,
    GStreamerApp,
    app_callback_class,
)

# -----------------------------------------------------------------------------------------------
# ユーザー定義のコールバッククラス
# -----------------------------------------------------------------------------------------------
# app_callback_classを継承したクラスを定義
class user_app_callback_class(app_callback_class):
    def __init__(self):
        super().__init__()
        self.new_variable = 42  # 新しい変数の例

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

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

# パイプラインからデータが利用可能になったときに呼び出されるコールバック関数
def app_callback(pad, info, user_data):
    # probe infoからGstBufferを取得
    buffer = info.get_buffer()
    # 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":
            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)
        # user_dataのnew_variableとnew_functionを使用する例
        # フレームにカスタムメッセージを表示
        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にセット
        user_data.set_frame(frame)

    # 結果をコンソールに出力
    print(string_to_print)
    return Gst.PadProbeReturn.OK


# -----------------------------------------------------------------------------------------------
# ユーザー定義のGStreamerアプリケーション
# -----------------------------------------------------------------------------------------------

# hailo_rpi_common.GStreamerAppクラスを継承したクラスを定義
class GStreamerDetectionApp(GStreamerApp):
    def __init__(self, args, user_data):
        # 親クラスのコンストラクタを呼び出す
        super().__init__(args, user_data)
        # 追加の初期化コードをここに追加できる

        # --- 変更点: ソースタイプとビデオソースを設定 ---
        self.source_type = args.source_type  # ソースタイプを設定（usb, rpi, file）
        self.video_source = args.video_source  # ビデオソースを設定（デバイスファイルやファイルパス）

        # モデルに基づいて設定すべきHailoパラメータを設定
        self.batch_size = 2
        self.network_width = 640
        self.network_height = 640
        self.network_format = "RGB"
        nms_score_threshold = 0.3  # NMSのスコアしきい値
        nms_iou_threshold = 0.45   # NMSのIoUしきい値

        # 一時的なコード: 新しいポストプロセスの.soファイルが存在するかチェック
        new_postprocess_path = os.path.join(self.current_path, '../resources/libyolo_hailortpp_post.so')
        if os.path.exists(new_postprocess_path):
            self.default_postprocess_so = new_postprocess_path
        else:
            self.default_postprocess_so = os.path.join(self.postprocess_dir, 'libyolo_hailortpp_post.so')

        if args.hef_path is not None:
            self.hef_path = args.hef_path
        # ネットワークに基づいてHEFファイルのパスを設定
        elif args.network == "yolov6n":
            self.hef_path = os.path.join(self.current_path, '../resources/yolov6n.hef')
        elif args.network == "yolov8s":
            self.hef_path = os.path.join(self.current_path, '../resources/yolov8s_h8l.hef')
        elif args.network == "yolox_s_leaky":
            self.hef_path = os.path.join(self.current_path, '../resources/yolox_s_leaky_h8l_mz.hef')
        else:
            assert False, "Invalid network type"

        # ユーザー定義のラベルJSONファイル
        if args.labels_json is not None:
            self.labels_config = f' config-path={args.labels_json} '
        else:
            self.labels_config = ''

        # コールバック関数を設定
        self.app_callback = app_callback

        # NMSのしきい値などを文字列として設定
        self.thresholds_str = (
            f"nms-score-threshold={nms_score_threshold} "
            f"nms-iou-threshold={nms_iou_threshold} "
            f"output-format-type=HAILO_FORMAT_TYPE_FLOAT32"
        )

        # プロセスタイトルを設定
        setproctitle.setproctitle("Hailo Detection App")

        # パイプラインを作成
        self.create_pipeline()

    # パイプラインの文字列を生成するメソッド
    def get_pipeline_string(self):
        # ソースタイプに応じてsource_elementを設定
        if self.source_type == "rpi":
            source_element = (
                "libcamerasrc name=src_0 ! "
                f"video/x-raw, format={self.network_format}, width=1536, height=864 ! "
                + QUEUE("queue_src_scale")
                + "videoscale ! "
                f"video/x-raw, format={self.network_format}, width={self.network_width}, height={self.network_height}, framerate=30/1 ! "
            )
        elif self.source_type == "usb":
            source_element = (
                f"v4l2src device={self.video_source} name=src_0 ! "
                "video/x-raw, width=640, height=480, framerate=30/1 ! "
            )
        else:
            source_element = (
                f"filesrc location=\"{self.video_source}\" name=src_0 ! "
                + QUEUE("queue_dec264")
                + " qtdemux ! h264parse ! avdec_h264 max-threads=2 ! "
                " video/x-raw, format=I420 ! "
            )
        # 共通のパイプライン要素を追加
        source_element += QUEUE("queue_scale")
        source_element += "videoscale n-threads=2 ! "
        source_element += QUEUE("queue_src_convert")
        source_element += "videoconvert n-threads=3 name=src_convert qos=false ! "
        source_element += f"video/x-raw, format={self.network_format}, width={self.network_width}, height={self.network_height}, pixel-aspect-ratio=1/1 ! "

        # 全体のパイプライン文字列を構築
        pipeline_string = (
            "hailomuxer name=hmux "
            + source_element
            + "tee name=t ! "
            + QUEUE("bypass_queue", max_size_buffers=20)
            + "hmux.sink_0 "
            + "t. ! "
            + QUEUE("queue_hailonet")
            + "videoconvert n-threads=3 ! "
            f"hailonet hef-path={self.hef_path} batch-size={self.batch_size} {self.thresholds_str} force-writable=true ! "
            + QUEUE("queue_hailofilter")
            + f"hailofilter so-path={self.default_postprocess_so} {self.labels_config} qos=false ! "
            + QUEUE("queue_hmuc")
            + "hmux.sink_1 "
            + "hmux. ! "
            + QUEUE("queue_hailo_python")
            + QUEUE("queue_user_callback")
            + "identity name=identity_callback ! "
            + QUEUE("queue_hailooverlay")
            + "hailooverlay ! "
            + QUEUE("queue_videoconvert")
            + "videoconvert n-threads=3 qos=false ! "
            + QUEUE("queue_hailo_display")
            + f"fpsdisplaysink video-sink={self.video_sink} name=hailo_display sync={self.sync} text-overlay={self.options_menu.show_fps} signal-fps-measurements=true "
        )
        # パイプライン文字列を出力
        print(pipeline_string)
        return pipeline_string

if __name__ == "__main__":
    # ユーザーコールバッククラスのインスタンスを作成
    user_data = user_app_callback_class()
    parser = get_default_parser()
    # 追加の引数をここで追加
    parser.add_argument(
        "--network",
        default="yolov6n",
        choices=['yolov6n', 'yolov8s', 'yolox_s_leaky'],
        help="使用するネットワーク。デフォルトはyolov6n。",
    )
    parser.add_argument(
        "--hef-path",
        default=None,
        help="HEFファイルへのパス。",
    )
    parser.add_argument(
        "--labels-json",
        default=None,
        help="カスタムラベルJSONファイルへのパス。",
    )
    # --- 変更点: ソースタイプとビデオソースの引数を追加 ---
    parser.add_argument(
        "--source-type",
        default="usb",
        choices=['usb', 'rpi', 'file'],
        help="ビデオソースのタイプ: 'usb'（USBカメラ）、'rpi'（Raspberry Piカメラ）、または'file'（ビデオファイル）",
    )
    parser.add_argument(
        "--video-source",
        default="/dev/video0",
        help="ビデオソースのデバイス（例: '/dev/video0'）またはファイルパス",
    )
    # 引数をパース
    args = parser.parse_args()
    # アプリケーションのインスタンスを作成
    app = GStreamerDetectionApp(args, user_data)
    # アプリケーションを実行
    app.run()

python realtime_webcam_yolo_detection.py

python realtime_webcam_yolo_detection.py --video-source /dev/video1

python realtime_webcam_yolo_detection.py --network yolov8s

python realtime_webcam_yolo_detection.py --source-type file --video-source /path/to/video.mp4