TimeSformer による人物動作認識（ソースコードと説明と利用ガイド）

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

# 人物動作認識プログラム（TimeSformer）
# 特徴技術名: TimeSformer
# 出典: Bertasius, G., Wang, H., & Torresani, L. (2021). Is space-time attention all you need for video understanding? Proceedings of the International Conference on Machine Learning (ICML)
# 特徴機能: 時空間分離注意機構による動作認識。空間注意と時間注意を分離適用し、動画フレーム間の時間的依存関係を効率的に学習してリアルタイム動作認識を実現
# 学習済みモデル: timesformer-base-finetuned-k400。Kinetics-400データセットで事前訓練された400動作クラス認識モデル。HuggingFace transformersライブラリにより自動ダウンロード
# 特徴技術および学習済モデルの利用制限: **Apache-2.0ライセンス（商用利用可能）。Facebook Research開発、HuggingFaceにて配布**
# 方式設計:
#   関連利用技術: OpenCV（動画処理、フレーム取得）、transformers（TimeSformerモデル）、torch（深層学習）、PIL（画像変換）
#   入力と出力: 入力: 動画（ユーザは「0:動画ファイル，1:カメラ，2:サンプル動画」のメニューで選択．0:動画ファイルの場合はtkinterでファイル選択．1の場合はOpenCVでカメラが開く．2の場合はhttps://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.aviを使用）；出力: OpenCV画面で動画再生、動作認識結果と信頼度をprint()で表示、result.txtファイルに保存
#   処理手順: 1.動画フレーム読み込み→2.フレームバッファ蓄積→3.TimeSformerモデル推論→4.動作クラス分類→5.信頼度算出→6.結果表示
#   前処理、後処理: 前処理：フレームリサイズ(224x224)、正規化、テンソル変換；後処理：softmax確率計算、上位5クラス抽出
#   追加処理: フレームバッファ管理、時間窓スライディング、重複フレーム削除
#   調整を必要とする設定値: フレーム数（8フレームまたは16フレーム選択可能）、信頼度閾値（デフォルト0.1）、フレームスキップ間隔（デフォルト2）
#   算出・計算処理の検証: 動作クラス確率は0-1範囲のsoftmax出力、上位5クラスの信頼度合計が正しく算出
# 将来方策: 複数人物対応、カスタム動作クラス追加、リアルタイム動作セグメンテーション
# その他の重要事項: フレーム蓄積後に推論実行のため初期遅延あり、GPU使用推奨
# 前準備: pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
# pip install transformers opencv-python pillow

import cv2
import torch
import numpy as np
from transformers import TimesformerForVideoClassification, AutoImageProcessor
from PIL import Image
import tkinter as tk
from tkinter import filedialog
import urllib.request
import time
from datetime import datetime

# グローバル変数
frame_count = 0
results_log = []
frame_buffer = []
current_action = None
action_start_time = None

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

try:
    print('TimeSformerモデル読み込み中...')
    model = TimesformerForVideoClassification.from_pretrained("facebook/timesformer-base-finetuned-k400")
    processor = AutoImageProcessor.from_pretrained("facebook/timesformer-base-finetuned-k400")
    model.to(device)
    model.eval()
    print('モデル読み込み完了')
    MODEL_NUM_FRAMES = int(getattr(model.config, "num_frames", 8))
    print(f'モデル設定のフレーム数: {MODEL_NUM_FRAMES}')
except Exception as e:
    print(f'モデル読み込みエラー: {e}')
    exit()

def preprocess_frames(video_frames):
    # AutoImageProcessorを使用した公式前処理
    inputs = processor(images=video_frames, return_tensors="pt")
    return inputs.pixel_values

def video_frame_processing(frame):
    global frame_count, frame_buffer, current_action, action_start_time, FRAME_COUNT

    current_time = time.time()
    frame_count += 1

    # フレームを224x224にリサイズ
    rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    pil_frame = Image.fromarray(rgb_frame)

    # フレームバッファに追加
    frame_buffer.append(pil_frame)

    # 設定フレーム数蓄積後に推論実行
    if len(frame_buffer) >= FRAME_COUNT:
        try:
            # 最新フレームを使用
            video_frames = frame_buffer[-FRAME_COUNT:]

            # AutoImageProcessorを使用した前処理
            pixel_values = preprocess_frames(video_frames).to(device)

            # 推論実行
            with torch.no_grad():
                outputs = model(pixel_values=pixel_values)
                probabilities = torch.nn.functional.softmax(outputs.logits, dim=-1)

            # 上位5クラス取得
            top5_prob, top5_indices = torch.topk(probabilities, 5)

            # Top-1動作取得
            top1_class_idx = top5_indices[0][0].item()
            top1_confidence = top5_prob[0][0].item()
            top1_action = model.config.id2label[top1_class_idx]

            # 動作継続時間管理
            if current_action != top1_action:
                current_action = top1_action
                action_start_time = current_time
                duration = 0.0
            else:
                duration = current_time - action_start_time if action_start_time else 0.0

            # 結果整形
            predicted_actions = []
            for i in range(5):
                class_idx = top5_indices[0][i].item()
                confidence = top5_prob[0][i].item()
                class_name = model.config.id2label[class_idx]
                predicted_actions.append(f"{class_name}({confidence:.3f})")

            top5_sum = top5_prob[0].sum().item()
            sum_ok = (0.0 <= top5_sum <= 1.0 + 1e-6)

            result = f"Top5: {', '.join(predicted_actions)} | Top5Sum={top5_sum:.3f} valid={sum_ok}"

            # OpenCV画面に表示する情報
            display_text = f"{top1_action} ({top1_confidence:.3f}) - {duration:.1f}sec"

        except Exception as e:
            result = f"推論エラー: {str(e)[:50]}"
            display_text = "推論エラー"

        # フレームバッファサイズ制限
        if len(frame_buffer) > FRAME_COUNT * 2:
            frame_buffer = frame_buffer[-FRAME_COUNT:]
    else:
        result = f"フレーム蓄積中: {len(frame_buffer)}/{FRAME_COUNT}"
        display_text = f"Loading... {len(frame_buffer)}/{FRAME_COUNT}"

    # OpenCV画面にテキスト表示
    display_frame = frame.copy()
    cv2.putText(display_frame, display_text, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)

    return display_frame, result, current_time

print("人物動作認識プログラム（TimeSformer）")
print("フレーム数を選択してください:")
print("0: 8フレーム")
print("1: 16フレーム")

frame_choice = input("選択: ")
if frame_choice == '1':
    requested = 16
else:
    requested = 8

# モデル設定に必ず合わせる
FRAME_COUNT = MODEL_NUM_FRAMES
if requested != MODEL_NUM_FRAMES:
    print(f"モデル設定に合わせてフレーム数を{MODEL_NUM_FRAMES}に統一します（選択値{requested}は無視）")
else:
    print(f"モデル設定と一致: {MODEL_NUM_FRAMES}")

print(f"設定フレーム数: {FRAME_COUNT}")
print("\n動画ソースを選択してください:")
print("0: 動画ファイル")
print("1: カメラ")
print("2: サンプル動画")

choice = input("選択: ")

if choice == '0':
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfilename()
    if not path:
        exit()
    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:
    # サンプル動画ダウンロード・処理
    SAMPLE_URL = 'https://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.avi'
    SAMPLE_FILE = 'vtest.avi'
    try:
        urllib.request.urlretrieve(SAMPLE_URL, SAMPLE_FILE)
        cap = cv2.VideoCapture(SAMPLE_FILE)
    except Exception as e:
        print(f'サンプル動画ダウンロードエラー: {e}')
        exit()

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

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

        MAIN_FUNC_DESC = f"TimeSformer動作認識({FRAME_COUNT}フレーム)"
        processed_frame, result, current_time = video_frame_processing(frame)
        cv2.imshow(MAIN_FUNC_DESC, processed_frame)
        if choice == '1':  # カメラの場合
            print(datetime.fromtimestamp(current_time).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3], result)
        else:  # 動画ファイルの場合
            print(frame_count, result)
        results_log.append(result)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
finally:
    print('\n=== プログラム終了 ===')
    cap.release()
    cv2.destroyAllWindows()
    if results_log:
        with open('result.txt', 'w', encoding='utf-8') as f:
            f.write('=== TimeSformer動作認識結果 ===\n')
            f.write(f'処理フレーム数: {frame_count}\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(f'\n処理結果をresult.txtに保存しました')