YOLO11によるインスタンスセグメンテーション（ソースコードと説明と利用ガイド）

"""
プログラム名: YOLO11物体検出・インスタンスセグメンテーションプログラム
特徴技術名: YOLO11 (You Only Look Once version 11)
出典: G. Jocher and J. Qiu, "Ultralytics YOLO11," GitHub repository, 2024. [Online]. Available: https://github.com/ultralytics/ultralytics
特徴機能: C3k2ブロックによる特徴抽出（カーネルサイズ2のCSPボトルネック構造により、計算効率と検出精度を両立）
学習済みモデル: yolo11n-seg.pt/yolo11s-seg.pt/yolo11m-seg.pt/yolo11l-seg.pt/yolo11x-seg.pt（COCOデータセット80クラス対応インスタンスセグメンテーションモデル、自動ダウンロード）
特徴技術および学習済モデルの利用制限: AGPL-3.0ライセンスに準拠（Ultralytics）。学術研究および商用利用が可能だが、商用利用時はEnterprise Licenseの取得を推奨。必ず利用者自身でUltralyticsのライセンス情報を確認すること。
方式設計:
  関連利用技術:
    - OpenCV: コンピュータビジョンライブラリ、画像・動画処理、カメラ制御
    - NumPy: 数値計算ライブラリ、配列処理、データ操作
    - Tkinter: GUIライブラリ、ファイル選択ダイアログ
    - PIL/Pillow: 画像処理ライブラリ、日本語テキスト描画
    - PyTorch: ディープラーニングフレームワーク、GPU/CPU自動選択
    - Ultralytics YOLO11: 物体検出・セグメンテーションフレームワーク
  入力と出力: 入力: 動画（ユーザは「0:動画ファイル，1:カメラg.，2:サンプル動画」のメニューで選択．0:動画ファイルの場合はtkinterでファイル選択．1の場合はOpenCVでカメラが開く．2の場合はhttps://github.com/opencv/opencv/blob/master/samples/data/vtest.aviを使用）、出力: OpenCV画面でリアルタイム表示、各フレームごとにprint()で処理結果を表示、プログラム終了時にresult.txtファイルに保存
  処理手順: 1.動画フレーム取得→2.CLAHE適用→3.YOLO11モデルによる推論実行→4.検出結果とセグメンテーションマスクの描画→5.リアルタイム表示
  前処理、後処理: 前処理：CLAHE適用（YUV色空間で輝度チャンネルのみ処理）、フレームバッファクリア（最新フレーム取得）、画像正規化（YOLO11内部処理）、後処理：検出結果可視化、信頼度フィルタリング、セグメンテーションマスク合成
  追加処理: DirectShowバックエンド使用によるカメラ安定化、バッファサイズ=1設定による遅延最小化、乱数シード設定による結果再現性確保
  調整を必要とする設定値: CONF_THRESHOLD（信頼度閾値、0.25、YOLO11公式推奨のデフォルト値、検出感度と精度のバランス調整）
将来方策: CONF_THRESHOLDの自動最適化機能実装（フレーム毎の検出数と信頼度分布を統計的に分析し、適合率と再現率のF値が最大となる閾値を動的に算出。実装方法：過去30フレームの検出結果を保持し、信頼度ヒストグラムから最適閾値を計算）
その他の重要事項: Windows環境対応、AGPL-3.0ライセンス、リアルタイム処理特化、COCOクラス検出可能
前準備:
pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
pip install ultralytics opencv-python numpy pillow
"""
import cv2
import tkinter as tk
from tkinter import filedialog
import os
import urllib.request
from ultralytics import YOLO
import numpy as np
from PIL import Image, ImageDraw, ImageFont
from collections import defaultdict
import torch
from datetime import datetime
import time
import sys
import io
import threading
import warnings

warnings.filterwarnings('ignore')

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', line_buffering=True)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f'デバイス: {str(device)}')

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

MODEL_INFO = {
    'n': {
        'name': 'yolo11n-seg.pt',
        'size': '5.9 MB',
        'desc': 'Nano (軽量・高速)'
    },
    's': {
        'name': 'yolo11s-seg.pt',
        'size': '19.7 MB',
        'desc': 'Small (バランス型)'
    },
    'm': {
        'name': 'yolo11m-seg.pt',
        'size': '43.3 MB',
        'desc': 'Medium (汎用)'
    },
    'l': {
        'name': 'yolo11l-seg.pt',
        'size': '53.5 MB',
        'desc': 'Large (高精度)'
    },
    'x': {
        'name': 'yolo11x-seg.pt',
        'size': '119 MB',
        'desc': 'Extra Large (最高精度)'
    }
}

CONF_THRESHOLD = 0.25
RANDOM_SEED = 42
CLAHE_CLIP_LIMIT = 2.0
CLAHE_TILE_SIZE = (8, 8)
WINDOW_NAME = 'YOLO11物体検出・インスタンスセグメンテーション'

COLOR_GREEN = (0, 255, 0)
COLOR_YELLOW = (0, 255, 255)
COLOR_GREEN_RGB = (0, 255, 0)
COLOR_YELLOW_RGB = (255, 255, 0)

clahe = cv2.createCLAHE(clipLimit=CLAHE_CLIP_LIMIT, tileGridSize=CLAHE_TILE_SIZE)

def bgr_to_rgb(color_bgr):
    return (color_bgr[2], color_bgr[1], color_bgr[0])

def generate_class_colors(num_classes):
    colors = []
    for i in range(num_classes):
        hue = int(180.0 * i / num_classes)
        hsv = np.uint8([[[hue, 255, 255]]])
        bgr = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)[0][0]
        colors.append((int(bgr[0]), int(bgr[1]), int(bgr[2])))
    return colors

CLASS_NAMES_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': '歯ブラシ'
}

FONT_PATH = 'C:/Windows/Fonts/meiryo.ttc'
FONT_SIZE = 24
try:
    font = ImageFont.truetype(FONT_PATH, FONT_SIZE)
except IOError:
    font = None

frame_count = 0
results_log = []
class_counts = {}
model = None


class ThreadedVideoCapture:
    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)
            self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
        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('=== YOLO11物体検出・インスタンスセグメンテーションプログラム ===')
    print('=' * 60)
    print('概要: CLAHEを適用し、リアルタイムでオブジェクトを検出します')
    print('機能: YOLO11による物体検出・セグメンテーション（COCOデータセット対応）')
    print('技術: CLAHE + YOLO11')
    print('操作: qキーで終了')
    print('出力: 各フレームごとに処理結果を表示し、終了時にresult.txtへ保存')
    print()


def draw_texts_with_pillow(bgr_frame, texts):
    img_pil = Image.fromarray(cv2.cvtColor(bgr_frame, cv2.COLOR_BGR2RGB))
    draw = ImageDraw.Draw(img_pil)

    for item in texts:
        text = item['text']
        x, y = item['org']
        color = item['color']
        draw.text((x, y), text, font=font, fill=color)

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


def draw_detection_results(frame, detection_info):
    texts_to_draw = []

    texts_to_draw.append({
        'text': f'Frame: {frame_count} | Device: {str(device).upper()}',
        'org': (10, 30),
        'color': COLOR_GREEN_RGB,
        'font_type': 'main'
    })

    if detection_info['total'] > 0:
        texts_to_draw.append({
            'text': f'検出物体数: {detection_info["total"]}',
            'org': (10, 60),
            'color': COLOR_YELLOW_RGB,
            'font_type': 'main'
        })

        y_offset = 90
        for class_name, count in detection_info['objects'].items():
            jp_name = CLASS_NAMES_JP.get(class_name, class_name)
            texts_to_draw.append({
                'text': f'{jp_name}: {count}個',
                'org': (10, y_offset),
                'color': COLOR_YELLOW_RGB,
                'font_type': 'main'
            })
            y_offset += 30

    if font:
        frame = draw_texts_with_pillow(frame, texts_to_draw)
    else:
        cv2.putText(frame, f'Frame: {frame_count} | Device: {str(device).upper()}',
                    (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, COLOR_GREEN, 2)
        if detection_info['total'] > 0:
            cv2.putText(frame, f'Detections: {detection_info["total"]}',
                        (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.7, COLOR_YELLOW, 2)
            y_offset = 90
            for class_name, count in detection_info['objects'].items():
                info_text = f'{class_name}: {count}'
                cv2.putText(frame, info_text, (10, y_offset), cv2.FONT_HERSHEY_SIMPLEX, 0.7, COLOR_YELLOW, 2)
                y_offset += 30

    return frame


def format_detection_output(detection_info, detections):
    if len(detections) == 0:
        return 'count=0'
    else:
        parts = []
        for det in detections:
            x1, y1, x2, y2 = det['x1'], det['y1'], det['x2'], det['y2']
            class_name = det['name']
            conf = det['conf']
            parts.append(f'class={class_name},conf={conf:.3f},box=[{x1},{y1},{x2},{y2}]')
        return f'count={len(detections)}; ' + ' | '.join(parts)


def detect_objects(frame):
    global model

    yuv_img = cv2.cvtColor(frame, cv2.COLOR_BGR2YUV)
    yuv_img[:,:,0] = clahe.apply(yuv_img[:,:,0])
    clahe_frame = cv2.cvtColor(yuv_img, cv2.COLOR_YUV2BGR)

    results = model(clahe_frame, conf=CONF_THRESHOLD, verbose=False)

    return results[0]


def process_video_frame(frame, timestamp_ms, is_camera):
    final_results = detect_objects(frame)

    processed_frame = final_results.plot(img=frame.copy())

    detection_info = {'objects': defaultdict(int), 'total': 0}
    detections = []
    if final_results.boxes is not None:
        detection_info['total'] = len(final_results.boxes)
        for box in final_results.boxes:
            if hasattr(box, 'cls') and box.cls is not None:
                cls_id = int(box.cls.item())
                if cls_id < len(model.names):
                    class_name = model.names[cls_id]
                    detection_info['objects'][class_name] += 1
                    xyxy = box.xyxy[0].cpu().numpy()
                    conf = float(box.conf.item())
                    detections.append({
                        'x1': int(xyxy[0]),
                        'y1': int(xyxy[1]),
                        'x2': int(xyxy[2]),
                        'y2': int(xyxy[3]),
                        'conf': conf,
                        'class': cls_id,
                        'name': class_name
                    })

    global class_counts
    for class_name, count in detection_info['objects'].items():
        if class_name not in class_counts:
            class_counts[class_name] = 0
        class_counts[class_name] += count

    processed_frame = draw_detection_results(processed_frame, detection_info)

    result = format_detection_output(detection_info, detections)

    return processed_frame, result


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


display_program_header()

np.random.seed(RANDOM_SEED)
torch.manual_seed(RANDOM_SEED)

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

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

MODEL_NAME = MODEL_INFO[model_choice]['name']
selected_size = MODEL_INFO[model_choice]['size']
print(f'\nモデル: {MODEL_NAME} ({selected_size}) を選択しました。')
print('=' * 60)

if not os.path.exists(MODEL_NAME):
    print(f'モデルファイル {MODEL_NAME} はローカルに存在しません。')
    print(f'Ultralyticsライブラリによる自動ダウンロードを試みます...')

print(f'YOLO11モデル {MODEL_NAME} をロード中...')
try:
    model = YOLO(MODEL_NAME)
    model.to(device)
    model.eval()
    print(f'モデルのロードが完了しました')
    print('')
except Exception as e:
    print(f'モデルのロードに失敗しました: {e}')
    raise SystemExit(1)

print("=== YOLO11リアルタイム物体検出・セグメンテーション（COCO対応） ===")
print("0: 動画ファイル")
print("1: カメラ")
print("2: サンプル動画")
choice = input("選択: ")

is_camera = (choice == '1')

if choice == '0':
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfilename()
    if path:
        cap = cv2.VideoCapture(path)
    else:
        raise SystemExit(1)
elif choice == '1':
    cap = ThreadedVideoCapture(0, is_camera=True)
elif choice == '2':
    SAMPLE_URL = 'https://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.avi'
    SAMPLE_FILE = 'vtest.avi'
    if not os.path.exists(SAMPLE_FILE):
        print(f'サンプル動画 {SAMPLE_FILE} をダウンロードします...')
        urllib.request.urlretrieve(SAMPLE_URL, SAMPLE_FILE)
        print('ダウンロードが完了しました。')
    cap = cv2.VideoCapture(SAMPLE_FILE)
else:
    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('操作方法:')
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:
            print('動画の終端に達したか、フレームを読み込めませんでした。')
            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 is_camera:
            timestamp = datetime.fromtimestamp(current_time).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
            print(timestamp, result)
        else:
            print(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=== プログラム終了 ===')
    if cap:
        cap.release()
    cv2.destroyAllWindows()
    if results_log:
        with open('result.txt', 'w', encoding='utf-8') as f:
            f.write('=== YOLO11物体検出・セグメンテーション結果 ===\n')
            f.write(f'処理フレーム数: {frame_count}\n')
            f.write(f'使用モデル: {MODEL_NAME}\n')
            f.write(f'モデル情報: {MODEL_INFO[model_choice]["desc"]}\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'画像処理: CLAHE適用（YUV色空間）\n')
            f.write(f'信頼度閾値: {CONF_THRESHOLD}\n')
            f.write(f'\n検出されたクラス一覧:\n')
            for class_name, count in sorted(class_counts.items()):
                jp_name = CLASS_NAMES_JP.get(class_name, class_name)
                f.write(f'  {jp_name} ({class_name}): {count}回\n')
            if is_camera:
                f.write('\n形式: タイムスタンプ, 検出結果\n')
            else:
                f.write('\n形式: フレーム番号, 検出結果\n')
            f.write('\n')
            f.write('\n'.join(results_log))
        print(f'処理結果をresult.txtに保存しました')
        print(f'検出されたクラス数: {len(class_counts)}')