RTMDet による物体検出（MMDetection を使用 ）（ソースコードと実行結果）

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_8xb32-300e_coco - CSPNeXtをバックボーンとしたRTMDet-Tiny。COCO 2017データセットで事前学習済み。性能と精度のバランスを実現。URL: https://download.openmmlab.com/mmdetection/v3.0/rtmdet/
- 方式設計
  - 関連利用技術:
    * 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画面でリアルタイムに表示．OpenCV画面内に処理結果をテキストで表示．プログラム終了時に処理結果をresult.txtファイルに保存し，「result.txtに保存」したことをprint()で表示．プログラム開始時に，プログラムの概要，ユーザが行う必要がある操作をprint()で表示
  - 処理手順: 1)DetInferencerでRTMDetモデル初期化、2)入力画像読み込み、3)単段階エンコーダー-デコーダーによる特徴抽出、4)動的ソフトラベル割り当てによる物体分類と位置回帰、5)検出結果の描画と表示
  - 前処理、後処理: 前処理: 画像の正規化とリサイズ、混合画像データ拡張（Mosaic+MixUp）、後処理: NMS(Non-Maximum Suppression)による重複検出の除去、信頼度閾値による結果フィルタリング
  - 追加処理: GPU/CPU自動デバイス選択による推論処理、検出結果の可視化処理、キャッシュ機能によるデータ拡張
  - 調整を必要とする設定値: model_choice（モデル選択）- 使用するRTMDetモデルの種類を指定。tiny/s/m/l/xで性能と速度のバランスを調整
- 将来方策: 複数の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 mmengine mmcv==2.1.0 mmdet opencv-python pillow tqdm
"""

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

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

# SSL証明書検証を無効化（モデルダウンロード用）
ssl._create_default_https_context = ssl._create_unverified_context

# 重要でないUserWarningを最小限に抑制
warnings.filterwarnings("once", category=UserWarning, module="mmdet")
warnings.filterwarnings("once", category=UserWarning, module="mmengine")

# 設定定数
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'
}
PRED_SCORE_THR = 0.3
FONT_PATH = 'C:/Windows/Fonts/meiryo.ttc'
FONT_SIZE = 20
SAMPLE_URL = 'https://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.avi'

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

# 変数
frame_count = 0
results_log = []
inferencer = None

def get_font():
    """フォントを取得"""
    try:
        return ImageFont.truetype(FONT_PATH, FONT_SIZE)
    except OSError:
        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:
        # 例外時は呼び出し元でraiseされる。ここではバッファを破棄するだけ。
        out_buf.close()
        err_buf.close()

def video_frame_processing(frame):
    """1フレームを推論し、可視化フレームと検出結果文字列リスト、タイムスタンプを返す"""
    global frame_count
    current_time = time.time()
    frame_count += 1

    # 進捗表示を抑止して推論
    result = infer_silently(frame)
    predictions = result['predictions'][0]
    vis_frame = result['visualization'][0]

    obj_lines = []
    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()

        # クラス名（メタ情報を優先し、なければCOCOの既定にフォールバック）
        classes = None
        if hasattr(predictions, 'metainfo') and isinstance(predictions.metainfo, dict):
            classes = predictions.metainfo.get('classes', None)
        if classes is None:
            classes = [
                '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','mouse','remote','keyboard','cell phone',
                'microwave','oven','toaster','sink','refrigerator','book','clock','vase','scissors',
                'teddy bear','hair drier','toothbrush'
            ]

        # スコア閾値を適用し、物体単位の出力行を作成
        for bbox, label, score in zip(bboxes, labels, scores):
            if score < PRED_SCORE_THR:
                continue
            x1, y1, x2, y2 = bbox
            class_name = classes[int(label)] if int(label) < len(classes) else str(int(label))
            obj_lines.append(
                f"{class_name} ({score:.2f}), x1={x1:.0f}, y1={y1:.0f}, x2={x2:.0f}, y2={y2:.0f}"
            )

        # 画面に日本語テキスト描画（検出数）
        font = get_font()
        img_pil = Image.fromarray(cv2.cvtColor(vis_frame, cv2.COLOR_BGR2RGB))
        draw = ImageDraw.Draw(img_pil)
        draw.text((10, 30), f"検出物体数: {len(obj_lines)}", font=font, fill=(0, 255, 0))
        vis_frame = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)

    return vis_frame, obj_lines, current_time

def init_inferencer(model_choice_key):
    """モデルを初期化"""
    selected_model = MODEL_CONFIGS[model_choice_key]
    print(f"選択されたモデル: RTMDet-{model_choice_key.upper()} ({selected_model})", flush=True)
    print("モデルを初期化中...", flush=True)
    inf = DetInferencer(model=selected_model, device=str(device))
    print("RTMDetモデルの初期化が完了した", flush=True)
    return inf

# ガイダンス表示
print("概要: MMDetection/RTMDetで物体検出を行う。各物体ごとに1行で出力する。", flush=True)
print("操作方法:", flush=True)
print("  1) モデルを選択する（tiny/s/m/l/x）", flush=True)
print("  2) 入力を選択する（0:動画ファイル, 1:カメラ, 2:サンプル動画）", flush=True)
print("  3) OpenCVウィンドウで結果を確認し、q キーで終了", flush=True)
print("注意事項: 初回実行時はモデルを自動ダウンロードする場合がある", flush=True)

# モデル選択
print("\nRTMDetモデルを選択してください:", flush=True)
print("tiny: RTMDet-Tiny (COCO2017, CSPNeXt)", flush=True)
print("s: RTMDet-S (COCO2017, CSPNeXt)", flush=True)
print("m: RTMDet-M (COCO2017, CSPNeXt)", flush=True)
print("l: RTMDet-L (COCO2017, CSPNeXt)", flush=True)
print("x: RTMDet-X (COCO2017, CSPNeXt)", flush=True)
model_choice = input("モデル選択 (tiny/s/m/l/x): ").lower().strip()
if model_choice not in MODEL_CONFIGS:
    print("無効な選択。RTMDet-Tinyを使用する", flush=True)
    model_choice = 'tiny'
inferencer = init_inferencer(model_choice)

# 入力選択
print("0: 動画ファイル", flush=True)
print("1: カメラ", flush=True)
print("2: サンプル動画", flush=True)
choice = input("選択: ").strip()

temp_file = None
if choice == '0':
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfilename(
        title="動画ファイルを選択",
        filetypes=[("Video files", "*.mp4 *.avi *.mov *.mkv")]
    )
    if not path:
        print("動画ファイルが選択されなかったため終了する", flush=True)
        raise SystemExit
    cap = cv2.VideoCapture(path)
elif choice == '1':
    cap = cv2.VideoCapture(0, cv2.CAP_DSHOW)
    if not cap.isOpened():
        cap = cv2.VideoCapture(0)
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
else:
    print("サンプル動画をダウンロード中...", flush=True)
    SAMPLE_FILE = 'vtest.avi'
    urllib.request.urlretrieve(SAMPLE_URL, SAMPLE_FILE)
    temp_file = SAMPLE_FILE
    cap = cv2.VideoCapture(SAMPLE_FILE)

if not cap.isOpened():
    print('動画ファイル・カメラを開けなかった', flush=True)
    if temp_file and os.path.exists(temp_file):
        os.remove(temp_file)
    raise SystemExit

# メイン処理
MAIN_FUNC_DESC = "RTMDet 物体検出"
print('\n=== 動画処理開始 ===', flush=True)
print('操作方法:', flush=True)
print('  q キー: プログラム終了', flush=True)
try:
    while True:
        ret, frame = cap.read()
        if not ret:
            break

        processed_frame, obj_lines, current_time = video_frame_processing(frame)
        cv2.imshow(MAIN_FUNC_DESC, processed_frame)

        # 物体単位1行。カメラ=日付を含む現地時刻(YYYY-MM-DD HH:MM:SS.mmm)、動画=フレーム番号(1開始)。
        if choice == '1':  # カメラ
            ts = datetime.fromtimestamp(current_time).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
            for line in obj_lines:
                print(ts, line, flush=True)
                results_log.append(f"{ts} {line}")
        else:  # 動画ファイル/サンプル動画
            for line in obj_lines:
                print(frame_count, line, flush=True)
                results_log.append(f"{frame_count} {line}")

        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
finally:
    cap.release()
    cv2.destroyAllWindows()
    if results_log:
        with open('result.txt', 'w', encoding='utf-8') as f:
            f.write('=== 結果 ===\n')
            f.write(f'処理フレーム数: {frame_count}\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('\n')
            f.write('\n'.join(results_log))
        print('処理結果をresult.txtに保存しました', flush=True)
    if temp_file and os.path.exists(temp_file):
        os.remove(temp_file)
    print('\n=== プログラム終了 ===', flush=True)