PyTorch モジュール化とスクリプトモード¶

%%writefile hello_world.py
print("hello world, machine learning is fun!")

import os

os.makedirs("going_modular", exist_ok=True)

import os
import zipfile
from pathlib import Path
import requests

# データフォルダのパス設定
data_path = Path("data/")
image_path = data_path / "pizza_steak_sushi"

# 画像フォルダが存在しない場合、ダウンロードして準備
if image_path.is_dir():
    print(f"{image_path} ディレクトリが存在します。")
else:
    print(f"{image_path} ディレクトリが見つかりません。作成中...")
    image_path.mkdir(parents=True, exist_ok=True)

# pizza、steak、sushiデータのダウンロード
with open(data_path / "pizza_steak_sushi.zip", "wb") as f:
    request = requests.get("https://github.com/vinsmoke-three/deeplearning-with-pytorch/raw/main/data/pizza_steak_sushi.zip")
    print("pizza、steak、sushiデータをダウンロード中...")
    f.write(request.content)

# zipファイルの解凍
with zipfile.ZipFile(data_path / "pizza_steak_sushi.zip", "r") as zip_ref:
    print("pizza、steak、sushiデータを解凍中...") 
    zip_ref.extractall(image_path)

# zipファイルの削除
os.remove(data_path / "pizza_steak_sushi.zip")

# 訓練用およびテスト用パスの設定
train_dir = image_path / "train"
test_dir = image_path / "test"

print(f"訓練データパス: {train_dir}")
print(f"テストデータパス: {test_dir}")

from torchvision import datasets, transforms

# シンプルな変換処理を作成
data_transform = transforms.Compose([ 
    transforms.Resize((64, 64)),    # 画像を64x64にリサイズ
    transforms.ToTensor(),          # テンソルに変換
])

# ImageFolderを使用してデータセットを作成
train_data = datasets.ImageFolder(root=train_dir,           # 画像の対象フォルダ
                                  transform=data_transform,  # 画像に適用する変換
                                  target_transform=None)     # ラベルに適用する変換（必要に応じて）

test_data = datasets.ImageFolder(root=test_dir, 
                                 transform=data_transform)

print(f"訓練データ:\n{train_data}\nテストデータ:\n{test_data}")

# クラス名をリストとして取得
class_names = train_data.classes
print(f"クラス名: {class_names}")

# クラス名と数値の辞書も取得可能
class_dict = train_data.class_to_idx
print(f"クラス辞書: {class_dict}")

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

from torch.utils.data import DataLoader

train_dataloader = DataLoader(dataset=train_data, 
                              batch_size=1,     # バッチあたりのサンプル数
                              num_workers=0,    # データローディング用のサブプロセス数
                              shuffle=True)     # データをシャッフルするか

test_dataloader = DataLoader(dataset=test_data, 
                             batch_size=1, 
                             num_workers=0, 
                             shuffle=False)    # テストデータは通常シャッフル不要

print(f"DataLoader作成完了")

# 単一画像のサイズ/形状を確認
img, label = next(iter(train_dataloader))

print(f"画像形状: {img.shape} -> [batch_size, color_channels, height, width]")
print(f"ラベル形状: {label.shape}")

%%writefile going_modular/data_setup.py
"""
画像分類データ用のPyTorch DataLoaderを作成する機能を提供します。
"""
import os

from torch.utils.data import DataLoader
from torchvision import datasets, transforms

NUM_WORKERS = os.cpu_count()

def create_dataloaders(
    train_dir: str, 
    test_dir: str, 
    transform: transforms.Compose, 
    batch_size: int, 
    num_workers: int=NUM_WORKERS
):
  """訓練用および評価用DataLoaderを作成します。

  訓練ディレクトリとテストディレクトリのパスを受け取り、
  PyTorch DatasetおよびDataLoaderに変換します。

  Args:
    train_dir: 訓練ディレクトリのパス
    test_dir: テストディレクトリのパス
    transform: 訓練・テストデータに適用するtorchvision変換
    batch_size: 各DataLoaderのバッチあたりのサンプル数
    num_workers: DataLoaderあたりのワーカー数

  Returns:
    (train_dataloader, test_dataloader, class_names)のタプル
    class_namesは対象クラスのリスト

    使用例:
      train_dataloader, test_dataloader, class_names = \
        = create_dataloaders(train_dir=path/to/train_dir,
                             test_dir=path/to/test_dir,
                             transform=some_transform,
                             batch_size=32,
                             num_workers=4)
  """
  # ImageFolderを使用してデータセットを作成
  train_data = datasets.ImageFolder(train_dir, transform=transform)
  test_data = datasets.ImageFolder(test_dir, transform=transform)

  # クラス名を取得
  class_names = train_data.classes

  # 画像をDataLoaderに変換
  train_dataloader = DataLoader(
      train_data,
      batch_size=batch_size,
      shuffle=True,
      num_workers=num_workers,
      pin_memory=True,
  )
  test_dataloader = DataLoader(
      test_data,
      batch_size=batch_size,
      shuffle=False,
      num_workers=num_workers,
      pin_memory=True,
  )

  return train_dataloader, test_dataloader, class_names

import torch
from torch import nn 

class TinyVGG(nn.Module):
    """TinyVGGアーキテクチャを作成します。

    CNN explainerウェブサイトのTinyVGGアーキテクチャをPyTorchで再現します。
    元のアーキテクチャ: https://poloclub.github.io/cnn-explainer/

    Args:
    input_shape: 入力チャンネル数を表す整数
    hidden_units: 層間の隠れユニット数を表す整数
    output_shape: 出力ユニット数を表す整数
    """
    def __init__(self, input_shape: int, hidden_units: int, output_shape: int) -> None:
        super().__init__()
        # 第1畳み込みブロック
        self.conv_block_1 = nn.Sequential(
          nn.Conv2d(in_channels=input_shape, 
                    out_channels=hidden_units, 
                    kernel_size=3, 
                    stride=1, 
                    padding=0),  
          nn.ReLU(),
          nn.Conv2d(in_channels=hidden_units, 
                    out_channels=hidden_units,
                    kernel_size=3,
                    stride=1,
                    padding=0),
          nn.ReLU(),
          nn.MaxPool2d(kernel_size=2, stride=2)
        )
        # 第2畳み込みブロック
        self.conv_block_2 = nn.Sequential(
          nn.Conv2d(hidden_units, hidden_units, kernel_size=3, padding=0),
          nn.ReLU(),
          nn.Conv2d(hidden_units, hidden_units, kernel_size=3, padding=0),
          nn.ReLU(),
          nn.MaxPool2d(2)
        )
        # 分類器
        self.classifier = nn.Sequential(
          nn.Flatten(),
          # このin_features形状はどこから来るのか？
          # ネットワークの各層が入力データの形状を圧縮・変更するため
          nn.Linear(in_features=hidden_units*13*13,
                    out_features=output_shape)
        )

    def forward(self, x: torch.Tensor):
        x = self.conv_block_1(x)
        x = self.conv_block_2(x)
        x = self.classifier(x)
        return x

import torch

device = "mps" if torch.mps.is_available() else "cpu"

# モデルのインスタンスを作成
torch.manual_seed(42)
model_0 = TinyVGG(input_shape=3,                     # カラーチャンネル数（RGBなので3）
                  hidden_units=10, 
                  output_shape=len(train_data.classes)).to(device)
print(model_0)

# 1. DataLoaderから画像とラベルのバッチを取得
img_batch, label_batch = next(iter(train_dataloader))

# 2. バッチから単一画像を取得し、モデルに適合する形状にunsqueeze
img_single, label_single = img_batch[0].unsqueeze(dim=0), label_batch[0]
print(f"単一画像形状: {img_single.shape}\n")

# 3. 単一画像で順伝播を実行
model_0.eval()
with torch.inference_mode():
    pred = model_0(img_single.to(device))

# 4. 結果を出力し、モデルlogits -> 予測確率 -> 予測ラベルに変換
print(f"出力logits:\n{pred}\n")
print(f"出力予測確率:\n{torch.softmax(pred, dim=1)}\n")
print(f"出力予測ラベル:\n{torch.argmax(torch.softmax(pred, dim=1), dim=1)}\n")
print(f"実際のラベル:\n{label_single}")

%%writefile going_modular/model_builder.py
"""
TinyVGGモデルをインスタンス化するPyTorchモデルコードを含みます。
"""
import torch
from torch import nn

class TinyVGG(nn.Module):
    """TinyVGGアーキテクチャを作成します。

    CNN explainerウェブサイトのTinyVGGアーキテクチャをPyTorchで再現します。
    元のアーキテクチャ: https://poloclub.github.io/cnn-explainer/

    Args:
    input_shape: 入力チャンネル数を表す整数
    hidden_units: 層間の隠れユニット数を表す整数
    output_shape: 出力ユニット数を表す整数
    """
    def __init__(self, input_shape: int, hidden_units: int, output_shape: int) -> None:
        super().__init__()
        self.conv_block_1 = nn.Sequential(
          nn.Conv2d(in_channels=input_shape, 
                    out_channels=hidden_units, 
                    kernel_size=3, 
                    stride=1, 
                    padding=0),  
          nn.ReLU(),
          nn.Conv2d(in_channels=hidden_units, 
                    out_channels=hidden_units,
                    kernel_size=3,
                    stride=1,
                    padding=0),
          nn.ReLU(),
          nn.MaxPool2d(kernel_size=2, stride=2)
        )
        self.conv_block_2 = nn.Sequential(
          nn.Conv2d(hidden_units, hidden_units, kernel_size=3, padding=0),
          nn.ReLU(),
          nn.Conv2d(hidden_units, hidden_units, kernel_size=3, padding=0),
          nn.ReLU(),
          nn.MaxPool2d(2)
        )
        # auto calculate flatten_size
        with torch.no_grad():
            dummy_input = torch.zeros(1, input_shape, image_size, image_size)
            x = self.conv_block_1(dummy_input)
            x = self.conv_block_2(x)
            flatten_size = x.view(1, -1).shape[1]  # or flatten_size = x[0].numel()

        self.classifier = nn.Sequential(
          nn.Flatten(),
          # このin_features形状の由来について：
          # ネットワークの各層が入力データの形状を圧縮・変更するため
          nn.Linear(in_features=flatten_size,
                    out_features=output_shape)
        )

    def forward(self, x: torch.Tensor):
        x = self.conv_block_1(x)
        x = self.conv_block_2(x)
        x = self.classifier(x)
        return x

import torch
from going_modular import model_builder

device = "mps" if torch.mps.is_available() else "cpu"

# "model_builder.py"スクリプトからモデルのインスタンスを作成
torch.manual_seed(42)
model_1 = model_builder.TinyVGG(input_shape=3,                   # カラーチャンネル数（RGBなので3）
                                hidden_units=10, 
                                output_shape=len(class_names)).to(device)
print(model_1)

%%writefile going_modular/engine.py
"""
PyTorchモデルの訓練・テスト用関数を含みます。
"""
from typing import Dict, List, Tuple

import torch
from tqdm import tqdm

def train_step(model: torch.nn.Module, 
               dataloader: torch.utils.data.DataLoader, 
               loss_fn: torch.nn.Module, 
               optimizer: torch.optim.Optimizer,
               device: torch.device) -> Tuple[float, float]:
    """PyTorchモデルを1エポック訓練します。

    対象PyTorchモデルを訓練モードにし、必要な訓練ステップ
    （順伝播、損失計算、オプティマイザーステップ）を実行します。

    Args:
    model: 訓練するPyTorchモデル
    dataloader: モデル訓練用のDataLoaderインスタンス
    loss_fn: 最小化するPyTorch損失関数
    optimizer: 損失関数最小化を支援するPyTorchオプティマイザー
    device: 計算対象デバイス（例："cuda"または"cpu"）

    Returns:
    訓練損失と訓練精度メトリクスのタプル
    形式：(train_loss, train_accuracy)。例：(0.1112, 0.8743)
    """
    # モデルを訓練モードに設定
    model.train()

    # 訓練損失と訓練精度の値を設定
    train_loss, train_acc = 0, 0

    # データローダーのデータバッチをループ
    for batch, (X, y) in enumerate(dataloader):
        # データを対象デバイスに送信
        X, y = X.to(device), y.to(device)

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

        # 2. 損失を計算・蓄積
        loss = loss_fn(y_pred, y)
        train_loss += loss.item() 

        # 3. オプティマイザーのゼロ勾配
        optimizer.zero_grad()

        # 4. 損失の逆伝播
        loss.backward()

        # 5. オプティマイザーステップ
        optimizer.step()

        # 全バッチで精度メトリクスを計算・蓄積
        y_pred_class = torch.argmax(torch.softmax(y_pred, dim=1), dim=1)
        train_acc += (y_pred_class == y).sum().item()/len(y_pred)

    # バッチあたりの平均損失と精度を取得するようメトリクスを調整
    train_loss = train_loss / len(dataloader)
    train_acc = train_acc / len(dataloader)
    return train_loss, train_acc

def test_step(model: torch.nn.Module, 
              dataloader: torch.utils.data.DataLoader, 
              loss_fn: torch.nn.Module,
              device: torch.device) -> Tuple[float, float]:
    """PyTorchモデルを1エポックテストします。

    対象PyTorchモデルを"eval"モードにし、テストデータセットで
    順伝播を実行します。

    Args:
    model: テストするPyTorchモデル
    dataloader: モデルテスト用のDataLoaderインスタンス
    loss_fn: テストデータの損失計算用PyTorch損失関数
    device: 計算対象デバイス（例："cuda"または"cpu"）

    Returns:
    テスト損失とテスト精度メトリクスのタプル
    形式：(test_loss, test_accuracy)。例：(0.0223, 0.8985)
    """
    # モデルを評価モードに設定
    model.eval() 

    # テスト損失とテスト精度の値を設定
    test_loss, test_acc = 0, 0

    # 推論コンテキストマネージャーをオン
    with torch.inference_mode():
        # DataLoaderバッチをループ
        for batch, (X, y) in enumerate(dataloader):
            # データを対象デバイスに送信
            X, y = X.to(device), y.to(device)

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

            # 2. 損失を計算・蓄積
            loss = loss_fn(test_pred_logits, y)
            test_loss += loss.item()

            # 精度を計算・蓄積
            test_pred_labels = test_pred_logits.argmax(dim=1)
            test_acc += ((test_pred_labels == y).sum().item()/len(test_pred_labels))

    # バッチあたりの平均損失と精度を取得するようメトリクスを調整
    test_loss = test_loss / len(dataloader)
    test_acc = test_acc / len(dataloader)
    return test_loss, test_acc

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) -> Dict[str, List[float]]:
    """PyTorchモデルを訓練・テストします。

    対象PyTorchモデルをtrain_step()およびtest_step()関数を通じて
    指定エポック数分実行し、同一エポックループ内でモデルを
    訓練・テストします。

    評価メトリクスを計算、出力、保存します。

    Args:
    model: 訓練・テストするPyTorchモデル
    train_dataloader: モデル訓練用DataLoaderインスタンス
    test_dataloader: モデルテスト用DataLoaderインスタンス
    optimizer: 損失関数最小化を支援するPyTorchオプティマイザー
    loss_fn: 両データセットの損失計算用PyTorch損失関数
    epochs: 訓練エポック数を表す整数
    device: 計算対象デバイス（例："cuda"または"cpu"）

    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)

    # エポック終了時に入力された結果を返す
    return results

%%writefile going_modular/utils.py
"""
PyTorchモデル訓練・保存用の各種ユーティリティ関数を含みます。
"""
from pathlib import Path
import torch

def save_model(model: torch.nn.Module,
               target_dir: str,
               model_name: str):
    """PyTorchモデルを対象ディレクトリに保存します。

    Args:
    model: 保存する対象PyTorchモデル
    target_dir: モデル保存用ディレクトリ
    model_name: 保存モデルのファイル名。".pth"または".pt"の
      ファイル拡張子を含む必要があります

    使用例:
    save_model(model=model_0,
               target_dir="models",
               model_name="05_going_modular_tingvgg_model.pth")
    """
    # 対象ディレクトリを作成
    target_dir_path = Path(target_dir)
    target_dir_path.mkdir(parents=True, exist_ok=True)

    # モデル保存パスを作成
    assert model_name.endswith(".pth") or model_name.endswith(".pt"), "model_nameは'.pt'または'.pth'で終わる必要があります"
    model_save_path = target_dir_path / model_name

    # モデルのstate_dict()を保存
    print(f"[INFO] モデルを以下に保存: {model_save_path}")
    torch.save(obj=model.state_dict(), f=model_save_path)

# ランダムシードを設定
torch.manual_seed(42) 
torch.cuda.manual_seed(42)

# エポック数を設定
NUM_EPOCHS = 5

# TinyVGGのインスタンスを再作成
model_0 = TinyVGG(input_shape=3,                     # カラーチャンネル数（RGBなので3）
                  hidden_units=10, 
                  output_shape=len(train_data.classes)).to(device)

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

# タイマーを開始
from timeit import default_timer as timer 
start_time = timer()

# model_0を訓練
model_0_results = train(model=model_0, 
                        train_dataloader=train_dataloader,
                        test_dataloader=test_dataloader,
                        optimizer=optimizer,
                        loss_fn=loss_fn, 
                        epochs=NUM_EPOCHS,
                        device=device)

# タイマーを終了し、所要時間を出力
end_time = timer()
print(f"[INFO] 総訓練時間: {end_time-start_time:.3f}秒")

# モデルを保存
save_model(model=model_0,
           target_dir="models",
           model_name="05_going_modular_cell_mode_tinyvgg_model.pth")

%%writefile going_modular/train.py
"""
デバイス非依存コードを使用してPyTorch画像分類モデルを訓練します。
"""

import os
import torch
from torchvision import transforms
import data_setup, engine, model_builder, utils

# ハイパーパラメータの設定
NUM_EPOCHS = 5
BATCH_SIZE = 32
HIDDEN_UNITS = 10
LEARNING_RATE = 0.001

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

# 対象デバイスの設定
device = "mps" if torch.mps.is_available() else "cpu"

# 変換処理を作成
data_transform = transforms.Compose([
  transforms.Resize((64, 64)),
  transforms.ToTensor()
])

# data_setup.pyを活用してDataLoaderを作成
train_dataloader, test_dataloader, class_names = data_setup.create_dataloaders(
    train_dir=train_dir,
    test_dir=test_dir,
    transform=data_transform,
    batch_size=BATCH_SIZE
)

# model_builder.pyを活用してモデルを作成
model = model_builder.TinyVGG(
    input_shape=3,
    hidden_units=HIDDEN_UNITS,
    output_shape=len(class_names)
).to(device)

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

# engine.pyを活用して訓練を開始
engine.train(model=model,
             train_dataloader=train_dataloader,
             test_dataloader=test_dataloader,
             loss_fn=loss_fn,
             optimizer=optimizer,
             epochs=NUM_EPOCHS,
             device=device)

# utils.pyを活用してモデルを保存
utils.save_model(model=model,
                 target_dir="models",
                 model_name="05_going_modular_script_mode_tinyvgg_model.pth")