import numpy as np
import torch
from torch.utils.data import Dataset
from torchvision import datasets, transforms이미지 데이터셋 다운로드
torchvision의 Image Transform 에 대하여 생소하다면 다음의 링크를 참고해 주시기 바랍니다.
- torchvision의 transform으로 이미지 정규화하기(평균, 표준편차를 계산하여 적용
- PyTorch 이미지 데이터셋(Image Dataset) 구성에 관한 거의 모든 것!
rps 데이터셋을 다운로드 받아서 tmp 폴더에 압축을 풀어 줍니다.
rps데이터셋은 가위바위보 이미지 데이터셋 입니다.
import urllib.request
import zipfile
# 데이터셋을 다운로드 합니다.
# 다운로드 후 tmp 폴더에 압축을 해제 합니다.
url = 'https://storage.googleapis.com/download.tensorflow.org/data/rps.zip'
urllib.request.urlretrieve(url, 'rps.zip')
local_zip = 'rps.zip'
zip_ref = zipfile.ZipFile(local_zip, 'r')
zip_ref.extractall('tmp/')
zip_ref.close()rps 데이터셋을 시각화 하기 위하여 임시 DataLoader를 생성합니다.
from torchvision.datasets import ImageFolder
from torch.utils.data import DataLoader
# 이미지 폴더로부터 데이터를 로드합니다.
dataset = ImageFolder(root='tmp/rps', # 다운로드 받은 폴더의 root 경로를 지정합니다.
transform=transforms.Compose([
transforms.ToTensor(),
]))
data_loader = DataLoader(dataset,
batch_size=32,
shuffle=True,
num_workers=8
)# ImageFolder로부터 로드한 dataset의 클래스를 확인합니다.
# 총 3개의 클래스로 구성되었음을 확인할 수 있습니다(paper, rock, scissors)
dataset.classes['paper', 'rock', 'scissors']# 1개의 배치를 추출합니다.
images, labels = next(iter(data_loader))# 이미지의 shape을 확인합니다. 300 X 300 RGB 이미지 임을 확인합니다.
images[0].shapetorch.Size([3, 300, 300])rps 데이터셋 시각화
- 총 3개의 class(가위/바위/보)로 구성된 사진 파일입니다.
- 출처: Laurence Moroney Dataset
import matplotlib.pyplot as plt
# ImageFolder의 속성 값인 class_to_idx를 할당
labels_map = {v:k for k, v in dataset.class_to_idx.items()}
figure = plt.figure(figsize=(12, 8))
cols, rows = 8, 4
# 이미지를 출력합니다. RGB 이미지로 구성되어 있습니다.
for i in range(1, cols * rows + 1):
sample_idx = torch.randint(len(images), size=(1,)).item()
img, label = images[sample_idx], labels[sample_idx].item()
figure.add_subplot(rows, cols, i)
plt.title(labels_map[label])
plt.axis("off")
# 본래 이미지의 shape은 (3, 300, 300) 입니다.
# 이를 imshow() 함수로 이미지 시각화 하기 위하여 (300, 300, 3)으로 shape 변경을 한 후 시각화합니다.
plt.imshow(torch.permute(img, (1, 2, 0)))
plt.show()
train / validation 데이터셋 split
현재 rps에 하나의 데이터셋으로 구성된 Image 파일을 2개의 데이터셋(train/test)으로 분할하도록 하겠습니다.
Image Augmentation 적용
Image Augmentation을 적용 합니다.
# Image Transform을 지정합니다.
image_transform = transforms.Compose([
transforms.Resize(224), # Resize (300 X 300) -> (224, 224) 크기 조정
transforms.RandomHorizontalFlip(0.5), # 50% 확률로 Horizontal Flip
transforms.ToTensor(), # Tensor 변환 (정규화)
# transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225)),
])# 이미지 폴더로부터 데이터를 로드합니다.
dataset = ImageFolder(root='tmp/rps', # 다운로드 받은 폴더의 root 경로를 지정합니다.
transform=image_transform) # Image Augmentation 적용 # Image Augmentation 이 적용된 DataLoader를 로드 합니다.
data_loader = DataLoader(dataset,
batch_size=32,
shuffle=True,
num_workers=8
)
# 1개의 배치를 추출합니다.
images, labels = next(iter(data_loader))# ImageFolder의 속성 값인 class_to_idx를 할당
labels_map = {v:k for k, v in dataset.class_to_idx.items()}
figure = plt.figure(figsize=(12, 8))
cols, rows = 8, 4
# 이미지를 출력합니다. RGB 이미지로 구성되어 있습니다.
for i in range(1, cols * rows + 1):
sample_idx = torch.randint(len(images), size=(1,)).item()
img, label = images[sample_idx], labels[sample_idx].item()
figure.add_subplot(rows, cols, i)
plt.title(labels_map[label])
plt.axis("off")
# 본래 이미지의 shape은 (3, 300, 300) 입니다.
# 이를 imshow() 함수로 이미지 시각화 하기 위하여 (300, 300, 3)으로 shape 변경을 한 후 시각화합니다.
plt.imshow(torch.permute(img, (1, 2, 0)))
plt.show()
from torch.utils.data import random_split
ratio = 0.8 # 학습셋(train set)의 비율을 설정합니다.
train_size = int(ratio * len(dataset))
test_size = len(dataset) - train_size
print(f'total: {len(dataset)}\ntrain_size: {train_size}\ntest_size: {test_size}')
# random_split으로 8:2의 비율로 train / test 세트를 분할합니다.
train_data, test_data = random_split(dataset, [train_size, test_size])total: 2520
train_size: 2016
test_size: 504torch.utils.data.DataLoader
DataLoader는 배치 구성과 shuffle등을 편하게 구성해 주는 util 입니다.
batch_size = 32 # batch_size 지정
num_workers = 8 # Thread 숫자 지정 (병렬 처리에 활용할 쓰레드 숫자 지정)
train_loader = DataLoader(train_data,
batch_size=batch_size,
shuffle=True,
num_workers=num_workers
)
test_loader = DataLoader(test_data,
batch_size=batch_size,
shuffle=False,
num_workers=num_workers
)train_loader의 1개 배치의 shape 출력
images, labels = next(iter(train_loader))
images.shape, labels.shape(torch.Size([32, 3, 224, 224]), torch.Size([32]))배치사이즈인 32가 가장 첫번째 dimension에 출력되고, 그 뒤로 채널(3), 세로(224px), 가로(224px) 순서로 출력이 됩니다.
즉, 224 X 224 RGB 컬러 이미지 32장이 1개의 배치로 구성이 되어 있습니다.
# 1개의 이미지의 shape를 확인합니다.
# 224 X 224 RGB 이미지가 잘 로드 되었음을 확인합니다.
images[0].shapetorch.Size([3, 224, 224])모델 정의
CUDA 설정이 되어 있다면 cuda를! 그렇지 않다면 cpu로 학습합니다.
(제 PC에는 GPU가 2대 있어서 cuda:0로 GPU 장비의 index를 지정해 주었습니다. 만약 다른 장비를 사용하고 싶다면 cuda:1 이런식으로 지정해 주면 됩니다)
# device 설정 (cuda:0 혹은 cpu)
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(device)cuda:0아래의 모델은 DNN으로 구성하였습니다. 추후, 모델 부분을 CNN이나 pre-trained model로 교체할 수 있습니다.
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
class CNNModel(nn.Module):
def __init__(self):
super(CNNModel, self).__init__()
self.sequential = nn.Sequential(
nn.Conv2d(in_channels=3, out_channels=32, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2, 2),
nn.Conv2d(in_channels=32, out_channels=64, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2, 2),
nn.Conv2d(in_channels=64, out_channels=64, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2, 2),
nn.Conv2d(in_channels=64, out_channels=128, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2, 2),
nn.Conv2d(in_channels=128, out_channels=128, kernel_size=3, padding=1),
nn.ReLU(),
nn.MaxPool2d(2, 2)
)
self.fc = nn.Linear(7*7*128, 3)
def forward(self, x):
x = self.sequential(x)
x = torch.flatten(x, 1)
x = self.fc(x)
return x model = CNNModel() # Model 생성
model.to(device) # device 에 로드 (cpu or cuda)CNNModel(
(sequential): Sequential(
(0): Conv2d(3, 32, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
(1): ReLU()
(2): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
(3): Conv2d(32, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
(4): ReLU()
(5): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
(6): Conv2d(64, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
(7): ReLU()
(8): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
(9): Conv2d(64, 128, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
(10): ReLU()
(11): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
(12): Conv2d(128, 128, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
(13): ReLU()
(14): MaxPool2d(kernel_size=2, stride=2, padding=0, dilation=1, ceil_mode=False)
)
(fc): Linear(in_features=6272, out_features=3, bias=True)
)torchsummary의 summary로 CNNModel의 구조와 paramter 수를 요약 출력 합니다.
- 설치되어 있지 않다면
pip install torchsummary로 설치할 수 있습니다.
from torchsummary import summary
summary(model, (3, 224, 224))----------------------------------------------------------------
Layer (type) Output Shape Param #
================================================================
Conv2d-1 [-1, 32, 224, 224] 896
ReLU-2 [-1, 32, 224, 224] 0
MaxPool2d-3 [-1, 32, 112, 112] 0
Conv2d-4 [-1, 64, 112, 112] 18,496
ReLU-5 [-1, 64, 112, 112] 0
MaxPool2d-6 [-1, 64, 56, 56] 0
Conv2d-7 [-1, 64, 56, 56] 36,928
ReLU-8 [-1, 64, 56, 56] 0
MaxPool2d-9 [-1, 64, 28, 28] 0
Conv2d-10 [-1, 128, 28, 28] 73,856
ReLU-11 [-1, 128, 28, 28] 0
MaxPool2d-12 [-1, 128, 14, 14] 0
Conv2d-13 [-1, 128, 14, 14] 147,584
ReLU-14 [-1, 128, 14, 14] 0
MaxPool2d-15 [-1, 128, 7, 7] 0
Linear-16 [-1, 3] 18,819
================================================================
Total params: 296,579
Trainable params: 296,579
Non-trainable params: 0
----------------------------------------------------------------
Input size (MB): 0.57
Forward/backward pass size (MB): 46.94
Params size (MB): 1.13
Estimated Total Size (MB): 48.65
----------------------------------------------------------------# 옵티마이저를 정의합니다. 옵티마이저에는 model.parameters()를 지정해야 합니다.
optimizer = optim.Adam(model.parameters(), lr=0.0005)
# 손실함수(loss function)을 지정합니다. Multi-Class Classification 이기 때문에 CrossEntropy 손실을 지정하였습니다.
loss_fn = nn.CrossEntropyLoss()훈련(Train)
from tqdm import tqdm # Progress Bar 출력def model_train(model, data_loader, loss_fn, optimizer, device):
# 모델을 훈련모드로 설정합니다. training mode 일 때 Gradient 가 업데이트 됩니다. 반드시 train()으로 모드 변경을 해야 합니다.
model.train()
# loss와 accuracy 계산을 위한 임시 변수 입니다. 0으로 초기화합니다.
running_loss = 0
corr = 0
# 예쁘게 Progress Bar를 출력하면서 훈련 상태를 모니터링 하기 위하여 tqdm으로 래핑합니다.
prograss_bar = tqdm(data_loader)
# mini-batch 학습을 시작합니다.
for img, lbl in prograss_bar:
# image, label 데이터를 device에 올립니다.
img, lbl = img.to(device), lbl.to(device)
# 누적 Gradient를 초기화 합니다.
optimizer.zero_grad()
# Forward Propagation을 진행하여 결과를 얻습니다.
output = model(img)
# 손실함수에 output, label 값을 대입하여 손실을 계산합니다.
loss = loss_fn(output, lbl)
# 오차역전파(Back Propagation)을 진행하여 미분 값을 계산합니다.
loss.backward()
# 계산된 Gradient를 업데이트 합니다.
optimizer.step()
# output의 max(dim=1)은 max probability와 max index를 반환합니다.
# max probability는 무시하고, max index는 pred에 저장하여 label 값과 대조하여 정확도를 도출합니다.
_, pred = output.max(dim=1)
# pred.eq(lbl).sum() 은 정확히 맞춘 label의 합계를 계산합니다. item()은 tensor에서 값을 추출합니다.
# 합계는 corr 변수에 누적합니다.
corr += pred.eq(lbl).sum().item()
# loss 값은 1개 배치의 평균 손실(loss) 입니다. img.size(0)은 배치사이즈(batch size) 입니다.
# loss 와 img.size(0)를 곱하면 1개 배치의 전체 loss가 계산됩니다.
# 이를 누적한 뒤 Epoch 종료시 전체 데이터셋의 개수로 나누어 평균 loss를 산출합니다.
running_loss += loss.item() * img.size(0)
# 누적된 정답수를 전체 개수로 나누어 주면 정확도가 산출됩니다.
acc = corr / len(data_loader.dataset)
# 평균 손실(loss)과 정확도를 반환합니다.
# train_loss, train_acc
return running_loss / len(data_loader.dataset), acc평가(Evaluate)
def model_evaluate(model, data_loader, loss_fn, device):
# model.eval()은 모델을 평가모드로 설정을 바꾸어 줍니다.
# dropout과 같은 layer의 역할 변경을 위하여 evaluation 진행시 꼭 필요한 절차 입니다.
model.eval()
# Gradient가 업데이트 되는 것을 방지 하기 위하여 반드시 필요합니다.
with torch.no_grad():
# loss와 accuracy 계산을 위한 임시 변수 입니다. 0으로 초기화합니다.
corr = 0
running_loss = 0
# 배치별 evaluation을 진행합니다.
for img, lbl in data_loader:
# device에 데이터를 올립니다.
img, lbl = img.to(device), lbl.to(device)
# 모델에 Forward Propagation을 하여 결과를 도출합니다.
output = model(img)
# output의 max(dim=1)은 max probability와 max index를 반환합니다.
# max probability는 무시하고, max index는 pred에 저장하여 label 값과 대조하여 정확도를 도출합니다.
_, pred = output.max(dim=1)
# pred.eq(lbl).sum() 은 정확히 맞춘 label의 합계를 계산합니다. item()은 tensor에서 값을 추출합니다.
# 합계는 corr 변수에 누적합니다.
corr += torch.sum(pred.eq(lbl)).item()
# loss 값은 1개 배치의 평균 손실(loss) 입니다. img.size(0)은 배치사이즈(batch size) 입니다.
# loss 와 img.size(0)를 곱하면 1개 배치의 전체 loss가 계산됩니다.
# 이를 누적한 뒤 Epoch 종료시 전체 데이터셋의 개수로 나누어 평균 loss를 산출합니다.
running_loss += loss_fn(output, lbl).item() * img.size(0)
# validation 정확도를 계산합니다.
# 누적한 정답숫자를 전체 데이터셋의 숫자로 나누어 최종 accuracy를 산출합니다.
acc = corr / len(data_loader.dataset)
# 결과를 반환합니다.
# val_loss, val_acc
return running_loss / len(data_loader.dataset), acc모델 훈련(training) & 검증
# 최대 Epoch을 지정합니다.
num_epochs = 10
min_loss = np.inf
# Epoch 별 훈련 및 검증을 수행합니다.
for epoch in range(num_epochs):
# Model Training
# 훈련 손실과 정확도를 반환 받습니다.
train_loss, train_acc = model_train(model, train_loader, loss_fn, optimizer, device)
# 검증 손실과 검증 정확도를 반환 받습니다.
val_loss, val_acc = model_evaluate(model, test_loader, loss_fn, device)
# val_loss 가 개선되었다면 min_loss를 갱신하고 model의 가중치(weights)를 저장합니다.
if val_loss < min_loss:
print(f'[INFO] val_loss has been improved from {min_loss:.5f} to {val_loss:.5f}. Saving Model!')
min_loss = val_loss
torch.save(model.state_dict(), 'DNNModel.pth')
# Epoch 별 결과를 출력합니다.
print(f'epoch {epoch+1:02d}, loss: {train_loss:.5f}, acc: {train_acc:.5f}, val_loss: {val_loss:.5f}, val_accuracy: {val_acc:.5f}')100% 63/63 [00:01<00:00, 40.66it/s][INFO] val_loss has been improved from inf to 0.21306. Saving Model!
epoch 01, loss: 0.87415, acc: 0.54167, val_loss: 0.21306, val_accuracy: 0.90873100% 63/63 [00:01<00:00, 41.15it/s][INFO] val_loss has been improved from 0.21306 to 0.04787. Saving Model!
epoch 02, loss: 0.10367, acc: 0.96974, val_loss: 0.04787, val_accuracy: 0.99008100% 63/63 [00:01<00:00, 40.99it/s][INFO] val_loss has been improved from 0.04787 to 0.01862. Saving Model!
epoch 03, loss: 0.05744, acc: 0.98710, val_loss: 0.01862, val_accuracy: 0.99603100% 63/63 [00:01<00:00, 41.19it/s][INFO] val_loss has been improved from 0.01862 to 0.01571. Saving Model!
epoch 04, loss: 0.01769, acc: 0.99603, val_loss: 0.01571, val_accuracy: 0.99206100% 63/63 [00:01<00:00, 40.45it/s][INFO] val_loss has been improved from 0.01571 to 0.00385. Saving Model!
epoch 05, loss: 0.00833, acc: 0.99653, val_loss: 0.00385, val_accuracy: 1.00000100% 63/63 [00:01<00:00, 40.82it/s]epoch 06, loss: 0.00504, acc: 0.99851, val_loss: 0.00403, val_accuracy: 1.00000100% 63/63 [00:01<00:00, 41.30it/s][INFO] val_loss has been improved from 0.00385 to 0.00156. Saving Model!
epoch 07, loss: 0.00179, acc: 1.00000, val_loss: 0.00156, val_accuracy: 1.00000100% 63/63 [00:01<00:00, 40.82it/s][INFO] val_loss has been improved from 0.00156 to 0.00061. Saving Model!
epoch 08, loss: 0.00055, acc: 1.00000, val_loss: 0.00061, val_accuracy: 1.00000100% 63/63 [00:01<00:00, 40.36it/s][INFO] val_loss has been improved from 0.00061 to 0.00031. Saving Model!
epoch 09, loss: 0.00033, acc: 1.00000, val_loss: 0.00031, val_accuracy: 1.00000100% 63/63 [00:01<00:00, 40.91it/s][INFO] val_loss has been improved from 0.00031 to 0.00028. Saving Model!
epoch 10, loss: 0.00021, acc: 1.00000, val_loss: 0.00028, val_accuracy: 1.00000저장한 가중치 로드 후 검증 성능 측정
# 모델에 저장한 가중치를 로드합니다.
model.load_state_dict(torch.load('DNNModel.pth'))# 최종 검증 손실(validation loss)와 검증 정확도(validation accuracy)를 산출합니다.
final_loss, final_acc = model_evaluate(model, test_loader, loss_fn, device)
print(f'evaluation loss: {final_loss:.5f}, evaluation accuracy: {final_acc:.5f}')evaluation loss: 0.00030, evaluation accuracy: 1.00000