株式会社MICINPublicationへの投稿
📹

Jetson Orin Nanoで物体×温度をRT推論

masatomoty
に公開
NVIDIA
YOLO
Jetson
TensorRT
tech

この記事は MICIN Advent Calendar 2025 の 13 日目の記事です。
https://adventar.org/calendars/11601

前回はInouey1008さんの、「【MICIN入社エントリ】治療用・診断アプリ開発を通して治療の選択肢を広げたい」 でした。

Jetson Orin Nanoとは

エッジAI向けに設計された小型・高性能GPU搭載ボードです。2023年3月に初版が発売され，最新版のsuperは2024年12月発売開始です。日本では秋月電子等で購入可能です。

https://www.nvidia.com/en-us/autonomous-machines/embedded-systems/jetson-orin/nano-super-developer-kit/

最大40TOPSクラスの推論性能を持ち，USBカメラや各種センサと組み合わせることで，

  • 物体検出
  • 画像認識
  • 異常検知
  • ロボティクス制御

といった処理を，クラウドに依存せず端末単体でリアルタイムに実行できます。

本記事では Jetson Orin Nano を使って，

  • YOLOv8 による物体検出（CPU / GPU）
  • MLX90640 サーマルカメラの熱画像取得
  • 検出対象の「平均温度」をリアルタイム表示

という構成を作成し，CPU実行とGPU実行の処理速度の違いを視覚的に比較しました。

基本仕様

項目 仕様
GPU 1024 CUDAコア + 32 Tensorコア
CPU 1.7 TFLOPS（FP16）
メモリ 8GB LPDDR5（102.4 GB/s 帯域，128-bit）
ストレージ microSDカードスロット
カメラ 22レーン MIPI CSI-2（最大4台カメラ接続可能）
USB USB Type-A × 4
USB Type-C × 1
DisplayPort 1 × DisplayPort 1.4（最大4K60対応）
消費電力 7W 〜 25W

Raspberry Piとの違い

Raspberry Piも映像処理は可能ですが，GPUによる本格的なAI推論には限界があります。

項目 Raspberry Pi Jetson Orin Nano
CPU ○ ○
GPU 弱い 強力（CUDA対応）
AI推論 低速 高速
TensorRT × ○
消費電力 低 中

特に TensorRT による最適化済み推論が使える点が，Jetson 最大の強みです。

MLライブラリ

YOLOv8

https://yolov8.com/

Ultralytics が開発している最新世代のリアルタイム物体検出モデル。軽量モデル（yolov8n）から高精度モデルまで幅広く揃っており，Jetson でもリアルタイム推論が可能です。

TensorRT

https://developer.nvidia.com/tensorrt

NVIDIA が提供する GPU 向け推論最適化ライブラリ。 PyTorch などで学習したモデルを最適化し，低レイテンシ・高速推論を実現できます。

センサ・ハードウェア系ライブラリ

Adafruit CircuitPython MLX90640

https://github.com/adafruit/Adafruit_CircuitPython_MLX90640

MLX90640 サーマルカメラを Python から簡単に扱うための公式ライブラリ。I2C 経由で 24×32 の温度データを取得可能。 本記事ではサーマル映像の取得および平均温度算出に使用しています。

実行環境

ハードウェア

  • Nvidia Jetson Orin Nano 8GB
  • USBカメラ：ロジクール Webカメラ 1080P
  • サーマルカメラ：MLX90640

配線

MLX90640 は Jetson の 40pin ヘッダに I2C 接続します。

MLX90640 Jetson
VIN 3.3V (Pin 1)
GND GND (Pin 6)
SDA SDA (Pin 3)
SCL SCL (Pin 5)

ソフトウェア

バージョン情報

===== SYSTEM =====  
JetPack/L4T: R36 (release), REVISION: 4.7
EABI: aarch64
OS: Ubuntu 22.04.5 LTS  

===== CUDA / TensorRT =====  
CUDA: Build cuda_12.6.r12.6/compiler.34714021_0  
TensorRT: 1.0-1

===== PYTHON =====  
Python: Python 3.10.12
pip: 25.3

===== AI LIB =====  
PyTorch: 2.8.0
Ultralytics: 8.3.235
OpenCV: 4.12.0

===== SENSOR / TOOLS =====  
MLX90640: Version: 1.3.7

実行コード

本構成では以下の処理を CPUとGPUで別ファイルに分けて実装しています。

  • USBカメラから RGB 画像取得
  • YOLOv8 による物体検出
  • MLX90640 から熱画像取得
  • 検出バウンディングボックス内の「平均温度」を算出
  • RGB映像とサーマル映像を横並びで表示
  • FPS，推論時間，レイテンシ，時刻のオーバーレイ表示
  • Flask による MJPEG ストリーミング配信

ライブラリ

# requirement.txt
flask
numpy
opencv-python
ultralytics
adafruit-blinka
adafruit-circuitpython-mlx90640
onnx
onnxruntime-gpu

便利関数

# util.py
#!/usr/bin/env python3

import cv2
import numpy as np

# =========================
# 設定定数
# =========================
# カメラ解像度
RGB_WIDTH = 640
RGB_HEIGHT = 480

# MLX90640 の解像度
THERM_H = 24
THERM_W = 32

# サーマル表示レンジ（必要に応じて調整）
VIS_MIN = 20.0  # ℃
VIS_MAX = 40.0  # ℃


# =========================
# ユーティリティ関数
# =========================
def get_thermal_frame(mlx, therm_h=THERM_H, therm_w=THERM_W):
    """
    MLX90640 から温度フレームを取得して 24x32 の 2D np.array[℃] を返す
    
    Args:
        mlx: MLX90640 インスタンス
        therm_h: サーマル画像の高さ（デフォルト: THERM_H）
        therm_w: サーマル画像の幅（デフォルト: THERM_W）
    
    Returns:
        24x32 の温度配列 [℃]
    """
    frame = [0.0] * (therm_w * therm_h)  # 768
    mlx.getFrame(frame)
    img = np.array(frame, dtype=np.float32).reshape((therm_h, therm_w))

    # 必要なら左右/上下反転する
    # img = np.fliplr(img)
    # img = np.flipud(img)

    return img


def thermal_to_color(thermal_img, out_size, vis_min=VIS_MIN, vis_max=VIS_MAX):
    """
    温度画像(24x32) → 疑似カラー(BGR)画像に変換
    VIS_MIN〜VIS_MAX で固定レンジ表示
    
    Args:
        thermal_img: 温度画像 (therm_h x therm_w)
        out_size: 出力サイズ (width, height)
        vis_min: 表示最小温度 [℃]（デフォルト: VIS_MIN）
        vis_max: 表示最大温度 [℃]（デフォルト: VIS_MAX）
    
    Returns:
        疑似カラー画像 (BGR)
    """
    img = np.nan_to_num(thermal_img, nan=np.nanmean(thermal_img))

    # 表示レンジでクリップ
    img = np.clip(img, vis_min, vis_max)

    # 0〜255 に線形マッピング
    img_norm = (img - vis_min) / (vis_max - vis_min)  # 0〜1
    img_norm = (img_norm * 255.0).astype(np.uint8)

    # 拡大
    img_norm = cv2.resize(img_norm, out_size, interpolation=cv2.INTER_CUBIC)

    # 疑似カラー
    img_color = cv2.applyColorMap(img_norm, cv2.COLORMAP_JET)

    # 画面にレンジを表示（おまけ）
    text = f"{vis_min:.1f}C - {vis_max:.1f}C"
    cv2.putText(
        img_color,
        text,
        (10, 30),
        cv2.FONT_HERSHEY_SIMPLEX,
        0.8,
        (255, 255, 255),
        2,
        cv2.LINE_AA,
    )
    cv2.putText(
        img_color,
        "MLX90640",
        (10, 60),
        cv2.FONT_HERSHEY_SIMPLEX,
        0.6,
        (255, 255, 255),
        1,
        cv2.LINE_AA,
    )

    return img_color


def bbox_to_thermal_region(thermal_img, x1, y1, x2, y2, rgb_w, rgb_h,
                            therm_h=THERM_H, therm_w=THERM_W):
    """
    RGB画像上の bbox(x1,y1,x2,y2) をサーマル画像(32x24)に単純スケールでマップし，
    その領域の平均温度[℃] を返す（簡易デモ用）
    
    Args:
        thermal_img: サーマル画像 (therm_h x therm_w)
        x1, y1, x2, y2: RGB画像上のバウンディングボックス座標
        rgb_w, rgb_h: RGB画像の幅と高さ
        therm_h: サーマル画像の高さ（デフォルト: THERM_H）
        therm_w: サーマル画像の幅（デフォルト: THERM_W）
    
    Returns:
        平均温度 [℃]，または None（領域が無効な場合）
    """
    tx1 = int(x1 / rgb_w * therm_w)
    tx2 = int(x2 / rgb_w * therm_w)
    ty1 = int(y1 / rgb_h * therm_h)
    ty2 = int(y2 / rgb_h * therm_h)

    tx1 = max(0, min(therm_w - 1, tx1))
    tx2 = max(0, min(therm_w, tx2))
    ty1 = max(0, min(therm_h - 1, ty1))
    ty2 = max(0, min(therm_h, ty2))

    if tx2 <= tx1 or ty2 <= ty1:
        return None

    region = thermal_img[ty1:ty2, tx1:tx2]
    if region.size == 0:
        return None

    return float(np.mean(region))


def put_text_top_right(img, text, y, scale=0.55,
                       color=(255, 255, 255),
                       thick=2,
                       shadow=True):
    """
    画像右上に視認性の高いテキストを表示
    
    Args:
        img: 画像
        text: 表示テキスト
        y: 上端からのオフセット（px）
        scale: フォントサイズ
        color: テキスト色 (B, G, R)
        thick: 線の太さ
        shadow: シャドウを表示するか
    """
    h, w = img.shape[:2]
    (tw, th), _ = cv2.getTextSize(
        text, cv2.FONT_HERSHEY_SIMPLEX, scale, thick
    )
    x = w - tw - 12
    y_text = y + th

    if shadow:
        cv2.putText(
            img, text,
            (x + 1, y_text + 1),
            cv2.FONT_HERSHEY_SIMPLEX,
            scale,
            (0, 0, 0),
            thick + 2,
            cv2.LINE_AA,
        )

    cv2.putText(
        img, text,
        (x, y_text),
        cv2.FONT_HERSHEY_SIMPLEX,
        scale,
        color,
        thick,
        cv2.LINE_AA,
    )


def put_text_bottom_right(img, text, y_offset, scale=0.8,
                          color=(255, 255, 255),
                          thick=3):
    """
    画像右下に，白色太字でテキストを表示
    
    Args:
        img: 画像
        text: 表示テキスト
        y_offset: 下からのオフセット（px）
        scale: フォントサイズ
        color: テキスト色 (B, G, R)
        thick: 線の太さ
    """
    h, w = img.shape[:2]
    (tw, th), _ = cv2.getTextSize(
        text, cv2.FONT_HERSHEY_SIMPLEX, scale, thick
    )
    x = w - tw - 12  # 右から12px余白
    y_text = h - y_offset  # 下から y_offset px の位置

    # 白色太字で表示
    cv2.putText(
        img, text,
        (x, y_text),
        cv2.FONT_HERSHEY_SIMPLEX,
        scale,
        color,
        thick,
        cv2.LINE_AA,
    )

CPU版

# cpu.py
#!/usr/bin/env python3
"""
CPU-only YOLOv8n + MLX90640 Thermal Dashboard

- GPU/TensorRT を使わずに CPU だけで YOLO 推論
- RGB + サーマルの2画面ダッシュボード
- 検出物体ごとの平均温度表示
"""

import time
from datetime import datetime

import cv2
import numpy as np
import board
import busio
import adafruit_mlx90640
from flask import Flask, Response
from ultralytics import YOLO

from util import (
    RGB_WIDTH,
    RGB_HEIGHT,
    THERM_H,
    THERM_W,
    VIS_MIN,
    VIS_MAX,
    bbox_to_thermal_region,
    get_thermal_frame,
    put_text_bottom_right,
    thermal_to_color,
)

# =========================
# 設定
# =========================
# YOLO モデルパス（CPU用）
MODEL_PATH_PT = "yolov8n.pt"  # 同ディレクトリに置いておく

# =========================
# Flask アプリ
# =========================
app = Flask(__name__)

# =========================
# MLX90640 初期化
# =========================
print("[MLX90640] init I2C...")
i2c = busio.I2C(board.SCL, board.SDA, frequency=400_000)
mlx = adafruit_mlx90640.MLX90640(i2c)
mlx.refresh_rate = adafruit_mlx90640.RefreshRate.REFRESH_8_HZ

# =========================
# カメラ初期化
# =========================
print("[Camera] opening /dev/video0 ...")
cap = cv2.VideoCapture(0)
if not cap.isOpened():
    raise RuntimeError("カメラが開けませんでした (/dev/video0)")

cap.set(cv2.CAP_PROP_FRAME_WIDTH, RGB_WIDTH)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, RGB_HEIGHT)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)  # レイテンシ低減

# =========================
# YOLO モデル (CPU)
# =========================
print("[YOLO] loading CPU model...")
model = YOLO(MODEL_PATH_PT)
print("[YOLO] model loaded (CPU)")


# =========================
# ストリーミング生成 (CPU)
# =========================
def gen_frames():
    """
    CPUのみで YOLO + サーマル の2画面ダッシュボードを生成して MJPEG で返す
    """
    print("[Stream] start generating frames (CPU-only)...")
    prev_time = time.time()
    frame_count = 0

    while True:
        # 古いフレームを捨てる（遅延対策）
        for _ in range(3):
            cap.grab()

        # RGB キャプチャ
        t_rgb_cap = time.time()
        ret, frame_rgb = cap.retrieve()
        if not ret:
            print("Camera read failed")
            break

        # 左右反転したい場合
        frame_rgb = cv2.flip(frame_rgb, 1)

        # サーマルフレーム取得
        try:
            t_th_cap = time.time()
            thermal_raw = get_thermal_frame(mlx)
        except Exception as e:
            print("MLX90640 error:", e)
            t_th_cap = time.time()
            thermal_raw = np.zeros((THERM_H, THERM_W), dtype=np.float32)

        # YOLO 推論 (CPU)
        t0 = time.time()
        results = model(
            frame_rgb,
            stream=False,
            device="cpu",   # ここで CPU を明示
        )
        infer_ms = (time.time() - t0) * 1000.0

        r = results[0]
        annotated_rgb = frame_rgb.copy()
        h, w, _ = annotated_rgb.shape

        # bbox + 平均温度ラベル
        if r.boxes is not None and len(r.boxes) > 0:
            boxes = r.boxes
            xyxy = boxes.xyxy.cpu().numpy().astype(int)
            clses = boxes.cls.cpu().numpy().astype(int)
            confs = boxes.conf.cpu().numpy()

            for box, cls_id, conf in zip(xyxy, clses, confs):
                x1, y1, x2, y2 = box.tolist()
                avg_temp = bbox_to_thermal_region(
                    thermal_raw, x1, y1, x2, y2, w, h
                )

                cv2.rectangle(
                    annotated_rgb,
                    (x1, y1),
                    (x2, y2),
                    (0, 255, 0),
                    2
                )

                name = model.names.get(cls_id, str(cls_id))
                if avg_temp is not None:
                    label = f"{name} {conf:.2f} {avg_temp:.1f}C"
                else:
                    label = f"{name} {conf:.2f}"

                cv2.putText(
                    annotated_rgb,
                    label,
                    (x1, max(0, y1 - 10)),
                    cv2.FONT_HERSHEY_SIMPLEX,
                    0.5,
                    (255, 255, 255),
                    2,
                    cv2.LINE_AA,
                )

        # FPS / レイテンシ計算
        frame_count = 1
        t_now = time.time()
        elapsed = t_now - prev_time
        fps = frame_count / elapsed if elapsed > 0 else 0.0

        lat_rgb_ms = (t_now - t_rgb_cap) * 1000.0
        lat_th_ms  = (t_now - t_th_cap) * 1000.0
        delta_ms   = lat_rgb_ms - lat_th_ms

        ts_rgb = datetime.fromtimestamp(t_rgb_cap).strftime("%H:%M:%S.%f")[:-3]
        ts_th  = datetime.fromtimestamp(t_th_cap).strftime("%H:%M:%S.%f")[:-3]

        # 左上にモデル種別とFPS/推論時間
        cv2.putText(
            annotated_rgb,
            f"YOLOv8n CPU  FPS:{fps:.3f}  inf:{infer_ms:.1f}ms",
            (10, 20),
            cv2.FONT_HERSHEY_SIMPLEX,
            0.5,
            (255, 255, 255),
            2,
            cv2.LINE_AA,
        )

        # 右下に RGB 側の時刻 & レイテンシ
        put_text_bottom_right(
            annotated_rgb,
            f"RGB cap: {ts_rgb}",
            y_offset=70,
            scale=0.9,
            color=(255, 255, 255),
            thick=3,
        )
        put_text_bottom_right(
            annotated_rgb,
            f"RGB lag: {lat_rgb_ms:.1f} ms",
            y_offset=30,
            scale=0.9,
            color=(255, 255, 255),
            thick=3,
        )

        # サーマル画像をカラー化
        thermal_color = thermal_to_color(
            thermal_raw,
            out_size=(w, h),
            vis_min=VIS_MIN,
            vis_max=VIS_MAX
        )

        # 右側（サーマル）の右下に時刻 & レイテンシ & 差分
        put_text_bottom_right(
            thermal_color,
            f"TH cap: {ts_th}",
            y_offset=110,
            scale=0.9,
            color=(255, 255, 255),
            thick=3,
        )
        put_text_bottom_right(
            thermal_color,
            f"TH lag: {lat_th_ms:.1f} ms",
            y_offset=70,
            scale=0.9,
            color=(255, 255, 255),
            thick=3,
        )
        put_text_bottom_right(
            thermal_color,
            f"RGB-TH Δ: {delta_ms:+.1f} ms",
            y_offset=30,
            scale=0.9,
            color=(255, 255, 255),
            thick=3,
        )

        # 2画面合成
        combined = cv2.hconcat([annotated_rgb, thermal_color])

        # MJPEG エンコード
        ret2, buffer = cv2.imencode(
            ".jpg",
            combined,
            [int(cv2.IMWRITE_JPEG_QUALITY), 70],
        )
        if not ret2:
            continue

        prev_time = t_now

        yield (
            b"--frame\r\n"
            b"Content-Type: image/jpeg\r\n\r\n" + buffer.tobytes() + b"\r\n"
        )


# =========================
# Flask ルーティング
# =========================
@app.route("/video")
def video():
    return Response(
        gen_frames(),
        mimetype="multipart/x-mixed-replace; boundary=frame",
    )


@app.route("/")
def index():
    return """
    <html>
      <head>
        <title>YOLO (CPU) + MLX90640 Dashboard</title>
        <style>
          body { margin: 0; background: #000; }
          img  { width: 100%; height: auto; display: block; }
        </style>
      </head>
      <body>
        <img src="/video" />
      </body>
    </html>
    """


if __name__ == "__main__":
    print("[Flask] starting on 0.0.0.0:5001 (CPU-only) ...")
    app.run(host="0.0.0.0", port=5001, debug=False)

GPU版

#!/usr/bin/env python3
"""
GPU (TensorRT or PyTorch CUDA) YOLOv8n + MLX90640 Thermal Dashboard

- GPU / TensorRT を使って YOLO 推論
- RGB + サーマルの2画面ダッシュボード
- 検出物体ごとの平均温度表示
"""

import time
from datetime import datetime

import cv2
import numpy as np
import board
import busio
import adafruit_mlx90640
from flask import Flask, Response
from ultralytics import YOLO

from util import (
    RGB_WIDTH,
    RGB_HEIGHT,
    THERM_H,
    THERM_W,
    VIS_MIN,
    VIS_MAX,
    bbox_to_thermal_region,
    put_text_bottom_right,
    thermal_to_color,
)

# =========================
# 設定
# =========================
# True: TensorRT エンジン (yolov8n.engine)
# False: PyTorch モデル (yolov8n.pt) を GPU(CUDA) で実行
USE_TRT_ENGINE = True
MODEL_PATH_TRT = "yolov8n.engine"
MODEL_PATH_PT = "yolov8n.pt"

# PyTorch モデル利用時の YOLO 入力解像度（エンジンは内部サイズ固定）
YOLO_IMGSZ = 416

# =========================
# Flask アプリ
# =========================
app = Flask(__name__)

# =========================
# MLX90640 初期化
# =========================
print("[MLX90640] init I2C...")
i2c = busio.I2C(board.SCL, board.SDA, frequency=400_000)
mlx = adafruit_mlx90640.MLX90640(i2c)
mlx.refresh_rate = adafruit_mlx90640.RefreshRate.REFRESH_8_HZ

# =========================
# カメラ初期化
# =========================
print("[Camera] opening /dev/video0 ...")
cap = cv2.VideoCapture(0)
if not cap.isOpened():
    raise RuntimeError("カメラが開けませんでした (/dev/video0)")

cap.set(cv2.CAP_PROP_FRAME_WIDTH, RGB_WIDTH)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, RGB_HEIGHT)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)  # バッファを極小にしてレイテンシ低減

# =========================
# YOLO モデル読み込み
# =========================
print("[YOLO] loading model...")
if USE_TRT_ENGINE:
    # TensorRT エンジン（入力サイズはエクスポート時の設定に依存、例: 640x640）
    model = YOLO(MODEL_PATH_TRT)
    print("[YOLO] TensorRT engine loaded")
else:
    # PyTorch モデル（GPU / CUDA）
    model = YOLO(MODEL_PATH_PT)
    print("[YOLO] PyTorch GPU model loaded")

# =========================
# サーマル取得キャッシュ（間引き更新用）
# =========================
last_thermal = np.zeros((THERM_H, THERM_W), dtype=np.float32)
last_thermal_ts = 0.0


def get_thermal_frame_cached(mlx_instance, interval_sec=0.15):
    """
    サーマルフレームをキャッシュして間引き更新する
    150msごとに更新（MLX90640は8Hzなので十分）
    """
    global last_thermal, last_thermal_ts
    now = time.time()
    if now - last_thermal_ts > interval_sec:  # 150ms ごとに更新
        try:
            frame = [0.0] * (THERM_W * THERM_H)
            mlx_instance.getFrame(frame)
            last_thermal = np.array(frame, dtype=np.float32).reshape(
                (THERM_H, THERM_W)
            )
            last_thermal_ts = now
        except Exception as e:
            print("MLX90640 error:", e)
            # エラー時は前回のフレームをそのまま返す
    return last_thermal


# =========================
# ストリーミング生成
# =========================
def gen_frames():
    """
    RGB + サーマル 2画面ダッシュボード用ストリーマー
    左: YOLO付きRGB
    右: サーマル（MLX90640）
    両方にキャプチャ時刻・レイテンシ・差分を表示
    """
    print("[Stream] start generating frames (GPU)...")
    prev_time = time.time()
    frame_count = 0

    fps_start_time = time.time()  # FPS計算用の開始時刻
    fps_frame_count = 0           # FPS計算用のフレームカウント

    last_results = None           # YOLO推論結果のキャッシュ（間引き推論用）
    YOLO_INFER_INTERVAL = 2       # Nフレームに1回だけ推論（2にすると半分）

    while True:
        try:
            # ---- 古いフレームを捨てる（レイテンシ低減）----
            for _ in range(3):
                cap.grab()

            # RGB キャプチャ
            t_rgb_cap = time.time()
            ret, frame_rgb = cap.retrieve()
            if not ret:
                print("Camera read failed")
                break

            # 左右反転（見た目合わせ）
            frame_rgb = cv2.flip(frame_rgb, 1)

            # ---- サーマルフレーム取得（キャッシュ版）----
            t_th_cap = time.time()
            thermal_raw = get_thermal_frame_cached(mlx)  # 24x32, 生温度[℃]

            frame_count += 1
            fps_frame_count += 1

            # ---- YOLO 推論（間引き）----
            # Nフレームに1回だけ推論、それ以外は前回の結果を使用
            if frame_count % YOLO_INFER_INTERVAL == 0 or last_results is None:
                try:
                    t0 = time.time()
                    if USE_TRT_ENGINE:
                        # TensorRT エンジンの場合は imgsz を渡さない
                        last_results = model(frame_rgb, stream=False)
                    else:
                        # PyTorch モデルの場合は imgsz と device を指定
                        last_results = model(
                            frame_rgb,
                            stream=False,
                            device="cuda:0",
                            imgsz=YOLO_IMGSZ,
                        )
                    infer_ms = (time.time() - t0) * 1000.0
                except Exception as e:
                    print(f"YOLO inference error: {e}")
                    # 推論エラー時は前回の結果を使用（なければスキップ）
                    if last_results is None:
                        continue
                    infer_ms = 0.0
            else:
                infer_ms = 0.0  # 今回は推論していない（キャッシュ結果）

            # last_resultsがNoneの場合はスキップ
            if last_results is None:
                continue

            r = last_results[0]
            annotated_rgb = frame_rgb.copy()
            h, w, _ = annotated_rgb.shape

            # バウンディングボックス + 平均温度ラベル
            if r.boxes is not None and len(r.boxes) > 0:
                boxes = r.boxes
                xyxy = boxes.xyxy.cpu().numpy().astype(int)
                clses = boxes.cls.cpu().numpy().astype(int)
                confs = boxes.conf.cpu().numpy()

                for box, cls_id, conf in zip(xyxy, clses, confs):
                    x1, y1, x2, y2 = box.tolist()
                    avg_temp = bbox_to_thermal_region(
                        thermal_raw, x1, y1, x2, y2, w, h
                    )

                    # 枠
                    cv2.rectangle(
                        annotated_rgb,
                        (x1, y1),
                        (x2, y2),
                        (0, 255, 0),
                        2,
                    )

                    # ラベルテキスト
                    name = model.names.get(cls_id, str(cls_id))
                    if avg_temp is not None:
                        label = f"{name} {conf:.2f} {avg_temp:.1f}C"
                    else:
                        label = f"{name} {conf:.2f}"

                    cv2.putText(
                        annotated_rgb,
                        label,
                        (x1, max(0, y1 - 10)),
                        cv2.FONT_HERSHEY_SIMPLEX,
                        0.5,
                        (255, 255, 255),
                        2,
                        cv2.LINE_AA,
                    )

            # ---- FPS / レイテンシ計算 ----
            t_now = time.time()
            elapsed = t_now - prev_time

            # FPS計算：1秒ごとにリセット
            fps_elapsed = t_now - fps_start_time
            if fps_elapsed >= 1.0:
                fps = fps_frame_count / fps_elapsed if fps_elapsed > 0 else 0.0
                fps_frame_count = 0
                fps_start_time = t_now
            else:
                fps = fps_frame_count / fps_elapsed if fps_elapsed > 0 else 0.0

            lat_rgb_ms = (t_now - t_rgb_cap) * 1000.0
            lat_th_ms = (t_now - t_th_cap) * 1000.0
            delta_ms = lat_rgb_ms - lat_th_ms  # RGB側がどれだけ遅いか（+ならRGBが遅延）

            # 表示用の時刻（ミリ秒まで）
            ts_rgb = datetime.fromtimestamp(t_rgb_cap).strftime("%H:%M:%S.%f")[:-3]
            ts_th = datetime.fromtimestamp(t_th_cap).strftime("%H:%M:%S.%f")[:-3]

            # ---- 左側 (RGB+YOLO) 上に情報オーバーレイ ----
            infer_status = (
                f"inf:{infer_ms:.1f}ms" if infer_ms > 0 else "inf:cached"
            )
            imgsz_label = (
                "engine" if USE_TRT_ENGINE else f"{YOLO_IMGSZ}x{YOLO_IMGSZ}"
            )
            backend_label = "TRT" if USE_TRT_ENGINE else "PT-GPU"

            cv2.putText(
                annotated_rgb,
                f"YOLOv8n {backend_label} ({imgsz_label})  FPS:{fps:.3f}  {infer_status}",
                (10, 20),
                cv2.FONT_HERSHEY_SIMPLEX,
                0.5,
                (255, 255, 255),
                2,
                cv2.LINE_AA,
            )

            # 右下に RGB 側の時刻 & レイテンシ
            put_text_bottom_right(
                annotated_rgb,
                f"RGB cap: {ts_rgb}",
                y_offset=70,
                scale=0.9,
                color=(255, 255, 255),
                thick=3,
            )
            put_text_bottom_right(
                annotated_rgb,
                f"RGB lag: {lat_rgb_ms:.1f} ms",
                y_offset=30,
                scale=0.9,
                color=(255, 255, 255),
                thick=3,
            )

            # ---- サーマル画像をカラー化 ----
            thermal_color = thermal_to_color(
                thermal_raw,
                out_size=(w, h),
                vis_min=VIS_MIN,
                vis_max=VIS_MAX,
            )

            # 右側（サーマル）の右下に時刻 & レイテンシ & 差分
            put_text_bottom_right(
                thermal_color,
                f"TH cap: {ts_th}",
                y_offset=110,
                scale=0.9,
                color=(255, 255, 255),
                thick=3,
            )
            put_text_bottom_right(
                thermal_color,
                f"TH lag: {lat_th_ms:.1f} ms",
                y_offset=70,
                scale=0.9,
                color=(255, 255, 255),
                thick=3,
            )
            put_text_bottom_right(
                thermal_color,
                f"RGB-TH Δ: {delta_ms:+.1f} ms",
                y_offset=30,
                scale=0.9,
                color=(255, 255, 255),
                thick=3,
            )

            # ---- 2画面合成（横並び）----
            combined = cv2.hconcat([annotated_rgb, thermal_color])

            # MJPEG用エンコード
            ret2, buffer = cv2.imencode(
                ".jpg",
                combined,
                [int(cv2.IMWRITE_JPEG_QUALITY), 70],
            )
            if not ret2:
                print("JPEG encode failed")
                continue

            prev_time = t_now

            yield (
                b"--frame\r\n"
                b"Content-Type: image/jpeg\r\n\r\n" + buffer.tobytes() + b"\r\n"
            )

        except Exception as e:
            print(f"Frame generation error: {e}")
            import traceback

            traceback.print_exc()
            # エラーが発生してもストリームを継続（少し待ってから再試行）
            time.sleep(0.1)
            continue

# =========================
# Flask ルーティング
# =========================
@app.route("/video")
def video():
    return Response(
        gen_frames(),
        mimetype="multipart/x-mixed-replace; boundary=frame",
    )


@app.route("/")
def index():
    # シンプルな2画面ダッシュボード
    return """
    <html>
      <head>
        <title>YOLO (GPU) + MLX90640 Dashboard</title>
        <style>
          body { margin: 0; background: #000; }
          img  { width: 100%; height: auto; display: block; }
        </style>
      </head>
      <body>
        <img src="/video" />
      </body>
    </html>
    """


if __name__ == "__main__":
    print("[Flask] starting on 0.0.0.0:5002 (GPU) ...")
    app.run(host="0.0.0.0", port=5002, debug=False)

実行コマンド

yolo export model=yolov8n.pt format=engine
python cpu.py # GPUの場合はpython gpu.py

実行結果

CPU 実行と GPU（TensorRT）実行の結果をGIFにしました。
GIFの仕様上，違いが限定的なのはご容赦ください....

CPU

CPU

  • 推論がカクつく
  • 検出フレーム落ちが発生
  • レイテンシが大きく，物体の動きへの追従が遅い

GPU

GPU

  • 推論が滑らか
  • フレームレートが安定
  • 物体の動きに遅れなく追従
  • 平均温度の追従性も大きく向上

ベンチマーク比較

ベンチマーク用のスクリプトを作成してFPS等についても比較を行いました。

# benchmark.py
#!/usr/bin/env python3
"""
cpu.py / gpu.py 相当パイプラインのベンチマークスクリプト

- カメラキャプチャ (/dev/video0)
- MLX90640 サーマル取得（キャッシュ付き）
- YOLOv8n 推論（CPU / GPU / TensorRT）
- bbox → サーマル平均温度マッピング
- サーマル疑似カラー変換

までを含めた「実アプリ寄りの処理時間 / FPS」を計測する。

Flask や JPEG エンコードは含めていないので、
cpu.py / gpu.py より少し有利な数字が出るが、
処理パイプラインの相対比較（CPU vs GPU）には十分。
"""

import time
import statistics as stats

import cv2
import numpy as np
import board
import busio
import adafruit_mlx90640
from ultralytics import YOLO

from util import (
    RGB_WIDTH,
    RGB_HEIGHT,
    THERM_H,
    THERM_W,
    VIS_MIN,
    VIS_MAX,
    bbox_to_thermal_region,
    thermal_to_color,
)

# PyTorch 用の YOLO 入力解像度（gpu.py / cpu.py に合わせる）
YOLO_IMGSZ = 416


# =========================
# MLX90640 初期化
# =========================
def init_mlx():
    print("[MLX90640] init I2C for benchmark...")
    i2c = busio.I2C(board.SCL, board.SDA, frequency=400_000)
    mlx = adafruit_mlx90640.MLX90640(i2c)
    mlx.refresh_rate = adafruit_mlx90640.RefreshRate.REFRESH_8_HZ
    return mlx


# =========================
# カメラ初期化
# =========================
def init_camera():
    print("[Camera] opening /dev/video0 for benchmark...")
    cap = cv2.VideoCapture(0)
    if not cap.isOpened():
        raise RuntimeError("カメラが開けませんでした (/dev/video0)")

    cap.set(cv2.CAP_PROP_FRAME_WIDTH, RGB_WIDTH)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, RGB_HEIGHT)
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
    return cap


# =========================
# サーマル取得キャッシュ（間引き更新用）
# =========================
last_thermal = np.zeros((THERM_H, THERM_W), dtype=np.float32)
last_thermal_ts = 0.0


def get_thermal_frame_cached(mlx_instance, interval_sec=0.15):
    """
    サーマルフレームをキャッシュして間引き更新する
    MLX90640 は 8Hz（約125ms）なので 150ms ごとに更新で十分。
    """
    global last_thermal, last_thermal_ts
    now = time.time()
    if now - last_thermal_ts > interval_sec:
        try:
            frame = [0.0] * (THERM_W * THERM_H)
            mlx_instance.getFrame(frame)
            last_thermal = np.array(frame, dtype=np.float32).reshape(
                (THERM_H, THERM_W)
            )
            last_thermal_ts = now
        except Exception as e:
            print("[WARN] MLX90640 error:", e)
            # エラー時は前回のフレームをそのまま返す
    return last_thermal


# =========================
# ベンチマーク本体
# =========================
def run_benchmark(
    mode: str = "gpu",
    duration_sec: int = 20,
    model_path_trt: str = "yolov8n.engine",
    model_path_pt: str = "yolov8n.pt",
):
    """
    mode:
        - "gpu"       : TensorRT エンジン (yolov8n.engine)
        - "gpu_torch" : PyTorch + CUDA (yolov8n.pt)
        - "cpu"       : PyTorch + CPU   (yolov8n.pt)
    duration_sec:
        - 何秒間ループを回して平均を取るか
    """

    print("========================================")
    print("Pipeline Benchmark (camera + thermal + YOLO)")
    print(f"Mode         : {mode}")
    if mode in ("cpu", "gpu_torch"):
        print(f"YOLO imgsz   : {YOLO_IMGSZ}")
    else:
        print("YOLO imgsz   : engine default (e.g. 640x640)")
    print(f"Duration     : {duration_sec} s")
    print("========================================")

    # ---- モデル読み込み ----
    if mode == "gpu":
        # TensorRT エンジンの場合は imgsz を渡さず、
        # エンジンが持っている入力サイズ（おそらく 640x640）に任せる
        print(f"[YOLO] loading TensorRT engine: {model_path_trt}")
        model = YOLO(model_path_trt)
        predict_kwargs = {}  # ★imgs値は渡さない
        device_label = "GPU (TensorRT)"
    elif mode == "gpu_torch":
        print(f"[YOLO] loading PyTorch model (GPU): {model_path_pt}")
        model = YOLO(model_path_pt)
        predict_kwargs = dict(imgsz=YOLO_IMGSZ, device="cuda:0")
        device_label = f"GPU (PyTorch, imgsz={YOLO_IMGSZ})"
    elif mode == "cpu":
        print(f"[YOLO] loading PyTorch model (CPU): {model_path_pt}")
        model = YOLO(model_path_pt)
        predict_kwargs = dict(imgsz=YOLO_IMGSZ, device="cpu")
        device_label = f"CPU (PyTorch, imgsz={YOLO_IMGSZ})"
    else:
        raise ValueError(f"Unknown mode: {mode}")

    print("[YOLO] model loaded.")

    # ---- ハードウェア初期化 ----
    mlx = init_mlx()
    cap = init_camera()

    # ---- ベンチマークループ ----
    t_start = time.time()
    frame_count = 0

    loop_times_ms = []   # 1ループ（カメラ + サーマル + YOLO + 後処理）
    infer_times_ms = []  # YOLO 推論だけ

    try:
        while True:
            now = time.time()
            if now - t_start > duration_sec:
                break

            t_loop0 = time.time()

            # 1) カメラ: 古いフレームを捨てて最新を取得
            for _ in range(3):
                cap.grab()
            t_rgb_cap = time.time()
            ret, frame_rgb = cap.retrieve()
            if not ret:
                print("[WARN] Camera read failed, stopping.")
                break

            # 左右反転（gpu.py / cpu.py と揃える）
            frame_rgb = cv2.flip(frame_rgb, 1)

            # 2) サーマル取得（キャッシュ版）
            t_th_cap = time.time()
            thermal_raw = get_thermal_frame_cached(mlx)

            # 3) YOLO 推論
            t_inf0 = time.time()
            results = model(frame_rgb, stream=False, **predict_kwargs)
            t_inf1 = time.time()
            infer_ms = (t_inf1 - t_inf0) * 1000.0

            # 4) bbox + 平均温度マッピング（簡易版）
            r = results[0]
            h, w, _ = frame_rgb.shape
            annotated_rgb = frame_rgb.copy()

            if r.boxes is not None and len(r.boxes) > 0:
                boxes = r.boxes
                xyxy = boxes.xyxy.cpu().numpy().astype(int)
                clses = boxes.cls.cpu().numpy().astype(int)
                confs = boxes.conf.cpu().numpy()

                for box, cls_id, conf in zip(xyxy, clses, confs):
                    x1, y1, x2, y2 = box.tolist()
                    avg_temp = bbox_to_thermal_region(
                        thermal_raw, x1, y1, x2, y2, w, h
                    )
                    # 枠とテキストも一応描画してコストに含める
                    cv2.rectangle(
                        annotated_rgb,
                        (x1, y1),
                        (x2, y2),
                        (0, 255, 0),
                        2,
                    )
                    name = r.names.get(cls_id, str(cls_id))
                    if avg_temp is not None:
                        label = f"{name} {conf:.2f} {avg_temp:.1f}C"
                    else:
                        label = f"{name} {conf:.2f}"
                    cv2.putText(
                        annotated_rgb,
                        label,
                        (x1, max(0, y1 - 10)),
                        cv2.FONT_HERSHEY_SIMPLEX,
                        0.5,
                        (255, 255, 255),
                        1,
                        cv2.LINE_AA,
                    )

            # 5) サーマル疑似カラー変換
            thermal_color = thermal_to_color(
                thermal_raw,
                out_size=(w, h),
                vis_min=VIS_MIN,
                vis_max=VIS_MAX,
            )

            # 6) ループ時間計測（ここまでが「ほぼ gpu.py/cpu.py の処理」）
            t_loop1 = time.time()
            loop_ms = (t_loop1 - t_loop0) * 1000.0

            frame_count += 1
            loop_times_ms.append(loop_ms)
            infer_times_ms.append(infer_ms)

    finally:
        cap.release()

    # ---- 結果集計 ----
    if frame_count == 0:
        print("[ERROR] No frames processed.")
        return

    avg_loop = stats.mean(loop_times_ms)
    med_loop = stats.median(loop_times_ms)
    avg_inf = stats.mean(infer_times_ms)
    med_inf = stats.median(infer_times_ms)
    fps = 1000.0 / avg_loop if avg_loop > 0 else 0.0

    print()
    print("===== Pipeline Benchmark Result =====")
    print(f"Device       : {device_label}")
    print(f"Duration     : {duration_sec} s")
    print(f"Frames       : {frame_count}")
    print("-------------------------------------")
    print(f"Loop avg     : {avg_loop:.2f} ms  →  {fps:.2f} FPS")
    print(f"Loop median  : {med_loop:.2f} ms")
    print(f"Infer avg    : {avg_inf:.2f} ms")
    print(f"Infer median : {med_inf:.2f} ms")
    print("=====================================\n")

    # Zenn 用コピペ出力
    print("Zenn 用コピペ:")
    print("```text")
    print(f"Device   : {device_label}")
    print(f"Duration : {duration_sec} s")
    print(f"Frames   : {frame_count}")
    print(f"Loop     : avg {avg_loop:.2f} ms  →  {fps:.2f} FPS (median {med_loop:.2f} ms)")
    print(f"Infer    : avg {avg_inf:.2f} ms (median {med_inf:.2f} ms)")
    print("```")


def main():
    import argparse

    parser = argparse.ArgumentParser(
        description="cpu.py / gpu.py に近いパイプラインのベンチマーク"
    )
    parser.add_argument(
        "--mode",
        type=str,
        default="gpu",
        choices=["gpu", "gpu_torch", "cpu"],
        help="gpu: TensorRT, gpu_torch: PyTorch+CUDA, cpu: PyTorch CPU",
    )
    parser.add_argument(
        "--duration",
        type=int,
        default=20,
        help="計測時間 [秒]",
    )
    args = parser.parse_args()

    run_benchmark(
        mode=args.mode,
        duration_sec=args.duration,
    )


if __name__ == "__main__":
    main()

実行結果

python benchmark.py --mode cpu --duration 20
python benchmark.py --mode gpu --duration 20
項目 CPU（PyTorch） GPU（TensorRT）
推論方式 PyTorch TensorRT
YOLO入力サイズ 416 Engine固定（640）
計測時間 20 s 20 s
フレーム数 38 64
平均ループ時間 533.58 ms 312.97 ms
実効FPS 1.87 FPS 3.20 FPS
推論平均時間 288.40 ms 64.48 ms

実行FPSが約1.7倍，推論平均時間が4.8倍の水準となっているのがわかります。

レイテンシ対策

レイテンシの可視化

FPSを描画するだけでなく，

  • RGB キャプチャ時刻
  • サーマル取得時刻
  • 推論後の現在時刻

をフレームに描画し，RGB とサーマルのレイテンシ差分（Δms）も同時に表示しています。

カメラの内部バッファを強制的に捨てる（cap.grab）

USBカメラは内部バッファにフレームが溜まりやすく，そのまま read() すると 数フレーム分遅れた映像を掴むことがあります。そこで毎ループの先頭で，古いフレームを強制的に捨てるようにしました。

# 古いフレームを捨てる（レイテンシ低減）
for _ in range(3):
    cap.grab()
ret, frame_rgb = cap.retrieve()

OpenCV のバッファサイズを最小化

OpenCV 側のフレームバッファもレイテンシの原因になります。以下の設定で 常に最新フレームのみを保持するようにしています。

cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)

Flask + MJPEG 配信による遅延の抑制

MJPEG ストリーミングは，長時間動かすと ブラウザ側だけどんどん遅延する問題が起きやすいです。

これを防ぐために，

  • 毎フレーム JPEG 品質を下げる（70）
  • フレームサイズを 640×480 に固定
cv2.imencode(
    ".jpg",
    combined,
    [int(cv2.IMWRITE_JPEG_QUALITY), 70]
)

として，転送量を削減しました。

詰まったポイント

Jetson 特有の環境差分や GPU まわりの依存関係でいくつか詰まりポイントが発生しました。

CUDAが有効化されない

import torch  
print(torch.cuda.is_available())
=> False

原因

  • Jetson 用の CUDA 対応 PyTorch ホイールではないものを入れてしまっている
  • pip から入れた通常の PyTorch（x86向け）が入っている
  • JetPack と PyTorch の対応バージョンがズレている

対策

# 1. 一度 PyTorch 系をすべて削除
sudo pip uninstall torch torchvision torchaudio -y  

# 2. NVIDIA 公式の Jetson 用 PyTorch をインストール  
# （※ JetPack のバージョンに対応したものを使用）
pip install torch torchvision

OpenCV の imshow が動かない（GUI無しビルド）

次のようなエラーで GUI 表示ができない問題。

The function is not implemented. Rebuild the library with GTK+ support

原因

  • pip install opencv-python版は GUI サポートが無効なことがある
  • Jetsonではapt版OpenCVを使うのが安全

対策

sudo apt install -y python3-opencv

サーマル画像が真っ赤になる（温度レンジミス）

MLX90640の表示がすべて真っ赤になる問題。

原因

  • 可視化時の温度レンジ（VIS_MIN, VIS_MAX）が
    実環境の温度と合っていなかった

対策

VIS_MIN = 20.0  
VIS_MAX = 40.0

このように室温帯に合わせたレンジ固定にすることで改善。

サーマルカメラの左右が反転して見える

MLX90640 は 物理向きと配線方向で左右が反転することがあります。

対策：

# 修正前
ret, frame_rgb = cap.read()

# 修正後
ret, frame_rgb = cap.read()
frame_rgb = cv2.flip(frame_rgb, 1)

で向きを調整。

まとめ

Jetson Orin Nano を使うことで，

  • エッジ端末単体で YOLO がリアルタイム実行できる
  • TensorRT による最適化で CPU と桁違いの速度差が出る
  • サーマルと組み合わせることで「温度 × 物体検出」という応用が可能

ということが，視覚的にも数値的にも確認できました。

生成AIやLLMが人気ですが，静かにエッジAIの世界も広がっています。
フィジカルAIの時代に向けて，エッジAIの可能性を探っていきたいです。

MICIN ではメンバーを大募集しています。
「とりあえず話を聞いてみたい」でも大歓迎ですので、お気軽にご応募ください！
https://recruit.micin.jp/

株式会社MICIN
株式会社MICINPublication

オンライン医療事業「curon(クロン)」、薬局向けサービス「curonお薬サポート」、臨床開発デジタルソリューション事業「MiROHA」、安心して治療生活を送っていただくために、患者さんのセルフケア習慣を支援するシステム「MedBridge(メドブリッジ)」など、様々なプロダクト開発を行なっています。

Discussion