MediaPipe前景・背景セグメンテーション（ソースコードと実行結果）

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

# プログラム名: MediaPipe前景・背景セグメンテーションプログラム
# 特徴技術名: MediaPipe
# 出典: MediaPipe Tasks - Google
# 特徴機能: MediaPipe Selfie Segmentationによる前景・背景分離。リアルタイムで動作する軽量な前景抽出
# 学習済みモデル: Selfie Segmentation内蔵モデル（前景・背景分離）
# 方式設計:
#   - 関連利用技術:
#     - MediaPipe: Googleが開発したマルチプラットフォーム機械学習ソリューション
#     - OpenCV: 画像処理、カメラ制御、描画処理、動画入出力管理
#   - 入力と出力: 入力: 動画（ユーザは「0:動画ファイル，1:カメラ，2:サンプル動画」のメニューで選択．0:動画ファイルの場合はtkinterでファイル選択．1の場合はOpenCVでカメラが開く．2の場合はhttps://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.aviを使用）、出力: OpenCV画面でリアルタイム表示（検出した前景を半透明マスクで表示）、1秒間隔でprint()による処理結果表示、プログラム終了時にresult.txtファイルに保存
#   - 処理手順: 1.フレーム取得、2.MediaPipe推論実行、3.前景・背景分離、4.信頼度閾値による選別、5.半透明マスク描画
#   - 前処理、後処理: 前処理：MediaPipe内部で自動実行。後処理：信頼度による閾値フィルタリングを実施
#   - 追加処理: セグメンテーションマスクの平均信頼度計算、前景領域の面積・重心計算
#   - 調整を必要とする設定値: CONF_THRESH（セグメンテーション信頼度閾値、デフォルト0.5）- 値を上げると誤検出が減少するが検出漏れが増加
# 将来方策: CONF_THRESHの動的調整機能。フレーム毎のマスク面積を監視し、面積が閾値を超えた場合は信頼度を上げ、面積が少ない場合は下げる適応的制御の実装
# その他の重要事項: Windows環境専用設計、複数人物が存在する場合も単一の前景マスクとして処理
# 前準備:
#   - pip install mediapipe opencv-python numpy

import cv2
import tkinter as tk
from tkinter import filedialog
import os
import numpy as np
import mediapipe as mp
import warnings
import time
import urllib.request

warnings.filterwarnings('ignore')

# ===== 設定・定数管理 =====
# MediaPipe設定
mp_selfie_segmentation = mp.solutions.selfie_segmentation

# モデル選択（0, 1から選択可能）
MODEL_SIZE = '0'  # 使用するモデルサイズ（0=一般モデル, 1=横向き画像最適化モデル）

# モデル情報
MODEL_INFO = {
    '0': {
        'name': 'General Model',
        'desc': '一般用途（推奨）',
        'model_selection': 0
    },
    '1': {
        'name': 'Landscape Model',
        'desc': '横向き画像最適化',
        'model_selection': 1
    }
}

# クラス名（前景）
CLASS_NAME = 'foreground'

# 前景用の色（緑）
FOREGROUND_COLOR = (0, 255, 0)

SAMPLE_URL = 'https://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.avi'
SAMPLE_FILE = 'vtest.avi'
RESULT_FILE = 'result.txt'

# カメラ設定
WINDOW_WIDTH = 1280  # カメラ解像度幅
WINDOW_HEIGHT = 720  # カメラ解像度高さ
FPS = 30  # フレームレート

# 検出パラメータ（調整可能）
CONF_THRESH = 0.5  # セグメンテーション信頼度閾値（0.0-1.0）

# 表示設定
PRINT_INTERVAL = 1.0  # 結果出力間隔（秒）


# プログラム概要表示
print('=== MediaPipe前景・背景セグメンテーションプログラム ===')
print('概要: リアルタイムで前景を抽出し、半透明マスクで表示します')
print('機能: MediaPipe Selfie Segmentationによる前景・背景分離')
print('操作: qキーで終了')
print('出力: 1秒間隔での処理結果表示、終了時にresult.txt保存')
print()

# システム初期化
print('システム初期化中...')
start_time = time.time()

# MediaPipeモデル初期化
segmentation = None
try:
    print(f'MediaPipe Selfie Segmentation {MODEL_INFO[MODEL_SIZE]["name"]}モデルを初期化中...')
    segmentation = mp_selfie_segmentation.SelfieSegmentation(
        model_selection=MODEL_INFO[MODEL_SIZE]["model_selection"]
    )
    print(f'MediaPipe Selfie Segmentation {MODEL_INFO[MODEL_SIZE]["name"]}モデルの初期化が完了しました')
    print(f'モデル: {MODEL_INFO[MODEL_SIZE]["name"]} ({MODEL_INFO[MODEL_SIZE]["desc"]})')
    print(f'セグメンテーション対象: 前景（人物等）')
except Exception as e:
    print('MediaPipe Selfie Segmentationモデルの初期化に失敗しました')
    print(f'エラー: {e}')
    exit()

print('CPUモード')
print('初期化完了')
print()

# グローバル変数
frame_count = 0
last_print_time = time.time()
results_log = []


def video_processing(frame):
    """フレーム処理メイン関数"""
    global frame_count, last_print_time, results_log

    frame_count += 1

    # RGB変換（MediaPipeはRGBを期待）
    rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

    # セグメンテーション実行
    results = segmentation.process(rgb_frame)

    foreground_detected = False
    confidence = 0.0
    area = 0
    centroid_x = 0
    centroid_y = 0

    if results.segmentation_mask is not None:
        # マスクを取得
        mask = results.segmentation_mask

        # 閾値処理
        binary_mask = (mask > CONF_THRESH).astype(np.uint8)

        # 前景が検出された場合
        if np.any(binary_mask):
            foreground_detected = True
            # マスクの平均信頼度を計算
            confidence = np.mean(mask[binary_mask == 1])

            # 面積計算（ピクセル数）
            area = np.sum(binary_mask)

            # 重心計算
            y_indices, x_indices = np.where(binary_mask == 1)
            if len(x_indices) > 0:
                centroid_x = int(np.mean(x_indices))
                centroid_y = int(np.mean(y_indices))

    # 1秒間隔での出力
    current_time = time.time()
    if current_time - last_print_time >= PRINT_INTERVAL:
        if foreground_detected:
            output = f'フレーム {frame_count}: 前景検出あり (信頼度: {confidence:.1%}, 面積: {area}px, 重心: ({centroid_x}, {centroid_y}))'
        else:
            output = f'フレーム {frame_count}: 前景検出なし'
        print(output)

        results_log.append(output)
        last_print_time = current_time

    # 描画処理
    output_frame = frame.copy()

    if foreground_detected and results.segmentation_mask is not None:
        mask = results.segmentation_mask

        # 半透明マスクの作成
        colored_mask = np.zeros_like(frame)
        colored_mask[:, :] = FOREGROUND_COLOR

        # マスクを適用（半透明）
        alpha = 0.4
        mask_3channel = np.stack([mask] * 3, axis=-1)
        output_frame = output_frame * (1 - mask_3channel * alpha) + colored_mask * mask_3channel * alpha
        output_frame = output_frame.astype(np.uint8)

        # ラベル表示
        label1 = f'{CLASS_NAME}'
        label2 = f'Conf:{confidence:.1%}'
        label3 = f'Area:{area}px'
        label4 = f'Centroid:({centroid_x},{centroid_y})'

        cv2.putText(output_frame, label1, (10, 90), cv2.FONT_HERSHEY_SIMPLEX, 0.7, FOREGROUND_COLOR, 2)
        cv2.putText(output_frame, label2, (10, 115), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (255, 255, 255), 1)
        cv2.putText(output_frame, label3, (10, 135), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (255, 255, 255), 1)
        cv2.putText(output_frame, label4, (10, 155), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (255, 255, 255), 1)

        # 重心に十字マーク表示
        cv2.drawMarker(output_frame, (centroid_x, centroid_y), (255, 0, 0), cv2.MARKER_CROSS, 10, 2)

    # システム情報表示
    status = '前景あり' if foreground_detected else '前景なし'
    info1 = f'MediaPipe (CPU) | Frame: {frame_count} | Status: {status}'
    info2 = 'Press: q=Quit'

    cv2.putText(output_frame, info1, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
    cv2.putText(output_frame, info2, (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 1)

    return output_frame


# 入力選択
print('0: 動画ファイル')
print('1: カメラ')
print('2: サンプル動画')

choice = input('選択: ')
temp_file = None

if choice == '0':
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfilename()
    if not path:
        exit()
    cap = cv2.VideoCapture(path)
    if not cap.isOpened():
        print(f'動画ファイルを開けませんでした: {path}')
        exit()
elif choice == '1':
    cap = cv2.VideoCapture(0, cv2.CAP_DSHOW)
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, WINDOW_WIDTH)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, WINDOW_HEIGHT)
    cap.set(cv2.CAP_PROP_FPS, FPS)
    if not cap.isOpened():
        print('カメラを開けませんでした')
        exit()
elif choice == '2':
    # サンプル動画ダウンロード・処理
    try:
        urllib.request.urlretrieve(SAMPLE_URL, SAMPLE_FILE)
        temp_file = SAMPLE_FILE
        cap = cv2.VideoCapture(SAMPLE_FILE)
        if not cap.isOpened():
            print(f'サンプル動画を開けませんでした: {SAMPLE_FILE}')
            exit()
        print('サンプル動画のダウンロードが完了しました')
    except Exception as e:
        print(f'動画のダウンロードに失敗しました: {SAMPLE_URL}')
        print(f'エラー: {e}')
        exit()
else:
    print('無効な選択です')
    exit()

# 動画処理開始メッセージ
print('\n=== 動画処理開始 ===')
print('操作方法:')
print('  q キー: プログラム終了')
print()

# メイン処理
try:
    while True:
        cap.grab()
        ret, frame = cap.retrieve()
        if not ret:
            break

        processed_frame = video_processing(frame)
        cv2.imshow('MediaPipe Foreground-Background Segmentation', processed_frame)

        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
finally:
    cap.release()
    cv2.destroyAllWindows()
    segmentation.close()

    # 結果保存
    if results_log:
        with open(RESULT_FILE, 'w', encoding='utf-8') as f:
            f.write('=== MediaPipe前景・背景セグメンテーション結果 ===\n')
            f.write(f'処理フレーム数: {frame_count}\n')
            f.write(f'使用デバイス: CPU\n')
            f.write('\n')
            f.write('\n'.join(results_log))

        print(f'\n処理結果を{RESULT_FILE}に保存しました')

    if temp_file and os.path.exists(temp_file):
        os.remove(temp_file)

print('\n=== プログラム終了 ===')