リアルタイム画像認識アルゴリズム YOLO v8の実装方法

conda create -n 環境名

conda create -n myenv python=3.10

//CUDAのGPUを用いる場合
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
//新しくインストールする場合は「 -U 」は不要
pip install -U ultralytics

from ultralytics import YOLO

model = YOLO("yolov8n.pt")
results = model.predict(source="bus.jpg", save=True)

from IPython.display import Image, display
display(Image(filename="runs/detect/predict/bus.jpg"))

  0: person
  1: bicycle
  2: car
  3: motorcycle
  4: airplane
  5: bus
  6: train
  7: truck
  8: boat
  9: traffic light
  10: fire hydrant
  11: stop sign
  12: parking meter
  13: bench
  14: bird
  15: cat
  16: dog
  17: horse
  18: sheep
  19: cow
  20: elephant
  21: bear
  22: zebra
  23: giraffe
  24: backpack
  25: umbrella
  26: handbag
  27: tie
  28: suitcase
  29: frisbee
  30: skis
  31: snowboard
  32: sports ball
  33: kite
  34: baseball bat
  35: baseball glove
  36: skateboard
  37: surfboard
  38: tennis racket
  39: bottle
  40: wine glass
  41: cup
  42: fork
  43: knife
  44: spoon
  45: bowl
  46: banana
  47: apple
  48: sandwich
  49: orange
  50: broccoli
  51: carrot
  52: hot dog
  53: pizza
  54: donut
  55: cake
  56: chair
  57: couch
  58: potted plant
  59: bed
  60: dining table
  61: toilet
  62: tv
  63: laptop
  64: mouse
  65: remote
  66: keyboard
  67: cell phone
  68: microwave
  69: oven
  70: toaster
  71: sink
  72: refrigerator
  73: book
  74: clock
  75: vase
  76: scissors
  77: teddy bear
  78: hair drier
  79: toothbrush

from ultralytics import YOLO

model = YOLO("yolov8n.pt")
model.train(
    data=r"WP_Roboflow_Dataset/data.yaml",
    imgsz=640,
    epochs=50,
    batch=16,   # VRAMに応じて調整（足りなければ 8/4 に）
    device=0    # GPU使用
)

import pyrealsense2 as rs
import numpy as np
import cv2, csv, time
from datetime import datetime
from ultralytics import YOLO

# ==== YOLOモデル ====
model = YOLO(r"runs/detect/train/weights/best.pt")  # 自分の学習済みモデルに置き換え
IMGSZ = 640 #画像サイズ
CONF_THR = 0.25 #信頼度閾値
SAMPLING_PERIOD = 0.05  # 20Hz
FPS=20
# ==== RealSense初期化 ====
pipe = rs.pipeline()
cfg = rs.config()
cfg.enable_stream(rs.stream.depth, 1024, 768, rs.format.z16, 30)
cfg.enable_stream(rs.stream.color, 1280, 720, rs.format.bgr8, 30)
profile = pipe.start(cfg)
align = rs.align(rs.stream.color)

depth_sensor = profile.get_device().first_depth_sensor()
depth_scale  = depth_sensor.get_depth_scale()  # 単位→m変換倍率, 深度画像の1カウントあたり、何メートルかを示し, メーカー(intel)が決めている.

# カメラ内部パラメータ（X,Y計算に使用）
intr = profile.get_stream(rs.stream.color).as_video_stream_profile().get_intrinsics()
#fx,fyは、画像の横縦方向（u,v方向）のスケールで「1m 離れた点が、画像上で何ピクセル分広がるか」を表す.cx,cyは基準となるピクセルを表す.
fx, fy, cx, cy = intr.fx, intr.fy, intr.ppx, intr.ppy

# ==== 保存用 ====
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
csv_path = f"realsense_yolov8_{stamp}.csv"

# ==== 動画保存用 ====
video_path = f"realsense_yolov8_{stamp}.mp4"   # or .avi
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
# 書き出しサイズは color の実サイズに合わせる（1280x720）
VIDEO_W, VIDEO_H = 1280, 720
# 実際の記録周期は 0.05s なので 20fps にしておくのが分かりやすい
video_writer = cv2.VideoWriter(video_path, fourcc, 20.0, (VIDEO_W, VIDEO_H))

video_writer = None
def init_writer_if_needed(frame, fps):
    global video_writer
    if video_writer is None:
        h, w = frame.shape[:2]
        fourcc = cv2.VideoWriter_fourcc(*"mp4v")
        video_writer = cv2.VideoWriter(str(video_path), fourcc, fps, (w, h))


def median_xyz(depth_np, x1, y1, x2, y2, margin=0.1):
    """bbox内の有効深度から(X,Y,Z)中央値[m]を求める"""
    #depth_npは深さの情報で、その形状はh*wある。
    h, w = depth_np.shape
    dx = int((x2 - x1) * margin)#端はノイズがあると考え、10%をdxと定義。
    dy = int((y2 - y1) * margin)
    x1, x2 = np.clip([x1+dx, x2-dx], 0, w-1)#bboxの中央80%が範囲外にはみ出さないための安全策。
    y1, y2 = np.clip([y1+dy, y2-dy], 0, h-1)
    roi = depth_np[int(y1):int(y2), int(x1):int(x2)]#80%範囲の深さ情報を抽出
    mask = roi > 0
    if np.count_nonzero(mask) == 0:#深度が一つもとれなかったらスキップする
        return None
    zs = roi[mask].astype(np.float32) * depth_scale
    us, vs = np.meshgrid(np.arange(x1, x2), np.arange(y1, y2))#bboxの中の3Ｄ化をするため(1点のみなら不要)
    us, vs = us[mask].astype(np.float32), vs[mask].astype(np.float32)
    Xs = (us - cx) * zs / fx
    Ys = (vs - cy) * zs / fy
    return float(np.median(Xs)), float(np.median(Ys)), float(np.median(zs))

# ==== メインループ ====
t_start = time.perf_counter()
t_prev = t_start
prev_xyz = None
PT0=None

with open(csv_path, "w", newline="", encoding="utf-8") as f:
    wcsv = csv.writer(f)
    wcsv.writerow(["t_elapsed_s", "dt_s", "Xm_m", "Ym_m", "Zm_m"])

    try:
        while True:
            # フレーム取得
            frames = pipe.wait_for_frames()#wait_for_frames() は色と深度の両方が揃ったタイミングまで待機する。
            aligned = align.process(frames)#深さの座標系を色の座標系に一致させる
            depth_fr = aligned.get_depth_frame()#深さ情報。まだnumpyじゃない。(pyrealsense2.depth_frame型)
            color_fr = aligned.get_color_frame()#色情報。まだnumpyじゃない。（pyrealsense2.video_frame型）
            if not depth_fr or not color_fr: #深度(depth) か カラー(color) のどちらか一方でも欠けていたらこのループの処理をスキップして次のフレームを待つ。
                continue

            depth_np = np.asanyarray(depth_fr.get_data())
            color_np = np.asanyarray(color_fr.get_data())

            # YOLO推論(verbose=Falseは, ログの出力をしないため)color_npは1枚なので最初の[0]のみしかない
            res = model(color_np, imgsz=IMGSZ, conf=CONF_THR, verbose=False)[0]

            # 時間制御（20Hz）
            t_now = time.perf_counter()
            elapsed = t_now - t_prev
            if elapsed < SAMPLING_PERIOD:
                time.sleep(SAMPLING_PERIOD - elapsed)
            t_now = time.perf_counter()
            dt = t_now - t_prev
            t_elapsed = t_now - t_start
            t_prev = t_now
            

            # バウンディングボックス処理
            Xm = Ym = Zm = None
            found_bbox = False
            x1 = y1 = x2 = y2 = None
            conf = None
            
            
            if (res.boxes is not None) and (len(res.boxes) > 0):
                confs = res.boxes.conf.cpu().numpy()#信頼度を抜き出してnumpyにする.array([0.83, 0.74, 0.65]) のようになる
                xyxy  = res.boxes.xyxy.cpu().numpy().astype(np.int32)#左上・右下の (x1,y1,x2,y2)）を numpy 配列に変換.
                #array([[100,  51, 300, 401],[120,  80, 261, 351],], dtype=int32)のようになる
                idx   = confs.argmax()#confsの中で一番大きい信頼度のインデックスを取得
                x1, y1, x2, y2 = xyxy[idx] #最も信頼度の高いオブジェクトの左上と右下の座標を取得
                found_bbox=True

                med = median_xyz(depth_np, x1, y1, x2, y2)
                if med is not None:
                    Xm, Ym, Zm = med
                    prev_xyz = med

                    # # 表示
                    # cv2.rectangle(color_np, (x1, y1), (x2, y2), (0,255,0), 2)
                    # cv2.putText(color_np, f"X:{Xm:.3f} Y:{Ym:.3f} Z:{Zm:.3f}",
                    #             (x1, max(0, y1-8)), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0,255,0), 2)

            # 検出なし時は前回値を保持
            hold_mode=False
            if Xm is None and prev_xyz is not None:
                Xm, Ym, Zm = prev_xyz
                hold_mode=True
                # cv2.putText(color_np, f"HOLD X:{Xm:.3f} Y:{Ym:.3f} Z:{Zm:.3f}",
                #             (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,255,255), 2)
            # 初期値の設定
            if (PT0 is None) and (Xm is not None):
                PT0 = (Xm, Ym, Zm)
            # 差分計算
            dX = dY = dZ = None
            if (Xm is not None) and (PT0 is not None):
                dX = Xm - PT0[0]
                dY = Ym - PT0[1]
                dZ = Zm - PT0[2]
            
            if found_bbox and (x1 is not None):
                cv2.rectangle(color_np, (x1, y1), (x2, y2), (0,255,0), 2)
                
            if Xm is not None and dX is not None:
                if hold_mode:
                    txt1 = f"HOLD  dX={dX:+.3f} dY={dY:+.3f} dZ={dZ:+.3f} [m]"
                else:
                    txt1 = f"dX={dX:+.3f} dY={dY:+.3f} dZ={dZ:+.3f} [m]"
                    txt2 = f"X={Xm:.3f} Y={Ym:.3f} Z={Zm:.3f} [m]"
            else:
                txt1 = "No valid 3D yet"
                txt2 = ""
            
            cv2.putText(color_np, txt1, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255,255,255), 2)
            if txt2:
                cv2.putText(color_np, txt2, (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (200,200,200), 2)
            cv2.putText(color_np, f"t={t_elapsed:.2f}s  dt={dt*1000:.1f}ms", (10, 90),
                        cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255,255,0), 2)       
                    
                
        
            # CSV記録
            if Xm is not None:
                wcsv.writerow([f"{t_elapsed:.6f}", f"{dt:.6f}", f"{dX:.6f}", f"{dY:.6f}", f"{dZ:.6f}"])

            # 表示
            cv2.imshow("YOLOv8 + RealSense", color_np)
            if cv2.waitKey(1) & 0xFF == 27:
                break
            # 動画を保存するため
            init_writer_if_needed(color_np, FPS)#color_npにテキストが書かれているので。
            video_writer.write(color_np)

    finally:
        pipe.stop()
        cv2.destroyAllWindows()
        try:
            if video_writer is not None:
                video_writer.release()
        except Exception:
            pass

print(f"CSV saved to: {csv_path}")