MMDetection と RTMDet による物体検出（COCO 80クラス）（ソースコードと説明と利用ガイド）

REM Python をシステム領域にインストール
winget install --scope machine --id Python.Python.3.12 -e --silent
REM Python のパス設定
set "PYTHON_PATH=C:\Program Files\Python312"
set "PYTHON_SCRIPTS_PATH=C:\Program Files\Python312\Scripts"
echo "%PATH%" | find /i "%PYTHON_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_SCRIPTS_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_SCRIPTS_PATH%" /M >nul

winget install --scope machine Codeium.Windsurf -e --silent

"""
- プログラム名: MMDetection と RTMDet による物体検出プログラム
- 特徴技術名: RTMDet
- 出典: Lyu, C., Zhang, W., Huang, H., Zhou, Y., Wang, Y., Liu, Y., ... & Chen, K. (2022). RTMDet: An empirical study of designing real-time object detectors. arXiv preprint arXiv:2212.07784.
- 特徴機能: 大型カーネル深度別分離畳み込みと動的ソフトラベル割り当て - リアルタイム性能とパラメータ-精度トレードオフを最適化。混合画像データ拡張キャッシュ機能により学習効率を向上
- 学習済みモデル: rtmdet_tiny/s/m/l/x - CSPNeXtをバックボーンとしたRTMDet。COCO 2017データセットで事前学習済み。URL: https://download.openmmlab.com/mmdetection/v3.0/rtmdet/
- 特徴技術および学習済モデルの利用制限: Apache 2.0ライセンス。オープンソースであり、商用利用も可能です。
- 方式設計
  - 関連利用技術:
    * MMDetection 3.3.0 - OpenMMLab物体検出フレームワーク、モジュラー設計と性能を提供
    * DetInferencer - MMDetection統一推論インターフェース、簡潔なAPIで推論実行
    * PyTorch - 深層学習フレームワーク、動的グラフとGPU加速をサポート
  - 入力と出力: 入力: 動画（ユーザは「0:動画ファイル，1:カメラ，2:サンプル動画」のメニューで選択．0:動画ファイルの場合はtkinterでファイル選択．1の場合はOpenCVでカメラが開く．2の場合はhttps://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.aviを使用）、出力: OpenCV画面でリアルタイムに表示．プログラム終了時に処理結果をresult.txtファイルに保存．
  - 処理手順: 1)DetInferencerでRTMDetモデル初期化、2)入力画像読み込み、3)単段階エンコーダー-デコーダーによる特徴抽出、4)動的ソフトラベル割り当てによる物体分類と位置回帰、5)検出結果の描画と表示
  - 前処理、後処理: 前処理: 画像の正規化とリサイズ、混合画像データ拡張（Mosaic+MixUp）はモデル内部で処理。後処理: NMS(Non-Maximum Suppression)による重複検出の除去、信頼度閾値による結果フィルタリング。
  - 調整を必要とする設定値: PRED_SCORE_THR（信頼度閾値、デフォルト0.3）- 検出感度を制御
- 将来方策: 複数のRTMDetモデル設定を自動ベンチマークし、ハードウェア性能に基づいて最適なmodel_choiceを推奨する機能の追加
- その他の重要事項: 初回実行時にモデルの自動ダウンロードが発生、CUDA対応GPU推奨、RTMDetは実時間性能に最適化
- 前準備: pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
  "C:\Program Files (x86)\Microsoft Visual Studio\2022\BuildTools\VC\Auxiliary\Build\vcvars64.bat"
  set DISTUTILS_USE_SDK=1
  pip install -U setuptools wheel
  pip install -U mmengine mmcv==2.1.0 mmdet opencv-python pillow tqdm matplotlib transformers tokenizers sentencepiece
"""

import os
import cv2
import time
import torch
import urllib.request
import ssl
import numpy as np
import tkinter as tk
from tkinter import filedialog
from datetime import datetime
from PIL import Image, ImageDraw, ImageFont
from mmdet.apis import DetInferencer
import warnings
from contextlib import redirect_stdout, redirect_stderr
from io import StringIO
import sys
import io
import threading

# Windows文字エンコーディング設定
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', line_buffering=True)

# tqdmプログレスバーを無効化
os.environ["TQDM_DISABLE"] = "1"

# SSL証明書検証を無効化（モデルダウンロード用、セキュリティリスクあり）
ssl._create_default_https_context = ssl._create_unverified_context

# 重要でないUserWarningを抑制
warnings.filterwarnings("ignore", category=UserWarning, module="mmengine.visualization.visualizer")
warnings.filterwarnings("ignore", category=UserWarning, module="mmdet.models.dense_heads.rtmdet_head")


# --- 設定定数 ---
# モデル設定
MODEL_CONFIGS = {
    'tiny': 'rtmdet_tiny_8xb32-300e_coco',
    's': 'rtmdet_s_8xb32-300e_coco',
    'm': 'rtmdet_m_8xb32-300e_coco',
    'l': 'rtmdet_l_8xb32-300e_coco',
    'x': 'rtmdet_x_8xb32-300e_coco'
}
MODEL_INFO = {
    'tiny': {'name': 'RTMDet-Tiny', 'params': '4.1M', 'mAP': 41.1, 'desc': '最速'},
    's': {'name': 'RTMDet-S', 'params': '7.9M', 'mAP': 44.5, 'desc': 'デフォルト'},
    'm': {'name': 'RTMDet-M', 'params': '20.6M', 'mAP': 49.3, 'desc': '中程度'},
    'l': {'name': 'RTMDet-L', 'params': '42.7M', 'mAP': 51.2, 'desc': '高精度'},
    'x': {'name': 'RTMDet-X', 'params': '69.8M', 'mAP': 52.8, 'desc': '最高精度'}
}

# 推論・表示設定
PRED_SCORE_THR = 0.3
FONT_PATH = 'C:/Windows/Fonts/meiryo.ttc'
FONT_SIZE = 16
WINDOW_NAME = "RTMDet Object Detection (COCO 80-Class)"
SAMPLE_URL = 'https://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.avi'
SAMPLE_FILE = 'vtest.avi'

# COCOクラス日本語マッピング
COCO_CLASSES_JP = {
    'person': '人', 'bicycle': '自転車', 'car': '車', 'motorcycle': 'バイク',
    'airplane': '飛行機', 'bus': 'バス', 'train': '電車', 'truck': 'トラック',
    'boat': 'ボート', 'traffic light': '信号機', 'fire hydrant': '消火栓',
    'stop sign': '停止標識', 'parking meter': 'パーキングメーター', 'bench': 'ベンチ',
    'bird': '鳥', 'cat': '猫', 'dog': '犬', 'horse': '馬', 'sheep': '羊',
    'cow': '牛', 'elephant': '象', 'bear': '熊', 'zebra': 'シマウマ', 'giraffe': 'キリン',
    'backpack': 'リュック', 'umbrella': '傘', 'handbag': 'ハンドバッグ', 'tie': 'ネクタイ',
    'suitcase': 'スーツケース', 'frisbee': 'フリスビー', 'skis': 'スキー板',
    'snowboard': 'スノーボード', 'sports ball': 'ボール', 'kite': '凧',
    'baseball bat': 'バット', 'baseball glove': 'グローブ', 'skateboard': 'スケートボード',
    'surfboard': 'サーフボード', 'tennis racket': 'テニスラケット', 'bottle': 'ボトル',
    'wine glass': 'ワイングラス', 'cup': 'カップ', 'fork': 'フォーク', 'knife': 'ナイフ',
    'spoon': 'スプーン', 'bowl': 'ボウル', 'banana': 'バナナ', 'apple': 'リンゴ',
    'sandwich': 'サンドイッチ', 'orange': 'オレンジ', 'broccoli': 'ブロッコリー',
    'carrot': 'ニンジン', 'hot dog': 'ホットドッグ', 'pizza': 'ピザ', 'donut': 'ドーナツ',
    'cake': 'ケーキ', 'chair': '椅子', 'couch': 'ソファ', 'potted plant': '鉢植え',
    'bed': 'ベッド', 'dining table': 'テーブル', 'toilet': 'トイレ', 'tv': 'テレビ',
    'laptop': 'ノートPC', 'mouse': 'マウス', 'remote': 'リモコン', 'keyboard': 'キーボード',
    'cell phone': '携帯電話', 'microwave': '電子レンジ', 'oven': 'オーブン',
    'toaster': 'トースター', 'sink': 'シンク', 'refrigerator': '冷蔵庫',
    'book': '本', 'clock': '時計', 'vase': '花瓶', 'scissors': 'ハサミ',
    'teddy bear': 'ぬいぐるみ', 'hair drier': 'ドライヤー', 'toothbrush': '歯ブラシ'
}


# --- グローバル変数 ---
frame_count = 0
results_log = []
class_counts = {}
inferencer = None
font = None

# --- GPU/CPU自動選択 ---
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f'デバイス: {str(device)}')

if device.type == 'cuda':
    torch.backends.cudnn.benchmark = True


class ThreadedVideoCapture:
    """スレッド化されたVideoCapture（常に最新フレームを取得）"""
    def __init__(self, src, is_camera=False):
        if is_camera:
            self.cap = cv2.VideoCapture(src, cv2.CAP_DSHOW)
            fourcc = cv2.VideoWriter_fourcc('M', 'J', 'P', 'G')
            self.cap.set(cv2.CAP_PROP_FOURCC, fourcc)
            self.cap.set(cv2.CAP_PROP_FPS, 60)
        else:
            self.cap = cv2.VideoCapture(src)

        self.grabbed, self.frame = self.cap.read()
        self.stopped = False
        self.lock = threading.Lock()
        self.thread = threading.Thread(target=self.update, args=())
        self.thread.daemon = True
        self.thread.start()

    def update(self):
        """バックグラウンドでフレームを取得し続ける"""
        while not self.stopped:
            grabbed, frame = self.cap.read()
            with self.lock:
                self.grabbed = grabbed
                if grabbed:
                    self.frame = frame

    def read(self):
        """最新フレームを返す"""
        with self.lock:
            return self.grabbed, self.frame.copy() if self.grabbed else None

    def isOpened(self):
        return self.cap.isOpened()

    def get(self, prop):
        return self.cap.get(prop)

    def release(self):
        self.stopped = True
        self.thread.join()
        self.cap.release()


def display_program_header():
    """プログラムのヘッダー情報を表示"""
    print('=' * 60)
    print('=== MMDetection/RTMDet 物体検出プログラム ===')
    print('=' * 60)
    print('概要: MMDetectionのRTMDetモデルを使用してリアルタイムで物体を検出します。')
    print('特徴: 高速かつ高精度なRTMDetを採用（COCOデータセット80クラス）。')
    print('ライセンス: Apache 2.0（商用利用可能）')
    print('操作: OpenCVウィンドウで "q" キーを押すと処理を終了します。')
    print('出力: 各フレームの検出結果をコンソールに出力し、終了時にresult.txtへ保存します。')
    print('注意: 初回実行時やモデル変更時には、モデルファイルのダウンロードが発生します。')
    print()


def get_font():
    """描画用のフォントを取得"""
    try:
        return ImageFont.truetype(FONT_PATH, FONT_SIZE)
    except OSError:
        print(f"警告: フォント '{FONT_PATH}' が見つかりません。デフォルトフォントを使用します。")
        return ImageFont.load_default()


def infer_silently(img):
    """DetInferencerの推論時の内部出力を抑制"""
    out_buf, err_buf = StringIO(), StringIO()
    try:
        with redirect_stdout(out_buf), redirect_stderr(err_buf):
            return inferencer(
                img,
                return_vis=True,
                no_save_vis=True,
                pred_score_thr=PRED_SCORE_THR
            )
    finally:
        out_buf.close()
        err_buf.close()


def parse_predictions(predictions):
    """推論結果を扱いやすい形式の辞書リストに変換"""
    parsed_dets = []
    if hasattr(predictions, 'pred_instances'):
        pred_instances = predictions.pred_instances
        bboxes = pred_instances.bboxes.cpu().numpy()
        labels = pred_instances.labels.cpu().numpy()
        scores = pred_instances.scores.cpu().numpy()
        classes_en = predictions.metainfo.get('classes', [])

        for bbox, label_id, score in zip(bboxes, labels, scores):
            if score < PRED_SCORE_THR:
                continue

            name_en = classes_en[int(label_id)] if int(label_id) < len(classes_en) else f"class_{int(label_id)}"
            parsed_dets.append({
                'x1': int(bbox[0]), 'y1': int(bbox[1]),
                'x2': int(bbox[2]), 'y2': int(bbox[3]),
                'score': float(score),
                'class_id': int(label_id),
                'name_en': name_en,
                'name_jp': COCO_CLASSES_JP.get(name_en, name_en)
            })
    return parsed_dets


def format_detection_output(detections):
    """物体検出のコンソール/ファイル出力用フォーマット"""
    if not detections:
        return 'count=0'

    parts = []
    for det in detections:
        box = f"[{det['x1']},{det['y1']},{det['x2']},{det['y2']}]"
        parts.append(f"class={det['name_jp']},conf={det['score']:.3f},box={box}")
    return f'count={len(detections)}; ' + ' | '.join(parts)


def draw_detection_info(frame, detections):
    """検出情報のテキストを描画"""
    global font
    img_pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
    draw = ImageDraw.Draw(img_pil)

    detected_classes = set(d['name_jp'] for d in detections)
    info_text = f"検出数: {len(detections)} | フレーム: {frame_count} | クラス種類: {len(detected_classes)}"
    draw.text((10, 10), info_text, font=font, fill=(0, 255, 0, 255))

    return cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)


def process_video_frame(frame, timestamp_ms, is_camera):
    """1フレームを推論し、可視化フレームと結果文字列を返す"""
    global class_counts

    result = infer_silently(frame)
    vis_frame = result['visualization'][0]
    predictions = result['predictions'][0]

    detections = parse_predictions(predictions)

    for det in detections:
        name = det['name_en']
        class_counts[name] = class_counts.get(name, 0) + 1

    vis_frame = draw_detection_info(vis_frame, detections)

    result_text = format_detection_output(detections)

    return vis_frame, result_text


def video_frame_processing(frame, timestamp_ms, is_camera):
    """動画フレーム処理（標準形式）"""
    global frame_count
    current_time = time.time()
    frame_count += 1

    processed_frame, result = process_video_frame(frame, timestamp_ms, is_camera)
    return processed_frame, result, current_time


def main():
    """メイン処理"""
    global inferencer, font, model_choice, frame_count, results_log, class_counts

    display_program_header()
    font = get_font()

    # --- モデル選択 ---
    print("=== RTMDetモデル選択 ===")
    print('使用するRTMDetモデルを選択してください:')
    for key, info in MODEL_INFO.items():
        print(f"{key}: {info['name']} ({info['params']} params, mAP {info['mAP']}) - {info['desc']}")

    model_choice = ''
    while model_choice not in MODEL_CONFIGS.keys():
        model_choice = input(f"選択 ({'/'.join(MODEL_CONFIGS.keys())}) [デフォルト: s]: ").strip().lower()
        if model_choice == '':
            model_choice = 's'
            break
        if model_choice not in MODEL_CONFIGS.keys():
            print("無効な選択です。もう一度入力してください。")

    # --- モデル初期化 ---
    selected_model_name = MODEL_CONFIGS[model_choice]
    print(f"\nモデルをロード中: {MODEL_INFO[model_choice]['name']} ({selected_model_name})")
    try:
        inferencer = DetInferencer(model=selected_model_name, device=str(device))
        print("モデルのロード完了")
    except Exception as e:
        print(f"モデルのロードに失敗しました: {e}")
        raise SystemExit(1)

    # --- 入力ソース選択 ---
    print("\n=== 入力ソースを選択してください ===")
    print('0: 動画ファイル')
    print('1: カメラ')
    print('2: サンプル動画')
    choice = input('選択: ').strip()

    is_camera = (choice == '1')

    if choice == '0':
        root = tk.Tk()
        root.withdraw()
        path = filedialog.askopenfilename(
            title="動画ファイルを選択",
            filetypes=[("Video files", "*.mp4 *.avi *.mov *.mkv")]
        )
        if not path:
            print("動画ファイルが選択されなかったため、プログラムを終了します。")
            raise SystemExit(1)
        cap = cv2.VideoCapture(path)
    elif choice == '1':
        cap = ThreadedVideoCapture(0, is_camera=True)
    else:
        if choice != '2':
            print("無効な選択です。サンプル動画を使用します。")
        print('サンプル動画をダウンロード中...')
        try:
            urllib.request.urlretrieve(SAMPLE_URL, SAMPLE_FILE)
            cap = cv2.VideoCapture(SAMPLE_FILE)
        except Exception as e:
            print(f"サンプル動画のダウンロードに失敗しました: {e}")
            raise SystemExit(1)

    if not cap.isOpened():
        print('動画ファイル・カメラを開けませんでした')
        raise SystemExit(1)

    # --- フレームレート取得とタイムスタンプ増分計算 ---
    if is_camera:
        actual_fps = cap.get(cv2.CAP_PROP_FPS)
        print(f'カメラのfps: {actual_fps}')
        timestamp_increment = int(1000 / actual_fps) if actual_fps > 0 else 33
    else:
        video_fps = cap.get(cv2.CAP_PROP_FPS)
        timestamp_increment = int(1000 / video_fps) if video_fps > 0 else 33

    # --- メインループ ---
    print('\n=== 動画処理開始 ===')
    print("操作方法: 'q' キーでプログラムを終了")

    start_time = time.time()
    last_info_time = start_time
    info_interval = 10.0
    timestamp_ms = 0
    total_processing_time = 0.0

    try:
        while True:
            ret, frame = cap.read()
            if not ret:
                break

            timestamp_ms += timestamp_increment

            processing_start = time.time()
            processed_frame, result, current_time = video_frame_processing(frame, timestamp_ms, is_camera)
            processing_time = time.time() - processing_start
            total_processing_time += processing_time
            cv2.imshow(WINDOW_NAME, processed_frame)

            if result:
                if is_camera:
                    timestamp = datetime.fromtimestamp(current_time).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
                    print(f'{timestamp}, {result}')
                else:
                    print(f'Frame {frame_count}: {result}')

                results_log.append(result)

            if is_camera:
                elapsed = current_time - last_info_time
                if elapsed >= info_interval:
                    total_elapsed = current_time - start_time
                    actual_fps = frame_count / total_elapsed if total_elapsed > 0 else 0
                    avg_processing_time = (total_processing_time / frame_count * 1000) if frame_count > 0 else 0
                    print(f'[情報] 経過時間: {total_elapsed:.1f}秒, 処理フレーム数: {frame_count}, 実測fps: {actual_fps:.1f}, 平均処理時間: {avg_processing_time:.1f}ms')
                    last_info_time = current_time

            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

    finally:
        print('\n=== プログラム終了 ===')
        cap.release()
        cv2.destroyAllWindows()

        if results_log:
            try:
                with open('result.txt', 'w', encoding='utf-8') as f:
                    f.write('=== RTMDet 物体検出結果 ===\n')
                    f.write(f'処理フレーム数: {frame_count}\n')
                    model_name = MODEL_CONFIGS[model_choice]
                    model_details = MODEL_INFO[model_choice]
                    f.write(f'使用モデル: {model_name}\n')
                    f.write(f'モデル情報: {model_details["name"]} ({model_details["params"]} params, mAP {model_details["mAP"]})\n')
                    f.write(f'使用デバイス: {str(device).upper()}\n')
                    if device.type == 'cuda':
                        f.write(f'GPU: {torch.cuda.get_device_name(0)}\n')
                    f.write(f'信頼度閾値: {PRED_SCORE_THR}\n')
                    f.write(f'\n検出されたクラス一覧（総検出回数）:\n')
                    if class_counts:
                        sorted_classes = sorted(class_counts.items(), key=lambda item: item[1], reverse=True)
                        for class_name_en, count in sorted_classes:
                            jp_name = COCO_CLASSES_JP.get(class_name_en, class_name_en)
                            f.write(f'  {jp_name} ({class_name_en}): {count}回\n')
                    else:
                        f.write('  検出された物体はありませんでした。\n')
                    f.write('\n--- ログ ---\n')
                    if is_camera:
                        f.write('形式: タイムスタンプ, 検出結果\n')
                    else:
                        f.write('形式: フレーム番号, 検出結果\n')
                    f.write('\n')
                    f.write('\n'.join(results_log))
                print('処理結果をresult.txtに保存しました')
            except Exception as e:
                print(f"エラー: 結果ファイルの書き込みに失敗しました - {e}")


if __name__ == '__main__':
    main()