[PyTorc]PyTorcを使用した画像分類器の作成:LeNet-5モデルの作成、MNISTデータセットの分類


動作原理(command-line)


Vscode端末でcmd->activate pytorch_py38を作成(作成した仮想環境を使用)
  • train: python main.py --mode test --download True --output_dir ./output
  • eval: python main.py --mode eval --download True --output_dir ./output --checkpoint ./output/model_epoch14.pt
  • test: python main.py --mode test --download True --output_dir ./output --checkpoint ./output/model_epoch14.pt
  • チェックポイントは、作成されたptファイルのいずれかです.

    main.py


    モジュール、関数の読み込み

    import torch
    import torch.nn as nn
        # 그래프의 기본적 building blocks를 제공
    from torchvision.datasets import MNIST
        # torchvision: 유명한 데이터셋, 모델 아키텍처, 이미지 변환 등을 제공
        # torchvision.datasets: 빌트인 데이터셋을 제공하는 모듈
    import torchvision.transforms as transforms
        # 이미지 transform을 제공하는 모듈
    from torch.utils.data.dataloader import DataLoader
        # torch.utils.data.DataLoader: 데이터 로딩 유틸리티 클래스
    import torch.optim as optim
        # optimization 알고리즘들을 제공하는 패키지
        
    import argparse
        # sys.argv를 어떻게 parsing할 지 파악하는 등의 역할을 함
    import sys, os
    
    from lenet5 import LeNet5
    from loss import *
    from tools import *

    command-lineパラメータの読み込み

    ### command-line에서 사용자가 입력한 arguments 들을 parsing
    ### -> mode, 데이터셋 download 여부, output 파일 저장 위치, checkpoint 파일 지정
    def parse_args():
        parser = argparse.ArgumentParser(description="MNIST")
            # ArgumentParser 객체: command-line을 파이썬 데이터형으로 parsing할 수 있는 객체
    
        parser.add_argument('--mode',
                            dest = 'mode',
                            help = 'train / eval / test',
                            default = None,
                            type = str)
            
        parser.add_argument('--download',
                            dest = 'download',
                            help = 'download Fashion MNIST dataset',
                            default = False,
                            type = bool)
        parser.add_argument('--output_dir',
                            dest = 'output_dir',
                            help = 'output directory',
                            default = './output',
                            type = str)
        parser.add_argument('--checkpoint',
                            dest = 'checkpoint',
                            help = 'checkpoint trained model',
                            default = None,
                            type = str)
            # add_argument() 메서드: ArgumentParser이 command-line의 문자을 객체로 어떻게 변환시킬지 지정
    
        ### 사용자가 args를 지정하지 않았다면 올바른 사용법 출력 후 종료
        if len(sys.argv) == 1:
            parser.print_help()
            sys.exit()
        
        args = parser.parse_args()  
            # ArgumentParser.parse_args() 메서드: 인자 문자열을 객체로 변환 & namespace의 attribute로 설정
            # 예: Namespace(checkpoint=None, download=True, mode='train', output_dir='./output') 식으로 나옴
            # args는 <class 'argparse.Namespace'> 타입을 가짐
    
        return args

    データの読み込み

    def get_data():
        download_root = "./mnist_dataset"
        my_transform = transforms.Compose([
            transforms.Resize([32, 32]), 
            transforms.ToTensor(), 
            transforms.Normalize((0.5,), (1.0,))
        ])
            # torchvision.transforms.Compose 클래스: 파라미터로써 compose를 할 Transform 객체의 리스트인 transforms를 받음
            # torchvision.transforms.Resize 클래스: 입력 이미지를 특정 사이즈로 변환함
            # torchvision.transforms.ToTensor 클래스: [0, 255] 범위의 (H X W X C)의 PIL 이미지나 numpy.ndarray를 [0.0, 1.0] 범위의 (C X H X W) float tensor로 변경함
            # torchvision.transforms.Normalize 클래스: 
            #   tensor 이미지를 mean & standard deviation으로 정규화시킴
            #   파라미터 mean = 각 채널의 mean의 sequence / 파라미터 std = 각 채널의 standard deviation의 sequence
    
        train_dataset = MNIST(
            root=download_root,
            transform=my_transform,
            train=True,
            download=args.download)
        eval_dataset = MNIST(
            root=download_root,
            transform=my_transform,
            train=False,
            download=args.download
        )
        test_dataset = MNIST(
            root=download_root,
            transform=my_transform,
            train=False,
            download=args.download
        )
            # torchvision.datasets.MNIST 클래스 파라미터
            #   root: 데이터들의 root 디렉터리
            #   train: train된 파일에서 불러와 데이터셋을 만들지 여부
            #   download: True로 설정 시, 인터넷에서 데이터를 다운받아 root dir에 넣음
            #   transform: PIL 이미지를 받아 변형된 형태로 리턴하는 함수나 transform
    
        return train_dataset, eval_dataset, test_dataset

    DataLoaderの作成

    def make_dataloader(train_dataset, eval_dataset, test_dataset):
        train_loader = DataLoader(train_dataset,
                                  batch_size=8,
                                  num_workers=0,
                                  pin_memory=True,
                                  drop_last=True,
                                  shuffle=True)
        eval_loader = DataLoader(eval_dataset,
                                 batch_size=1,
                                 num_workers=0,
                                 pin_memory=True,
                                 drop_last=False,
                                 shuffle=False)
        test_loader = DataLoader(test_dataset,
                                 batch_size=1,
                                 num_workers=0,
                                 pin_memory=True,
                                 drop_last=False,
                                 shuffle=False)
            # torch.utils.data.DataLoader 클래스: 데이터셋과 sampler를 결합하고, 주어진 데이터셋에 대해 iterable을 제공
            #   사용 파라미터:
            #       batch_size: 한 에폭마다 들어갈 배치의 수. 기본은 1
            #       num_workers: 데이터 로딩에 쓸 subprocesses의 개수. 0이면 메인 프로세스에서 불러옴을 의미. 본인의 CPU, GPU 상황에 따라 주는 값을 달리하기
            #       drop_last : True로 설정 시, batch 사이즈를 맞추고 남은 데이터들을 drop 시킴. 디폴트는 False로, 남는 데이터들은 조금 작은 배치 크기를 형성함
            #       pin_memory: True로 설정 시, 데이터 로더는 Tensor를 CUDA pinned momory로 복사함.
            #       shuffle: False가 default이며, True로 설정 시 한 에폭이 끝날 때마다 데이터들의 순서를 섞음
            #   자료형 구성:
            #       [0]: tensor 형의 img가 batch_size개 있음
            #       [1]: tensor 형의 정답 라벨이 batch_size개 있음
            # eval, test는 평가 용이기 때문에 batch_size를 1로. drop_last, shuffle이 굳이 필요 없음
        
        return train_loader, eval_loader, test_loader

    学習モデル

    def train_model(_model, device, train_loader, epoch=15):
        model = _model(batch=8, n_classes=10, in_channel=1, in_width=32, in_height=32, is_train=True)
        model.to(device)
            # Module.to() 메서드: 파라미터와 버퍼들을 옮김(또는 cast). device 파라미터에는 torch.device 객체가 들어감
        model.train()
            # Module.train() 메서드: 모듈을 training 모드로 설정함. 
    
        ### optimizer(최적화): 최적(가장 좋은 성능을 보이는)의 파라미터를 찾는 방식 설정
        optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9)
            # torch.optim 패키지를 사용하기 위해선 현재 상태(state)를 저장 & 파라미터들을 업데이트(계산된 기울기에 따라)할 optimizer 객체를 선언해야 함
            # Module.parameters() 메서드: 모듈의 파라미터들에 대한 iterator를 리턴함. 보통 optimizer에게 전달됨.
            #   nn.Parameter: Tensor 형을 가짐. Module에 속성으로 할당 시 자동으로 매개변수로 등록됨
            # torch.optim.SGD 클래스: stochastic gradient descent를 적용함
            #   lr(=learning rate), momentum(default: 0)
    
        ### learning rate scheduler: 미리 학습 일정을 정해두고 그에 따라 학습률을 조정함 -> 상황에 맞게 학습률을 가변적으로 적당히 변경
        scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=5, gamma=0.1)
            # torch.optim.lr_scheduler: 에폭의 수에 따라 학습률을 조정하도록 함. 최적화 업데이트 이후 수행되어야 함
            # torch.optim.lr_scheduler.StepLR 클래스:
            #   지정한 step_size 에폭 단위로 학습률에 gamma를 곱해 학습률을 감소시킴
            #   step_size: 학습률 decay의 주기 / gamma: 기본은 0.1, 학습률 decay의 mulitplicative 인자
    
        ### criterion: 손실함수 클래스 불러오기
        criterion = FashionMNISTloss(device = device)
    
        ### 학습에 사용될 각종 수 설정
        epoch = epoch
        iter = 0
    
        ### 각 epoch마다 학습 진행
        for e in range(epoch):
            total_loss = 0
            for i, batch in enumerate(train_loader):
                ### DataLoader에서 불러온 데이터셋을 가져온 뒤 디바이스에 올리기
                img = batch[0]  # shape 예: torch.Size([8, 1, 32, 32]) -> [배치, 채널, 세로, 가로]
                gt = batch[1]   # (예) tensor([8, 4, 0, 9, 9, 2, 9, 2])
                img = img.to(device)
                gt = gt.to(device)
    
                ### 모델에서의 output을 받아옴
                out = model(img)
                    # torch.Size([8, 10]) ==> [배치, 클래스 수] 형태로 각 배치에서 클래스별 결과(내놓은 답)을 출력
                
                ### output과 정답을 비교해서 손실함수를 사용해 손실값을 구함
                loss_val = criterion(out, gt)
    
                ### 손실값을 사용해 오차역전파를 사용해 파라미터를 개선시킴
                loss_val.backward()
                    # Module 클래스에서 forward 함수만 정의하면,
                    # backward 함수는 'torch.autograd' 기능(신경망 학습을 지원하는 PyTorch의 자동 미분 엔진)을 통해 자동으로 정의됨
                optimizer.step()
                    # torch.optim.SGD.setp() 메서드: single 최적화 setp을 수행 -> 가중치 업데이트
                optimizer.zero_grad()
                    # 모든 매개변수의 gradient 버퍼를 0으로 만듦
                    #  ==> 기존의 계산된 기울기 값을 누적시키지 않을 때, 기존에 계산된 기울기를 0으로 만듦
    
                ### 전체 손실값을 누적시킴
                total_loss += loss_val.item()
                    # torch.tensor.item(): tensor의 값을 파이썬 표준 number로 리턴함. tensor에 하나의 elem만 있어야 함
                
                ### 일정 iter마다 loss를 출력해봄
                if iter % 1000 == 0:
                    print("{} epoch {} iter loss: {}".format(e, iter, loss_val.item()))
    
                iter += 1
            
            ### 한 에폭이 끝남 -> 평균 손실을 구하며 학습률 갱신, 텐서 저장
            mean_loss = total_loss / i  # 평균 손실값을 구함, i = train_loader size:  7499
            scheduler.step()    # 학습률을 갱신함
            print("-> {} epoch mean loss: {}".format(e, mean_loss))
            torch.save(model.state_dict(), args.output_dir + "/model_epoch"+str(e)+".pt")
                # torch.save() 함수: 파일로 객체를 저장함
                #   obj: 저장할 객체 / f: 파일 이름(관습적으로 *.pt 파일을 사용해 tensor를 저장함)
                # Module.state_dict() 메서드: 모듈의 모든 state를 담고 있는 딕셔너리를 리턴함
                #   파라미터들과 버퍼들을 포함함. key는 파라미터와 버퍼의 이름
                #   (ex) module.state_dict().keys()의 출력 = ['bias', 'weight']
        print("===== END Train =====")

    評価モデル

    def eval_model(_model, device, eval_loader):
        model = _model(batch=1, n_classes=10, in_channel=1, in_width=32, in_height=32)
        checkpoint = torch.load(args.checkpoint)    # trained model을 load함
        model.load_state_dict(checkpoint)
            # Module.load_state_dict() 메서드: state_dict로부터 해당 모듈과 그 자식들로 파라미터와 버퍼들을 복사해옴
        model.to(device)
        model.eval()
            # Module.eval() 메서드: 모듈을 evaluation mode로 설정함
    
        ### 평가에 쓰일 수 정의
        accuracy = 0    # 정확도
        num_eval = 0    # 평가 횟수
    
        ### load한 데이터셋마다 학습 진행
        for batch in eval_loader:
            img = batch[0]
            gt = batch[1]
            img = img.to(device)
    
            out = model(img)
            out = out.cpu()
                # Module.cpu() 메서드: 모든 모델의 파라미터 & 버퍼를 CPU로 옮김
    
            if out == gt:
                accuracy += 1    # 모델이 낸 답이 정답과 같으면 정확도 +1
    
            num_eval += 1
        
        print("Evaluation scroe: {} / {}".format(accuracy, num_eval))
            # 총 몇 개의 정답을 맞추었는가? (예) Evaluation scroe: 8682 / 10000

    モデルのテスト

    def test_model(_model, device, test_loader):
        model = _model(batch=1, n_classes=10, in_channel=1, in_width=1, in_height=1)
        checkpoint = torch.load(args.checkpoint)
        model.load_state_dict(checkpoint)
        model.to(device)
        model.eval()
    
        for batch in test_loader:
            img = batch[0]
            img = img.to(device)
    
            out = model(img)
            out = out.cpu()
    
            ### 결과를 이미지로 출력함
            show_img(img.cpu().numpy(), str(out.item()))
                # Tensor.numpy(): tensor를 ndarray로 변경함
                # 이미지 위에 모델이 낸 답을 씀

    main関数mainかんすう:フルファンクション

    def main():
        ### output_dir로 설정한 디렉터리가 존재하는 지 확인 -> 존재하지 않는다면 만들기
        if not os.path.isdir(args.output_dir):
            os.mkdir(args.output_dir)
    
        ### 어떤 device에 올릴지 지정
        if torch.cuda.is_available():
            device = torch.device("cuda")
        else:
            device = torch.device("cpu")
    
        ### FashionMNIST dataset을 불러옴
        train_dataset, eval_dataset, test_dataset = get_data()
    
        ### DataLoader를 만듦
        train_loader, eval_loader, test_loader = make_dataloader(train_dataset, eval_dataset, test_dataset)
    
    
        ### LeNet5 모델을 불러옮
        _model = LeNet5
    
        ### 모델 학습
        if args.mode == "train":
            train_model(_model, device, train_loader, epoch=15)
        elif args.mode == "eval":
            eval_model(_model, device, eval_loader)
        elif args.mode == "test":
            test_model(_model, device, test_loader)
    
    if __name__=="__main__":
        args = parse_args()
        main()

    lenet5.py

    import torch 
    import torch.nn as nn
    
    class LeNet5(nn.Module):
        # torch.nn.Module 클래스: 
        #   모든 신경망 모듈을 위한 base class. 모델을 만들려면 이 클래스를 subclass 해야 함
        #   Module 클래스는 다른 Module 들을 포함할 수 있음
        def __init__(self, batch, n_classes, in_channel, in_width, in_height, is_train=False):
            super().__init__()
    
            self.batch = batch
            self.n_classes = n_classes
            self.in_channel = in_channel
            self.in_width = in_width
            self.in_height = in_height
            self.is_train = is_train
    
            ### convolution & pooling layer
            # ※ convolution output = {(W - K + 2P) / S} + 1
            self.conv0 = nn.Conv2d(self.in_channel, 6, kernel_size=5, stride=1, padding=0)
                # torch.nn.Conv2d 클래스: 입력 신호에 대해 2D Convolution을 적용함
                #   필수 파라미터: input_channel, output_channel
                #   보통 입력 크기는 (B, C_in, H, W), 출력 크기는 (B, C_out, H_out, W_out)
                #   자료형으로 TensorFloat32를 사용
            self.pool0 = nn.AvgPool2d(2, stride=2)
                # torch.nn.AvgPool2d 클래스: 입력 신호에 대해 2D average pooling을 적용함
                #   필수 파라미터: kernel_size
            self.conv1 = nn.Conv2d(6, 16, kernel_size=5, stride=1, padding=0)
            self.pool1 = nn.AvgPool2d(2, stride=2)
            self.conv2 = nn.Conv2d(16, 120, kernel_size=5, stride=1, padding=0)
    
            ### fully-connected layer
            self.fc3 = nn.Linear(120, 84)
                # torch.nn.Linear 클래스: 입력 데이터에 대해 선형 변환을 적용. y = x * A.t + b
                #   필수 파라미터: in_features(각 input sample의 사이즈), out_features
            self.fc4 = nn.Linear(84, self.n_classes)
    
        def forward(self, x):
            # nn.Module.forward() 메서드: 매 call마다 수행될 내용. subclass에서 재정의되어야 함. 재정의한 부분이 하단
            #   Model 객체를 데이터와 함께 호출하면 자동으로 실행됨. 따라서 my_model = LeNet(input)으로 선언/호출 해도 자동으로 forward 수행됨
            # ※ x의 shape: [B, C, H, W]
    
            x = self.conv0(x)
            x = torch.tanh(x)
            x = self.pool0(x)
            x = self.conv1(x)
            x = torch.tanh(x)
            x = self.pool1(x)
            x = self.conv2(x)
            x = torch.tanh(x)
            
            x = torch.flatten(x, start_dim=1)
                # 4차원을 2차원으로 바꿈 ([b, c, h, w] -> [B, C*H*W])
    
            x = self.fc3(x)
            x = torch.tanh(x)
            x = self.fc4(x)
    
            x = x.view(self.batch, -1)
                # Tensor.view() 메서드: 인자로 주어진 tensor의 shape를 변경해 새 tensor로 리턴
                #   파라미터 shape에 -1 입력 시, 다른 dimension으로부터 값을 자동으로 추정함
    
            x = nn.functional.softmax(x, dim=1)
                # torch.nn.functional.softmax() 함수: [0, 1]의 범위를 갖도록 softmax 함수를 적용
    
            if self.is_train is False:
                x = torch.argmax(x, dim=1)
                    # torch.argmax() 함수: 입력 tensor에 대해 가장 큰 값을 리턴함
    
            return x

    loss.py

    import torch
    import torch.nn as nn
    import sys
    
    class MNISTloss(nn.Module):
        def __init__(self, device = torch.device('cpu')):
            super(MNISTloss, self).__init__()
                # super(자식 클래스, self)로 부모 클래스의 메서드 호출. super()와 동일. 현재 클래스가 어떤 클래스인지 명확히 표시
            self.loss = nn.CrossEntropyLoss().to(device)
                # torch.nn.CrossEntropyLoss 클래스: input과 target 간의 cross entropy loss를 계산하는 criterion. 분류 문제에 많이 사용됨.
    
        def forward(self, out, gt):
            # nn.Module.forward() 메서드: 매 call마다 수행될 내용. subclass에서 재정의되어야 함. 재정의한 부분이 하단
            #   Model 객체를 데이터와 함께 호출하면 자동으로 실행됨. 따라서 my_model = LeNet(input)으로 선언/호출 해도 자동으로 forward 수행됨
            loss_val = self.loss(out, gt)
                # loss(input, target) 식으로 사용
                # 클래스(분류 목록)에 대한 확률로 target이 주어진다면, shape는 input과 같아야 하고 각 값은 [0, 1] 범위의 값이어야 함
                # CrossEntropyLoss() 선언 시 reduction이 none으로 설정되었다면 target과 같은 shape를, 그렇지 않다면 sclar 반환
            
            return loss_val

    tools.py

    from PIL import Image, ImageDraw
        # PIL: Python Image Libaray, 파이썬 인터프리터에서 이미지를 처리할 수 있도록 함
        #   Image 모듈: PIL 이미지를 나타낼 수 있도록 함
        #   ImageDraw 모듈: Image형 객체에 대해 2D 그래픽 제공. 새 이미지 만들거나 annotate 하는 등 가능
    import numpy as np
    import matplotlib.pyplot as plt
    
    def show_img(img_data, text):
        _img_data = img_data * 255
            # 기존에 [0, 1] 범위였던 픽셀값들을 [0, 255] 범위로 바꿈
            # (1, 1, 32, 32) 형상을 가짐
    
        _img_data = np.array(_img_data[0, 0], dtype=np.uint8)
            # 4차원 데이터를 2차원으로 바꿈. 첫 번째 배치 첫 번째 채널이 [32, 32] 이므로
        img_data = Image.fromarray(_img_data)
            # PIL.Image.fromarray() 함수: array 객체로부터 이미지 메모리를 생성
        draw = ImageDraw.Draw(img_data)
            # PIL.ImageDraw.Draw() 함수: 입력된 이미지에 draw를 할 수 있는 객체를 생성
    
        cx, cy = _img_data.shape[0] / 2, _img_data.shape[1] / 2
        if text is not None:
            draw.text((cx, cy), text)   # 이미지의 중심에 text 내용을 써넣음
    
        plt.imshow(img_data)
        plt.show()