코딩도상국

안녕하세요. WB LOS/NLOS Classification Using Deep Learning Method(1)에서 UWB CIR Dataset을 생성하였다면, 

2편으로 논문에서 제시한 CNN_LSTM 네트워크를 약간 변형하여 구성하겠습니다. 

coding-yoon.tistory.com/138

1편을 보고 오시는 것을 추천드립니다. 이는 1편처럼 Dataset이 준비됐다는 가정 하에 진행됩니다.

Columns : 1016 (Sampling CIR)

Label : 42000(LOS : 21000, NLOS : 21000)  

먼저 위 논문은 CNN-LSTM 구조로 LOS/NLOS를 학습하는 모델입니다.

(epoch : 10, learning rate : 0.001, dropout : 0.5, Train Sample : 35000, Test Sample : 7000)

CNN에서 CIR Featur을 추출, Redundant information을 제거하고, LSTM을 이용하여 분류합니다.

( CNN+stacked-LSTM Accuracy : 82.14% )

Implemnet ( Dataset : df_uwb_data 준비 )

1. Import

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
from torch.utils.tensorboard import SummaryWriter

from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt 

import time
import random

import uwb_dataset

print("Pytorch Version :", torch.__version__)  # Pytorch Version : 1.7.1+cu110
writer = SummaryWriter('runs/UWB_CIR_Classfication')

%matplotlib inline

2. Hyper-Parameters

# random seed
random_seed = 42

num_epoch = 10
batch_size = 64
in_channels = 1
num_classes = 2
num_layers = 2
fully_connected = 128

lr = 0.001
weight_decay = 0.0

# Parameters
view_train_iter = 50
view_val_iter = 5
save_point = 0.90

3. Random Seed

def torch_random_seed(on_seed=False, random_seed=1):
    if on_seed:
        torch.manual_seed(random_seed)
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False

        np.random.seed(random_seed)
        random.seed(random_seed)
        
torch_random_seed(on_seed=True, random_seed=random_seed)

4. Model Evaluation Function

def get_clf_eval(y_true, y_pred, average='weighted'):
    accuracy = accuracy_score(y_true, y_pred)
    precision = precision_score(y_true, y_pred, average=average)
    recall = recall_score(y_true, y_pred, average=average)
    f1 = f1_score(y_true, y_pred, average=average)

    return accuracy, precision, recall, f1

5. Split (Train, Validation, Test) X, label Data

# sklearn의 train_test_split은 stratify 파라미터를 통해 Label의 비율을 유지하면서 Split
x_train, x_test, y_train, y_test = train_test_split(df_uwb_data.values, df_uwb['NLOS'].values, test_size=0.1, random_state=42, stratify=df_uwb['NLOS'].values)
x_train, x_val, y_train, y_val = train_test_split(x_train, y_train, test_size=0.1, random_state=random_seed, stratify=y_train)

print("x_train shape :", x_train.shape, y_train.shape)
print("x_val shape :", x_val.shape, y_val.shape)
print("x_test shape :", x_test.shape, y_test.shape)

print("Train NLOS 0 count :", len(y_train[y_train==0]))
print("Train NLOS 1 count :", len(y_train[y_train==1]))
print("Validation NLOS 0 count :", len(y_val[y_val==0]))
print("Validation NLOS 1 count :", len(y_val[y_val==1]))
print("Test NLOS 0 count :", len(y_test[y_test==0]))
print("Test NLOS 0 count :", len(y_test[y_test==1]))

7. Dataset & DataLoader

def generating_loader(x_data, y_data, batch_size=batch_size, shuffle=True, drop_last=True):
    # preprocessing x_data
    x_data = np.expand_dims(x_data, axis=1)
    x_tensor = torch.tensor(x_data, dtype=torch.float32)
    # preprocessing y_data
    y_tensor = torch.tensor(y_data, dtype=torch.long).view(-1)

    return DataLoader(TensorDataset(x_tensor, y_tensor), batch_size=batch_size, shuffle=shuffle, drop_last=drop_last)

trainloader = generating_loader(x_train, y_train, batch_size=batch_size, shuffle=True, drop_last=True)
validationloader = generating_loader(x_val, y_val, batch_size=batch_size, shuffle=False, drop_last=True)
testloader = generating_loader(x_val, y_val, batch_size=batch_size, shuffle=False, drop_last=True)

for x, label in trainloader:
    print(x.shape, label.shape)
    break

8. Create Model

class CNN_LSTM(nn.Module):
    def __init__(self, in_channels, out_channels, batch_size, num_layers, fully_connected, device):
        super(CNN_LSTM, self).__init__()
        self.batch_size = batch_size
        self.conv1d_layer = nn.Sequential(
            nn.Conv1d(in_channels=in_channels, out_channels=10, kernel_size=4, stride=1, padding=0),
            nn.ReLU(),
            nn.Conv1d(in_channels=10, out_channels=20, kernel_size=5, stride=1, padding=0),
            nn.ReLU(),
            nn.MaxPool1d(kernel_size=2, stride=2),
        ) 
        self.lstm = nn.LSTM(input_size = 504, 
                            hidden_size = 32, 
                            num_layers = num_layers,
                            bias = False,
                            dropout = 0.5,
                            bidirectional = True,
                            batch_first=True)

        self.hidden_state, self.cell_state = self.init_hidden()
        
        self.bn2 = nn.BatchNorm1d(20)
        self.bn0 = nn.BatchNorm1d(64)
        self.bn1 = nn.BatchNorm1d(128)

        self.fc_layer = nn.Linear(64, 128)
        self.relu = nn.ReLU()
        self.fc_layer_class = nn.Linear(128, out_channels)


    def init_hidden(self):
        hidden_state = torch.zeros(num_layers*2, self.batch_size, 32).to(device)
        cell_state = torch.zeros(num_layers*2, self.batch_size, 32).to(device)

        return hidden_state, cell_state
        
    def forward(self, x):
        x = self.conv1d_layer(x)
        x, _ = self.lstm(x,(self.hidden_state, self.cell_state))
        x = x[:, -1 :].view(x.size(0), -1)
        x = self.bn0(x)
        x = self.fc_layer(x)
        x = self.bn1(x)
        x = self.relu(x)
        x = self.fc_layer_class(x)
        x = self.relu(x)

        return x

9. Loss Function, Optimizer

device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
model = CNN_LSTM(
    in_channels=in_channels,\
    device=device,\
    out_channels=num_classes,\
    batch_size=batch_size,\
    fully_connected=fully_connected,\
    num_layers=num_layers).to(device)
loss_function = nn.CrossEntropyLoss()  
optimizer = optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)  # optimizer

# tensorboard
images, labels = next(iter(trainloader))
writer.add_graph(model, images.to(device))

# lr_scheduler = optim.lr_scheduler.MultiStepLR(optimizer=optimizer, milestones=[int(num_epoch * 0.5), int(num_epoch * 0.75)], gamma=0.1, last_epoch=-1)

10. Train, Validation 

start = time.time()

correct = 0
total = 0
train_acc = []
tmp_acc = 0
loss_arr = []

print("*Train Start!!*")
if torch.cuda.device_count() == True:
    print("epoch : {}, learing rate : {}, device : {}".format(num_epoch, lr, torch.cuda.get_device_name(0)))
    print("Model : {}".format(model._get_name()))
    print("Loss function : {}".format(loss_function._get_name()))
    print("Optimizer : {}".format(str(optimizer).replace("\n", " ").replace("     ", ", ")))
else:
    print("epoch : {}, learing rate : {}, device : {}".format(num_epoch, lr, device))
    print("Model : {}".format(model._get_name()))
    print("Loss function : {}".format(loss_function._get_name()))
    print("Optimizer : {}".format(str(optimizer).replace("\n", " ").replace("     ", ", ")))
print("*"*100)

# train
for epoch in range(num_epoch):
    epoch += 1
    for train_iter, (train_x, train_y_true) in enumerate(trainloader):
        model.train()  # Train mode
        model.zero_grad()  # model zero initialize
        optimizer.zero_grad()  # optimizer zero initialize
        

        train_x, train_y_true = train_x.to(device), train_y_true.to(device)  # device(gpu)
        train_y_pred = model.forward(train_x)  # forward
        loss = loss_function(train_y_pred, train_y_true)  # loss function
        loss.backward()  # backward
        optimizer.step()  # optimizer
        _, pred_index = torch.max(train_y_pred, 1)
        
        if train_iter % view_train_iter == 0:
            loss_arr.append(loss.item())
            total += train_y_true.size(0)  # y.size(0)
            correct += (pred_index == train_y_true).sum().float()  # correct
            tmp_acc = correct / total  # accuracy
            train_acc.append(tmp_acc)

            writer.add_scalar("Loss/train", loss, epoch)
            writer.add_scalar("Accuracy/train",tmp_acc, epoch)

            print("[Train] ({}, {}) Time={:.2f}[s], loss = {:.5f}, Accuracy = {:.4f}, lr={:.6f}".format(epoch, train_iter, time.time()-start, loss.item(), tmp_acc, optimizer.param_groups[0]['lr']))
    # lr_scheduler.step()
    # validation 
    if epoch % view_val_iter == 0: 
        val_acc_tmp, val_precision_tmp, val_recall_tmp, val_f1_tmp = [], [], [], []
        val_acc_result, val_precision_result, val_recall_result, val_f1_result = [], [], [], []
        val_time = time.time()
        for val_iter, (val_x, val_y_true) in enumerate(validationloader):
            model.eval()
            val_x, val_y_true = val_x.to(device), val_y_true.to(device)  # device(gpu)
            val_y_pred = model.forward(val_x)  # forward
            _, val_pred_index = torch.max(val_y_pred, 1)

            val_pred_index_cpu = val_pred_index.cpu().detach().numpy()
            val_y_true_cpu = val_y_true.cpu().detach().numpy()
            
            val_acc, val_precision, val_recall, val_f1 = get_clf_eval(val_y_true_cpu, val_pred_index_cpu)

            val_acc_tmp.append(val_acc), val_acc_result.append(val_acc)
            val_precision_tmp.append(val_precision), val_precision_result.append(val_precision)
            val_recall_tmp.append(val_recall), val_recall_result.append(val_recall)
            val_f1_tmp.append(val_f1), val_f1_result.append(val_f1)

        val_acc_mean = sum(val_acc_tmp, 0.0)/len(val_acc_tmp)
        val_precision_mean = sum(val_precision_tmp, 0.0)/len(val_precision_tmp)
        val_recall_mean = sum(val_recall_tmp, 0.0)/len(val_recall_tmp)
        val_f1_mean = sum(val_f1_tmp, 0.0)/len(val_f1_tmp)

        print("-"*100)
        print("|  Validation {:.2f}[s], Accuracy : {:.4f}, Precision : {:.4f}, Recall : {:.4f}, F1 Score : {:.4f}   |".format(
            time.time()-val_time, val_acc_mean, val_precision_mean, val_recall_mean, val_f1_mean))
        print("-"*100)
        if val_acc_mean >= save_point:
            epoch_str = str(epoch)
            lr_str = str(lr)
            batch_str= str(batch_size)
            acc_str= str((int(val_acc_mean*100)))
            model_name = "["+model._get_name()+"](epoch-"+epoch_str+")-"+"(init_lr-"+lr_str+")-"+"(batch-"+batch_str+")-"+"(acc-"+acc_str+").pt"
            save_path = os.path.join(path, dir_ ,model_name)
            parameters = {'epoch' : epoch, 'model_state_dict' : model.state_dict(), 'optimizer_state_dict' : optimizer.state_dict(), 'loss' : loss}
            torch.save(parameters, save_path)
            print('[INFO] Model Saved : '+ save_path)
writer.flush()
writer.close()

fig = plt.figure(figsize=[16, 8])
loss_plt = plt.subplot(2,1,1)
acc_plt = plt.subplot(2,1,2)

loss_plt.plot(loss_arr, color='red', marker="*")
loss_plt.set_title("Train - Loss", fontsize=15)
loss_plt.legend(['Train-Loss'])
loss_plt.grid(True, axis='y')

acc_plt.plot(train_acc, color='green', marker="*")
acc_plt.set_title("Train - Accuracy", fontsize=15)
acc_plt.legend(['Train-Accuracy'])
acc_plt.set_ylim((0.0, 1.05))
acc_plt.grid(True, axis='y')

plt.show()

11. Model Evaluation

test_start = time.time()

model.eval()
with torch.no_grad():
    test_acc_tmp, test_precision_tmp, test_recall_tmp, test_f1_tmp = [], [], [], []
    for test_iter, (test_x, test_y_true) in enumerate(testloader):
        test_x, test_y_true = test_x.to(device), test_y_true.to(device)
        test_y_pred = model.forward(test_x)  # forward

        _, test_pred_index = torch.max(test_y_pred, 1)

        test_pred_index_cpu = test_pred_index.cpu().detach().numpy()
        test_y_true_cpu = test_y_true.cpu().detach().numpy()
            
        test_acc, test_precision, test_recall, test_f1 = get_clf_eval(test_y_true_cpu, test_pred_index_cpu)

        test_acc_tmp.append(test_acc), test_precision_tmp.append(test_precision), test_recall_tmp.append(test_recall), test_f1_tmp.append(test_f1)

    test_acc_mean = sum(test_acc_tmp, 0.0)/len(test_acc_tmp)
    test_precision_mean = sum(test_precision_tmp, 0.0)/len(test_precision_tmp)
    test_recall_mean = sum(test_recall_tmp, 0.0)/len(test_recall_tmp)
    test_f1_mean = sum(test_f1_tmp, 0.0)/len(test_f1_tmp)
    print("[Evaluation] {:.2f}[s], Test Accuracy : {:.4f}, Precision : {:.4f}, Recall : {:.4f}, F1 Score : {:.4f}".format(
        time.time()-test_start, test_acc_mean, test_precision_mean, test_recall_mean, test_f1_mean))
    print("[Model Performance] Model Performance : {:.5f}".format(test_acc_mean))

모델에 제시된 파라미터는 그대로 사용하고, 약간 변형하여 모델을 구축하였는데 높은 Accuracy를 보여줍니다. 

하지만 LOS/NLOS의 분류를 통해 UWB 성능을 올리는 방법을 제시하였지만, 논문의 Limitations으로 실제로 이 분류기를 통해 UWB 성능을 검증하지 못했습니다. 

그리고 제가 생각하는 또 다른 문제는 오픈소스의 데이터라고 생각합니다. 

이 데이터를 보았을 때, 굳이 딥러닝, 머신러닝을 사용할 필요가 있을까? 의문이 듭니다. 

1차원적으로 생각하였을 때 Threshold를 10000에서 잘라버리면, NLOS를 쉽게 지워버릴 수 있습니다. 

현 데이터는 허수(방향) 부분을 제외한 오직 크기의 성질만을 가지고 학습하였기 때문에, 과연 실제 환경 속에서 제대로 작동할지 의문이 듭니다. 

UWB 특성상 Nanosecond로 시간을 재는 방식이기 때문에 데이터 추출하는 것이 굉장히 굉장히 어려움이 있어 구현하는 것은 어려움이 있습니다. 그렇기 때문에 5년 전 오픈소스이지만, 2020년에도 이를 이용해 논문을 작성했을 것이라고 생각합니다.

안녕하세요. 오늘은 Deep Learning 분야에서 CNN의 BottleNeck구조에 대해 알아보겠습니다. 

대표적으로 ResNet에서 BottleNeck을 사용했습니다. 

ResNet에서 왼쪽은 BottleNeck 구조를 사용하지 않았고, 오른쪽은 BottleNeck 구조를 사용했습니다.

BottleNeck을 설명하기 전, Convolution의 Parameters을 계산할 줄 알아야 합니다. 이 부분은 다른 글에서 자세히 설명하겠습니다.

Convolution Parameters = Kernel Size x Kernel Size x Input Channel x Output Channel

BottleNeck의 핵심은 1x1 Convolution입니다. ( Pointwise Convolution 이라고도 합니다. 이는 Depthwise Separable Convolution에서도 똑같은 원리로 이용되기 때문에 잘 알아두면 좋습니다.)

1x1 Convolution의 Parameters는 1 x 1 x Input Channel x Output Channel입니다.

대게 1x1 Convolution은 연산량이 작기 때문에 Feature Map(Output Channel)을 줄이거나 키울 때 사용됩니다.

BottleNeck 구조

1. Input Channel = 256인 320x320  Input Image가 있다고 가정합니다. 

2. Channel Compression ( 채널 압축 )

Input Channel 256 -> Output Channel 64

256을 64로 채널을 강제로 축소한 이유는 오로지 연산량을 줄이기 위함입니다. 

1x1 Convolution에는 Spatial(공간적인) 특징을 가지고 있지 않습니다. Convolution 연산이 공간적인 특징을 추출하기 위해선 Kernel이 최소 2 이상 되어야 합니다. 

3. 특징 추출

Input Channel 64 -> Output Channel 64 

3x3 Convolution은 특성을 추출하는 역할을 합니다.

3x3 Convolution 연산은 = 3 x 3 x Input Channel x Output Channel 입니다. ( 3 x 3 x 64 x 64 )

3x3 Convolution은 1x1 Convolution 보다 9배 연산량이 많기 때문에, 1x1 Convolution에서 채널을 줄인 후에 

3x3 Convolution에서 특성을 추출합니다. 

4. Channel Increase( 채널 증가 )

Input Channel 64 -> Output Channel 256

CNN은 Feature Map의 특성이 많으면 많을수록 학습이 잘 되기 때문에, 1x1 Convolution으로 강제적으로 채널을 증가시켜줍니다. 

BottleNeck의 구조는 1x1 Convolution으로 장난을 치면서 연산량을 최소화하는 것입니다. 

하지만 강제로 채널을 줄이고 늘리는 것은 정보 손실을 일으킵니다. 

정보 손실은 모델의 정확성을 떨어뜨립니다. 

연산량과 정보손실은 서로 tradeoff 관계이기 때문에 서로의 합의점을 찾는 것이 중요합니다. 

ResNet이 제시한 두 구조를 Pytorch로 구현해 Parameter를 확인하겠습니다. 

Standard는 Channel 수가 적을지라도, 3x3 Convolution을 두 번 통과했고, 

BottleNeck은 1x1, 3x3, 1x1 순으로 Convolution을 통과하고, Channel 수는 4배 정도 많지만, Parameter가 세 배 정도 적습니다. 

그리고 형성된 것이 도로의 병목 현상과 비슷하다 하여 BottleNeck 구조라고 불립니다.

cf) Pytorch Code

# standard
class Standard(nn.Module):
    def __init__(self, in_dim=256, mid_dim=64, out_dim=64):
        super(BuildingBlock, self).__init__()
        self.building_block = nn.Sequential(
            nn.Conv2d(in_channels=in_dim, out_channels=mid_dim, kernel_size=3, padding=1, bias=False),
            nn.ReLU(),
            nn.Conv2d(in_channels=mid_dim, out_channels=out_dim, kernel_size=3, padding=1, bias=False),
        )
        self.relu = nn.ReLU()

    def forward(self, x):
        fx = self.building_block(x)  # F(x)
        out = fx + x  # F(x) + x
        out = self.relu(out)
        return out

# BottleNeck
class BottleNeck(nn.Module):
    def __init__(self, in_dim=256, mid_dim=64, out_dim=256):
        super(BottleNeck, self).__init__()
        self.bottleneck = nn.Sequential(
            nn.Conv2d(in_channels=in_dim, out_channels=mid_dim, kernel_size=1, bias=False),
            nn.ReLU(),
            nn.Conv2d(in_channels=mid_dim, out_channels=mid_dim, kernel_size=3, padding=1, bias=False),
            nn.ReLU(),
            nn.Conv2d(in_channels=mid_dim, out_channels=in_dim, kernel_size=1, bias=False),
        )

        self.relu = nn.ReLU()

    def forward(self, x):
        fx = self.bottleneck(x)  # F(x)
        out = fx + x  # F(x) + x
        out = self.relu(out)
        return out