PyTorchでコンピュータビジョン¶

# PyTorchのインポート
import torch
from torch import nn

# torchvisionのインポート
import torchvision
from torchvision import datasets
from torchvision.transforms import ToTensor

# 可視化用ライブラリ
import matplotlib.pyplot as plt

# バージョン確認
# 注意：PyTorchは1.10.0以上、torchvisionは0.11以上が必要
print(f"PyTorch version: {torch.__version__}\ntorchvision version: {torchvision.__version__}")

# 訓練用データの設定
train_data = datasets.FashionMNIST(
    root="data",           # データの保存先
    train=True,           # 訓練用データを取得
    download=True,        # データが存在しない場合はダウンロード
    transform=ToTensor(), # PIL画像をPyTorchテンソルに変換
    target_transform=None # ラベルの変換（今回は使用しない）
)

# テスト用データの設定
test_data = datasets.FashionMNIST(
    root="data",
    train=False,          # テスト用データを取得
    download=True,
    transform=ToTensor()
)

# 最初のサンプルを確認
image, label = train_data[0]
print(f"画像の形状: {image.shape}, ラベル: {label}")

# データセットのサイズを確認
print(f"訓練データ数: {len(train_data)}, テストデータ数: {len(test_data)}")

# クラス名を確認
class_names = train_data.classes
print(f"クラス名: {class_names}")

# 単一画像の表示
image, label = train_data[0]
print(f"画像の形状: {image.shape}")
plt.figure(figsize=(6, 6))
plt.imshow(image.squeeze(), cmap="gray")  # squeeze()で次元を削除
plt.title(f"{class_names[label]} (ラベル: {label})")
plt.axis("off")
plt.show()

# 複数画像のランダム表示
torch.manual_seed(42)
fig = plt.figure(figsize=(9, 9))
rows, cols = 4, 4

for i in range(1, rows * cols + 1):
    # ランダムにサンプルを選択
    random_idx = torch.randint(0, len(train_data), size=[1]).item()
    img, label = train_data[random_idx]

    # サブプロットを作成
    fig.add_subplot(rows, cols, i)
    plt.imshow(img.squeeze(), cmap="gray")
    plt.title(class_names[label])
    plt.axis(False)

plt.tight_layout()
plt.show()

from torch.utils.data import DataLoader

# バッチサイズの設定（ハイパーパラメータ）
BATCH_SIZE = 32

# DataLoaderの作成
train_dataloader = DataLoader(
    train_data,           # データセット
    batch_size=BATCH_SIZE, # バッチサイズ
    shuffle=True          # エポックごとにデータをシャッフル
)

test_dataloader = DataLoader(
    test_data,
    batch_size=BATCH_SIZE,
    shuffle=False         # テストデータはシャッフル不要
)

# DataLoaderの情報を確認
print(f"訓練DataLoader: {len(train_dataloader)} バッチ (各バッチ {BATCH_SIZE} サンプル)")
print(f"テストDataLoader: {len(test_dataloader)} バッチ (各バッチ {BATCH_SIZE} サンプル)")

# バッチデータの確認
train_features_batch, train_labels_batch = next(iter(train_dataloader))
print(f"バッチ画像の形状: {train_features_batch.shape}")
print(f"バッチラベルの形状: {train_labels_batch.shape}")

# バッチから単一サンプルを表示
torch.manual_seed(42)
random_idx = torch.randint(0, len(train_features_batch), size=[1]).item()
img, label = train_features_batch[random_idx], train_labels_batch[random_idx]

plt.figure(figsize=(6, 6))
plt.imshow(img.squeeze(), cmap="gray")
plt.title(class_names[label])
plt.axis("off")
plt.show()

print(f"単一画像の形状: {img.shape}")
print(f"ラベル: {label} ({class_names[label]})")

from torch import nn

class FashionMNISTModelV0(nn.Module):
    def __init__(self, input_shape: int, hidden_units: int, output_shape: int):
        super().__init__()
        self.layer_stack = nn.Sequential(
            nn.Flatten(),  # 画像を1次元ベクトルに変換
            nn.Linear(in_features=input_shape, out_features=hidden_units),
            nn.Linear(in_features=hidden_units, out_features=output_shape)
        )

    def forward(self, x):
        return self.layer_stack(x)

# Flattenレイヤーの動作を確認
flatten_model = nn.Flatten()
x = train_features_batch[0]  # 単一サンプル

print(f"Flatten前の形状: {x.shape} -> [カラーチャンネル, 高さ, 幅]")
output = flatten_model(x)
print(f"Flatten後の形状: {output.shape} -> [カラーチャンネル, 高さ×幅]")

torch.manual_seed(42)

# モデルのインスタンス化
model_0 = FashionMNISTModelV0(
    input_shape=784,              # 28×28=784 ピクセル
    hidden_units=10,              # 隠れ層のニューロン数
    output_shape=len(class_names) # クラス数（10）
)

print(model_0)

# ヘルパー関数のインポート（正確度計算用）
from helper_functions import accuracy_fn

# 損失関数と最適化器の設定
loss_fn = nn.CrossEntropyLoss()  # 多クラス分類用
optimizer = torch.optim.SGD(params=model_0.parameters(), lr=0.1)

from timeit import default_timer as timer

def print_train_time(start: float, end: float, device: torch.device = None):
    """訓練時間を表示する関数

    Args:
        start (float): 開始時間
        end (float): 終了時間
        device (torch.device): 計算デバイス

    Returns:
        float: 経過時間（秒）
    """
    total_time = end - start
    print(f"訓練時間 on {device}: {total_time:.3f} 秒")
    return total_time

from tqdm import tqdm

# シードと時間測定の開始
torch.manual_seed(42)
train_time_start_on_cpu = timer()

epochs = 3

# 訓練とテストのループ
for epoch in tqdm(range(epochs)):
    print(f"エポック: {epoch}\n-------")

    ### 訓練フェーズ ###
    train_loss = 0
    model_0.train()  # 訓練モードに設定

    for batch, (X, y) in enumerate(train_dataloader):
        # 1. 順伝播
        y_pred = model_0(X)

        # 2. 損失計算
        loss = loss_fn(y_pred, y)
        train_loss += loss

        # 3. 勾配をゼロに
        optimizer.zero_grad()

        # 4. 逆伝播
        loss.backward()

        # 5. パラメータ更新
        optimizer.step()

        # 進捗表示
        if batch % 400 == 0:
            print(f"処理済みサンプル: {batch * len(X)}/{len(train_dataloader.dataset)}")

    # バッチあたりの平均損失を計算
    train_loss /= len(train_dataloader)

    ### テストフェーズ ###
    test_loss, test_acc = 0, 0
    model_0.eval()  # 評価モードに設定

    with torch.inference_mode():
        for X, y in test_dataloader:
            # 1. 順伝播
            test_pred = model_0(X)

            # 2. 損失計算
            test_loss += loss_fn(test_pred, y)

            # 3. 正確度計算
            test_acc += accuracy_fn(y_true=y, y_pred=test_pred.argmax(dim=1))

        # バッチあたりの平均値を計算
        test_loss /= len(test_dataloader)
        test_acc /= len(test_dataloader)

    print(f"\n訓練損失: {train_loss:.5f} | テスト損失: {test_loss:.5f}, テスト正確度: {test_acc:.2f}%\n")

# 訓練時間の計算
train_time_end_on_cpu = timer()
total_train_time_model_0 = print_train_time(
    start=train_time_start_on_cpu,
    end=train_time_end_on_cpu,
    device="cpu"
)

def eval_model(model: torch.nn.Module, 
               data_loader: torch.utils.data.DataLoader, 
               loss_fn: torch.nn.Module, 
               accuracy_fn,
               device: torch.device = "cpu"):
    """モデルの評価を行う関数

    Args:
        model: 評価対象のPyTorchモデル
        data_loader: 評価用データローダー
        loss_fn: 損失関数
        accuracy_fn: 正確度計算関数
        device: 計算デバイス

    Returns:
        dict: 評価結果（モデル名、損失、正確度）
    """
    loss, acc = 0, 0
    model.eval()

    with torch.inference_mode():
        for X, y in data_loader:
            X, y = X.to(device), y.to(device)

            # 予測の実行
            y_pred = model(X)

            # 損失と正確度の累積
            loss += loss_fn(y_pred, y)
            acc += accuracy_fn(y_true=y, y_pred=y_pred.argmax(dim=1))

        # バッチあたりの平均値を計算
        loss /= len(data_loader)
        acc /= len(data_loader)

    return {
        "model_name": model.__class__.__name__,
        "model_loss": loss.item(),
        "model_acc": acc
    }

# `model_0`の結果を計算
torch.manual_seed(42)
model_0_results = eval_model(
    model=model_0,
    data_loader=test_dataloader,
    loss_fn=loss_fn,
    accuracy_fn=accuracy_fn
)

print(model_0_results)

# デバイスの自動選択
device = "mps" if torch.mps.is_available() else "cpu"
print(f"使用デバイス: {device}")

class FashionMNISTModelV1(nn.Module):
    def __init__(self, input_shape: int, hidden_units: int, output_shape: int):
        super().__init__()
        self.layer_stack = nn.Sequential(
            nn.Flatten(),
            nn.Linear(in_features=input_shape, out_features=hidden_units),
            nn.ReLU(),  # 非線形活性化関数を追加
            nn.Linear(in_features=hidden_units, out_features=output_shape),
            nn.ReLU()   # 出力層にも追加
        )

    def forward(self, x: torch.Tensor):
        return self.layer_stack(x)

# モデルのインスタンス化（GPUに送信）
torch.manual_seed(42)
model_1 = FashionMNISTModelV1(
    input_shape=784,
    hidden_units=10,
    output_shape=len(class_names)
).to(device)

print(f"モデルのデバイス: {next(model_1.parameters()).device}")

loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(params=model_1.parameters(), lr=0.1)

def train_step(model: torch.nn.Module,
               data_loader: torch.utils.data.DataLoader,
               loss_fn: torch.nn.Module,
               optimizer: torch.optim.Optimizer,
               accuracy_fn,
               device: torch.device = device):
    """訓練ステップを実行する関数"""
    train_loss, train_acc = 0, 0
    model.to(device)
    model.train()

    for batch, (X, y) in enumerate(data_loader):
        # データをGPUに送信
        X, y = X.to(device), y.to(device)

        # 1. 順伝播
        y_pred = model(X)

        # 2. 損失計算
        loss = loss_fn(y_pred, y)
        train_loss += loss
        train_acc += accuracy_fn(y_true=y, y_pred=y_pred.argmax(dim=1))

        # 3. 勾配をゼロに
        optimizer.zero_grad()

        # 4. 逆伝播
        loss.backward()

        # 5. パラメータ更新
        optimizer.step()

    # バッチあたりの平均値を計算
    train_loss /= len(data_loader)
    train_acc /= len(data_loader)
    print(f"訓練損失: {train_loss:.5f} | 訓練正確度: {train_acc:.2f}%")

def test_step(data_loader: torch.utils.data.DataLoader,
              model: torch.nn.Module,
              loss_fn: torch.nn.Module,
              accuracy_fn,
              device: torch.device = device):
    """テストステップを実行する関数"""
    test_loss, test_acc = 0, 0
    model.to(device)
    model.eval()

    with torch.inference_mode():
        for X, y in data_loader:
            X, y = X.to(device), y.to(device)

            # 1. 順伝播
            test_pred = model(X)

            # 2. 損失と正確度の計算
            test_loss += loss_fn(test_pred, y)
            test_acc += accuracy_fn(y_true=y, y_pred=test_pred.argmax(dim=1))

        # バッチあたりの平均値を計算
        test_loss /= len(data_loader)
        test_acc /= len(data_loader)
        print(f"テスト損失: {test_loss:.5f} | テスト正確度: {test_acc:.2f}%\n")

torch.manual_seed(42)
train_time_start_on_gpu = timer()

epochs = 3
for epoch in tqdm(range(epochs)):
    print(f"エポック: {epoch}\n---------")
    train_step(data_loader=train_dataloader, 
               model=model_1, 
               loss_fn=loss_fn,
               optimizer=optimizer,
               accuracy_fn=accuracy_fn)
    test_step(data_loader=test_dataloader,
              model=model_1,
              loss_fn=loss_fn,
              accuracy_fn=accuracy_fn)

train_time_end_on_gpu = timer()
total_train_time_model_1 = print_train_time(start=train_time_start_on_gpu,
                                            end=train_time_end_on_gpu,
                                            device=device)

# デバイス対応版のeval_model関数を使用
torch.manual_seed(42)
model_1_results = eval_model(
    model=model_1,
    data_loader=test_dataloader,
    loss_fn=loss_fn,
    accuracy_fn=accuracy_fn,
    device=device
)

print("`model_1`の結果:", model_1_results)
print("`model_0`の結果:", model_0_results)

class FashionMNISTModelV2(nn.Module):
    """
    TinyVGGアーキテクチャを模倣したCNNモデル
    参考: https://poloclub.github.io/cnn-explainer/
    """
    def __init__(self, input_shape: int, hidden_units: int, output_shape: int):
        super().__init__()

        # 1つ目の畳み込みブロック
        self.block_1 = nn.Sequential(
            nn.Conv2d(in_channels=input_shape, 
                      out_channels=hidden_units, 
                      kernel_size=3,    # 3x3のフィルター
                      stride=1,         # ストライド（デフォルト）
                      padding=1),       # パディング（出力サイズを維持）
            nn.ReLU(),
            nn.Conv2d(in_channels=hidden_units, 
                      out_channels=hidden_units,
                      kernel_size=3,
                      stride=1,
                      padding=1),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2,    # 2x2の最大プーリング
                         stride=2)         # サイズを半分に
        )

        # 2つ目の畳み込みブロック
        self.block_2 = nn.Sequential(
            nn.Conv2d(hidden_units, hidden_units, 3, padding=1),
            nn.ReLU(),
            nn.Conv2d(hidden_units, hidden_units, 3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2)
        )

        # 分類器（全結合層）
        self.classifier = nn.Sequential(
            nn.Flatten(),
            # 28x28 → 14x14 → 7x7 なので、最終的にhidden_units*7*7の特徴量
            nn.Linear(in_features=hidden_units*7*7, 
                      out_features=output_shape)
        )

    def forward(self, x: torch.Tensor):
        x = self.block_1(x)
        x = self.block_2(x)
        x = self.classifier(x)
        return x

# モデルのインスタンス化
torch.manual_seed(42)
model_2 = FashionMNISTModelV2(
    input_shape=1,                 # グレースケール画像（1チャンネル）
    hidden_units=10, 
    output_shape=len(class_names)
).to(device)

print(model_2)

# テスト用のダミーデータを作成
torch.manual_seed(42)
images = torch.randn(size=(32, 3, 64, 64))  # [バッチサイズ, チャンネル, 高さ, 幅]
test_image = images[0]

print(f"バッチ画像の形状: {images.shape}")
print(f"単一画像の形状: {test_image.shape}")

# 畳み込み層の作成
conv_layer = nn.Conv2d(in_channels=3,      # 入力チャンネル数
                       out_channels=10,     # 出力チャンネル数
                       kernel_size=3,       # フィルターサイズ
                       stride=1,            # ストライド
                       padding=0)           # パディング

# 畳み込み層を通した後の形状
output = conv_layer(test_image.unsqueeze(dim=0))  # バッチ次元を追加
print(f"畳み込み後の形状: {output.shape}")

# 異なるパラメータでの畳み込み層
conv_layer_2 = nn.Conv2d(in_channels=3,
                         out_channels=10,
                         kernel_size=(5, 5),  # より大きなフィルター
                         stride=2,            # より大きなストライド
                         padding=0)

output_2 = conv_layer_2(test_image.unsqueeze(dim=0))
print(f"異なる設定での畳み込み後の形状: {output_2.shape}")

# 重みとバイアスの形状確認
print(f"重み形状: {conv_layer_2.weight.shape}")
print(f"バイアス形状: {conv_layer_2.bias.shape}")

# 最大プーリング層の動作確認
max_pool_layer = nn.MaxPool2d(kernel_size=2)

# 畳み込み → 最大プーリングの流れ
test_conv_output = conv_layer(test_image.unsqueeze(dim=0))
test_pooled_output = max_pool_layer(test_conv_output)

print(f"畳み込み後: {test_conv_output.shape}")
print(f"最大プーリング後: {test_pooled_output.shape}")

# 小さなテンソルでの動作確認
torch.manual_seed(42)
small_tensor = torch.randn(size=(1, 1, 2, 2))
print(f"\n元のテンソル:\n{small_tensor}")

pooled_tensor = max_pool_layer(small_tensor)
print(f"\n最大プーリング後:\n{pooled_tensor}")
print(f"形状変化: {small_tensor.shape} → {pooled_tensor.shape}")

# 損失関数と最適化器の設定
loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(params=model_2.parameters(), lr=0.1)

# 訓練の実行
torch.manual_seed(42)
train_time_start_model_2 = timer()

epochs = 3
for epoch in tqdm(range(epochs)):
    print(f"エポック: {epoch}\n---------")
    train_step(data_loader=train_dataloader, 
               model=model_2, 
               loss_fn=loss_fn,
               optimizer=optimizer,
               accuracy_fn=accuracy_fn,
               device=device)
    test_step(data_loader=test_dataloader,
              model=model_2,
              loss_fn=loss_fn,
              accuracy_fn=accuracy_fn,
              device=device)

train_time_end_model_2 = timer()
total_train_time_model_2 = print_train_time(start=train_time_start_model_2,
                                           end=train_time_end_model_2,
                                           device=device)

# `model_2`の結果を取得
model_2_results = eval_model(
    model=model_2,
    data_loader=test_dataloader,
    loss_fn=loss_fn,
    accuracy_fn=accuracy_fn,
    device=device
)

print("`model_2`（CNN）の結果:", model_2_results)

import pandas as pd

# 結果をDataFrameにまとめ
compare_results = pd.DataFrame([model_0_results, model_1_results, model_2_results])

# 訓練時間も追加
compare_results["training_time"] = [
    total_train_time_model_0,
    total_train_time_model_1,
    total_train_time_model_2
]

print("モデル比較結果:")
print(compare_results)

# 正確度の可視化
compare_results.set_index("model_name")["model_acc"].plot(kind="barh")
plt.xlabel("正確度 (%)")
plt.ylabel("モデル")
plt.title("モデル別正確度比較")
plt.show()

def make_predictions(model: torch.nn.Module, data: list, device: torch.device = device):
    """モデルで予測を実行する関数

    Args:
        model: 訓練済みモデル
        data: 予測対象のデータリスト
        device: 計算デバイス

    Returns:
        torch.Tensor: 予測確率のテンソル
    """
    pred_probs = []
    model.eval()

    with torch.inference_mode():
        for sample in data:
            # サンプルの前処理（バッチ次元の追加とデバイス移動）
            sample = torch.unsqueeze(sample, dim=0).to(device)

            # 順伝播（ロジット出力）
            pred_logit = model(sample)

            # 確率への変換（ソフトマックス適用）
            pred_prob = torch.softmax(pred_logit.squeeze(), dim=0)

            # CPUに移動して結果リストに追加
            pred_probs.append(pred_prob.cpu())

    return torch.stack(pred_probs)

import random
random.seed(42)

# ランダムに9個のテストサンプルを選択
test_samples = []
test_labels = []
for sample, label in random.sample(list(test_data), k=9):
    test_samples.append(sample)
    test_labels.append(label)

print(f"テストサンプル画像形状: {test_samples[0].shape}")
print(f"テストサンプルラベル: {test_labels[0]} ({class_names[test_labels[0]]})")

# 予測の実行
pred_probs = make_predictions(model=model_2, data=test_samples)

# 予測確率から予測クラスへの変換
pred_classes = pred_probs.argmax(dim=1)

print("予測確率（最初の2サンプル）:")
print(pred_probs[:2])
print(f"\n予測クラス: {pred_classes}")
print(f"正解ラベル: {test_labels}")

# 予測結果の可視化
plt.figure(figsize=(9, 9))
nrows, ncols = 3, 3

for i, sample in enumerate(test_samples):
    plt.subplot(nrows, ncols, i+1)

    # 画像の表示
    plt.imshow(sample.squeeze(), cmap="gray")

    # 予測と正解のラベル取得
    pred_label = class_names[pred_classes[i]]
    truth_label = class_names[test_labels[i]]

    # タイトルテキストの作成
    title_text = f"予測: {pred_label} | 正解: {truth_label}"

    # 正解/不正解に応じて色を変更
    if pred_label == truth_label:
        plt.title(title_text, fontsize=10, c="g")  # 正解なら緑
    else:
        plt.title(title_text, fontsize=10, c="r")  # 不正解なら赤

    plt.axis(False)

plt.tight_layout()
plt.show()

from tqdm import tqdm

# 全テストデータで予測を実行
y_preds = []
model_2.eval()

with torch.inference_mode():
    for X, y in tqdm(test_dataloader, desc="予測実行中"):
        X, y = X.to(device), y.to(device)

        # 順伝播
        y_logit = model_2(X)

        # ロジット → 確率 → 予測ラベル
        y_pred = torch.softmax(y_logit, dim=1).argmax(dim=1)

        # CPUに移動
        y_preds.append(y_pred.cpu())

# 予測結果を結合
y_pred_tensor = torch.cat(y_preds)
print(f"予測結果のサイズ: {y_pred_tensor.shape}")

from torchmetrics import ConfusionMatrix
from mlxtend.plotting import plot_confusion_matrix

# 混同行列の計算
confmat = ConfusionMatrix(num_classes=len(class_names), task='multiclass')
confmat_tensor = confmat(preds=y_pred_tensor, target=test_data.targets)

# 混同行列の可視化
fig, ax = plot_confusion_matrix(
    conf_mat=confmat_tensor.numpy(),
    class_names=class_names,
    figsize=(10, 7),
    show_normed=False,  # 正規化しない（実際の数を表示）
    colorbar=True
)
plt.title("FashionMNIST 混同行列")
plt.show()

from pathlib import Path

# モデル保存ディレクトリの作成
MODEL_PATH = Path("models")
MODEL_PATH.mkdir(parents=True, exist_ok=True)

# 保存パスの設定
MODEL_NAME = "03_pytorch_computer_vision_model_2.pth"
MODEL_SAVE_PATH = MODEL_PATH / MODEL_NAME

# モデルの状態辞書を保存
print(f"モデルを保存中: {MODEL_SAVE_PATH}")
torch.save(obj=model_2.state_dict(), f=MODEL_SAVE_PATH)

# 新しいモデルインスタンスの作成
loaded_model_2 = FashionMNISTModelV2(
    input_shape=1,
    hidden_units=10,
    output_shape=10
)

# 保存された状態辞書の読み込み
loaded_model_2.load_state_dict(torch.load(f=MODEL_SAVE_PATH))

# デバイスに移動
loaded_model_2 = loaded_model_2.to(device)

print("モデルの読み込み完了")

# 読み込んだモデルの評価
torch.manual_seed(42)
loaded_model_2_results = eval_model(
    model=loaded_model_2,
    data_loader=test_dataloader,
    loss_fn=loss_fn,
    accuracy_fn=accuracy_fn,
    device=device
)

print("読み込んだモデルの結果:", loaded_model_2_results)
print("元のモデルの結果:", model_2_results)

# 結果の一致確認
results_match = torch.isclose(
    torch.tensor(model_2_results["model_loss"]),
    torch.tensor(loaded_model_2_results["model_loss"]),
    atol=1e-08,
    rtol=0.0001
)
print(f"結果が一致しているか: {results_match}")