転移学習(書きかけ）

まずクローラーでデータを集める
ディレクトリのプレフィクスは以下の通り

arai→あらいぐま
tanuki→タヌキ
train→トレーニングデータ
val→検証データ

#ディレクトリを作ります

!md tanuki_train
!md tanuki_val
!md arai_train
!md arai_val

Bing用クローラーのモジュールをインポート

以下はdはディレクトリ
kdは検索キーワード
これをディレクトリと検索キーワード変えて
実行する

from icrawler.builtin import BingImageCrawler
d="tanuki_tarin"#ディレクトリ
kd="たぬき"#検索キーワード
bing_crawler = BingImageCrawler(
    downloader_threads=4,    # ダウンローダーのスレッド数
    storage={'root_dir': d}) # ダウンロード先のディレクトリ名

# クロール（キーワード検索による画像収集）の実行

bing_crawler.crawl(
    keyword=kd,   # 検索キーワード（日本語もOK）
    max_num=100)  # ダウンロードする画像の最大枚数

パッケージのimport

import glob
import os.path as osp
import random
import numpy as np
import json
from PIL import Image
from tqdm import tqdm
import matplotlib.pyplot as plt
%matplotlib inline

import torch
import torch.nn as nn
import torch.optim as optim
import torch.utils.data as data
import torchvision
from torchvision import models, transforms

Transform作成

class ImageTransform():
def init(self, resize, mean, std):
self.data_transform = {
'train': transforms.Compose([
transforms.RandomResizedCrop(
resize, scale=(0.5, 1.0)), # データオーギュメンテーション
transforms.RandomHorizontalFlip(), # データオーギュメンテーション
transforms.ToTensor(), # テンソルに変換
transforms.Normalize(mean, std) # 標準化
]),
'val': transforms.Compose([
transforms.Resize(resize), # リサイズ
transforms.CenterCrop(resize), # 画像中央をresize×resizeで切り取り
transforms.ToTensor(), # テンソルに変換
transforms.Normalize(mean, std) # 標準化
])
}

def __call__(self, img, phase='train'):
    return self.data_transform[phase](img)

1. 画像読み込み

image_file_path = './arai_trarin/000001.jpg'
img = Image.open(image_file_path) # [高さ][幅][色RGB]

2. 元の画像の表示

plt.imshow(img)
plt.show()

3. 画像の前処理と処理済み画像の表示

size = 224
mean = (0.485, 0.456, 0.406)
std = (0.229, 0.224, 0.225)

transform = ImageTransform(size, mean, std)
img_transformed = transform(img, phase="train") # torch.Size([3, 224, 224])

(色、高さ、幅)を (高さ、幅、色)に変換し、0-1に値を制限して表示

img_transformed = img_transformed.numpy().transpose((1, 2, 0))
img_transformed = np.clip(img_transformed, 0, 1)
plt.imshow(img_transformed)
plt.show()

def make_datapath_list(phase="train"):
path_train=["arai_train","tanuki_train"]
path_val=["arai_val","tanuki_val"]
path_list=[]
if phase=="train":
path_=path_train
else:
path_=path_val
# globを利用してサブディレクトリまでファイルパスを取得する
for target_path in path_:
for path in glob.glob(target_path+"/*.jpg"):
path_list.append(path)
return path_list

実行確認

train_list = make_datapath_list(phase="train")
val_list = make_datapath_list(phase="val")

print("train_list=",train_list)
print("train_list=",val_list)

class TeniDataset(data.Dataset):
"""
画像のDatasetクラス。PyTorchのDatasetクラスを継承。

Attributes
----------
file_list : リスト
    画像のパスを格納したリスト
transform : object
    前処理クラスのインスタンス
phase : 'train' or 'val'
    訓練か検証かを設定
"""

def __init__(self, file_list, transform=None, phase='train'):
    self.file_list = file_list  # ファイルパスのリスト
    self.transform = transform  # 前処理クラスのインスタンス
    self.phase = phase  # train or valの指定

def __len__(self):
    '''画像の枚数を返す'''
    return len(self.file_list)

def __getitem__(self, index):
    # index番目の画像をロード
    img_path = self.file_list[index]
    img = Image.open(img_path)  # [高さ][幅][色RGB]

    # 画像の前処理を実施
    img_transformed = self.transform(
        img, self.phase)  # torch.Size([3, 224, 224])
    print("img_path=",img_path)

    if img_path[0:2] in 'arai':
        label=0
    elif img_path[0:4] in 'tanuki':
        label=1
    else:
        label=2
        print("label=",label)
    return img_transformed,label

###------------------
size = 224
mean = (0.485, 0.456, 0.406)
std = (0.229, 0.224, 0.225)

実行

train_dataset = TeniDataset(
file_list=train_list, transform=ImageTransform(size, mean, std), phase='train')

val_dataset = TeniDataset(
file_list=val_list, transform=ImageTransform(size, mean, std), phase='val')

動作確認

index = 0
print(train_dataset.getitem(index)[0].size())
print(train_dataset.getitem(index)[1])
#---------------------------------------------------------------------

ミニバッチのサイズを指定

batch_size = 32

DataLoaderを作成

train_dataloader = torch.utils.data.DataLoader(
train_dataset, batch_size=batch_size, shuffle=True)

val_dataloader = torch.utils.data.DataLoader(
val_dataset, batch_size=batch_size, shuffle=False)

辞書型変数にまとめる

dataloaders_dict = {"train": train_dataloader, "val": val_dataloader}

動作確認

batch_iterator = iter(dataloaders_dict["train"]) # イテレータに変換
inputs, labels = next(
batch_iterator) # 1番目の要素を取り出す
print(inputs.size())
print(labels)

学習済みのVGG-16モデルをロード

VGG-16モデルのインスタンスを生成

use_pretrained = True # 学習済みのパラメータを使用
net = models.vgg16(pretrained=use_pretrained)

VGG16の最後の出力層の出力ユニットをアリとハチの2つに付け替える

net.classifier[6] = nn.Linear(in_features=4096, out_features=2)

訓練モードに設定

net.train()

print('ネットワーク設定完了：学習済みの重みをロードし、訓練モードに設定しました')

損失関数の設定

criterion = nn.CrossEntropyLoss()

転移学習で学習させるパラメータを、変数params_to_updateに格納する

params_to_update = []

学習させるパラメータ名

update_param_names = ["classifier.6.weight", "classifier.6.bias"]

学習させるパラメータ以外は勾配計算をなくし、変化しないように設定

for name, param in net.named_parameters():
if name in update_param_names:
param.requires_grad = True
params_to_update.append(param)
print(name)
else:
param.requires_grad = False

params_to_updateの中身を確認

print("-----------")
print(params_to_update)

最適化手法の設定

optimizer = optim.SGD(params=params_to_update, lr=0.001, momentum=0.9)

モデルを学習させる関数を作成

def train_model(net, dataloaders_dict, criterion, optimizer, num_epochs):

# epochのループ
for epoch in range(num_epochs):
    print('Epoch {}/{}'.format(epoch+1, num_epochs))
    print('-------------')

    # epochごとの学習と検証のループ
    for phase in ['train', 'val']:
        if phase == 'train':
            net.train()  # モデルを訓練モードに
        else:
            net.eval()   # モデルを検証モードに

        epoch_loss = 0.0  # epochの損失和
        epoch_corrects = 0  # epochの正解数

        # 未学習時の検証性能を確かめるため、epoch=0の訓練は省略
        if (epoch == 0) and (phase == 'train'):
            continue

        # データローダーからミニバッチを取り出すループ
        for inputs, labels in tqdm(dataloaders_dict[phase]):

            # optimizerを初期化
            optimizer.zero_grad()

            # 順伝搬（forward）計算
            with torch.set_grad_enabled(phase == 'train'):
                outputs = net(inputs)
                loss = criterion(outputs, labels)  # 損失を計算
                _, preds = torch.max(outputs, 1)  # ラベルを予測
                

                # 訓練時はバックプロパゲーション
                if phase == 'train':
                    loss.backward()
                    optimizer.step()

                # イタレーション結果の計算
                # lossの合計を更新
                epoch_loss += loss.item() * inputs.size(0)  
                # 正解数の合計を更新
                epoch_corrects += torch.sum(preds == labels.data)

        # epochごとのlossと正解率を表示
        epoch_loss = epoch_loss / len(dataloaders_dict[phase].dataset)
        epoch_acc = epoch_corrects.double(
        ) / len(dataloaders_dict[phase].dataset)

        print('{} Loss: {:.4f} Acc: {:.4f}'.format(
            phase, epoch_loss, epoch_acc))

学習・検証を実行する

num_epochs=2
train_model(net, dataloaders_dict, criterion, optimizer, num_epochs=num_epochs)