Layumi's Person Re-ID Baseline with Swin Transformer による画像全体の特徴量算出（ソースコードと実行結果）

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

# Layumi's Person Re-ID Baseline with Swin Transformer (人体全身特徴抽出)
# 実装基盤: Layumi's PyTorch ReID Baseline
#   - 小さく、使いやすく、強力なPyTorch実装のperson/vehicle re-ID基本モデル
#   - GitHub: https://github.com/layumi/Person_reID_baseline_pytorch
#   - 著者: Liang Zheng, Zhedong Zheng, Yi Yang
#
# バックボーン技術: Swin Transformer
#   - 出典: Liu, Z., Lin, Y., Cao, Y., Hu, H., Wei, Y., Zhang, Z., ... & Guo, B. (2021).
#     Swin transformer: Hierarchical vision transformer using shifted windows.
#     In Proceedings of the IEEE/CVF international conference on computer vision (pp. 10012-10022).
#   - 特徴: 階層的シフテッドウィンドウ自己注意機構
#
# Person Re-ID関連論文:
#   - Zheng, L., Shen, L., Tian, L., Wang, S., Wang, J., & Tian, Q. (2015).
#     Scalable person re-identification: A benchmark.
#     In Proceedings of the IEEE international conference on computer vision (pp. 1116-1124).
#   - Sun, Y., Zheng, L., Yang, Y., Tian, Q., & Wang, S. (2018).
#     Beyond part models: Person retrieval with refined part pooling (and a strong convolutional baseline).
#     In Proceedings of the European conference on computer vision (ECCV) (pp. 480-496).
#
# === 学習済みモデルの詳細 ===
# データセット: Market-1501 (人体全身画像データセット)
#   - 1501人の歩行者画像
#   - 画像サイズ: 128×64ピクセル (縦長の人体全身)
#   - 6台のカメラで撮影された32,668枚の人体画像
#   - 顔認識用ではなく、人体全体の外観認識用
#
# 学習内容:
#   - 751人分の人物ID分類 (訓練用)
#   - 服装の色・パターン
#   - 体型・身長の相対的特徴
#   - 持ち物（カバン、リュックなど）
#   - 姿勢・全体的な外観
#   - 顔の詳細は解像度的に認識不可能
#
# Layumiベースライン実装の特徴:
#   - Person Re-Identification (人物再識別) 専用
#   - 異なるカメラ間での同一人物追跡を目的
#   - Strong Baseline手法を採用
#   - BNNeck構造による特徴学習
#   - 学習済みモデル: https://drive.google.com/open?id=1XVEYb0TN2SbBYOqf8SzazfYZlpH9CxyE
#   - 性能: Rank@1精度 92.75%, mAP 79.70%
#
# 方式設計:
#   関連利用技術: PyTorch、timm、OpenCV、PIL、gdown
#   入力と出力: 入力: 動画フレーム、出力: 512次元人体特徴ベクトル、フレーム間類似度
#   処理手順: 学習済みモデルダウンロード、ft_net_swin読み込み、フレーム処理、特徴抽出、類似度計算
#   前処理、後処理: リサイズ(224,224)、ImageNet正規化、テンソル変換
#   追加処理: ClassBlock (BNNeck)による512次元特徴ベクトル生成
#   調整を必要とする設定値: linear_num=512（特徴ベクトル次元数）、droprate=0.5
#   算出・計算処理の検証: コサイン類似度によるフレーム間の外観変化検出
#
# 重要事項:
#   - 本来は人体検出後の切り出し画像に適用すべき
#   - 現在は画像全体を処理（背景含む）
#   - 人体が映っている場合に効果的
#
# 前準備: pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
# pip install timm opencv-python pillow requests gdown

import cv2
import tkinter as tk
from tkinter import filedialog
import urllib
import urllib.request
import torch
import torch.nn as nn
import torch.nn.functional as F
import torchvision.transforms as transforms
from PIL import Image, ImageDraw, ImageFont
import timm
import time
from datetime import datetime
import os
import numpy as np
import json

# gdownモジュールのインポート処理
try:
    import gdown
    GDOWN_AVAILABLE = True
except ImportError:
    GDOWN_AVAILABLE = False
    print("警告: gdownモジュールが見つかりません。pip install gdownを実行してください。")

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

# 日本語フォント設定
FONT_PATH = 'C:/Windows/Fonts/meiryo.ttc'
FONT_SIZE = 20
try:
    font_japanese = ImageFont.truetype(FONT_PATH, FONT_SIZE)
except:
    font_japanese = None
    print("日本語フォントの読み込みに失敗しました。英語表示になります。")

frame_count = 0
results_log = []
feature_vectors_log = []  # 特徴ベクトルを保存
previous_feature = None  # 前フレームの特徴ベクトル保存用
previous_frame_number = None  # 前フレーム番号

def weights_init_kaiming(m):
    """Kaiming初期化 (He initialization)
    参考: He, K., Zhang, X., Ren, S., & Sun, J. (2015).
    Delving deep into rectifiers: Surpassing human-level performance on imagenet classification."""
    classname = m.__class__.__name__
    if classname.find('Linear') != -1:
        nn.init.kaiming_normal_(m.weight, a=0, mode='fan_out')
        nn.init.constant_(m.bias, 0.0)
    elif classname.find('Conv') != -1:
        nn.init.kaiming_normal_(m.weight, a=0, mode='fan_in')
        if m.bias is not None:
            nn.init.constant_(m.bias, 0.0)
    elif classname.find('BatchNorm') != -1:
        if m.affine:
            nn.init.constant_(m.weight, 1.0)
            nn.init.constant_(m.bias, 0.0)

def weights_init_classifier(m):
    """分類層の初期化"""
    classname = m.__class__.__name__
    if classname.find('Linear') != -1:
        nn.init.normal_(m.weight, std=0.001)
        if m.bias is not None:
            nn.init.constant_(m.bias, 0.0)

class ClassBlock(nn.Module):
    """BNNeck構造を含む分類ブロック
    参考: Luo, H., Gu, Y., Liao, X., Lai, S., & Jiang, W. (2019).
    Bag of tricks and a strong baseline for deep person re-identification."""
    def __init__(self, input_dim, class_num, droprate, relu=False, bnorm=True, linear=512, return_f=False):
        super(ClassBlock, self).__init__()
        self.return_f = return_f
        add_block = []
        if linear > 0:
            add_block += [nn.Linear(input_dim, linear)]
        else:
            linear = input_dim
        if bnorm:
            add_block += [nn.BatchNorm1d(linear)]  # BNNeck
        if relu:
            add_block += [nn.LeakyReLU(0.1)]
        if droprate > 0:
            add_block += [nn.Dropout(p=droprate)]
        add_block = nn.Sequential(*add_block)
        add_block.apply(weights_init_kaiming)

        classifier = []
        classifier += [nn.Linear(linear, class_num)]
        classifier = nn.Sequential(*classifier)
        classifier.apply(weights_init_classifier)

        self.linear_num = linear
        self.add_block = add_block
        self.classifier = classifier

    def forward(self, x):
        x = self.add_block(x)
        if self.return_f:
            f = x
            x = self.classifier(x)
            return [x, f]
        else:
            x = self.classifier(x)
            return x

class ft_net_swin(nn.Module):
    """Layumi's ReID Baseline with Swin Transformer backbone
    Layumiベースラインアーキテクチャ + Swin Transformerバックボーン"""
    def __init__(self, class_num=751, droprate=0.5, stride=2, circle=False, linear_num=512):
        super(ft_net_swin, self).__init__()
        # Swin Transformerバックボーン
        model_ft = timm.create_model('swin_base_patch4_window7_224', pretrained=True, drop_path_rate=0.2)
        model_ft.head = nn.Sequential()  # 元の分類ヘッドを削除
        self.model = model_ft
        self.circle = circle
        self.avgpool1d = nn.AdaptiveAvgPool1d(1)
        self.avgpool2d = nn.AdaptiveAvgPool2d((1,1))
        # Layumiベースラインの分類ブロック（BNNeck付き）
        # Market-1501データセット751人分の人体特徴を学習
        self.classifier = ClassBlock(1024, class_num, droprate, linear=linear_num, return_f=circle)

    def forward(self, x):
        # Swin Transformerで人体全体の階層的特徴を抽出
        x = self.model.forward_features(x)
        # Global Average Pooling
        if x.dim() == 3:
            x = self.avgpool1d(x.permute((0, 2, 1)))
        else:
            x = self.avgpool2d(x.permute((0, 3, 1, 2)))
        x = x.view(x.size(0), x.size(1))
        # 512次元の人体特徴ベクトルを生成（BNNeck経由）
        x = self.classifier(x)
        return x

def download_pretrained_model():
    """Layumi's ReID Baseline学習済みモデルのダウンロード"""
    model_path = 'layumi_swin_pretrained.pth'
    if not os.path.exists(model_path):
        if not GDOWN_AVAILABLE:
            print("gdownが利用できないため、学習済みモデルをダウンロードできません。")
            print("pip install gdownを実行してください。")
            return None
        print("Layumi's ReID Baseline (Market-1501学習済み) モデルをダウンロード中...")
        try:
            file_id = '1XVEYb0TN2SbBYOqf8SzazfYZlpH9CxyE'
            gdown.download(f'https://drive.google.com/uc?id={file_id}', model_path, quiet=False)
            print("ダウンロード完了（751人分の人体特徴を学習済み）")
        except Exception as e:
            print(f"ダウンロードエラー: {e}")
            print("ImageNet事前学習のみで実行します")
            return None
    return model_path

# Layumi's Person Re-IDモデルの初期化（751人分の人体ID分類）
model = ft_net_swin(class_num=751, linear_num=512, circle=False).to(device)

# Layumi's ReID Baseline学習済みモデルのロード
pretrained_path = download_pretrained_model()
if pretrained_path and os.path.exists(pretrained_path):
    try:
        checkpoint = torch.load(pretrained_path, map_location=device)
        model.load_state_dict(checkpoint, strict=False)
        print("Layumi's ReID Baseline (Market-1501) モデルをロード完了")
        print("学習内容: 服装、体型、持ち物、姿勢などの人体全体特徴")
    except Exception as e:
        print(f"モデルロードエラー: {e}")
        print("ImageNet事前学習のみで実行")

model.eval()

# 入力画像を224x224にリサイズして正規化
transform = transforms.Compose([
    transforms.Resize((224, 224), interpolation=3),
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])

def calculate_cosine_similarity(feat1, feat2):
    """コサイン類似度の計算

    定義: cos(θ) = (A・B) / (||A|| × ||B||)
    - A・B: ベクトルAとBの内積
    - ||A||, ||B||: ベクトルAとBのL2ノルム（ユークリッドノルム）

    値域: [-1, 1]
    - 1: 完全に同一方向（同一特徴）
    - 0: 直交（無関係）
    - -1: 完全に逆方向（正反対の特徴）

    Person Re-IDでの解釈:
    - 0.9以上: ほぼ同一人物の可能性が高い
    - 0.7-0.9: 類似した外観（同一人物の可能性あり）
    - 0.5-0.7: 部分的に類似
    - 0.5未満: 異なる人物の可能性が高い

    参考: Hermans, A., Beyer, L., & Leibe, B. (2017).
    In defense of the triplet loss for person re-identification.
    """
    feat1_norm = feat1 / np.linalg.norm(feat1)
    feat2_norm = feat2 / np.linalg.norm(feat2)
    return np.dot(feat1_norm, feat2_norm)

def calculate_euclidean_distance(feat1, feat2):
    """ユークリッド距離の計算

    定義: d = √(Σ(ai - bi)²)

    Person Re-IDでの使用:
    特徴空間での距離を測定。小さいほど類似度が高い。
    """
    return np.linalg.norm(feat1 - feat2)

def draw_japanese_text(img, text, position, font, color=(0, 255, 0)):
    """OpenCV画像に日本語テキストを描画"""
    if font_japanese is None:
        # フォントが利用できない場合は英語で描画
        cv2.putText(img, text, position, cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
        return img

    # PILで日本語描画
    img_pil = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
    draw = ImageDraw.Draw(img_pil)
    draw.text(position, text, font=font, fill=color[::-1])  # BGRをRGBに変換
    return cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)

def video_frame_processing(frame):
    global frame_count, previous_feature, feature_vectors_log, previous_frame_number
    current_time = time.time()
    frame_count += 1

    try:
        # フレーム全体をPIL画像に変換（本来は人体検出後の画像を使用すべき）
        pil_image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        input_tensor = transform(pil_image).unsqueeze(0).to(device)

        # Layumi's ReID Baselineで人体特徴ベクトルを抽出
        with torch.no_grad():
            outputs = model(input_tensor)
            if isinstance(outputs, list):
                features = outputs[1]  # circle=Trueの場合
            else:
                features = outputs

            # 512次元の人体特徴ベクトル（BNNeck後）
            feature_vector = features.cpu().numpy().flatten()

        # 現在のフレームの512次元ベクトルの統計情報
        feat_mean = np.mean(feature_vector)  # 現在のフレームの512次元の平均
        feat_std = np.std(feature_vector)    # 現在のフレームの512次元の標準偏差
        feat_min = np.min(feature_vector)    # 現在のフレームの512次元の最小値
        feat_max = np.max(feature_vector)    # 現在のフレームの512次元の最大値
        feat_norm = np.linalg.norm(feature_vector)  # 現在のフレームの512次元のL2ノルム

        # 前フレームとの比較
        comparison_text = "比較対象: なし（初回フレーム）"
        appearance_change = "計測開始"
        cosine_sim = None  # 初回は比較不可
        euclidean_dist = None  # 初回は比較不可

        if previous_feature is not None and previous_frame_number is not None:
            # 現在のフレームと直前のフレームの比較
            cosine_sim = calculate_cosine_similarity(feature_vector, previous_feature)
            euclidean_dist = calculate_euclidean_distance(feature_vector, previous_feature)
            comparison_text = f"比較対象: フレーム{frame_count} vs フレーム{previous_frame_number}"

            # 外観変化の判定（コサイン類似度に基づく）
            if cosine_sim > 0.95:
                appearance_change = "同一外観"
            elif cosine_sim > 0.85:
                appearance_change = "微小変化"
            elif cosine_sim > 0.70:
                appearance_change = "中程度変化"
            elif cosine_sim > 0.50:
                appearance_change = "大幅変化"
            else:
                appearance_change = "完全に異なる"

        # 特徴ベクトルをログに保存
        feature_data = {
            'frame': frame_count,
            'timestamp': current_time,
            'feature_vector': feature_vector.tolist(),
            'statistics': {
                'mean': float(feat_mean),
                'std': float(feat_std),
                'min': float(feat_min),
                'max': float(feat_max),
                'l2_norm': float(feat_norm),
                'description': '現在のフレームの512次元ベクトルの統計値'
            },
            'comparison': {
                'compared_with_frame': previous_frame_number if previous_frame_number else None,
                'cosine': float(cosine_sim) if cosine_sim is not None else None,
                'euclidean_distance': float(euclidean_dist) if euclidean_dist is not None else None,
                'description': comparison_text
            }
        }
        feature_vectors_log.append(feature_data)

        # 現在のフレームの特徴を次回比較用に保存
        previous_feature = feature_vector.copy()
        previous_frame_number = frame_count

        if cosine_sim is not None and euclidean_dist is not None:
            result = f"フレーム{frame_count}: {comparison_text}, コサイン類似度={cosine_sim:.4f}, ユークリッド距離={euclidean_dist:.3f}, {appearance_change}"
        else:
            result = f"フレーム{frame_count}: {comparison_text}, {appearance_change}"

        # 画面表示（日本語対応）
        frame = draw_japanese_text(frame, "人物再識別システム", (10, 30), font_japanese, (0, 255, 0))
        frame = draw_japanese_text(frame, f"現在のフレーム: {frame_count}", (10, 60), font_japanese, (0, 255, 0))
        frame = draw_japanese_text(frame, comparison_text, (10, 90), font_japanese, (0, 255, 0))
        if cosine_sim is not None:
            frame = draw_japanese_text(frame, f"コサイン類似度: {cosine_sim:.4f}", (10, 120), font_japanese, (0, 255, 0))
        if euclidean_dist is not None:
            frame = draw_japanese_text(frame, f"ユークリッド距離: {euclidean_dist:.3f}", (10, 150), font_japanese, (0, 255, 0))
        frame = draw_japanese_text(frame, f"状態: {appearance_change}", (10, 180), font_japanese, (0, 255, 0))
        frame = draw_japanese_text(frame, f"512次元統計: 平均={feat_mean:.3f}, 標準偏差={feat_std:.3f}", (10, 210), font_japanese, (0, 255, 0))

        return frame, result, current_time

    except Exception as e:
        result = f"エラー: {str(e)}"
        return frame, result, current_time

print("\n" + "="*70)
print("Layumi's Person Re-ID Baseline with Swin Transformer")
print("GitHub: https://github.com/layumi/Person_reID_baseline_pytorch")
print("="*70)

print("\n【比較方法】")
print("比較対象: 現在のフレーム(N) と 直前のフレーム(N-1) の512次元特徴ベクトル")
print("比較タイミング: 各フレーム処理時に直前フレームと比較")

print("\n【類似度メトリクスの定義】")
print("1. コサイン類似度 (Cosine Similarity):")
print("   定義: cos(θ) = (A・B) / (||A|| × ||B||)")
print("   比較対象: フレームNの512次元ベクトル と フレームN-1の512次元ベクトル")
print("   値域: [-1, 1] (1=完全一致, 0=無関係, -1=正反対)")
print("   Person Re-IDでの解釈:")
print("     > 0.95: ほぼ同一人物・同一姿勢")
print("     0.85-0.95: 同一人物の可能性高（姿勢変化）")
print("     0.70-0.85: 類似外観（同一人物の可能性あり）")
print("     0.50-0.70: 部分的類似")
print("     < 0.50: 異なる人物の可能性大")

print("\n2. ユークリッド距離 (Euclidean Distance):")
print("   定義: d = √(Σ(ai - bi)²)")
print("   比較対象: フレームNの512次元ベクトル と フレームN-1の512次元ベクトル")
print("   値域: [0, ∞) (0=完全一致, 大きいほど相違)")

print("\n【統計値の対象】")
print("平均・標準偏差・最小値・最大値・L2ノルム:")
print("  対象: 各フレームの512次元特徴ベクトル（そのフレーム内の512個の値）")
print("  注意: 全フレームの統計ではなく、個別フレームの512次元の統計")

print("\n【重要】このモデルについて:")
print("  - 実装: Layumi's PyTorch ReID Baseline")
print("  - 対象: 人体全身（顔ではありません）")
print("  - 学習データ: Market-1501 (1501人の歩行者画像)")
print("  - 認識内容: 服装、体型、持ち物、姿勢などの外観特徴")
print("  - 特徴ベクトル: 512次元（BNNeck後の特徴表現）")
print("  - 注意: 現在は画像全体を処理（本来は人体検出が必要）")
print("\n" + "="*70)

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'
    urllib.request.urlretrieve(SAMPLE_URL, SAMPLE_FILE)
    cap = cv2.VideoCapture(SAMPLE_FILE)

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 = "人物再識別 (Person Re-ID)"
        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('=== 結果 ===\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に保存しました')

    # 特徴ベクトルログの保存
    if feature_vectors_log:
        with open('feature_vectors.json', 'w', encoding='utf-8') as f:
            json.dump(feature_vectors_log, f, ensure_ascii=False, indent=2)
        print(f'特徴ベクトルをfeature_vectors.jsonに保存しました')