【実践事例】ResNet × PyTorchでロゴを含む画像を見分ける

from PIL import Image, ImageDraw, ImageFont
import os, random
import numpy as np
from PIL import ImageFilter

def generate_image(path, logo_list, size=(1024, 512)):
    """
    ランダムな背景線とロゴ文字列を描いた画像を生成する

    Parameters:
    - path: 未使用（互換性のため保持。実際は return された画像オブジェクトを使って保存）
    - logo_list: ロゴ候補の文字列リスト（例: ["◎◎◎", "ABCD"]）。この中からランダムに1つ選んで描画される
    - size: 出力画像のサイズ（幅, 高さ）。デフォルトは 1024×512

    Returns:
    - PIL.Image.Image オブジェクト（加工済みのロゴ画像）

    使用例:
        logo_patterns = ["◎AB◎", "◎◎◎◎", "■●■●"]
        img = generate_image(None, logo_patterns)
        img.save("sample.jpg")
    """


    img = Image.new("RGB", size, (255, 255, 255))
    draw = ImageDraw.Draw(img)

    # 背景に縦線と横線（線幅ランダム）
    for x in range(0, size[0], 40):
        width = random.randint(1, 5)
        draw.line([(x, 0), (x, size[1])], fill=(200, 200, 200), width=width)

    for y in range(0, size[1], 40):
        width = random.randint(1, 5)
        draw.line([(0, y), (size[0], y)], fill=(200, 200, 200), width=width)

    # ランダムなフォントサイズ・位置で「◎◎◎」を描画
    font_size = random.randint(50, 150)
    try:        
        font = ImageFont.truetype(r"c:\WINDOWS\Fonts\MEIRYOB.TTC", font_size)
    except:
        font = ImageFont.load_default()

    text = random.choice(logo_list)  # 🔹 ロゴパターンをランダムに選ぶ
    text_size = draw.textbbox((0, 0), text, font=font)[2:]
    max_x = size[0] - text_size[0]
    max_y = size[1] - text_size[1]
    pos = (random.randint(0, max_x), random.randint(0, max_y))
    draw.text(pos, text, font=font, fill=(0, 0, 0))

    # 全体に軽いノイズ（点状）を追加
    np_img = np.array(img).astype(np.int16)
    noise = np.random.randint(-20, 20, size=np_img.shape).astype(np.int16)
    noisy_img = np.clip(np_img + noise, 0, 255).astype(np.uint8)
    img = Image.fromarray(noisy_img)

    # 軽いぼかしで“にじみ”を加える
    img = img.filter(ImageFilter.GaussianBlur(radius=0.7))

    return img

def generate_dataset(folder, logo_list, prefix="logo", count=10):
    """
    指定フォルダに分類用画像を保存する関数

    Parameters:
    - folder: 画像を保存するベースフォルダ（例："d:/DataSet/train"）
    - logo_list: ロゴ候補の文字列リスト（例：["◎◎◎", "ABCD", ...]）
    - prefix: ファイル名の先頭に付けるPrefix（例："logo", "none"）
    - count: 生成する画像の枚数（例：100）
    """
    os.makedirs(folder, exist_ok=True)

    for i in range(count):
        img = generate_image(path=None, logo_list=logo_list)
        filename = f"{prefix}_{i}.jpg"
        img.save(os.path.join(folder, filename))


# ================================================================================
# sampleデータの生成
# ================================================================================
gen_cnt=50 # 生成する枚数
root_folder=r"P:\logo_classification\dataset" # <== ご自身の環境に変更してください。
logo = ["◎AB◎"]
none = ["◎●◎●◎","◎◎◎◎","■■■■","■◎■◎","●●●●","ABCD","A◎B◎"]

# 修正例：順序を揃えて呼び出す
generate_dataset(os.path.join(root_folder, "train/logo"), logo, prefix="logo", count=gen_cnt)
generate_dataset(os.path.join(root_folder, "train/none"), none, prefix="none", count=gen_cnt)
generate_dataset(os.path.join(root_folder, "val/logo"), logo, prefix="logo", count=gen_cnt)
generate_dataset(os.path.join(root_folder, "val/none"), none, prefix="none", count=gen_cnt)
generate_dataset(os.path.join(root_folder, "test"), logo, prefix="logo", count=int(gen_cnt / 2))
generate_dataset(os.path.join(root_folder, "test"), none, prefix="none", count=int(gen_cnt / 2))

import torch
import torch.nn as nn
from torchvision import datasets, models, transforms
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt
from sklearn.metrics import classification_report, confusion_matrix
import seaborn as sns
import numpy as np
from matplotlib import rcParams
rcParams['font.family'] = 'Meiryo'

# ---------------------------------------
# デバイス設定（GPUが使えれば使用）
# ---------------------------------------
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("使用デバイス:", device)

# ---------------------------------------
# データ変換と読み込み
# ---------------------------------------
transform = transforms.Compose([
    transforms.Resize((224, 224)),
    transforms.ToTensor(),
])

# 学習データを格納したフォルダを指定
train_ds = datasets.ImageFolder(r"P:\logo_classification\dataset\train", transform=transform) # ← 環境に合わせて変更
# 検証データを格納したフォルダを指定
val_ds   = datasets.ImageFolder(r"P:\logo_classification\dataset\val", transform=transform) # ← 環境に合わせて変更

train_loader = DataLoader(train_ds, batch_size=16, shuffle=True)
val_loader   = DataLoader(val_ds, batch_size=16)

# ---------------------------------------
# モデル構築（ResNet18 → 2クラス分類）
# ---------------------------------------
model = models.resnet18(pretrained=True)
model.fc = nn.Linear(model.fc.in_features, 2)
model = model.to(device)

criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

# ---------------------------------------
# 学習ループ（エポックあり）
# ---------------------------------------
num_epochs = 10  # ← 必要に応じて変更
train_losses = []
train_accuracies = []
val_accuracies = []

for epoch in range(num_epochs):
    model.train()
    running_loss = 0
    correct = 0
    total = 0

    for images, labels in train_loader:
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad()
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

        running_loss += loss.item()
        preds = torch.argmax(outputs, dim=1)
        correct += (preds == labels).sum().item()
        total += labels.size(0)

    train_loss = running_loss / len(train_loader)
    train_acc = correct / total
    train_losses.append(train_loss)
    train_accuracies.append(train_acc)

    # 検証精度
    correct = total = 0
    model.eval()
    with torch.no_grad():
        for images, labels in val_loader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            preds = torch.argmax(outputs, dim=1)
            correct += (preds == labels).sum().item()
            total += labels.size(0)

    val_acc = correct / total
    val_accuracies.append(val_acc)

    print(f"Epoch {epoch+1}/{num_epochs} | Loss: {train_loss:.4f} | Train Acc: {train_acc:.2%} | Val Acc: {val_acc:.2%}")

# ---------------------------------------
# モデル保存
# ---------------------------------------
torch.save(model.state_dict(), "model.pth")
print("\nモデルを model.pth に保存しました。")

# ---------------------------------------
# 検証セットでの詳細レポート＋混同行列
# ---------------------------------------
y_true, y_pred = [], []
model.eval()
with torch.no_grad():
    for images, labels in val_loader:
        images = images.to(device)
        outputs = model(images)
        preds = torch.argmax(outputs, dim=1)
        y_true.extend(labels.numpy())
        y_pred.extend(preds.cpu().numpy())  # predsはGPU上のTensorなのでcpu()が必要

print("\n=== 混同行列 ===")
cm = confusion_matrix(y_true, y_pred)
print(cm)

print("\n=== レポート ===")
print(classification_report(y_true, y_pred, target_names=val_ds.classes))

# ---------------------------------------
# 学習曲線プロット（Loss / Accuracy）
# ---------------------------------------
plt.figure(figsize=(12, 4))

plt.subplot(1, 2, 1)
plt.plot(train_losses, label="Train Loss")
plt.xlabel("Epoch")
plt.ylabel("Loss")
plt.title("Training Loss")
plt.legend()

plt.subplot(1, 2, 2)
plt.plot(train_accuracies, label="Train Accuracy")
plt.plot(val_accuracies, label="Val Accuracy")
plt.xlabel("Epoch")
plt.ylabel("Accuracy")
plt.title("Accuracy")
plt.legend()

plt.tight_layout()
plt.show()

# ---------------------------------------
# 混同行列のヒートマップ
# ---------------------------------------
labels = val_ds.classes
plt.figure(figsize=(5, 4))
sns.heatmap(cm, annot=True, fmt="d", cmap="Blues", xticklabels=labels, yticklabels=labels)
plt.xlabel("予測ラベル")
plt.ylabel("正解ラベル")
plt.title("混同行列（Validation）")
plt.tight_layout()
plt.show()

# インスタンスの生成
clf = LogoClassifier()

# 学習
clf.train(data_dir=r"P:\logo_classification\dataset/train",
            val_dir=r"P:\logo_classification\dataset/val",
            epochs=10)

# 評価
clf.evaluate(data_dir=r"P:\logo_classification\dataset\val",
            class_names=["logo", "none"],results=r"P:\logo_classification\dataset\results")

# モデルの保存
clf.save(r"P:\logo_classification\dataset/logo_classifier.pt")

# モデルの読込
clf.load(r"P:\logo_classification\dataset/logo_classifier.pt")

# 分類の実行
clf.classify_folder(
    input_dir=r"P:\logo_classification\dataset\test",
    output_dir=r"P:\logo_classification\result",
    class_names=["logo", "none"]
)

import os
import torch
import torch.nn as nn
import pandas as pd
import torch.optim as optim
from torchvision import datasets, transforms, models
from torch.utils.data import DataLoader
from sklearn.metrics import classification_report, confusion_matrix
from PIL import Image
import matplotlib.pyplot as plt
from matplotlib import rcParams
import seaborn as sns 
import shutil
import time 

rcParams['font.family'] = 'Meiryo'

class LogoClassifier:
    def __init__(self, num_classes=2, image_size=(224, 224), device=None):
        """
        LogoClassifier クラスの初期化

        Parameters:
        - num_classes (int): 分類するクラスの数（デフォルトは2）
        - image_size (tuple): 入力画像のサイズ（デフォルトは224×224）
        - device (torch.device or None): 使用するデバイス（Noneの場合は自動判定）

        処理内容:
        - ResNet18 をベースとしたモデルを構築し、最終層を num_classes に合わせて再定義
        - 指定デバイス（GPUまたはCPU）にモデルを移動
        - 使用デバイスを標準出力に表示
        """

        self.image_size = image_size
        self.num_classes = num_classes
        self.device = device or torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = models.resnet18(pretrained=True)
        self.model.fc = nn.Linear(self.model.fc.in_features, num_classes)
        self.model = self.model.to(self.device)
        print(f"使用デバイス: {self.device}")

    def train(self, data_dir, val_dir=None, epochs=10, batch_size=8, lr=1e-4):
        """
        学習を実行。val_dir を指定すると各エポック後に検証精度も記録されます。

        Parameters:
        - data_dir (str): ImageFolder形式の学習データパス
        - val_dir (str): ImageFolder形式の検証データパス（省略可）
        - epochs (int): エポック数
        - batch_size (int): バッチサイズ
        - lr (float): 学習率
        """
        transform = transforms.Compose([
            transforms.Resize(self.image_size),
            transforms.RandomHorizontalFlip(),
            transforms.ColorJitter(brightness=0.2, contrast=0.2),
            transforms.ToTensor()
        ])

        train_dataset = datasets.ImageFolder(data_dir, transform=transform)
        train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

        val_loader = None
        if val_dir:
            val_dataset = datasets.ImageFolder(val_dir, transform=transforms.Compose([
                transforms.Resize(self.image_size),
                transforms.ToTensor()
            ]))
            val_loader = DataLoader(val_dataset, batch_size=batch_size)

        criterion = nn.CrossEntropyLoss()
        optimizer = optim.Adam(self.model.parameters(), lr=lr)

        self.train_losses = []
        self.train_accuracies = []
        self.val_accuracies = []  # ← ここが追加ポイント

        start = time.time()
        self.model.train()

        for epoch in range(epochs):
            running_loss = 0.0
            correct = total = 0

            for images, labels in train_loader:
                images, labels = images.to(self.device), labels.to(self.device)
                optimizer.zero_grad()
                outputs = self.model(images)
                loss = criterion(outputs, labels)
                loss.backward()
                optimizer.step()

                running_loss += loss.item()
                preds = torch.argmax(outputs, 1)
                correct += (preds == labels).sum().item()
                total += labels.size(0)

            avg_loss = running_loss / len(train_loader)
            acc = correct / total

            self.train_losses.append(avg_loss)
            self.train_accuracies.append(acc)

            log_msg = f"Epoch {epoch+1}/{epochs}: Loss={avg_loss:.4f}, Accuracy={acc:.2%}"

            # バリデーション精度の記録（オプション）
            if val_loader:
                self.model.eval()
                val_correct = val_total = 0
                with torch.no_grad():
                    for images, labels in val_loader:
                        images, labels = images.to(self.device), labels.to(self.device)
                        outputs = self.model(images)
                        preds = torch.argmax(outputs, 1)
                        val_correct += (preds == labels).sum().item()
                        val_total += labels.size(0)
                val_acc = val_correct / val_total
                self.val_accuracies.append(val_acc)
                log_msg += f" | Val Accuracy={val_acc:.2%}"
                self.model.train()
            else:
                self.val_accuracies = []

            print(log_msg)

        print(f"学習完了（所要時間: {time.time() - start:.1f} 秒）")

    def evaluate(self, data_dir, batch_size=8, class_names=None, results=None):
        """
        検証用データセットに対して分類精度を評価し、結果を出力・保存します。

        Parameters:
        - data_dir (str): ImageFolder形式の検証データが格納されたディレクトリ
        - batch_size (int): DataLoaderのバッチサイズ（デフォルト: 8）
        - class_names (list of str or None): クラス名リスト（インデックスに対応）。Noneなら整数で扱う。
        - results (str): 評価結果（混同行列・分類曲線・CSV）を保存する出力フォルダパス

        処理内容:
        - モデルを検証データで推論し、全体の精度を表示
        - classification_report を標準出力に表示（class_names 指定時）
        - 混同行列のヒートマップ画像を保存（confusion_matrix.png）
        - 学習曲線をPNG画像として保存（training_curve.png）
        - 各画像の正解・予測・判定結果をCSVとして保存（result.csv）

        Returns:
        - accuracy (float): 全体の分類精度（百分率）
        """

        transform = transforms.Compose([
            transforms.Resize(self.image_size),
            transforms.ToTensor()
        ])
        dataset = datasets.ImageFolder(data_dir, transform=transform)
        loader = DataLoader(dataset, batch_size=batch_size)

        correct = 0
        total = 0
        all_preds = []
        all_labels = []
        all_paths = []

        self.model.eval()
        with torch.no_grad():
            for imgs, labels in loader:
                imgs, labels = imgs.to(self.device), labels.to(self.device)
                outputs = self.model(imgs)
                _, preds = torch.max(outputs, 1)
                correct += (preds == labels).sum().item()
                total += labels.size(0)
                all_preds += preds.cpu().numpy().tolist()
                all_labels += labels.cpu().numpy().tolist()
                all_paths += [p for p, _ in dataset.samples[len(all_paths):len(all_paths)+len(preds)]]

        accuracy = 100 * correct / total
        print(f"評価精度: {accuracy:.2f}%")

        os.makedirs(results, exist_ok=True)

        if class_names:
            print("\n" + classification_report(all_labels, all_preds, target_names=class_names))

            cm = confusion_matrix(all_labels, all_preds)
            plt.figure(figsize=(6, 5))
            sns.heatmap(cm, annot=True, fmt="d", cmap="Blues",
                        xticklabels=class_names, yticklabels=class_names)
            plt.xlabel("Predicted")
            plt.ylabel("True")
            plt.title("Confusion Matrix")
            plt.tight_layout()
            plt.savefig(os.path.join(results,"confusion_matrix.png"))

        self.plot_training_curve(os.path.join(results,"training_curve.png"))

        df = pd.DataFrame({
            "ファイル名": all_paths,
            "正解値": [class_names[l] for l in all_labels],
            "予測値": [class_names[p] for p in all_preds],
            "判定": ["〇" if t == p else "×" for t, p in zip(all_labels, all_preds)]
        })

        df.to_csv(os.path.join(results,"result.csv"), index=False, encoding="shift-jis")
        print(f"{results} に結果を保存しました")

        return accuracy

    def plot_training_curve(self,path=None):
        """
        学習過程の損失と分類精度を可視化します。

        Parameters:
        - path (str or None): 保存パスを指定するとPNG画像として保存。Noneなら画面に表示。

        描画内容:
        - Train Loss（オレンジ）
        - Train Accuracy（緑）
        - Val Accuracy（青, 存在する場合）

        注意:
        - train() 実行後にのみ使用可能
        """

        if not hasattr(self, 'train_losses') or not hasattr(self, 'train_accuracies'):
            print("学習履歴が見つかりません。train() 実行後に使用してください。")
            return

        plt.figure(figsize=(12, 4))

        # ─── 損失 ───
        plt.subplot(1, 2, 1)
        plt.plot(self.train_losses, label="Train Loss", color="orange")
        plt.xlabel("Epoch")
        plt.ylabel("Loss")
        plt.title("学習損失")
        plt.grid(True)
        plt.legend()

        # ─── 精度 ───
        plt.subplot(1, 2, 2)
        plt.plot(self.train_accuracies, label="Train Accuracy", color="green")
        if hasattr(self, 'val_accuracies') and self.val_accuracies:
            plt.plot(self.val_accuracies, label="Val Accuracy", color="blue")
        plt.xlabel("Epoch")
        plt.ylabel("Accuracy")
        plt.title("分類精度")
        plt.grid(True)
        plt.legend()

        plt.tight_layout()
        if path:
            plt.savefig(path)
        else:
            plt.show()

    def save(self, model_path):
        """
        学習済みモデルを指定パスに保存します。

        Parameters:
        - model_path (str): 保存先のファイルパス（例: "model/logo_classifier.pt"）

        備考:
        - 保存前にCPUへ移動させ、保存後に再び self.device に戻します。
        """

        os.makedirs(os.path.dirname(model_path), exist_ok=True)
        self.model.cpu()
        torch.save(self.model.state_dict(), model_path)
        print(f"モデル保存済: {model_path}")
        self.model = self.model.to(self.device)

    def load(self, model_path):
        """
        指定パスのモデルファイルを読み込み、使用可能な状態にします。

        Parameters:
        - model_path (str): 読み込むファイルのパス（例: "model/logo_classifier.pt"）
        """

        self.model.load_state_dict(torch.load(model_path, map_location=self.device))
        self.model = self.model.to(self.device)
        self.model.eval()
        print(f"モデル読み込み済: {model_path}")

    def predict(self, image_path, class_names=None):
        """
        1枚の画像ファイルに対してクラス分類を実行します。

        Parameters:
        - image_path (str): 入力画像のファイルパス
        - class_names (list or None): インデックスに対応するクラス名（省略時は数字で返す）

        Returns:
        - str or int: 推論されたクラス（ラベル名またはインデックス）
        """

        transform = transforms.Compose([
            transforms.Resize(self.image_size),
            transforms.ToTensor()
        ])
        img = Image.open(image_path).convert("RGB")
        input_tensor = transform(img).unsqueeze(0).to(self.device)

        self.model.eval()
        with torch.no_grad():
            output = self.model(input_tensor)
            _, predicted = torch.max(output, 1)
            class_index = predicted.item()
            return class_names[class_index] if class_names else class_index

    def classify_folder(self, input_dir, output_dir, class_names=None, extensions=("jpg", "jpeg", "png", "bmp", "tif", "tiff")):
        """
        フォルダ内の画像をクラスごとに分類し、出力フォルダにコピー整理します。

        Parameters:
        - input_dir (str): 分類対象の画像が格納されたディレクトリ
        - output_dir (str): 分類結果を保存するルートディレクトリ
        - class_names (list of str): クラス名（インデックスに対応）
        - extensions (tuple): 対象とする画像拡張子の一覧（デフォルトは一般的な画像形式）

        出力例:
        - output_dir/logo/xxx.jpg
        - output_dir/none/yyy.png
        """

        transform = transforms.Compose([
            transforms.Resize(self.image_size),
            transforms.ToTensor()
        ])

        os.makedirs(output_dir, exist_ok=True)
        for cname in class_names:
            os.makedirs(os.path.join(output_dir, cname), exist_ok=True)

        image_files = [f for f in os.listdir(input_dir) if f.lower().endswith(extensions)]

        for fname in image_files:
            path = os.path.join(input_dir, fname)
            img = Image.open(path).convert("RGB")
            input_tensor = transform(img).unsqueeze(0).to(self.device)

            self.model.eval()
            with torch.no_grad():
                output = self.model(input_tensor)
                _, pred = torch.max(output, 1)
                predicted_class = class_names[pred.item()] if class_names else str(pred.item())

            dest = os.path.join(output_dir, predicted_class, fname)
            shutil.copy2(path, dest)
            print(f">>> {fname} を {predicted_class} にコピーしました。")