PyTorch実験追跡 - FoodVision Mini実験プロジェクト¶

# 必要なライブラリのインポート
import matplotlib.pyplot as plt
import torch
import torchvision

from torch import nn
from torchvision import transforms
from torchinfo import summary
from going_modular import data_setup, engine

# デバイス設定（Apple SiliconのMacではMPS、それ以外はCPU）
device = "mps" if torch.mps.is_available() else "cpu"
device

def set_seeds(seed: int=42):
    """PyTorchの操作用にランダムシードを設定する

    Args:
        seed (int, optional): 設定するランダムシード。デフォルトは42
    """
    # 一般的なtorch操作用のシード設定
    torch.manual_seed(seed)
    # MPSでのtorch操作用のシード設定
    torch.mps.manual_seed(seed)

import os
import zipfile
from pathlib import Path
import requests

def download_data(source: str, 
                  destination: str,
                  remove_source: bool = True) -> Path:
    """ソースからzipファイルをダウンロードし、指定先に展開する

    Args:
        source (str): zipファイルのURL
        destination (str): 展開先ディレクトリ
        remove_source (bool): ダウンロード後にソースファイルを削除するか

    Returns:
        pathlib.Path: ダウンロードされたデータへのパス
    """
    # データフォルダのパス設定
    data_path = Path("data/")
    image_path = data_path / destination

    # 画像フォルダが存在しない場合、ダウンロードして準備
    if image_path.is_dir():
        print(f"[INFO] {image_path} ディレクトリが存在するため、ダウンロードをスキップします。")
    else:
        print(f"[INFO] {image_path} ディレクトリが見つからないため、作成します...")
        image_path.mkdir(parents=True, exist_ok=True)

        # データをダウンロード
        target_file = Path(source).name
        with open(data_path / target_file, "wb") as f:
            request = requests.get(source)
            print(f"[INFO] {source}から{target_file}をダウンロード中...")
            f.write(request.content)

        # データを展開
        with zipfile.ZipFile(data_path / target_file, "r") as zip_ref:
            print(f"[INFO] {target_file}データを展開中...") 
            zip_ref.extractall(image_path)

        # zipファイルを削除
        if remove_source:
            os.remove(data_path / target_file)

    return image_path

# データのダウンロード
image_path = download_data(
    source="https://github.com/vinsmoke-three/deeplearning-with-pytorch/raw/main/data/pizza_steak_sushi.zip",
    destination="pizza_steak_sushi"
)

# ディレクトリの設定
train_dir = image_path / "train"
test_dir = image_path / "test"

# ImageNet正規化レベルの設定
normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                 std=[0.229, 0.224, 0.225])

# 変換パイプラインを手動で作成
manual_transforms = transforms.Compose([
    transforms.Resize((224, 224)),
    transforms.ToTensor(),
    normalize
])           
print(f"手動作成された変換: {manual_transforms}")

# データローダーを作成
train_dataloader, test_dataloader, class_names = data_setup.create_dataloaders(
    train_dir=train_dir,
    test_dir=test_dir,
    transform=manual_transforms,
    batch_size=32
)

# 事前訓練された重みの設定
weights = torchvision.models.EfficientNet_B0_Weights.DEFAULT

# 重みから変換を取得
automatic_transforms = weights.transforms() 
print(f"自動作成された変換: {automatic_transforms}")

# データローダーを作成
train_dataloader, test_dataloader, class_names = data_setup.create_dataloaders(
    train_dir=train_dir,
    test_dir=test_dir,
    transform=automatic_transforms,
    batch_size=32
)

# EfficientNet_B0の事前訓練重みをダウンロード
weights = torchvision.models.EfficientNet_B0_Weights.DEFAULT
model = torchvision.models.efficientnet_b0(weights=weights).to(device)

# ベースレイヤーをフリーズ（requires_grad=Falseに設定）
for param in model.features.parameters():
    param.requires_grad = False

# 新しいレイヤーを作成するためシードを設定
set_seeds() 

# 分類ヘッドを問題に合わせて更新
model.classifier = torch.nn.Sequential(
    nn.Dropout(p=0.2, inplace=True),
    nn.Linear(in_features=1280, 
              out_features=len(class_names),
              bias=True).to(device)
)

try:
    from torch.utils.tensorboard import SummaryWriter
except:
    print("[INFO] tensorboardが見つかりません...インストール中。")
    !pip install -q tensorboard
    from torch.utils.tensorboard import SummaryWriter

# デフォルト設定でwriterを作成
writer = SummaryWriter()

from typing import Dict, List
from tqdm import tqdm
from going_modular.engine import train_step, test_step

def train(model: torch.nn.Module, 
          train_dataloader: torch.utils.data.DataLoader, 
          test_dataloader: torch.utils.data.DataLoader, 
          optimizer: torch.optim.Optimizer,
          loss_fn: torch.nn.Module,
          epochs: int,
          device: torch.device,
          writer: torch.utils.tensorboard.writer.SummaryWriter = None
          ) -> Dict[str, List]:
    """PyTorchモデルを訓練し、TensorBoardで結果を追跡する

    Args:
        model: 訓練・テストするPyTorchモデル
        train_dataloader: 訓練用DataLoader
        test_dataloader: テスト用DataLoader
        optimizer: PyTorchオプティマイザー
        loss_fn: PyTorch損失関数
        epochs: 訓練エポック数
        device: 計算デバイス
        writer: SummaryWriter インスタンス

    Returns:
        訓練・テスト損失および精度のメトリクスを含む辞書
    """
    # 空の結果辞書を作成
    results = {"train_loss": [],
               "train_acc": [],
               "test_loss": [],
               "test_acc": []}

    # エポック数分のループで訓練・テストを実行
    for epoch in tqdm(range(epochs)):
        train_loss, train_acc = train_step(model=model,
                                           dataloader=train_dataloader,
                                           loss_fn=loss_fn,
                                           optimizer=optimizer,
                                           device=device)
        test_loss, test_acc = test_step(model=model,
                                        dataloader=test_dataloader,
                                        loss_fn=loss_fn,
                                        device=device)

        # 進行状況を印刷
        print(
          f"Epoch: {epoch+1} | "
          f"train_loss: {train_loss:.4f} | "
          f"train_acc: {train_acc:.4f} | "
          f"test_loss: {test_loss:.4f} | "
          f"test_acc: {test_acc:.4f}"
        )

        # 結果辞書を更新
        results["train_loss"].append(train_loss)
        results["train_acc"].append(train_acc)
        results["test_loss"].append(test_loss)
        results["test_acc"].append(test_acc)

        ### 新機能: 実験追跡 ###
        # writerが存在する場合、結果をログに記録
        if writer:
            # 損失結果をSummaryWriterに追加
            writer.add_scalars(main_tag="Loss", 
                               tag_scalar_dict={"train_loss": train_loss,
                                                "test_loss": test_loss},
                               global_step=epoch)

            # 精度結果をSummaryWriterに追加
            writer.add_scalars(main_tag="Accuracy", 
                               tag_scalar_dict={"train_acc": train_acc,
                                                "test_acc": test_acc}, 
                               global_step=epoch)

            # PyTorchモデルアーキテクチャを追跡
            writer.add_graph(model=model, 
                             input_to_model=torch.randn(32, 3, 224, 224).to(device))

    # writerを閉じる
    if writer:
        writer.close()

    # エポック終了時に結果を返す
    return results

# 損失関数とオプティマイザーの定義
loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

# モデルの訓練
set_seeds()
results = train(model=model.to(device),
                train_dataloader=train_dataloader,
                test_dataloader=test_dataloader,
                optimizer=optimizer,
                loss_fn=loss_fn,
                epochs=5,
                device=device,
                writer=writer)

def create_writer(experiment_name: str, 
                  model_name: str, 
                  extra: str=None) -> torch.utils.tensorboard.writer.SummaryWriter():
    """特定のlog_dirに保存するtorch.utils.tensorboard.writer.SummaryWriter()インスタンスを作成する

    log_dirは runs/timestamp/experiment_name/model_name/extra の組み合わせになります。

    timestampは現在の日付をYYYY-MM-DD形式で表示します。

    Args:
        experiment_name (str): 実験の名前
        model_name (str): モデルの名前
        extra (str, optional): ディレクトリに追加する任意の内容。デフォルトはNone

    Returns:
        torch.utils.tensorboard.writer.SummaryWriter(): log_dirに保存するwriterのインスタンス

    使用例:
        # "runs/2022-06-04/data_10_percent/effnetb2/5_epochs/"に保存するwriterを作成
        writer = create_writer(experiment_name="data_10_percent",
                               model_name="effnetb2",
                               extra="5_epochs")
        # 上記は以下と同じです:
        writer = SummaryWriter(log_dir="runs/2022-06-04/data_10_percent/effnetb2/5_epochs/")
    """
    from datetime import datetime
    import os

    # 現在の日付のタイムスタンプを取得（特定の日のすべての実験は同じフォルダに保存）
    timestamp = datetime.now().strftime("%Y-%m-%d") # 現在の日付をYYYY-MM-DD形式で返す

    if extra:
        # ログディレクトリパスを作成
        log_dir = os.path.join("runs", timestamp, experiment_name, model_name, extra)
    else:
        log_dir = os.path.join("runs", timestamp, experiment_name, model_name)

    print(f"[INFO] SummaryWriterを作成しました。保存先: {log_dir}...")
    return SummaryWriter(log_dir=log_dir)

# サンプルwriterを作成
example_writer = create_writer(experiment_name="data_10_percent",
                               model_name="effnetb0",
                               extra="5_epochs")

from typing import Dict, List
from tqdm import tqdm

# train()にwriterパラメータを追加
def train(model: torch.nn.Module, 
          train_dataloader: torch.utils.data.DataLoader, 
          test_dataloader: torch.utils.data.DataLoader, 
          optimizer: torch.optim.Optimizer,
          loss_fn: torch.nn.Module,
          epochs: int,
          device: torch.device, 
          writer: torch.utils.tensorboard.writer.SummaryWriter = None
          ) -> Dict[str, List]:
    """PyTorchモデルを訓練・テストします。

    指定されたエポック数の間、train_step()とtest_step()関数を通して
    対象のPyTorchモデルを処理し、同じエポックループ内でモデルを
    訓練・テストします。

    評価メトリクスを全体を通して計算、印刷、保存します。

    writerが存在する場合、指定されたwriter log_dirにメトリクスを保存します。

    Args:
      model: 訓練・テストするPyTorchモデル
      train_dataloader: モデルを訓練するためのDataLoaderインスタンス
      test_dataloader: モデルをテストするためのDataLoaderインスタンス
      optimizer: 損失関数を最小化するためのPyTorchオプティマイザー
      loss_fn: 両方のデータセットで損失を計算するためのPyTorch損失関数
      epochs: 何エポック訓練するかを示す整数
      device: 計算するターゲットデバイス（例："cuda"または"cpu"）
      writer: モデル結果をログ出力するSummaryWriter()インスタンス

    Returns:
      訓練・テスト損失および訓練・テスト精度メトリクスの辞書。
      各メトリクスにはエポックごとのリストの値があります。
      形式: {train_loss: [...],
                train_acc: [...],
                test_loss: [...],
                test_acc: [...]} 
      例：epochs=2で訓練した場合: 
              {train_loss: [2.0616, 1.0537],
                train_acc: [0.3945, 0.3945],
                test_loss: [1.2641, 1.5706],
                test_acc: [0.3400, 0.2973]} 
    """
    # 空の結果辞書を作成
    results = {"train_loss": [],
               "train_acc": [],
               "test_loss": [],
               "test_acc": []}

    # 指定されたエポック数だけ訓練・テストステップをループ
    for epoch in tqdm(range(epochs)):
        train_loss, train_acc = train_step(model=model,
                                          dataloader=train_dataloader,
                                          loss_fn=loss_fn,
                                          optimizer=optimizer,
                                          device=device)
        test_loss, test_acc = test_step(model=model,
          dataloader=test_dataloader,
          loss_fn=loss_fn,
          device=device)

        # 進行状況を印刷
        print(
          f"Epoch: {epoch+1} | "
          f"train_loss: {train_loss:.4f} | "
          f"train_acc: {train_acc:.4f} | "
          f"test_loss: {test_loss:.4f} | "
          f"test_acc: {test_acc:.4f}"
        )

        # 結果辞書を更新
        results["train_loss"].append(train_loss)
        results["train_acc"].append(train_acc)
        results["test_loss"].append(test_loss)
        results["test_acc"].append(test_acc)


        ### 新機能: writerパラメータを使用して実験を追跡 ###
        # writerがあるかどうか確認し、ある場合はログ出力
        if writer:
            # 結果をSummaryWriterに追加
            writer.add_scalars(main_tag="Loss", 
                               tag_scalar_dict={"train_loss": train_loss,
                                                "test_loss": test_loss},
                               global_step=epoch)
            writer.add_scalars(main_tag="Accuracy", 
                               tag_scalar_dict={"train_acc": train_acc,
                                                "test_acc": test_acc}, 
                               global_step=epoch)

            # writerを閉じる
            writer.close()
        else:
            pass
    ### 新機能終了 ###

    # エポック終了時に結果を返す
    return results

# 10%と20%の訓練データをダウンロード（必要に応じて）
data_10_percent_path = download_data(
    source="https://github.com/vinsmoke-three/deeplearning-with-pytorch/raw/main/data/pizza_steak_sushi.zip",
    destination="pizza_steak_sushi"
)

data_20_percent_path = download_data(
    source="https://github.com/vinsmoke-three/deeplearning-with-pytorch/raw/main/data/pizza_steak_sushi_20_percent.zip",
    destination="pizza_steak_sushi_20_percent"
)

# 訓練ディレクトリパスの設定
train_dir_10_percent = data_10_percent_path / "train"
train_dir_20_percent = data_20_percent_path / "train"

# テストディレクトリパスの設定（注意：結果を比較するため両方で同じテストデータセットを使用）
test_dir = data_10_percent_path / "test"

# ディレクトリを確認
print(f"訓練ディレクトリ 10%: {train_dir_10_percent}")
print(f"訓練ディレクトリ 20%: {train_dir_20_percent}")
print(f"テストディレクトリ: {test_dir}")

from torchvision import transforms

# データ分布をImageNetと一致させるための変換を作成
normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406], # 色チャンネルごとの値 [red, green, blue]
                                 std=[0.229, 0.224, 0.225]) # 色チャンネルごとの値 [red, green, blue]

# 変換をパイプラインに構成
simple_transform = transforms.Compose([
    transforms.Resize((224, 224)), # 1. 画像をリサイズ
    transforms.ToTensor(), # 2. 画像を0から1の値を持つテンソルに変換
    normalize # 3. ImageNetデータセットと分布が一致するように画像を正規化
])

BATCH_SIZE = 32

# 10%訓練・テストDataLoaderを作成
train_dataloader_10_percent, test_dataloader, class_names = data_setup.create_dataloaders(
    train_dir=train_dir_10_percent,
    test_dir=test_dir, 
    transform=simple_transform,
    batch_size=BATCH_SIZE
)

# 20%訓練・テストDataLoaderを作成
train_dataloader_20_percent, test_dataloader, class_names = data_setup.create_dataloaders(
    train_dir=train_dir_20_percent,
    test_dir=test_dir,
    transform=simple_transform,
    batch_size=BATCH_SIZE
)

# 各dataloader内のサンプル数/バッチ数を確認（両実験で同じtest_dataloaderを使用）
print(f"10%訓練データ内のサイズ{BATCH_SIZE}のバッチ数: {len(train_dataloader_10_percent)}")
print(f"20%訓練データ内のサイズ{BATCH_SIZE}のバッチ数: {len(train_dataloader_20_percent)}")
print(f"テストデータ内のサイズ{BATCH_SIZE}のバッチ数: {len(test_dataloader)} （すべての実験で同じテストセットを使用）")
print(f"クラス数: {len(class_names)}、クラス名: {class_names}")

import torchvision
from torch import nn

# 出力特徴数を取得（ピザ、ステーキ、寿司の各クラスに1つずつ）
OUT_FEATURES = len(class_names)

# EffNetB0特徴抽出器を作成
def create_effnetb0():
    # 1. 事前訓練重みでベースモデルを取得し、ターゲットデバイスに送信
    weights = torchvision.models.EfficientNet_B0_Weights.DEFAULT
    model = torchvision.models.efficientnet_b0(weights=weights).to(device)

    # 2. ベースモデルレイヤーをフリーズ
    for param in model.features.parameters():
        param.requires_grad = False

    # 3. シードを設定
    set_seeds()

    # 4. 分類ヘッドを変更
    model.classifier = nn.Sequential(
        nn.Dropout(p=0.2),
        nn.Linear(in_features=1280, out_features=OUT_FEATURES)
    ).to(device)

    # 5. モデルに名前を付ける
    model.name = "effnetb0"
    print(f"[INFO] 新しい{model.name}モデルを作成しました。")
    return model

# EffNetB2特徴抽出器を作成
def create_effnetb2():
    # 1. 事前訓練重みでベースモデルを取得し、ターゲットデバイスに送信
    weights = torchvision.models.EfficientNet_B2_Weights.DEFAULT
    model = torchvision.models.efficientnet_b2(weights=weights).to(device)

    # 2. ベースモデルレイヤーをフリーズ
    for param in model.features.parameters():
        param.requires_grad = False

    # 3. シードを設定
    set_seeds()

    # 4. 分類ヘッドを変更
    model.classifier = nn.Sequential(
        nn.Dropout(p=0.3),
        nn.Linear(in_features=1408, out_features=OUT_FEATURES)
    ).to(device)

    # 5. モデルに名前を付ける
    model.name = "effnetb2"
    print(f"[INFO] 新しい{model.name}モデルを作成しました。")
    return model

effnetb0 = create_effnetb0() 

# EffNetB0特徴抽出器モデルのレイヤーの出力要約を取得（完全な出力を表示するにはコメントアウトを外す）
# summary(model=effnetb0, 
#         input_size=(32, 3, 224, 224), # これは"input_size"であることを確認、"input_shape"ではない
#         # col_names=["input_size"], # より小さな出力にはコメントアウトを外す
#         col_names=["input_size", "output_size", "num_params", "trainable"],
#         col_width=20,
#         row_settings=["var_names"]
# ) 

effnetb2 = create_effnetb2()

# EffNetB2特徴抽出器モデルのレイヤーの出力要約を取得（完全な出力を表示するにはコメントアウトを外す）
# summary(model=effnetb2, 
#         input_size=(32, 3, 224, 224), # これは"input_size"であることを確認、"input_shape"ではない
#         # col_names=["input_size"], # より小さな出力にはコメントアウトを外す
#         col_names=["input_size", "output_size", "num_params", "trainable"],
#         col_width=20,
#         row_settings=["var_names"]
# ) 

# 1. エポックリストを作成
num_epochs = [5, 10]

# 2. モデルリストを作成（各実験用に新しいモデルを作成する必要がある）
models = ["effnetb0", "effnetb2"]

# 3. 様々なdataloaderのためのdataloaders辞書を作成
train_dataloaders = {"data_10_percent": train_dataloader_10_percent,
                     "data_20_percent": train_dataloader_20_percent}

%%time
from going_modular.utils import save_model

# 1. ランダムシードを設定
set_seeds(seed=42)

# 2. 実験番号を追跡
experiment_number = 0

# 3. 各DataLoaderをループ
for dataloader_name, train_dataloader in train_dataloaders.items():

    # 4. 各エポック数をループ
    for epochs in num_epochs: 

        # 5. 各モデル名をループし、名前に基づいて新しいモデルを作成
        for model_name in models:

            # 6. 情報印刷出力を作成
            experiment_number += 1
            print(f"[INFO] 実験番号: {experiment_number}")
            print(f"[INFO] モデル: {model_name}")
            print(f"[INFO] DataLoader: {dataloader_name}")
            print(f"[INFO] エポック数: {epochs}")  

            # 7. モデルを選択
            if model_name == "effnetb0":
                model = create_effnetb0() # 毎回新しいモデルを作成（各実験がスクラッチから始まるため重要）
            else:
                model = create_effnetb2() # 毎回新しいモデルを作成（各実験がスクラッチから始まるため重要）

            # 8. すべてのモデル用に新しい損失とオプティマイザーを作成
            loss_fn = nn.CrossEntropyLoss()
            optimizer = torch.optim.Adam(params=model.parameters(), lr=0.001)

            # 9. ターゲットDataLoaderでターゲットモデルを訓練し、実験を追跡
            train(model=model,
                  train_dataloader=train_dataloader,
                  test_dataloader=test_dataloader, 
                  optimizer=optimizer,
                  loss_fn=loss_fn,
                  epochs=epochs,
                  device=device,
                  writer=create_writer(experiment_name=dataloader_name,
                                       model_name=model_name,
                                       extra=f"{epochs}_epochs"))

            # 10. 最高のモデルを戻せるようにモデルをファイルに保存
            save_filepath = f"07_{model_name}_{dataloader_name}_{epochs}_epochs.pth"
            save_model(model=model,
                       target_dir="models",
                       model_name=save_filepath)
            print("-"*50 + "\n")

# Jupyter/Colabノートブックでの場合
# %load_ext tensorboard
# %tensorboard --logdir runs

# 最高のモデルファイルパスを設定
best_model_path = "models/08_effnetb2_data_20_percent_10_epochs.pth"

# 新しいEffNetB2のインスタンスを作成（保存されたstate_dict()をロードするため）
best_model = create_effnetb2()

# 保存された最高のモデルstate_dict()をロード
best_model.load_state_dict(torch.load(best_model_path))

# モデルファイルサイズを確認
from pathlib import Path

# モデルサイズをバイトで取得してからメガバイトに変換
effnetb2_model_size = Path(best_model_path).stat().st_size // (1024*1024)
print(f"EfficientNetB2特徴抽出器モデルサイズ: {effnetb2_model_size} MB")

# 画像で予測を行いプロットする関数をインポート
from going_modular.predictions import pred_and_plot_image

# 20%テストセットからランダムに3つの画像のリストを取得
import random
num_images_to_plot = 3
test_image_path_list = list(Path(data_20_percent_path / "test").glob("*/*.jpg")) # 20%データセットからすべてのテスト画像パスを取得
test_image_path_sample = random.sample(population=test_image_path_list,
                                       k=num_images_to_plot) # k個の画像をランダムに選択

# ランダムなテスト画像パスを反復し、予測を行いプロット
for image_path in test_image_path_sample:
    pred_and_plot_image(model=best_model,
                        image_path=image_path,
                        class_names=class_names,
                        image_size=(224, 224))

# カスタム画像パス
custom_image_path = Path("data/sushi.jpg")

# カスタム画像で予測
pred_and_plot_image(model=best_model,
                    image_path=custom_image_path,
                    class_names=class_names)