CHAPTER5 이미지 처리 능력이 탁월한 CNN 1

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torchvision import transforms, datasets

USE_CUDA = torch.cuda.is_available()
DEVICE = torch.device("cuda" if USE_CUDA else "cpu")

# 하이퍼파라미터인 epoch 와 batch 사이즈 정해주기
EPOCHS = 40
BATCH_SIZE = 64

# 학습용, 테스트용 데이터셋 불러오기
# transforms를 이용한 전처리는 파이토치 텐서화, 정규화만 진행
train_loader = torch.utils.data.DataLoader(
    datasets.FashionMNIST('./.data',
                         train = True,
                         download = True,
                         transform = transforms.Compose([
                             transforms.ToTensor(),
                             transforms.Normalize((0.1307, ), (0.3081,))
                         ])),
    batch_size = BATCH_SIZE, shuffle = True
)

test_loader = torch.utils.data.DataLoader(
    datasets.FashionMNIST('./.data',
                         train = False,
                         download = True,
                         transform = transforms.Compose([
                             transforms.ToTensor(),
                             transforms.Normalize((0.1307, ), (0.3081, ))
                         ])),
    batch_size = BATCH_SIZE, shuffle = True
)

tmp_transform = transforms.Compose([transforms.ToTensor()])
tmp_trainset = datasets.FashionMNIST('./data', train = True, download = True, transform = tmp_transform)
print(list(tmp_trainset.train_data.size()))
print(tmp_trainset.train_data.float().mean()/255)
print(tmp_trainset.train_data.float().std()/255)

tmp_transform = transforms.Compose([transforms.ToTensor()])
tmp_trainset = datasets.MNIST('./data', train = True, download = True, transform = tmp_transform)
print(list(tmp_trainset.train_data.size()))
print(tmp_trainset.train_data.float().mean()/255)
print(tmp_trainset.train_data.float().std()/255)

class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        # Conv2D(입력 채널 수, 출력 채널 수)
        # 흑백이라 색상 채널이 1개 뿐.
        # 1개 받아서 10개의 특징 맵 생성하기
        self.conv1 = nn.Conv2d(1, 10, kernel_size = 5)
        # 10개 받아서 20개 생성하기
        self.conv2 = nn.Conv2d(10, 20, kernel_size = 5)

        #  컨볼루션 결과로 나온 출력값에 드롭아웃 해주기.
        # 이번엔 함수 사용하지 않고 nnDropout2d 모듈 사용해서 인스턴스 만들어보기.
        self.drop = nn.Dropout2d()

        # 일반 신경망
        self.fc1 = nn.Linear(320, 50) # 50은 그냥 임의로 한거임.
        self.fc2 = nn.Linear(50, 10) # 10개로 분류할거니까

    # 입력부터 출력까지 데이터가 지나가는 길 만들기
    def forward(self, x):
        # 컨볼루션 계층 거치고, 맥스 풀링 거치게 하기.
        # 맥스 풀링 후 relu 활성화 함수 거치기
        x = F.relu(F.max_pool2d(self.conv1(x), 2)) # 2 = 커널 크기
        x = F.relu(F.max_pool2d(self.conv2(x), 2))

        # 일반 신경망은 1차원의 입력을 받기 때문에 펴주야함.
        x = x.view(-1, 320)

        # 이제 특징들은 다 추출했으니까, 얘를 받아 분류하는 신경망 계층 구성.
        x = F.relu(self.fc1(x)) # relu 거치고
        x = self.drop(x, training = self.training) # 드롭아웃 해주고
        x = self.fc2(x)
        return x

# 모델 인스턴스 생성
model = Net().to(DEVICE)

# 최적화 알고리즘 = optim.SGD
# momentum = 미분 그래프가 3차 이상의 그래프라 최소점이 2개 이상 생기는 경우
# SGD 에 이전의 이동값을 고려하도록 설계함. (momentum = 관성)
# 지역 최소값에 도달하더라도 앞으로 나아가서 지역 최소값을 탈출할 수 있도록 설정.
optimizer = optim.SGD(model.parameters(), lr = 0.01, momentum = 0.5)

def train(model, train_loader, optimizer, epoch):
    model.train()
    for batch_idx, (data, target) in enumerate(train_loader):
        data, target = data.to(DEVICE), target.to(DEVICE)
        optimizer.zero_grad()
        output = model(data)
        loss = F.cross_entropy(output, target)
        loss.backward()
        optimizer.step()

        if batch_idx % 200 == 0:
            print(f"Train Epoch: {epoch} [{batch_idx * len(data)}/{len(train_loader.dataset)} ({100. * batch_idx / len(train_loader):.0f}%)]\tLoss: {loss.item():.6f}")

def evaluate(model, test_loader):
    model.eval()
    test_loss = 0
    correct = 0

    with torch.no_grad():
        for data, target in test_loader:
            data, target = data.to(DEVICE), target.to(DEVICE)
            output = model(data)

            # 배치 오차를 합산
            test_loss += F.cross_entropy(output, target,
                                        reduction = 'sum').item()

            # 가장 높은 값을 가진 인덱스가 예측값
            pred = output.max(1, keepdim = True)[1]
            correct += pred.eq(target.view_as(pred)).sum().item()

    test_loss /= len(test_loader.dataset)
    test_accuracy = 100. * correct / len(test_loader.dataset)
    return test_loss, test_accuracy


# 모델 실행
for epoch in range(1, EPOCHS + 1):
    train(model, train_loader, optimizer, epoch)
    test_loss, test_accuracy = evaluate(model, test_loader)

    print(f"[{epoch}] Test Loss: {test_loss:.4f}, Accuracy: {test_accuracy:.2f}%")