# =================================================== # 肺炎検出システムの実装 # =================================================== import torch import torch.nn as nn import torch.optim as optim from torchvision import models, transforms, datasets from torch.utils.data import DataLoader import matplotlib.pyplot as plt import numpy as np # =================================================== # 1. データ拡張の定義 # =================================================== # transforms.Compose(): 複数の変換を順番に適用 # 訓練用: データ拡張を含む train_transform = transforms.Compose([ # Resize(): 画像サイズを統一（ResNetは224×224を期待） transforms.Resize((224, 224)), # RandomHorizontalFlip(): 50%の確率で左右反転 # X線画像は左右反転しても医学的に意味がある transforms.RandomHorizontalFlip(), # RandomRotation(): ランダムに±10度回転 # 撮影角度のばらつきをシミュレート transforms.RandomRotation(10), # ColorJitter(): 明るさとコントラストをランダムに変更 # 撮影条件の違いをシミュレート transforms.ColorJitter(brightness=0.2, contrast=0.2), # ToTensor(): PIL画像をPyTorchテンソルに変換 # 値の範囲を[0, 255]から[0, 1]に正規化 transforms.ToTensor(), # Normalize(): ImageNetの統計値で標準化 # 事前学習モデルは同じ正規化を期待する transforms.Normalize( mean=[0.485, 0.456, 0.406], # ImageNetの平均 std=[0.229, 0.224, 0.225] # ImageNetの標準偏差 ) ]) # 検証・テスト用: データ拡張なし val_transform = transforms.Compose([ transforms.Resize((224, 224)), transforms.ToTensor(), transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) ]) print(“データ拡張パイプラインを定義しました”)

# =================================================== # 2. データセットのロード # =================================================== # datasets.ImageFolder(): フォルダ構造からデータセットを作成 # フォルダ名がクラス名として自動的に認識される # chest_xray/train/NORMAL/ → クラス0 # chest_xray/train/PNEUMONIA/ → クラス1 train_dataset = datasets.ImageFolder( ‘chest_xray/train’, # 訓練データのフォルダ transform=train_transform # データ拡張を適用 ) val_dataset = datasets.ImageFolder( ‘chest_xray/val’, transform=val_transform # 検証データには拡張なし ) test_dataset = datasets.ImageFolder( ‘chest_xray/test’, transform=val_transform # テストデータにも拡張なし ) # DataLoader(): ミニバッチでデータを効率的にロード # shuffle=True: 訓練時はデータをシャッフル（過学習防止） # num_workers: 並列でデータを読み込むワーカー数 train_loader = DataLoader( train_dataset, batch_size=32, # 一度に32枚の画像を処理 shuffle=True, # データをシャッフル num_workers=4 # 4つの並列ワーカー ) val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False, num_workers=4) test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False, num_workers=4) # データセットの統計を表示 print(f”訓練データ: {len(train_dataset)}枚”) print(f”検証データ: {len(val_dataset)}枚”) print(f”テストデータ: {len(test_dataset)}枚”) print(f”\nクラス: {train_dataset.classes}”)

# =================================================== # 3. 転移学習モデルの定義 # =================================================== # デバイスの設定（GPUがあれば使用） device = torch.device(“cuda” if torch.cuda.is_available() else “cpu”) print(f”使用デバイス: {device}”) # =================================================== # ResNet50を事前学習済みモデルとしてロード # =================================================== # models.resnet50(): ResNet50アーキテクチャをロード # pretrained=True: ImageNetで事前学習した重みを使用 # これにより、少ないデータでも高い精度を達成できる model = models.resnet50(pretrained=True) # 最終層の入力特徴数を取得 # ResNet50のfc層は2048次元の特徴を受け取る num_features = model.fc.in_features print(f”元の最終層の入力次元: {num_features}”) # 最終層を置き換え（2クラス分類用） # nn.Sequential(): 複数の層を順番に接続 model.fc = nn.Sequential( # Dropout(): 過学習防止（50%のニューロンをランダムに無効化） nn.Dropout(0.5), # Linear(): 全結合層（2048次元 → 2クラス） nn.Linear(num_features, 2) ) # モデルをデバイス（GPU/CPU）に転送 model = model.to(device) print(“転移学習モデルを準備しました”)

# =================================================== # 4. 損失関数とオプティマイザの設定 # =================================================== # nn.CrossEntropyLoss(): 分類タスクの標準的な損失関数 # 内部でSoftmaxを計算するため、モデル出力はlogitsでOK criterion = nn.CrossEntropyLoss() # optim.Adam(): 学習率を自動調整する最適化アルゴリズム # lr=0.0001: 学習率（転移学習では小さめに設定） # 事前学習の重みを大きく壊さないため optimizer = optim.Adam(model.parameters(), lr=0.0001) # 学習率スケジューラー: 学習が停滞したら学習率を下げる # mode=’min’: val_lossが下がらなくなったら発動 # patience=3: 3エポック改善なしで学習率を下げる # factor=0.5: 学習率を半分にする scheduler = optim.lr_scheduler.ReduceLROnPlateau( optimizer, mode=’min’, patience=3, factor=0.5 ) print(“損失関数とオプティマイザを設定しました”)

# =================================================== # 5. 訓練関数の定義 # =================================================== def train_epoch(model, dataloader, criterion, optimizer, device): “”” 1エポック分の訓練を実行 Args: model: 訓練するモデル dataloader: 訓練データのDataLoader criterion: 損失関数 optimizer: 最適化アルゴリズム device: 使用デバイス（GPU/CPU） Returns: epoch_loss: エポック平均損失 epoch_acc: エポック精度（%） “”” # model.train(): 訓練モードに設定 # Dropout、BatchNormが訓練用の動作をする model.train() running_loss = 0.0 correct = 0 total = 0 for inputs, labels in dataloader: # データをデバイスに転送 inputs, labels = inputs.to(device), labels.to(device) # optimizer.zero_grad(): 勾配を初期化 # 前のバッチの勾配が残らないようにする optimizer.zero_grad() # 順伝播: 入力から予測を計算 outputs = model(inputs) # 損失を計算 loss = criterion(outputs, labels) # 逆伝播: 勾配を計算 loss.backward() # optimizer.step(): パラメータを更新 optimizer.step() # 統計を更新 running_loss += loss.item() # outputs.max(1): 最大値とそのインデックスを取得 # インデックスが予測クラス _, predicted = outputs.max(1) total += labels.size(0) correct += predicted.eq(labels).sum().item() epoch_loss = running_loss / len(dataloader) epoch_acc = 100. * correct / total return epoch_loss, epoch_acc

# =================================================== # 6. 検証関数の定義 # =================================================== def validate(model, dataloader, criterion, device): “”” 検証/テストを実行 Returns: epoch_loss, epoch_acc: 損失と精度 all_preds: すべての予測ラベル all_labels: すべての正解ラベル all_probs: 肺炎クラスの確率（ROC-AUC計算用） “”” # model.eval(): 評価モードに設定 # Dropout無効化、BatchNormは推論用の動作 model.eval() running_loss = 0.0 correct = 0 total = 0 all_preds = [] all_labels = [] all_probs = [] # torch.no_grad(): 勾配計算を無効化 # メモリ使用量を削減し、推論を高速化 with torch.no_grad(): for inputs, labels in dataloader: inputs, labels = inputs.to(device), labels.to(device) outputs = model(inputs) loss = criterion(outputs, labels) running_loss += loss.item() # Softmaxで確率に変換 probs = torch.softmax(outputs, dim=1) _, predicted = outputs.max(1) total += labels.size(0) correct += predicted.eq(labels).sum().item() # 評価指標計算用に保存 all_preds.extend(predicted.cpu().numpy()) all_labels.extend(labels.cpu().numpy()) # 肺炎クラス（インデックス1）の確率 all_probs.extend(probs[:, 1].cpu().numpy()) epoch_loss = running_loss / len(dataloader) epoch_acc = 100. * correct / total return epoch_loss, epoch_acc, all_preds, all_labels, all_probs

# =================================================== # 7. 訓練の実行 # =================================================== num_epochs = 20 best_val_loss = float(‘inf’) # 記録用リスト train_losses = [] train_accs = [] val_losses = [] val_accs = [] print(“訓練を開始します…”) print(“=” * 50) for epoch in range(num_epochs): # 訓練 train_loss, train_acc = train_epoch( model, train_loader, criterion, optimizer, device ) # 検証 val_loss, val_acc, _, _, _ = validate( model, val_loader, criterion, device ) # 学習率スケジューラーを更新 scheduler.step(val_loss) # 記録 train_losses.append(train_loss) train_accs.append(train_acc) val_losses.append(val_loss) val_accs.append(val_acc) # 進捗表示 print(f”Epoch {epoch+1:2d}/{num_epochs}”) print(f” Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%”) print(f” Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%”) # 最良モデルを保存 if val_loss < best_val_loss: best_val_loss = val_loss torch.save(model.state_dict(), 'best_pneumonia_model.pth') print(" → 最良モデルを保存しました！") print() print("=" * 50) print("訓練が完了しました！")

# =================================================== # 8. テストセットでの評価 # =================================================== from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score import seaborn as sns # 最良モデルをロード model.load_state_dict(torch.load(‘best_pneumonia_model.pth’)) # テストセットで評価 test_loss, test_acc, test_preds, test_labels, test_probs = validate( model, test_loader, criterion, device ) print(f”テスト精度: {test_acc:.2f}%”) # 詳細な評価レポート print(“\n” + “=” * 50) print(“Classification Report:”) print(“=” * 50) print(classification_report( test_labels, test_preds, target_names=[‘NORMAL’, ‘PNEUMONIA’] )) # ROC-AUC（分類性能の総合指標） roc_auc = roc_auc_score(test_labels, test_probs) print(f”ROC-AUC: {roc_auc:.4f}”)

# =================================================== # 9. Confusion Matrixの可視化 # =================================================== # Confusion Matrix（混同行列）を計算 cm = confusion_matrix(test_labels, test_preds) # ヒートマップで可視化 plt.figure(figsize=(8, 6)) sns.heatmap( cm, annot=True, # 数値を表示 fmt=’d’, # 整数で表示 cmap=’Blues’, # 青系のカラーマップ xticklabels=[‘NORMAL’, ‘PNEUMONIA’], yticklabels=[‘NORMAL’, ‘PNEUMONIA’] ) plt.title(‘Confusion Matrix’, fontsize=14) plt.ylabel(‘True Label（正解）’) plt.xlabel(‘Predicted Label（予測）’) plt.tight_layout() plt.savefig(‘pneumonia_confusion_matrix.png’, dpi=150, bbox_inches=’tight’) plt.show() # 結果の解釈 print(“\nConfusion Matrixの解釈:”) print(f”True Negative (正常→正常): {cm[0][0]}”) print(f”False Positive (正常→肺炎): {cm[0][1]}”) print(f”False Negative (肺炎→正常): {cm[1][0]} ← これを最小化したい！”) print(f”True Positive (肺炎→肺炎): {cm[1][1]}”)

# =================================================== # 表面欠陥検出（YOLOv8アプローチ） # =================================================== from ultralytics import YOLO import cv2 import numpy as np from pathlib import Path # =================================================== # 1. マスク画像からBounding Boxを抽出 # =================================================== def mask_to_bbox(mask_path): “”” セグメンテーションマスクからBounding Boxを抽出 MVTec ADのground_truthは画素マスク形式 これをYOLO形式のBounding Boxに変換する Args: mask_path: マスク画像のパス（白=欠陥、黒=正常） Returns: bboxes: Bounding Boxのリスト [[x, y, w, h], …] “”” # グレースケールで読み込み mask = cv2.imread(mask_path, cv2.IMREAD_GRAYSCALE) if mask is None: return [] # cv2.connectedComponentsWithStats(): 連結成分を検出 # 白い領域（欠陥）を個別の領域として認識 num_labels, labels, stats, centroids = cv2.connectedComponentsWithStats( mask, connectivity=8 # 8近傍で連結を判定 ) bboxes = [] # ラベル0は背景なので1から開始 for i in range(1, num_labels): # stats[i]: [x, y, width, height, area] x, y, w, h, area = stats[i] # 小さすぎる領域（ノイズ）は除外 if area < 100: continue bboxes.append([x, y, w, h]) return bboxes # 使用例 mask_path = "mvtec_ad/bottle/ground_truth/broken_large/000_mask.png" bboxes = mask_to_bbox(mask_path) print(f"検出されたBounding Box: {len(bboxes)}個")

# =================================================== # 2. MVTec ADをYOLO形式に変換 # =================================================== def convert_mvtec_to_yolo(mvtec_dir, output_dir, category=’bottle’): “”” MVTec ADデータセットをYOLO形式に変換 Args: mvtec_dir: MVTec ADのルートディレクトリ output_dir: 出力先ディレクトリ category: 変換するカテゴリ（例: ‘bottle’） “”” import os category_dir = Path(mvtec_dir) / category out_dir = Path(output_dir) / category # ディレクトリ作成 (out_dir / ‘images’ / ‘train’).mkdir(parents=True, exist_ok=True) (out_dir / ‘images’ / ‘val’).mkdir(parents=True, exist_ok=True) (out_dir / ‘labels’ / ‘train’).mkdir(parents=True, exist_ok=True) (out_dir / ‘labels’ / ‘val’).mkdir(parents=True, exist_ok=True) # 正常画像を訓練データとして使用（負例として） good_dir = category_dir / ‘train’ / ‘good’ for img_file in good_dir.glob(‘*.png’): # 画像をコピー img = cv2.imread(str(img_file)) out_path = out_dir / ‘images’ / ‘train’ / img_file.name cv2.imwrite(str(out_path), img) # 空のラベルファイル（欠陥なし） label_path = out_dir / ‘labels’ / ‘train’ / f”{img_file.stem}.txt” label_path.touch() # 空ファイルを作成 # テストセットの異常画像を処理 test_dir = category_dir / ‘test’ for defect_type in test_dir.iterdir(): if not defect_type.is_dir() or defect_type.name == ‘good’: continue for img_file in defect_type.glob(‘*.png’): # 対応するマスクファイル mask_file = category_dir / ‘ground_truth’ / defect_type.name / f”{img_file.stem}_mask.png” if not mask_file.exists(): continue # Bounding Boxを抽出 bboxes = mask_to_bbox(str(mask_file)) if len(bboxes) == 0: continue # 画像を読み込み img = cv2.imread(str(img_file)) h, w = img.shape[:2] # 画像を保存 out_img_path = out_dir / ‘images’ / ‘val’ / f”{defect_type.name}_{img_file.name}” cv2.imwrite(str(out_img_path), img) # YOLO形式のラベルを保存 out_label_path = out_dir / ‘labels’ / ‘val’ / f”{defect_type.name}_{img_file.stem}.txt” with open(out_label_path, ‘w’) as f: for bbox in bboxes: x, y, bw, bh = bbox # YOLO形式に変換（正規化座標） x_center = (x + bw / 2) / w y_center = (y + bh / 2) / h bw_norm = bw / w bh_norm = bh / h # クラスID=0（defect） f.write(f”0 {x_center:.6f} {y_center:.6f} {bw_norm:.6f} {bh_norm:.6f}\n”) print(f”処理: {out_img_path.name}”) print(f”\n変換完了: {out_dir}”) # 変換を実行 convert_mvtec_to_yolo(‘mvtec_ad’, ‘mvtec_yolo’, category=’bottle’)

# =================================================== # 3. YOLOv8で訓練 # =================================================== # データ設定ファイル（data.yaml）の内容: “”” train: mvtec_yolo/bottle/images/train val: mvtec_yolo/bottle/images/val nc: 1 names: [‘defect’] “”” # モデルをロード model = YOLO(‘yolov8n.pt’) # 訓練 results = model.train( data=’mvtec_defect.yaml’, # データ設定ファイル epochs=100, imgsz=640, batch=16, name=’mvtec_defect_detection’, patience=20, # Early Stopping ) print(“訓練が完了しました”)

# =================================================== # 異常検知アプローチ（Anomalib使用） # =================================================== # !pip install anomalib from anomalib.data import MVTec from anomalib.models import Padim from anomalib.engine import Engine # =================================================== # 1. データモジュールの準備 # =================================================== # MVTec(): Anomalib用のMVTecデータローダー # 自動的に正常画像で訓練、異常画像でテストを行う datamodule = MVTec( root=’./mvtec_ad’, # データセットのパス category=’bottle’, # カテゴリを指定 image_size=(256, 256), # 画像サイズ train_batch_size=32, eval_batch_size=32, num_workers=8, ) print(f”カテゴリ: {datamodule.category}”)

# =================================================== # 2. PaDiMモデルの定義 # =================================================== # PaDiM: Patch Distribution Modeling # 正常画像のパッチ分布を学習し、異常を検出 # # 仕組み: # 1. 事前学習CNNで特徴を抽出 # 2. 各位置の特徴分布（平均、共分散）を計算 # 3. テスト時は分布からの距離で異常度を計算 model = Padim( backbone=’resnet18′, # 特徴抽出に使うCNN layers=[‘layer1’, ‘layer2’, ‘layer3’], # 使用する層 ) print(“PaDiMモデルを準備しました”)

# =================================================== # 3. 訓練と評価 # =================================================== # Engine(): Anomalibの訓練エンジン engine = Engine( max_epochs=1, # PaDiMは1エポックで十分 accelerator=’auto’, # GPU/CPU自動選択 devices=1, ) # 訓練（正常画像のみで学習） engine.fit(model=model, datamodule=datamodule) # テスト（異常画像で評価） test_results = engine.test(model=model, datamodule=datamodule) print(“\nテスト結果:”) print(f”Image AUROC: {test_results[0][‘image_AUROC’]:.4f}”) print(f”Pixel AUROC: {test_results[0][‘pixel_AUROC’]:.4f}”)

# =================================================== # リアルタイム顔認識システムの実装 # =================================================== # !pip install face_recognition import face_recognition import cv2 import numpy as np import pickle import sqlite3 from datetime import datetime from pathlib import Path # =================================================== # 1. データベースのセットアップ # =================================================== def setup_database(): “”” SQLiteデータベースを初期化 テーブル構造: – persons: 登録された人物（ID、名前、作成日時） – face_encodings: 顔埋め込み（人物ID、128次元ベクトル） – recognition_logs: 認識ログ（人物ID、時刻、信頼度） “”” # sqlite3.connect(): データベースに接続（なければ作成） conn = sqlite3.connect(‘face_recognition.db’) cursor = conn.cursor() # 人物テーブル cursor.execute(”’ CREATE TABLE IF NOT EXISTS persons ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ”’) # 顔埋め込みテーブル # encoding: 128次元のベクトルをバイナリで保存 cursor.execute(”’ CREATE TABLE IF NOT EXISTS face_encodings ( id INTEGER PRIMARY KEY AUTOINCREMENT, person_id INTEGER, encoding BLOB NOT NULL, FOREIGN KEY (person_id) REFERENCES persons(id) ) ”’) # 認識ログテーブル cursor.execute(”’ CREATE TABLE IF NOT EXISTS recognition_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, person_id INTEGER, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, confidence REAL, FOREIGN KEY (person_id) REFERENCES persons(id) ) ”’) conn.commit() conn.close() print(“データベースを初期化しました”) # データベースを作成 setup_database()

# =================================================== # 2. 顔の登録機能 # =================================================== def register_person(name, image_dir): “”” 人物を登録（複数の顔画像から） Args: name: 登録する人物の名前 image_dir: 顔画像が入っているフォルダ Returns: person_id: 登録された人物のID 使用例: register_person(“田中太郎”, “faces/tanaka”) ※ faces/tanaka/ に田中太郎の顔写真を複数枚入れておく “”” conn = sqlite3.connect(‘face_recognition.db’) cursor = conn.cursor() # 人物をデータベースに追加 try: cursor.execute(“INSERT INTO persons (name) VALUES (?)”, (name,)) person_id = cursor.lastrowid print(f”新規登録: {name} (ID: {person_id})”) except sqlite3.IntegrityError: # 既に登録されている場合 print(f”{name} は既に登録されています。埋め込みを追加します。”) cursor.execute(“SELECT id FROM persons WHERE name = ?”, (name,)) person_id = cursor.fetchone()[0] # 画像フォルダから顔を抽出 image_dir = Path(image_dir) encoding_count = 0 for image_file in image_dir.glob(‘*.jpg’): # 画像を読み込み（RGB形式） # face_recognition.load_image_file(): RGB形式で読み込み image = face_recognition.load_image_file(str(image_file)) # 顔を検出 # face_locations(): 画像内の顔の位置を検出 # 戻り値: [(top, right, bottom, left), …] face_locations = face_recognition.face_locations(image) if len(face_locations) == 0: print(f” 警告: {image_file.name} に顔が検出されませんでした”) continue if len(face_locations) > 1: print(f” 警告: {image_file.name} に複数の顔が検出されました（最初の顔を使用）”) # 顔埋め込みを計算 # face_encodings(): 128次元の顔埋め込みベクトルを計算 # この埋め込みで顔の「特徴」を数値化 face_encodings = face_recognition.face_encodings(image, face_locations) for encoding in face_encodings: # pickle.dumps(): NumPy配列をバイナリに変換 # SQLiteにはバイナリ（BLOB）として保存 encoding_blob = pickle.dumps(encoding) cursor.execute( “INSERT INTO face_encodings (person_id, encoding) VALUES (?, ?)”, (person_id, encoding_blob) ) encoding_count += 1 print(f” 処理: {image_file.name}”) conn.commit() conn.close() print(f”\n{name} を登録しました（{encoding_count}枚の顔画像）”) return person_id # 使用例（実行時はコメントを外す） # register_person(“田中太郎”, “faces/tanaka”) # register_person(“山田花子”, “faces/yamada”)

# =================================================== # 3. データベースから顔埋め込みをロード # =================================================== def load_known_faces(): “”” データベースから登録済みの顔をロード Returns: known_face_encodings: 顔埋め込みのリスト known_face_names: 対応する名前のリスト person_ids: 対応する人物IDのリスト “”” conn = sqlite3.connect(‘face_recognition.db’) cursor = conn.cursor() # JOINでpersonsとface_encodingsを結合 cursor.execute(“”” SELECT p.id, p.name, fe.encoding FROM persons p JOIN face_encodings fe ON p.id = fe.person_id “””) rows = cursor.fetchall() conn.close() known_face_encodings = [] known_face_names = [] person_ids = [] for person_id, name, encoding_blob in rows: # pickle.loads(): バイナリをNumPy配列に復元 encoding = pickle.loads(encoding_blob) known_face_encodings.append(encoding) known_face_names.append(name) person_ids.append(person_id) print(f”登録済み人物: {len(set(known_face_names))}人”) print(f”顔データ: {len(known_face_encodings)}件”) return known_face_encodings, known_face_names, person_ids

# =================================================== # 4. リアルタイム顔認識 # =================================================== def real_time_recognition(): “”” Webカメラでリアルタイム顔認識を実行 操作: – ‘q’キー: 終了 “”” # 登録済みの顔をロード known_face_encodings, known_face_names, person_ids = load_known_faces() if len(known_face_encodings) == 0: print(“登録された顔がありません。先に顔を登録してください。”) return # Webカメラを開く # cv2.VideoCapture(0): デフォルトカメラを使用 cap = cv2.VideoCapture(0) if not cap.isOpened(): print(“カメラを開けませんでした”) return # データベース接続（ログ記録用） conn = sqlite3.connect(‘face_recognition.db’) cursor = conn.cursor() # 重複ログ防止用（同じ人を連続で記録しない） last_recognition_time = {} print(“認識を開始します… (‘q’で終了)”) while True: # フレームを取得 ret, frame = cap.read() if not ret: break # 処理速度向上のため画像を縮小 # 0.25倍にリサイズ（4倍速くなる） small_frame = cv2.resize(frame, (0, 0), fx=0.25, fy=0.25) # BGRからRGBに変換（face_recognitionはRGB形式） rgb_small_frame = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB) # 顔を検出 face_locations = face_recognition.face_locations(rgb_small_frame) # 顔が検出された場合のみ埋め込みを計算 if len(face_locations) > 0: face_encodings = face_recognition.face_encodings( rgb_small_frame, face_locations ) # 各顔について処理 for (top, right, bottom, left), face_encoding in zip( face_locations, face_encodings ): # 登録済みの顔と比較 # compare_faces(): 閾値以下なら True # tolerance=0.6: デフォルト値（小さいほど厳格） matches = face_recognition.compare_faces( known_face_encodings, face_encoding, tolerance=0.6 ) name = “Unknown” confidence = 0 person_id = None # 距離を計算（小さいほど似ている） # face_distance(): ユークリッド距離を計算 face_distances = face_recognition.face_distance( known_face_encodings, face_encoding ) if len(face_distances) > 0: # 最も距離が小さい（最も似ている）顔を選択 best_match_index = np.argmin(face_distances) if matches[best_match_index]: name = known_face_names[best_match_index] distance = face_distances[best_match_index] # 信頼度 = 1 – 距離（距離0で信頼度1） confidence = 1 – distance person_id = person_ids[best_match_index] # ログを記録（1分以内の重複は無視） current_time = datetime.now() if person_id not in last_recognition_time or \ (current_time – last_recognition_time[person_id]).seconds > 60: cursor.execute( “INSERT INTO recognition_logs (person_id, confidence) VALUES (?, ?)”, (person_id, confidence) ) conn.commit() last_recognition_time[person_id] = current_time print(f”認識: {name} (信頼度: {confidence:.3f})”) # 元の画像サイズに戻す（4倍） top *= 4 right *= 4 bottom *= 4 left *= 4 # Bounding Boxを描画 color = (0, 255, 0) if name != “Unknown” else (0, 0, 255) cv2.rectangle(frame, (left, top), (right, bottom), color, 2) # 名前と信頼度を表示 text = f”{name}” if confidence > 0: text += f” ({confidence:.2f})” cv2.rectangle(frame, (left, bottom – 35), (right, bottom), color, cv2.FILLED) cv2.putText(frame, text, (left + 6, bottom – 6), cv2.FONT_HERSHEY_DUPLEX, 0.6, (255, 255, 255), 1) # 画面に表示 cv2.imshow(‘Face Recognition’, frame) # ‘q’キーで終了 if cv2.waitKey(1) & 0xFF == ord(‘q’): break # 後片付け cap.release() cv2.destroyAllWindows() conn.close() print(“認識を終了しました”) # 実行 # real_time_recognition()