はじめに

Raspberry Pi zero2 Wで @PINTO03091 さんのYOLOXの最小最速モデルを動かします。
画像処理約50msec、推論約50msecとなっており実質速度は100msec(10fps)程度をたたき出しました。

この記事の方法では、指定フレーム数の画像データを取得し、取得完了後にmp4データを出力しています。この方法ではリアルタイムに推論画像を確認できないため、こちらの記事で動画をストリーミング出力する様改造しています。
ストリーミング版ではmp4ファイル出力は行っていませんので、用途に応じて修正するか用途によって使い分けてください。

環境

ハードはRPiにカメラを付けるだけで、操作はすべてPCからのヘッドレス運用を行います。

ハードウェア

PC : Raspberry Pi zero2 W
Camera : Arducam IMX708搭載 Raspberry Pi用 固定焦点カメラモジュール（標準）
カメラケーブル : FFC 0.5mmピッチ 22PIN Bタイプ
microSD : 16GB以上程度の適当なもの
USB micro to USB type A ケーブル : データ通信用の適当なもの

RPiとカメラとカメラケーブルだけ新規購入するだけだと¥7880で実行可能。

※カメラは個人的趣味によりV3相当の固定焦点カメラを使用していますが、カメラモジュールV3でもそのまま動くはずです。Raspberry Pi zero2のカメラ設定を変更し、コードのimage キャプチャ部を修正すればUSBカメラやカメラモジュールV2でも動くはずです。

ソフトウェア

RPiのOSはBullseye 64bit lite を使用します。
tfliteのruntimeはYOLOXのモデルは @PINTO03091 さんのリポジトリより使用させていただきます。
それぞれこちらのリポジトリの2.15.0とこちらのリポジトリのyolox_ti_body_head_hand_n_1x3x128x160_bgr_uint8.tfliteを使用します。

環境構築

Raspberry Pi OS

RPiのOSはRaspberry Pi Imagerを使用してください。
使用するOSは、[Raspberry Pi OS (othrer)]⇒[Raspberry Pi OS(Legacy, 64-bit) Lite]です。
イメージの焼き方は紹介サイトがたくさんあると思いますので参考にしてください。
書き込む際に、[設定を編集する]⇒[サービス]からSSHを有効にしておきます。
また、ホスト名を付けて覚えておきます。
wifi接続設定もしておきます。

ヘッドレス設定

Raspberry Pi zero2 Wにはmicro HDMIやUSBポートが付いていますので、キーボードマウスやディスプレイを直接接続して作業できますが、面倒なのでUSBケーブルを通じたSSH接続をするのが便利です。
wifi設定をしておけばwifi接続でのSSHも可能ですが、ルートを二本確保しておいた方が何かと便利です。

USB SSHを使ったヘッドレス運用についてはこちらのサイトやこちらのサイトに詳しいため、ポイントのみ記録しておきます。

RPi側

OSを焼いたmicroSDをPCで読み取り、config.txt とcmdline.txt を編集します。
config.txtは、一番最後に次の一行を追記します。

dtoverlay=dwc2

USB SSHするだけならこの行だけで良いのですが、ついでにIMX708を使うための設定も書き込んでおきます。

dtoverlay=imx708

cmdline.txtは、rootwait と quietの間に次の様に[modules-load=dwc2,g_ether]という文字列を挿入します。

.....rootwait modules-load=dwc2,g_ether quiet......

PC側

PCはwin10、win11 を想定しています。

こちらのサイトの、「Acer Incorporated. - Other hardware - USB Ethernet/RNDIS Gadget」の「Windows 7, Windows 8, Windows 8.1 and later drivers」をダウンロードします。

コマンドプロンプトでダウンロードしたフォルダに入り、次のコマンドで解凍します。

expand '対象ファイル名' -F:* D:\OTGdriver

Raspberry Pi zero2 WにmicroSDカードを指し、USB microケーブルでPCと接続します。USB microコネクタは2個ついていますが、左側（HDMIコネクタ側）に挿します。
デバイスマネージャからRPiが刺さっているUSBシリアルデバイスを探し、[ドライバーの更新]から先ほどダウンロードしたドライバを指定してインストールします。

動作の確認

USB micro ケーブルでPCと接続した状態で適当なツールからSSH接続します。OSを焼いた際に決めたhost名.local をターゲットとしてアクセスできます。

以降の作業はSSH接続したRPi上での作業となります。

swap 無効化と/ tmp RAM化

動画取得時のレイテンシを安定化させるためswapを無効化します。

sudo systemctl stop dphys-swapfile
sudo systemctl disable dphys-swapfile

動画撮影のレイテンシを小さくするためtmpフォルダをRAMディスク化します。

sudo nano /etc/fstab

以下の行を追加

tmpfs /tmp tmpfs defaults,size=64m,noatime,mode=1777 0 0

SD上の/tmpを削除

sudo rm -rf /tmp

カメラモジュールv3をデバイスツリーに追加

先ほどUSB SSHの際にconfig.txtに書き込んでいる場合は不要です。
config.txt にデバイスツリーを追加する設定を書き込みます。

sudo nano /boot/config.txt

以下の行を追加

dtoverlay=imx708

ここまで作業したら一度reboot します。

pipをインストール

sudo apt update && sudo apt  -y upgrade
sudo apt -y install python3-dev python3-pip

picamera2をインストール

Raspberry Pi カメラモジュールV3についているIMX708を制御する為picamera2、opencv、依存ライブラリをインストールします。

pip3 install picamera2
pip3 install opencv-python
sudo apt -y install libgl1-mesa-dev

tflite runtimeをインストール

詳細情報はリポジトリをご覧ください。

sudo apt install -y swig libjpeg-dev zlib1g-dev python-is-python3 unzip wget curl git cmake make
pip install -U pip
pip install numpy
TFVER=2.15.0.post1
PYVER=39
ARCH=aarch64
pip install --no-cache-dir https://github.com/PINTO0309/TensorflowLite-bin/releases/download/v${TFVER}/tflite_runtime-${TFVER/-/}-cp${PYVER}-none-linux_${ARCH}.whl

YOLOXモデルをダウンロード

適当な作業場所を作って作業します。

mkdir ~/YOLOX
cd ~/YOLOX

モデルダウンロード用のスクリプトをダウンロードして実行します。

wget https://raw.githubusercontent.com/PINTO0309/PINTO_model_zoo/main/426_YOLOX-Body-Head-Hand/download_custom_yolox_for_quantization.sh
sudo chmod 755 download_custom_yolox_for_quantization.sh
./download_custom_yolox_for_quantization.sh

種々のモデルが含まれており、全部で1.4GBほどあります。しばらくダウンロードを待ちます。
今回使用するモデルは、以下の三つの8bit量子化されたモデルです。
yolox_ti_body_head_hand_n_1x3x128x160_bgr_uint8.tflite
yolox_ti_body_head_hand_n_1x3x256x320_bgr_uint8.tflite
yolox_ti_body_head_hand_n_1x3x480x640_bgr_uint8.tflite

推論する

作業フォルダ内で以下のコードを実行すると10秒程度の動画を撮影してtest.mp4と言うファイルに保存します。

"""
This demo code is designed to run a lightweight model for edge devices
at high speed instead of degrading accuracy due to INT8 quantization.

runtime: https://github.com/PINTO0309/TensorflowLite-bin
code cited from: https://qiita.com/UnaNancyOwen/items/650d79c88a58a3cc30ce
                https://github.com/PINTO0309/PINTO_model_zoo/blob/main/426_YOLOX-Body-Head-Hand/demo/demo_yolox-ti_tflite.py
"""
import cv2
import time
import numpy as np
from typing import List
from picamera2 import Picamera2
import os

# params
WEIGHTS = "yolox_ti_body_head_hand_n_1x3x128x160_bgr_uint8.tflite"
# WEIGHTS = "yolox_ti_body_head_hand_n_1x3x256x320_bgr_uint8.tflite"
# WEIGHTS = "yolox_ti_body_head_hand_n_1x3x480x640_bgr_uint8.tflite"

NUM_CLASSES = 3
SCORE_THRESHOLD = 0.50
IOU_THRESHOLD = 0.4
CAP_WIDTH   = 320                #出力動画の幅
CAP_HEIGHT = 240                #出力動画の高さ
LAW_WIDTH   = 2304              #カメラ内のraw画像の幅
LAW_HEIGHT = 1296              #カメラ内のraw画像の高さ
count = 0

folder_path ="/tmp/img"
movie_length = 100           #撮影するフレーム数
time_list = []
num_threads = 4                 #スレッド数 1-4を選択
exposure_time = 5000       #イメージセンサの露出時間
analog_gain = 20.0               #イメージセンサのgain
coutner = 0

movie_file_name = "test.mp4"                  #出力ファイル名
codec = cv2.VideoWriter_fourcc('m', 'p', '4', 'v')

# detection model class for yolox
class DetectionModel:
    # constructor
    def __init__(
        self,
        *,
        weight: str,
    ):
        self.__initialize(weight=weight)

    # initialize
    def __initialize(
        self,
        *,
        weight: str,
    ):
        from tflite_runtime.interpreter import Interpreter # type: ignore
        self._interpreter = Interpreter(model_path=weight, num_threads=num_threads)
        self._input_details = self._interpreter.get_input_details()
        self._output_details = self._interpreter.get_output_details()
        self._input_shapes = [
            input.get('shape', None) for input in self._input_details
        ]
        self._input_names = [
            input.get('name', None) for input in self._input_details
        ]
        self._output_shapes = [
            output.get('shape', None) for output in self._output_details
        ]
        self._output_names = [
            output.get('name', None) for output in self._output_details
        ]
        self._model = self._interpreter.get_signature_runner()
        self._h_index = 1
        self._w_index = 2
        strides = [8, 16, 32]
        self.grids, self.expanded_strides = \
            self.__create_grids_and_expanded_strides(strides=strides)

    # create grids and expanded strides
    def __create_grids_and_expanded_strides(
        self,
        *,
        strides: List[int],
    ):
        grids = []
        expanded_strides = []

        hsizes = [self._input_shapes[0][self._h_index] // stride for stride in strides]
        wsizes = [self._input_shapes[0][self._w_index] // stride for stride in strides]

        for hsize, wsize, stride in zip(hsizes, wsizes, strides):
            xv, yv = np.meshgrid(np.arange(wsize), np.arange(hsize))
            grid = np.stack((xv, yv), 2).reshape(1, -1, 2)
            grids.append(grid)
            shape = grid.shape[:2]
            expanded_strides.append(np.full((*shape, 1), stride))

        grids = np.concatenate(grids, 1)
        expanded_strides = np.concatenate(expanded_strides, 1)

        return grids, expanded_strides

    # detect objects
    def __call__(
        self,
        *,
        image: np.ndarray,
        score_threshold: float,
        iou_threshold: float,
    ):
        self.image_shape = image.shape
        prep_image, resize_ratio_w, resize_ratio_h = self.__preprocess(image=image)
        datas = {
            f'{input_name}': input_data \
                for input_name, input_data in zip(self._input_names, [np.asarray([prep_image], dtype=np.uint8)])
        }
        outputs = [
            output for output in \
                self._model(
                    **datas
                ).values()
        ][0]
        boxes, scores, class_ids = \
            self.__postprocess(
                output_blob=outputs,
                resize_ratio_w=resize_ratio_w,
                resize_ratio_h=resize_ratio_h,
            )
        boxes, scores, class_ids = \
            self.__nms(
                boxes=boxes,
                scores=scores,
                class_ids=class_ids,
                score_threshold=score_threshold,
                iou_threshold=iou_threshold,
            )
        return class_ids, scores, boxes

    # preprocess
    def __preprocess(
        self,
        *,
        image: np.ndarray,
    ):
        resize_ratio_w = self._input_shapes[0][self._w_index] / self.image_shape[1]
        resize_ratio_h = self._input_shapes[0][self._h_index] / self.image_shape[0]
        resized_image = \
            cv2.resize(
                image,
                dsize=(self._input_shapes[0][self._w_index], self._input_shapes[0][self._h_index])
            )
        return resized_image, resize_ratio_w, resize_ratio_h

    # postprocess
    def __postprocess(
        self,
        *,
        output_blob: np.ndarray,
        resize_ratio_w: float,
        resize_ratio_h: float,
    ):
        output_blob[..., :2] = (output_blob[..., :2] + self.grids) * self.expanded_strides
        output_blob[..., 2:4] = np.exp(output_blob[..., 2:4]) * self.expanded_strides

        predictions: np.ndarray = output_blob[0]
        boxes = predictions[:, :4]
        boxes_xywh = np.ones_like(boxes)

        # yolox-ti
        boxes[:, 0] = boxes[:, 0] / resize_ratio_w
        boxes[:, 1] = boxes[:, 1] / resize_ratio_h
        boxes[:, 2] = boxes[:, 2] / resize_ratio_w
        boxes[:, 3] = boxes[:, 3] / resize_ratio_h
        boxes_xywh[:, 0] = (boxes[:, 0] - boxes[:, 2] * 0.5)
        boxes_xywh[:, 1] = (boxes[:, 1] - boxes[:, 3] * 0.5)
        boxes_xywh[:, 2] = ((boxes[:, 0] + boxes[:, 2] * 0.5) - boxes_xywh[:, 0])
        boxes_xywh[:, 3] = ((boxes[:, 1] + boxes[:, 3] * 0.5) - boxes_xywh[:, 1])

        scores = predictions[:, 4:5] * predictions[:, 5:]
        class_ids = scores.argmax(1)
        scores = scores[np.arange(len(class_ids)), class_ids]

        return boxes_xywh, scores, class_ids

    # non maximum suppression
    def __nms(
        self,
        *,
        boxes: np.ndarray,
        scores: np.ndarray,
        class_ids: np.ndarray,
        score_threshold: float,
        iou_threshold: float,
    ):
        indices = \
            cv2.dnn.NMSBoxesBatched(
                bboxes=boxes,
                scores=scores,
                class_ids=class_ids,
                score_threshold=score_threshold,
                nms_threshold=iou_threshold,
            ) # OpenCV 4.7.0 or later

        keep_boxes = []
        keep_scores = []
        keep_class_ids = []
        for index in indices:
            keep_boxes.append(boxes[index])
            keep_scores.append(scores[index])
            keep_class_ids.append(class_ids[index])

        if len(keep_boxes) > 0:
            keep_boxes = np.vectorize(int)(keep_boxes)

        return keep_boxes, keep_scores, keep_class_ids

# get raudom colors
def get_colors(num: int):
    colors = []
    np.random.seed(0)
    for _ in range(num):
        color = np.random.randint(0, 256, [3]).astype(np.uint8)
        colors.append(color.tolist())
    return colors


# save mp4
def movie_save():
    global count
    average_time = 0
    for i in range(count - 1):
        average_time += time_list[i + 1]
    average_time /= (count)
    fps = 1 / average_time
    print(fps)
    video = cv2.VideoWriter(movie_file_name, codec, fps, (CAP_WIDTH, CAP_HEIGHT))
    if not video.isOpened():
        print("cant't be opened") 
        sys.exit()
    for i in range(count):
        frame = cv2.imread(folder_path + "image_" + str(i) + ".jpg")
        video.write(frame)
    video.release()

# main
def main():
    global count
    print(count)
    # check folder path
    if not os.path.exists(folder_path):
        os.makedirs(folder_path)
        print("Image folder created")
    else:
        print("Image folder exists")


    # read image
    cap = Picamera2()
    config = cap.create_still_configuration(main={"size":(CAP_WIDTH, CAP_HEIGHT)},raw={"size":(LAW_WIDTH,LAW_HEIGHT)}) 
    cap.configure(config)
    cap.set_controls({"ExposureTime":exposure_time, "AnalogueGain": analog_gain})
    cap.start()


    # create detection model class for yolox
    model = DetectionModel(weight=WEIGHTS)

    # detect objects
    score_threshold = SCORE_THRESHOLD
    iou_threshold = IOU_THRESHOLD

    for i in range(movie_length):
        start_time_frame = time.perf_counter()

        #res, image = cap.read(i)
        image = cap.capture_array()
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        start_time = time.perf_counter()
        class_ids, scores, boxes = \
            model(
                image=image,
                score_threshold=score_threshold,
                iou_threshold=iou_threshold,
            )
        elapsed_time = time.perf_counter() - start_time
        cv2.putText(
            image,
            f'{elapsed_time*1000:.2f} ms',
            (10, 30),
            cv2.FONT_HERSHEY_SIMPLEX,
            0.7,
            (255, 255, 255),
            2,
            cv2.LINE_AA,
        )
        
        cv2.putText(
            image,
            f'{elapsed_time*1000:.2f} ms',
            (10, 30),
            cv2.FONT_HERSHEY_SIMPLEX,
            0.7,
            (0, 0, 255),
            1,
            cv2.LINE_AA,
        )
        
        # draw objects
        num_classes = NUM_CLASSES
        colors = get_colors(num_classes)
        for box, score, class_id in zip(boxes, scores, class_ids):
            color = colors[class_id]
            thickness = 2
            line_type = cv2.LINE_AA
            cv2.rectangle(image, box, color, thickness, line_type)

        # save movie
        cv2.imwrite(folder_path + "image_" + str(count) + ".jpg", image)
        elapsed_time_frame = time.perf_counter() - start_time_frame
        print("frame_number = " + str(count) + " / time = " + str(elapsed_time_frame))
        time_list.append(elapsed_time_frame)
        count += 1

    movie_save()

if __name__ == '__main__':
    main()