PyTorchを使ってDeep Learningのお勉強 画像認識編（MNIST）

今回からいよいよ画像認識編です

ようやくこの辺りから、Xi IoTの実装に近づいてきそうです
（カメラで撮った画像を解析する的な？）

PyTorch お勉強シリーズ
第1回 PyTorchを使ってDeep Learningのお勉強 基礎編
第2回 PyTorchを使ったDeep Learningのお勉強 PyTorch Lightning編
第3回 PyTorchを使ったDeep Learningのお勉強 TensorBoard編
第4回 PyTorchを使ったDeep Learningのお勉強 画像認識編（MNIST）（イマココ）
第5回 PyTorchを使ったDeep Learningのお勉強 画像認識編（CIFAR-10）

というわけでディープラーニングの世界で画像処理といえば、CNN！
（いや他の手法もあるんですが）
CNN（Convolutional Neural Network）がどういうものか、、、という説明はググる方向にお任せします
またCNNをベースとして派生形（進化型）はいろいろあるのですが、
まずはCNNの基礎となる畳み込み、プーリング、全結合層からなる処理を実装して
特徴量を抽出し、画像の認識をして分類を行ってみたいと思います

たまには環境をちゃんと書いてみる

・Windows 10
・anaconda：4.8.3
・PyTorch：1.4.0
・PyTorch_Lighgning：0.7.3
・GPU：NVIDIA Geforce GTX 1070
　 Cuda cuda_10.2.89_441.22_win10
　 CuDNN v7.6.5.32

CNN大まかな流れを分解して書く

１．データ読み込み（MNIST）

まずは使う画像データは機械学習でド定番の手書きの数字データ（学習用に60,000個、検証用に10,000個）を使います

pytorch datasetでダウンロードをしてくれる模様

import torch, torchvision
import torch.nn as nn
from torchvision import transforms

# 画像ファイルを読み込むための準備(Channel x H x W)
transform = transforms.Compose([
    transforms.ToTensor()
])

# MNISTデータセットのダウンロード
train = torchvision.datasets.MNIST(
    root='.', 
    train=True,
    download=True,
    transform=transform)

print(len(train), type(train))

60,000枚あることが確認できます

Output

60000

画像一枚あたり、1チャネル（白黒）、28 x 28（784画素）なのが分かります

train[0][0].shape

Output

torch.Size([1, 28, 28])

ちょっと画像を表示して、確認してみましょう

import numpy as np
import matplotlib.pyplot as plt

img = np.transpose(train[0][0], (1, 2, 0))
img = img.reshape(img.shape[0], img.shape[1])
print("正解ラベル：", train[0][1])
plt.imshow(img, cmap='gray')

こんな感じの画像と正解ラベルがセットになっています

２．畳み込み（Convolution）

畳み込み層に必要な各ハイパーパラメータを下記で設定してやります
・入力チャネル：1（白黒、カラーRGBなら3になる）
・出力チャネル：4（特徴量を抽出するフィルタの数）
・カーネルサイズ：3 x 3
・ストライド：1（カーネルをずらすサイズ）
・パディング：1（画像の周りを0埋めサイズ）

PyTorchでは nn.Conv2d一つで簡単に定義できます
（手で書くとかなり複雑になります）

# 畳み込み層の定義
conv = nn.Conv2d(
    in_channels=1, 
    out_channels=4,
    kernel_size=3,
    stride=1,
    padding=1)

重みとバイアスの初期値がセットされてます

conv, conv.weight, conv.bias

Output

Conv2d(1, 4, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))

Parameter containing:
tensor([[[[ 0.1061, -0.2007, 0.0837],
[-0.2116, -0.1864, -0.3124],
[ 0.2976, -0.3223, 0.0535]]],

[[[ 0.0756, -0.1733, -0.0979],
[-0.0471, 0.3147, -0.0986],
[ 0.2859, -0.1177, 0.0985]]],

[[[ 0.2931, 0.2343, 0.2719],
[ 0.2293, -0.2391, -0.1068],
[-0.1981, -0.2732, 0.1047]]],

[[[ 0.0633, 0.2615, 0.0747],
[-0.1364, 0.0661, -0.0697],
[-0.1794, -0.2842, -0.2955]]]], requires_grad=True)
tensor([ 0.3227, 0.0761, -0.1957, 0.2157], requires_grad=True)

では、画像一枚だけピックアップして初期値のパラメータのままで一度畳み込んでみます

# とりあえず画像一枚だけ畳み込んでみる（batchsize=1のイメージ）
x = train[0][0] # 一枚だけ xに代入 [1, 28, 28]
x = x.reshape(1, 1, 28, 28) # バッチサイズ, チャネル数, H, W [1, 1, 28, 28] 　
x = conv(x) # [1, 4, 28, 28] # 畳み込み！

4チャネルの画像（4枚の特徴マップ）が出力されています
（カーネルサイズにより、H, Wのサイズは変わります）

x.shape

Output
torch.Size([1, 4, 28, 28])

畳み込みした4枚の特徴マップも確認することもできます

for i in range(4):
    img2 = x[0][i].to('cpu').detach().numpy().copy()
    plt.subplot(1, 5, i+1)
    plt.imshow(img2)
    plt.axis('off')
plt.show()

畳み込み処理ではカーネルを学習させて、目的の特徴量（特徴マップ）を抽出して圧縮していくことになります

３．プーリング（Pooling）

プーリング処理では畳み込まれた特徴マップに統計的処理を施して縮小していきます

MAXプーリングを用いてハイパーパラメータはこれで設定
・カーネルサイズ：2x2
・ストライド：2

# プーリング処理
x = F.max_pool2d(x, kernel_size=2, stride=2)

28x28 から 14x14 に縮小されているのが分かります

x.shape

Output

torch.Size([1, 4, 14, 14])

４．全結合層

抽出された特徴マップを数値として、1つの次元にまとめて
ニューラルネットワークに入力します

チャネル x H x Wが入力値になります

# 全結合層の定義
x_shape = x.shape[1] * x.shape[2] * x.shape[3] # [784]
x = x.view(-1, x_shape) # [1, 784]

x.shape
>>> 出力：
torch.Size([1, 784])

入力に合わせて、出力は10クラス分類（0-9）になるので 10 に設定

fc = nn.Linear(x_shape, 10) # 784 => 10
x = fc(x)

x
>>> 出力：
tensor([[ 0.2561,  0.0623, -0.0122,  0.1757, -0.2418, -0.1730,  0.0067,  0.0611,
          0.0405,  0.1923]], grad_fn=<AddmmBackward>)

10個の数値が出力されました
今回の元の画像「5」だったので、5に対応する 6個目の数値が最も大きくなるように、重みとバイアスを学習させていきます
（今回の出力では1つ目の数値が大きいので「0」で間違い）

pytorch lightningフレームワークでCNN

というところで、一通り CNN の処理過程を分解してみました
ここからは PyTorch Lightning を使って実際に学習させてみたいと思います

まず前半は、同じくデータセットを準備

from torchvision import transforms

transform = transforms.Compose([
    transforms.ToTensor()
])

# データセットの取得
train_val = torchvision.datasets.MNIST(
    root='.',
    train=True,
    download=True,
    transform=transform)

test = torchvision.datasets.MNIST(
    root='.',
    train=False,
    download=True,
    transform=transform)

# train : val = 80% : 20%
n_train = int(len(train_val) * 0.8)
n_val = len(train_val) - n_train

# データをランダムに分割
train, val = torch.utils.data.random_split(train_val, [n_train, n_val])

# 分割後のサンプル数を確認
len(train), len(val), len(test)
>>> 出力：
(48000, 12000, 10000)

PyTorch Lightningの学習モデルの構築
いじるのは、”class Net” に畳み込みとプーリング処理を追加するだけです

import pytorch_lightning as pl
from pytorch_lightning import Trainer
# 学習データ用クラス
class TrainNet(pl.LightningModule):
    
    @pl.data_loader
    def train_dataloader(self):
        return torch.utils.data.DataLoader(train, self.batch_size, shuffle=True)
    
    def training_step(self, batch, batch_nb):
        x, t = batch
        y = self.forward(x)
        loss = self.lossfun(y, t)
        y_label = torch.argmax(y, dim=1)
        acc = torch.sum(t == y_label) * 1.0 / len(t)
        tensorboard_logs = {'train/train_loss': loss, 'train/train_acc': acc} # tensorboard
        results = {'loss': loss, 'log': tensorboard_logs}
        #results = {'loss': loss}
        return results

    
# 検証データ用クラス
class ValidationNet(pl.LightningModule):

    @pl.data_loader
    def val_dataloader(self):
        return torch.utils.data.DataLoader(val, self.batch_size)

    def validation_step(self, batch, batch_nb):
        x, t = batch
        y = self.forward(x)
        loss = self.lossfun(y, t)
        y_label = torch.argmax(y, dim=1)
        acc = torch.sum(t == y_label) * 1.0 / len(t)
        results = {'val_loss': loss, 'val_acc': acc}
        return results

    def validation_end(self, outputs):
        avg_loss = torch.stack([x['val_loss'] for x in outputs]).mean()
        avg_acc = torch.stack([x['val_acc'] for x in outputs]).mean()
        tensorboard_logs = {'val/avg_loss': avg_loss, 'val/avg_acc': avg_acc}
        results = {'val_loss': avg_loss, 'val_acc': avg_acc, 'log': tensorboard_logs}        
        #results = {'val_loss': avg_loss, 'val_acc': avg_acc}
        return results

    
# テストデータ用クラス
class TestNet(pl.LightningModule):

    @pl.data_loader
    def test_dataloader(self):
        return torch.utils.data.DataLoader(test, self.batch_size)

    def test_step(self, batch, batch_nb):
        x, t = batch
        y = self.forward(x)
        loss = self.lossfun(y, t)
        y_label = torch.argmax(y, dim=1)
        acc = torch.sum(t == y_label) * 1.0 / len(t)
        results = {'test_loss': loss, 'test_acc': acc}
        return results

    def test_end(self, outputs):
        avg_loss = torch.stack([x['test_loss'] for x in outputs]).mean()
        avg_acc = torch.stack([x['test_acc'] for x in outputs]).mean()
        results = {'test_loss': avg_loss, 'test_acc': avg_acc}
        return results

    
# 学習データ、検証データ、テストデータクラスの継承クラス
class Net(TrainNet, ValidationNet, TestNet):
    def __init__(self, input_size=784, hidden_size=100, output_size=10, batch_size=256):
        super(Net, self).__init__()
        self.conv = nn.Conv2d(in_channels=1, out_channels=4, kernel_size=3, padding=1, stride=1)
        self.L1 = nn.Linear(input_size, hidden_size)
        self.L2 = nn.Linear(hidden_size, output_size)
        self.batch_size = batch_size
        
    def forward(self, x):
        x = self.conv(x)
        x = F.max_pool2d(x, 2, 2)
        x = x.view(-1, 784)
        x = self.L1(x)
        x = F.relu(x)
        x = self.L2(x)
        return x
    
    def lossfun(self, y, t):
        return F.cross_entropy(y, t)
        #return nn.CrossEntropyLoss(y, t)
    
    def configure_optimizers(self):
        return torch.optim.SGD(self.parameters(), lr=0.1)

さぁ、学習ループのスタート！
の、前に注意点です
この画像認識の学習には結構な時間がかかります
そこで Trainer の設定値について注目です
　・gpus：CPUでもできなくないけど、100%になります
　・epoch：デフォルト1,000になっているので、数時間かかる

gpus

CPUで学習開始すると瞬間でぶぉーーーーとなって 100% にでした

gpus を使えば、静かなもんで計算も速し

GPUを使えるかどうか確認

torch.cuda.is_available()
True

epoch

学習前は一体どれくらいの数が適正なのか？？？は正直あんまりよく分かりません
デフォルト値の 1,000 は、GPUを使っても 1時間たっても終わりませんでした
ので、学習開始してみてTensorBoard でリアルタイムに確認しながら途中で止めるというのが、良さそうです
もちろん途中で止めて、後で途中まで学習したパラメータを再利用できます

以上を踏まえた上で、学習開始です

# モデルのインスタンス化
net = Net()
net

Net(
(conv): Conv2d(1, 4, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
(L1): Linear(in_features=784, out_features=100, bias=True)
(L2): Linear(in_features=100, out_features=10, bias=True)
(bn): BatchNorm1d(784, eps=1e-05, momentum=0.1, affine=True, track_running_stats=True)
)

# GPUで学習
trainer = Trainer(max_epochs=100, gpu=1)

# モデルの学習
trainer.fit(net)

TensorBoardで学習状況の確認

epoch数 50くらいで頭打ちになっているのが分かります

PyTorch Lightning では途中で止めたとしても、途中まで学習したパラメータを簡単に保存して、また後で使って推論することもできます

パラメータの保存方法
（CPUで保存しておかないと、後でGPUのないマシンで使えないので）

net = net.to('cpu')
torch.save(net.state_dict(), './xxx.pth')

推論

保存したパラメータを読み込み

net = Net()
net.load_state_dict(torch.load('xxx.pth'))

適当な画像をテストしてみる

import numpy as np
import matplotlib.pyplot as plt

sample = test[random.randint(0, 10000)]
sample_label = sample[1]

img = np.transpose(sample[0], (1, 2, 0))
img = img.reshape(img.shape[0], img.shape[1])
print("正解ラベル：", sample_label)
plt.imshow(img, cmap='gray')

推論してみる

sample_img = sample[0].unsqueeze(0) # [1, 28, 28] > [1, 1, 28, 28] 
y_predict = net.forward(sample_img)
print(y_predict)
print("推論結果: ", y_predict.argmax())

わーい、結果 "7"と出てきました

tensor([[ -2.5076,   2.4008,  10.9549,   6.3163,  -5.5979, -21.0443, -15.4997,
          39.9161,  -8.9796,  10.5217]], grad_fn=<AddmmBackward>)
推論結果:  tensor(7)

あと10個くらい試してみましたが、一個間違いがったらの出たので
Accuracy: 97% くらいになっているので、まぁそんなもんですね

aaa

次は、もうちょっと複雑な画像を使ってみます