ResNet18特徴抽出による画像異常検知（ソースコードと実行結果）

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

# プログラム名: ResNet18特徴抽出による動画異常検知プログラム

# 特徴技術名: ResNet18によるCNN特徴抽出
# 出典: He, K., Zhang, X., Ren, S., & Sun, J. (2016). Deep residual learning for image recognition. In Proceedings of the IEEE conference on computer vision and pattern recognition (pp. 770-778).

# 特徴機能: 残差学習（Residual Learning）による深層ネットワークの学習
#   残差学習は、入力を出力に直接接続するスキップ接続により、深いネットワークでも勾配消失問題を解決し、特徴抽出を実現する機能である。

# 学習済みモデル: ResNet18 ImageNet事前学習モデル
#   概要: ImageNet-1Kデータセットで事前学習された18層の残差ネットワーク
#   特徴: 11.7Mパラメータ、69.8%のTop-1精度を持つ特徴抽出器
#   URL: https://pytorch.org/vision/main/models/generated/torchvision.models.resnet18.html

# 方式設計
#   - 関連利用技術:
#     * OpenCV: 動画処理とリアルタイム表示（フレーム取得、画像処理、GUI表示機能）
#     * PyTorch: 深層学習フレームワーク（テンソル演算、GPU加速、モデル推論機能）
#     * NumPy: 数値計算ライブラリ（配列操作、距離計算、統計処理機能）
#     * PIL/Pillow: 画像処理ライブラリ（日本語フォント描画、画像変換機能）

#   - 入力と出力:
#     入力: 動画（ユーザは「0:動画ファイル，1:カメラ，2:サンプル動画」のメニューで選択．0:動画ファイルの場合はtkinterでファイル選択．1の場合はOpenCVでカメラが開く．2の場合はhttps://github.com/opencv/opencv/blob/master/samples/data/vtest.aviを使用）
#     出力: OpenCV画面でリアルタイムに異常スコアと判定結果を表示．各フレームごとにprint()により処理結果を表示．プログラム終了時にresult.txtファイルに保存

#   - 処理手順:
#     1. 動画フレームを224x224にリサイズ
#     2. ResNet18で特徴ベクトル（512次元）を抽出
#     3. 初期30フレームを正常サンプルとして蓄積
#     4. 以降のフレームで正常サンプルとの最小距離を計算
#     5. 最小距離を異常スコアとして算出
#     6. 閾値との比較により正常/異常を判定

#   - 前処理、後処理:
#     前処理: ImageNet正規化（平均値[0.485, 0.456, 0.406]、標準偏差[0.229, 0.224, 0.225]）による入力データの標準化
#     後処理: ユークリッド距離による類似度計算と最小距離による異常スコア算出

#   - 追加処理:
#     * 初期30フレームによる正常パターン学習：システム起動時の正常状態を自動学習することで、環境固有の正常パターンを確立し検出精度を向上
#     * GPU自動選択機能：CUDA利用可能時に自動的にGPU処理に切り替え、推論速度を向上

#   - 調整を必要とする設定値:
#     anomaly_threshold（異常判定閾値）: 正常/異常を判定する距離の閾値値（現在50.0）。環境や用途に応じて調整が必要なパラメータ

# 将来方策: 異常判定閾値の自動調整機能
#   正常データの統計的分析（平均、標準偏差）に基づく閾値自動設定機能の実装により、環境適応性を向上

# その他の事項:
#   * Windows環境での動作を想定
#   * 日本語フォント表示にmeiryo.ttcを使用
#   * CUDA対応GPU使用時の処理対応

# 前準備:
#   pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
#   pip install opencv-python pillow numpy

import cv2
import numpy as np
import time
from datetime import datetime
import tkinter as tk
from tkinter import filedialog
import urllib.request
from PIL import Image, ImageDraw, ImageFont
import torch
import torch.nn as nn
import torchvision.models as models
from torchvision.models import ResNet18_Weights
import torchvision.transforms as transforms

# 定数定義
LEARNING_FRAMES = 30  # 正常パターン学習用フレーム数
ANOMALY_THRESHOLD = 50.0  # 異常判定閾値
RESIZE_SIZE = 224  # リサイズサイズ
FONT_PATH = 'C:/Windows/Fonts/meiryo.ttc'  # フォントパス
FONT_SIZE = 30  # フォントサイズ

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

# モデル初期化
model = models.resnet18(weights=ResNet18_Weights.IMAGENET1K_V1)
model.fc = nn.Identity()
model.to(device)
model.eval()

# 画像変換処理
transform = transforms.Compose([
    transforms.ToPILImage(),
    transforms.Resize((RESIZE_SIZE, RESIZE_SIZE)),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                         std=[0.229, 0.224, 0.225])
])

# グローバル変数
embed_db = []
frame_count = 0
results_log = []

def video_frame_processing(frame):
    global embed_db, frame_count
    current_time = time.time()
    frame_count += 1

    # 特徴抽出
    resized = cv2.resize(frame, (RESIZE_SIZE, RESIZE_SIZE))
    with torch.no_grad():
        tensor = transform(resized).unsqueeze(0).to(device)
        embedding = model(tensor).cpu().numpy().flatten()

    # 正常パターン学習フェーズ
    if frame_count <= LEARNING_FRAMES:
        embed_db.append(embedding)
        label = '登録中（正常サンプル）'
        score = 0.0
        if frame_count == LEARNING_FRAMES:
            embed_db = np.array(embed_db)
    else:
        # 異常検知フェーズ
        distances = np.linalg.norm(embed_db - embedding, axis=1)
        score = distances.min()
        label = '正常' if score < ANOMALY_THRESHOLD else '異常'

    # 画面表示用テキスト描画
    try:
        font = ImageFont.truetype(FONT_PATH, FONT_SIZE)
    except (OSError, IOError):
        font = ImageFont.load_default()

    img_pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
    draw = ImageDraw.Draw(img_pil)

    if frame_count <= LEARNING_FRAMES:
        display_text = f'{label}'
        text_color = (0, 255, 0)
    else:
        display_text = f'Score: {score:.2f} - {label}'
        text_color = (0, 255, 0) if label == '正常' else (255, 0, 0)

    draw.text((10, 30), display_text, font=font, fill=text_color)
    processed_frame = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)

    # 結果文字列
    if frame_count <= LEARNING_FRAMES:
        result = f'{label}'
    else:
        result = f'Score = {score:.2f} → {label}'

    return processed_frame, result, current_time

# メイン処理
print('\n=== ResNet18特徴抽出による動画異常検知システム ===')
print('概要: 動画フレームから特徴抽出を行い、初期30フレームを正常パターンとして学習し、')
print('      その後のフレームで異常を検知します')
print('異常検知方式: 正常サンプルとの最小距離により異常度を判定')
print(f'異常判定閾値: {ANOMALY_THRESHOLD}')
print()
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 = "ResNet18異常検知"
        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に保存しました')