論文の勉強5 DenseNet

論文の勉強をメモ書きレベルですがのせていきます。あくまでも自分の勉強目的です。
構造部分に注目し、その他の部分は書いていません。ご了承ください。
本当にいい加減・不正確な部分が多数あると思いますのでご理解ください。

今回は、以下の論文のDense Netの実装を行います。

タイトル：Densely Connected Convolutional Networks

https://arxiv.org/abs/1608.06993

### DenseNets

### 構造

$\boldsymbol{x}_0$が畳み込みネットワークを通過することを考える。
ネットワークは$L$層含む場合、第$l$層での非線形変換(Batch Normalization,ReLU,Pooling,Convolutionなどを組み合わせたもの)を$H_l(・)$、
第$l$層の出力を$\boldsymbol{x}_l$と表す。

#### Dense connectivity(dense block)
第$l$層でその前のすべての出力$\boldsymbol{x}_0,\cdots,\boldsymbol{x}_{l-1}$を受け取る。

```math
\boldsymbol{x}_l=H_l([\boldsymbol{x}_0,\cdots,\boldsymbol{x}_{l-1}])
```

ここで、$[\boldsymbol{x}_0,\cdots,\boldsymbol{x}_{l-1}]$は第0~$l-1$層での出力を結合したものである。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2358656/5e1585e5-c9c2-b45b-c2d0-39cde32af69b.png)

#### composite function
Batch Normalization(BN)、ReLU、3×3畳み込み(Conv)を組み合わせて使用する。  

#### Pooilng layers(transition layer)
特徴マップのダウンサンプリングを行うため、
dense blockの間に、畳み込みとプーリングを行うtransition layerを挿入します。
この層はBN、1×1Conv、2×2Pooligで構成されます。
入力のチャンネル数が$m$のとき、transition layerでの出力を$m\theta$とします。
ここで、$0<\theta\leq1$とし、$0<\theta<1$のときDenseNet-Cと呼びここでは$\theta=0.5$とします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2358656/62cb2b20-97ef-bf94-4817-6b0f6ad988df.png)

#### Growth rate
$H_l$が$k$個の特徴マップを作り出す場合、第$l$層への入力は$k_0+k(l-1)$チャンネルとなります。ここで、$k_0$は入力層のチャンネル数である。
この$k$をgrowth rateと呼ぶ。

#### bottleneck layers
dense blockは$k$チャンネルの特徴マップを出力するが、これは入力と比べて少ない。
dense blockの3×3Convの前に1×1Convを追加して(bottleneck)、3×3Convへの入力チャンネル数を減らして計算効率を改善する。
bottleneck layersの構造は、BN-ReLU-1×1Conv-BN-ReLU-3×3Convとなり、DenseNet-Bと表します。
今回は1×1Convで$4×k$個の特徴マップを作ることとする。
  
#### Implementation Details
最初のdense blockに入力する前に、入力を16チャンネル(ConvNet-BCの場合は、growth rateの2倍)に変換します。
今回は、strideが2の7×7Convと3×3のmax poolingを使用します。
最後のdense blockの後にはGlobal Avarage Poolng、そしてsoftmaxを使用して出力を出します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2358656/b93814e5-a15a-9efa-329f-ba2f95f6c2c9.png)

#### 学習
SGDで学習し、学習率は0.1とする。エポックが全体のエポック数の50%、75%となったら0.1を掛けて減衰させる。
weight decayは0.0001とし、Nesterov momentumを0.9とする。

rateを0.2としたdropout層を各畳み込み層の後に入れる。

#### 実装
$k=32$とし、DenseNet-169を実装します。使用する構造はbottleneck構造を持ったDenseNet-BCです。

#### keras
必要なライブラリのインポートを行います。

```python
import tensorflow.keras as keras
import tensorflow as tf
from tensorflow.keras.models import Sequential, Model
from tensorflow.keras.layers import Input, Conv2D, Activation, MaxPooling2D, AveragePooling2D, Flatten, Dense, Dropout, GlobalAveragePooling2D, BatchNormalization, Add
from keras.layers.merge import concatenate
from tensorflow.keras import backend as K
from tensorflow.keras.optimizers import SGD
from tensorflow.keras.callbacks import LearningRateScheduler
from keras.datasets import cifar10
import numpy as np
import cv2
```

まず、BN-ReLU-Convのお決まりの流れを定義しておきます。

```python
class bn_relu_conv(Model):
    def __init__(self, out_channels, kernel_size=1, strides=1, name=None):
        super(bn_relu_conv, self).__init__(name=name)        
        
        self.bn = BatchNormalization()
        self.relu = Activation("relu")
        self.conv = Conv2D(out_channels, kernel_size=kernel_size, strides=strides, padding='same')
    
    def call(self, x):
        x = self.bn(x)
        x = self.relu(x)
        x = self.conv(x)
        return x
```

Dense Layerの定義です。
2回の畳み込み処理を行います。

```python
class dense_layer(Model):
    """dense block"""
    def __init__(self, growth_rate, n):
        super(dense_layer, self).__init__(name='layer'+'_'+str(n))        
        
        self.conv1 = bn_relu_conv(out_channels=growth_rate*4, kernel_size=1, strides=1, name='layer'+'_'+str(n)+'conv1')
        self.conv2 = bn_relu_conv(out_channels=growth_rate, kernel_size=3, strides=1, name='layer'+'_'+str(n)+'conv2')

    def call(self, x):
        out = self.conv1(x)
        out = self.conv2(out)
        return out
```

Dense Blockの定義です。
Dense Layerを繰り返し、各Layerの出力を最後に結合します。

```python
class dense_block(Model):
    """dense block"""
    def __init__(self, num_layers, growth_rate, name):
        super(dense_block, self).__init__(name=name)
        
        self.dense_layers = []
        for i in range(num_layers):
            self.dense_layers += [dense_layer(growth_rate=growth_rate, n=i)]
    
    def call(self, x):
        for layer in self.dense_layers:
            main = x
            x = layer(x)
            x = concatenate([main, x])
        return x
```

transition Layerの定義です。
畳み込みとプーリングを行い、特徴マップを変形します。

```python
class transition_layer(Model):
    def __init__(self, input_channels, name):
        super(transition_layer, self).__init__(name=name)
        self.conv = bn_relu_conv(out_channels=input_channels, kernel_size=1, strides=1, name=name+'conv1')
        self.pool = AveragePooling2D((2, 2))
    
    def call(self, x):
        out = self.conv(x)
        out = self.pool(out)
        return out
```

DenseNetの本体の実装を行います。

```python
class dense_net(Model):
    """dense block"""
    def __init__(self, num_layers, growth_rate, num_classes=1000):
        super(dense_net, self).__init__()
        
        self.layer_dense = []
        
        in_channels=  2 * growth_rate
        
        self.layer_dense += [bn_relu_conv(out_channels=in_channels, kernel_size=7, strides=2, name='conv1'),
                             MaxPooling2D((3,3), strides=2, padding='same')]
        
        for i in range(len(num_layers)):
            self.layer_dense += [dense_block(num_layers=num_layers[i], growth_rate=growth_rate, name='block_'+str(i))]
            
            in_channels += growth_rate * num_layers[i]
            
            if i != (len(num_layers)-1):
                self.layer_dense += [transition_layer(in_channels//2, name='trans_'+str(i))]
                in_channels = in_channels//2
            
        self.layer_dense += [GlobalAveragePooling2D(),
                             Dense(num_classes, activation="softmax")]
        
    def call(self, x):
        for layer in self.layer_dense:
            x = layer(x)
        return x
```

構造の確認をします。

```python
model = dense_net(num_layers=[6,12,32,32], growth_rate=32)
model.build((None, 224, 224, 3))  # build with input shape.
dummy_input = Input(shape=(224, 224, 3))  # declare without batch demension.
model_summary = Model(inputs=[dummy_input], outputs=model.call(dummy_input), name="pretrained")
model_summary.summary()
```

    Model: "pretrained"
    _________________________________________________________________
    Layer (type)                 Output Shape              Param #   
    =================================================================
    input_1 (InputLayer)         [(None, 224, 224, 3)]     0         
    _________________________________________________________________
    conv1 (bn_relu_conv)         (None, 112, 112, 64)      9484      
    _________________________________________________________________
    max_pooling2d (MaxPooling2D) (None, 56, 56, 64)        0         
    _________________________________________________________________
    block_0 (dense_block)        (None, 56, 56, 256)       339264    
    _________________________________________________________________
    trans_0 (transition_layer)   (None, 28, 28, 128)       33920     
    _________________________________________________________________
    block_1 (dense_block)        (None, 28, 28, 512)       931968    
    _________________________________________________________________
    trans_1 (transition_layer)   (None, 14, 14, 256)       133376    
    _________________________________________________________________
    block_2 (dense_block)        (None, 14, 14, 1280)      4377600   
    _________________________________________________________________
    trans_2 (transition_layer)   (None, 7, 7, 640)         824960    
    _________________________________________________________________
    block_3 (dense_block)        (None, 7, 7, 1664)        5999616   
    _________________________________________________________________
    global_average_pooling2d (Gl (None, 1664)              0         
    _________________________________________________________________
    dense (Dense)                (None, 1000)              1665000   
    =================================================================
    Total params: 14,315,188
    Trainable params: 14,160,238
    Non-trainable params: 154,950
    _________________________________________________________________
    

学習の設定を行い、実行はせず実装を終わりにします。

```python
# 学習率を返す関数を用意する
def lr_schedul(epoch):
    x = 0.1
    if epoch >= 20:
        x = 0.1*0.1
    if epoch >= 30:
        x = 0.1*(0.1**2)
    return x


lr_decay = LearningRateScheduler(
    lr_schedul,
    verbose=1,
)

sgd = SGD(lr=0.1, momentum=0.9, decay=1e-4, nesterov=True)
model.compile(loss=['categorical_crossentropy'], optimizer=sgd, metrics=['accuracy'])
```


```python
batch_size = 64
epochs = 40

history=model.fit(x_train, y_train, batch_size=batch_size, epochs=epochs, validation_data=(x_test, y_test), callbacks=[lr_decay])
```

#### 転移学習(keras)
学習済みのモデルが提供されています。
出力層を付け加えて使用します。

```python
from keras.applications.densenet import DenseNet169

base_model = DenseNet169(weights='imagenet', include_top=False, classes=1000)

x = base_model.output
x = GlobalAveragePooling2D()(x)
x = Dense(1024, activation='relu')(x)
output = Dense(10, activation='softmax')(x)

model = Model(inputs=base_model.input, outputs=output)

for layer in base_model.layers:
    layer.trainable = False
```


```python
# 学習率を返す関数を用意する
def lr_schedul(epoch):
    x = 0.1
    if epoch >= 20:
        x = 0.1*0.1
    if epoch >= 30:
        x = 0.1*(0.1**2)
    return x

lr_decay = LearningRateScheduler(
    lr_schedul,
    verbose=1,
)

sgd = SGD(lr=0.1, momentum=0.9, decay=1e-4, nesterov=True)
model.compile(loss=['categorical_crossentropy'], optimizer=sgd, metrics=['accuracy'])
```


```python
history=model.fit(x_train, y_train, batch_size=batch_size, epochs=epochs, validation_data=(x_test, y_test), callbacks=[lr_decay])
```

#### pytorch
実装の内容はkerasとほぼ同様です。
必要なライブラリのインポートを行います。

```python
import torch
import torch.nn as nn
import torch.optim as optim
from torchsummary import summary

import pytorch_lightning as pl
from torchmetrics import Accuracy as accuracy
```

各モジュールの定義を行います。

```python
class bn_relu_conv(nn.Module):
    def __init__(self, in_channels, out_channels, kernel_size, stride,padding=1):
        super(bn_relu_conv, self).__init__()
        self.bn = nn.BatchNorm2d(in_channels)
        self.relu = nn.ReLU(inplace=True)
        self.conv = nn.Conv2d(in_channels, out_channels, kernel_size=kernel_size, stride=stride, padding=padding)
        self.drop = nn.Dropout(0.2)
        
    def forward(self, x):
        out = self.bn(x)
        out = self.relu(out)
        out = self.conv(out)
        out = self.drop(out)
        return out
```


```python
class dense_layer(nn.Module):
    def __init__(self, in_channels, growth_rate):
        super(dense_layer, self).__init__()        
        
        self.conv1 = bn_relu_conv(in_channels, out_channels=growth_rate*4, kernel_size=1, stride=1, padding=0)
        self.conv2 = bn_relu_conv(growth_rate*4, out_channels=growth_rate, kernel_size=3, stride=1)

    def forward(self, x):
        out = self.conv1(x)
        out = self.conv2(out)
        return out
```


```python
class dense_block(nn.Module):
    """dense block"""
    def __init__(self, in_channels, num_layers, growth_rate):
        super(dense_block, self).__init__()
        
        dense_layers = []
        for i in range(num_layers):
            dense_layers += [dense_layer(in_channels=int(in_channels + i*growth_rate), growth_rate=growth_rate)]
        self.dense_layers = nn.Sequential(*dense_layers)
        
    def forward(self, x):
        for layer in self.dense_layers:
            out = layer(x)
            x = torch.cat([x, out], 1)
        return x
```


```python
class transition_layer(nn.Module):
    def __init__(self, in_channels, out_channels):
        super(transition_layer, self).__init__()
        self.conv = bn_relu_conv(in_channels=in_channels, out_channels=out_channels, kernel_size=1, stride=1,padding=0)
        self.pool = nn.AvgPool2d(kernel_size=2, stride=2)
    
    def forward(self, x):
        out = self.conv(x)
        out = self.pool(out)
        return out
```

DenseNet本体の実装です。

```python
class dense_net(nn.Module):
    """dense block"""
    def __init__(self, num_layers, growth_rate, num_classes=1000):
        super(dense_net, self).__init__()
        
        in_channels=  2 * growth_rate
        self.first_layer = nn.Sequential(*[bn_relu_conv(in_channels=3, out_channels=in_channels, kernel_size=7, stride=2, padding=3),
                                            nn.MaxPool2d(kernel_size=3, stride=2, padding=1)])
        layer_dense = []
        for i in range(len(num_layers)):
            layer_dense += [dense_block(in_channels=in_channels, num_layers=num_layers[i], growth_rate=growth_rate)]
            
            in_channels += growth_rate * num_layers[i]
            if i != (len(num_layers)-1):
                layer_dense += [transition_layer(in_channels=in_channels, out_channels=in_channels//2)]
                in_channels = in_channels//2
            
        layer_dense += [nn.AdaptiveAvgPool2d((1,1))]
        self.layer_dense = nn.Sequential(*layer_dense)
        self.classifier = nn.Linear(in_channels, num_classes)
        
    def forward(self, x):
        out = self.first_layer(x)
        out = self.layer_dense(out)
        out = out.view(out.shape[0], -1)
        out = self.classifier(out)
        return x
```

モデルの構造を確認します。

```python
summary(dense_net(num_layers=[6,12,32,32], growth_rate=32), (3,224,224))
```

    

    ----------------------------------------------------------------
            Layer (type)               Output Shape         Param #
    ================================================================
           BatchNorm2d-1          [-1, 3, 224, 224]               6
                  ReLU-2          [-1, 3, 224, 224]               0
                Conv2d-3         [-1, 64, 112, 112]           9,472
               Dropout-4         [-1, 64, 112, 112]               0
    中略
              Conv2d-929             [-1, 32, 7, 7]          36,896
             Dropout-930             [-1, 32, 7, 7]               0
        bn_relu_conv-931             [-1, 32, 7, 7]               0
         dense_layer-932             [-1, 32, 7, 7]               0
         dense_block-933           [-1, 1664, 7, 7]               0
    AdaptiveAvgPool2d-934           [-1, 1664, 1, 1]               0
              Linear-935                 [-1, 1000]       1,665,000
    ================================================================
    Total params: 14,160,238
    Trainable params: 14,160,238
    Non-trainable params: 0
    ----------------------------------------------------------------
    Input size (MB): 0.57
    Forward/backward pass size (MB): 471.02
    Params size (MB): 54.02
    Estimated Total Size (MB): 525.61
    ----------------------------------------------------------------
    
学習の設定を行い、実装を終わりにします。

```python
class DenseTrainer(pl.LightningModule):
    def __init__(self):
        super().__init__()
        self.model = dense_net(num_layers=[6,12,32,32], growth_rate=32)
        
    def forward(self, x):
        x = self.model(x)
        return x
    
    def training_step(self, batch, batch_idx):
        x, y = batch 
        #x, y = x.to(device), y.to(device)
        y_hat = self.forward(x)
        loss = nn.CrossEntropyLoss()(y_hat, y)
        return {'loss': loss, 'y_hat':y_hat, 'y':y, 'batch_loss': loss.item()*x.size(0)}
    
    def validation_step(self, batch, batch_idx):
        x, y = batch
        #x, y = x.to(device), y.to(device)
        y_hat = self.forward(x)
        loss = nn.CrossEntropyLoss()(y_hat, y)
        return {'y_hat':y_hat, 'y':y, 'batch_loss': loss.item()*x.size(0)}
    
    def test_step(self, batch, batch_nb):
        x, y = batch
        #x, y = x.to(device), y.to(device)
        y_hat = self.forward(x)
        loss = nn.CrossEntropyLoss()(y_hat, y)
        y_label = torch.argmax(y_hat, dim=1)
        acc = accuracy()(y_label, y)
        return {'test_loss': loss, 'test_acc': acc}
    
    def training_epoch_end(self, train_step_output):
        y_hat = torch.cat([val['y_hat'] for val in train_step_outputs], dim=0)
        y = torch.cat([val['y'] for val in train_step_outputs], dim=0)
        epoch_loss = sum([val['batch_loss'] for val in train_step_outputs]) / y_hat.size(0)
        preds = torch.argmax(y_hat, dim=1)
        acc = accuracy()(preds, y)
        self.log('train_loss', epoch_loss, prog_bar=True, on_epoch=True)
        self.log('train_acc', acc, prog_bar=True, on_epoch=True)
        
        print('---------- Current Epoch {} ----------'.format(self.current_epoch + 1))
        print('train Loss: {:.4f} train Acc: {:.4f}'.format(epoch_loass, acc))
    
    def validation_epoch_end(self, val_step_outputs):
        y_hat = torch.cat([val['y_hat'] for val in val_step_outputs], dim=0)
        y = torch.cat([val['y'] for val in val_step_outputs], dim=0)
        epoch_loss = sum([val['batch_loss'] for val in val_step_outputs]) / y_hat.size(0)
        preds = torch.argmax(y_hat, dim=1)
        acc = accuracy()(preds, y)
        self.log('val_loss', epoch_loss, prog_bar=True, on_epoch=True)
        self.log('val_acc', acc, prog_bar=True, on_epoch=True)
        
        print('valid Loss: {:.4f} valid Acc: {:.4f}'.format(epoch_loss, acc))
    
    # New: テストデータに対するエポックごとの処理
    def test_epoch_end(self, test_step_outputs):
        y_hat = torch.cat([val['y_hat'] for val in test_step_outputs], dim=0)
        y = torch.cat([val['y'] for val in test_step_outputs], dim=0)
        epoch_loss = sum([val['batch_loss'] for val in test_step_outputs]) / y_hat.size(0)
        preds = torch.argmax(y_hat, dim=1)
        acc = accuracy()(preds, y)
        self.log('test_loss', epoch_loss, prog_bar=True, on_epoch=True)
        self.log('test_acc', acc, prog_bar=True, on_epoch=True)
        
        print('test Loss: {:.4f} test Acc: {:.4f}'.format(epoch_loss, acc))
        
    def configure_optimizers(self):
        optimizer = optim.SGD(self.parameters(), lr=0.1, momentum=0.9, weight_decay=1e-4)
        scheduler = torch.optim.lr_scheduler.MultiStepLR(optimizer, milestones=[150, 225], gamma=0.1)
        return {'optimizer': optimizer, 'lr_scheduler': scheduler}
```

#### 転移学習(pytorch)
pytorchでも学習済のモデルが提供されています。

```python
from torchvision import models

dense = models.densenet161(pretrained=True)
```
```python
class DenseTrainer(pl.LightningModule):
    def __init__(self):
        super(VGGTrainer, self).__init__()
        dense = models.densenet161(pretrained=True)
        
        dense.classifier = nn.Linear(in_features=2208, out_features=10)
        self.model = dense
        
        update_param_names = ['classifier.weight', 'classifier.bias']
        
        for name, param in self.model.named_parameters():
            if name in update_param_names:
                param.requires_grad = True
            else:
                param.requires_grad = False
        
    def forward(self, x):
        x = self.model(x)
        return x
    
    def training_step(self, batch, batch_idx):
        x, y = batch 
        #x, y = x.to(device), y.to(device)
        y_hat = self.forward(x)
        loss = nn.CrossEntropyLoss()(y_hat, y)
        return {'loss': loss, 'y_hat':y_hat, 'y':y, 'batch_loss': loss.item()*x.size(0)}
    
    def validation_step(self, batch, batch_idx):
        x, y = batch
        #x, y = x.to(device), y.to(device)
        y_hat = self.forward(x)
        loss = nn.CrossEntropyLoss()(y_hat, y)
        return {'y_hat':y_hat, 'y':y, 'batch_loss': loss.item()*x.size(0)}
    
    def test_step(self, batch, batch_nb):
        x, y = batch
        #x, y = x.to(device), y.to(device)
        y_hat = self.forward(x)
        loss = nn.CrossEntropyLoss()(y_hat, y)
        y_label = torch.argmax(y_hat, dim=1)
        acc = accuracy()(y_label, y)
        return {'test_loss': loss, 'test_acc': acc}
    
    def training_epoch_end(self, train_step_output):
        y_hat = torch.cat([val['y_hat'] for val in train_step_outputs], dim=0)
        y = torch.cat([val['y'] for val in train_step_outputs], dim=0)
        epoch_loss = sum([val['batch_loss'] for val in train_step_outputs]) / y_hat.size(0)
        preds = torch.argmax(y_hat, dim=1)
        acc = accuracy()(preds, y)
        self.log('train_loss', epoch_loss, prog_bar=True, on_epoch=True)
        self.log('train_acc', acc, prog_bar=True, on_epoch=True)
        
        print('---------- Current Epoch {} ----------'.format(self.current_epoch + 1))
        print('train Loss: {:.4f} train Acc: {:.4f}'.format(epoch_loass, acc))
    
    def validation_epoch_end(self, val_step_outputs):
        y_hat = torch.cat([val['y_hat'] for val in val_step_outputs], dim=0)
        y = torch.cat([val['y'] for val in val_step_outputs], dim=0)
        epoch_loss = sum([val['batch_loss'] for val in val_step_outputs]) / y_hat.size(0)
        preds = torch.argmax(y_hat, dim=1)
        acc = accuracy()(preds, y)
        self.log('val_loss', epoch_loss, prog_bar=True, on_epoch=True)
        self.log('val_acc', acc, prog_bar=True, on_epoch=True)
        
        print('valid Loss: {:.4f} valid Acc: {:.4f}'.format(epoch_loss, acc))
    
    # New: テストデータに対するエポックごとの処理
    def test_epoch_end(self, test_step_outputs):
        y_hat = torch.cat([val['y_hat'] for val in test_step_outputs], dim=0)
        y = torch.cat([val['y'] for val in test_step_outputs], dim=0)
        epoch_loss = sum([val['batch_loss'] for val in test_step_outputs]) / y_hat.size(0)
        preds = torch.argmax(y_hat, dim=1)
        acc = accuracy()(preds, y)
        self.log('test_loss', epoch_loss, prog_bar=True, on_epoch=True)
        self.log('test_acc', acc, prog_bar=True, on_epoch=True)
        
        print('test Loss: {:.4f} test Acc: {:.4f}'.format(epoch_loss, acc))
        
    def configure_optimizers(self):
        optimizer = optim.SGD(self.parameters(), lr=0.1, momentum=0.9, weight_decay=1e-4)
        scheduler = torch.optim.lr_scheduler.MultiStepLR(optimizer, milestones=[20, 30], gamma=0.1)
        return {'optimizer': optimizer, 'lr_scheduler': scheduler}
```

これでDenseNet論文のメモ書きを終わります。