Custom Dataset, Dataloader, Transform

1. landmark가 있는 Custom Dataset

일반적이지 않은 데이터셋으로부터 데이터를 읽어오고 전처리하는 튜토리얼

dataset download url : https://download.pytorch.org/tutorial/faces.zip

이 데이터셋은 landmark가 있는 데이터셋이다.

Dataset 클래스

len(dataset) 에서 호츨되는 len 은 데이터셋의 크기를 리턴해야 한다.

dataset[i] 에서 호출되는 getitem 은 i번쨰 샘플을 찾는데 사용된다.

__init__을 사용하여 CSV 파일 안에 있는 데이터를 읽지만, __getitem__을 이용해서 이미지를 판독해야 한다. 이 방법은 모든 이미지를 메모리에 저장하지 않고 필요할때마다 읽기 때문에 메모리에 효율적이다.

데이터 샘플은 {‘image’: image, ‘landmarks’: landmarks}의 사전 형태를 갖는다.

from __future__ import print_function, division
import os
import torch
import pandas as pd
from skimage import io, transform
from skimage.transform import rescale, resize
import numpy as np
import matplotlib.pyplot as plt
from torch.utils.data import Dataset, DataLoader
from torchvision import transforms, utils
from PIL import Image
import cv2

import warnings
warnings.filterwarnings("ignore")

plt.ion() # 반응형 모드

class FaceLandmarksDataset(Dataset):
  """ Face Landmarks dataset """
  
  def __init__(self, csv_file, root_dir, transform = None):
    """ Args:
              csv_file (string): csv 파일 경로
              root_dir (string): 모든 이미지가 존재하는 디렉토리 경로
              transform (callable): 샘플에 적용할 optional transform
    """
    self.landmarks_frame = pd.read_csv(csv_file)  # csv_file을 불러온다.
    self.root_dir = root_dir                      # 모든 이미지가 존재하는 디렉토리 
    self.transform = transform

  def __len__(self):
    return len(self.landmarks_frame)              # 데이터의 개수, 즉 image 개수라고 할 수 있다.

  def __getitem__(self, idx):
    if torch.is_tensor(idx):
      idx = idx.tolist()                          # idx가 tensor이면 list로 바꾼다.

    img_name = os.path.join(self.root_dir, self.landmarks_frame.iloc[idx, 0]) # root_dir 안의 csv_file에서 모든 이미지를 불러와 할당
    image = io.imread(img_name)                   # 이미지 읽기

    landmarks = self.landmarks_frame.iloc[idx, 1:]      # csv_file에서 각 idx 마다의 데이터들을 다 불러온다.
    landmarks = np.array([landmarks])                   # reshape를 위해 list가 아닌 numpy
    landmarks = landmarks.astype('float').reshape(-1,2) # (k,2) 배열로 변환
    sample = {'image': image, 'landmarks': landmarks}   # 각 image와 landmarks(keypoint)를 연결

    if self.transform:
      sample = self.transform(sample)
    
    return sample

''' landmark를 보기 위한 함수 '''
def show_landmarks(image, landmarks):
    """Show image with landmarks"""
    """ 랜드마크(landmark)와 이미지를 보여줍니다. """
    plt.imshow(image)
    plt.scatter(landmarks[:, 0], landmarks[:, 1], s=10, marker='.', c='r')
    plt.pause(0.001)  # 갱신이 되도록 잠시 멈춥니다.

face_dataset = FaceLandmarksDataset(csv_file = '/content/drive/MyDrive/data/faces/face_landmarks.csv',
                                   root_dir = '/content/drive/MyDrive/data/faces/') # __init__ 에 대한 args들을 받으면 len, getitem을 계산해놓고, 추후에 len이나 dataset[i]를 실행하면 리턴해줌

fig = plt.figure()

for i in range(len(face_dataset)):
  sample = face_dataset[i] # i만큼 idx를 지정

  print("number: {} \t image shape: {} \t label shape: {} \n len image: {} \t len label: {}"
        .format(i, sample['image'].shape, sample['landmarks'].shape, len(sample['image']),len(sample['landmarks'])))
  ''' 
      image - 1개의 이미지에 대해 324 row(높이) x 215 col(넓이) x 3 dim(RGB)
      label - 총 68개의 landmarks(keypoint) 68 row x 2 col
  '''

  ax = plt.subplot(1, 4, i+1)
  plt.tight_layout()
  ax.set_title('Sample #{}'.format(i))
  ax.axis('off')
  show_landmarks(**sample) # **sample = io.imread(os.path.join('/content/drive/MyDrive/data/faces', img_name)),landmarks

  if i == 3:
    plt.show()
    break

Q. 왜 landmarks shape이 (68,2)가 되어야 하지?

데이터가 columns를 보면 x,y가 존재한다. 따라서 1개의 landmark(keypoint)를 표현하기 위해 x,y값을 묶어야 한다. => 묶어서 보았을 때 이미지 상에 총 68개 keypoint가 존재한다.

Q. image와 label의 size를 맞춰야 하는 것 아닌가?

지금 출력하고 있는 것은 각각의 이미지의 height x weight x depth 이다. 따라서 총 이미지 크기가 아니기 때문에 맞지 않았던 것이다. 총 images 들의 len과 labels 들의 len은 맞는다.

Transform 클래스

샘플들을 data augmentation 시킨다.

• rescale: 이미지의 크기를 조절
• randomCrop: 무작위 자르기, data augmentation
• ToTensor: numpy이미지를 torch이미지로 변경( 축 변환 )

클래스를 작성할 때는 call 함수를 구현해야 한다. 필요하다면 init 함수도 구현해야 한다.

class Rescale(object):
  '''
  주어진 사이즈로 샘플 크기 조정

  Args: output size(tuple or int): 원하는 사이즈 값
                                   tuple인 경우 해당 tuple이 결과물의 크기가 되고, 
                                   int라면 비율을 유지하면서 길이가 작은 쪽이 output size가 된다.
  '''

  def __init__(self, output_size):
    assert isinstance(output_size, (int, tuple))    # assert = 뒤의 조건이 True 가 아니면 에러를 발생시킨다. output size가 int나 tuple 타입이면 true
    self.output_size = output_size

  def __call__(self,sample):
    image, landmarks = sample['image'], sample['landmarks']

    h, w = image.shape[:2]                # 각 image.shape는 [h,w,c] 로 구성되어 있다. 따라서 2번째까지만 추출
    if isinstance(self.output_size, int): # args - output_size가 int라면
      if h > w:                           # h/w 비율을 곱하여 new_h가 더 큰 값이 되도록
        new_h, new_w = self.output_size * h / w, self.output_size 
      else:
        new_h, new_w = self.output_size, self.output_size * w / h
    
    else:                                 # tuple이라면 그대로 출력
      new_h, new_w = self.output_size
    

    new_h, new_w = int(new_h), int(new_w)           # 크기는 int로 출력해야 되기에

    img = cv2.resize(image, (new_h, new_w))   # image를 내가 원하는 사이즈의 값으로 크기 조정
    # 강의에서는 transform.resize 로 되어있는데 실행되지 않아 cv2로 바꾸어 실행해보았다.

    # print("landmarks data: {} \nlandmarks shape: {} \nnew_w data: {} \nnew_h data: {}".format(landmarks, landmarks.shape, new_w, new_h) )

    landmarks = landmarks * [ new_w / w, new_h / h ]

    return {'image': img, 'landmarks': landmarks}

class RandomCrop(object):
  '''
  샘플데이터를 무작위로 자름

  Args: output_size (tuple or int): 줄이고자 하는 크기로 int면 정사각형, tuple이라면 지정된 크기로
  '''

  def __init__(self, output_size):
    assert isinstance(output_size, (int, tuple))
    if isinstance(output_size, int):                # int
      self.output_size = (output_size, output_size)  # 그대로
    else:                                           # tuple
      assert len(output_size) == 2                  # size가 2일때만
      self.output_size = output_size

  def __call__(self, sample):
    image, landmarks = sample['image'], sample['landmarks']

    h, w = image.shape[:2]                        # 256 이라고 하면
    new_h, new_w = self.output_size               # new = 224,224
    
    top = np.random.randint(0, h - new_h)         # (0 ~ (256 - 224)) 값 사이의 random추출
    left = np.random.randint(0, w - new_w)        # 원본을 넘게 자르도록 설정되지 않도록 하기 위함
    
    image = image[top: top + new_h,               # image의 높이를 새로 설정한 크기와 random하게 추출한 값을 더함
                  left: left + new_w]             # 원본보다는 무조건 작거나 같다.
    
    landmarks = landmarks - [left,top]            # keypoint의 위치도 left,top을 뺌


    return {'image': image, 'landmarks': landmarks}

class ToTensor(object):
  '''
  numpy array를 tensor(torch)로 변환
  '''

  def __call__(self,sample):
    image, landmarks = sample['image'], sample['landmarks']

    # swap color axis ( numpy image: HxWxC => torch image: CxHxW )
    image = image.transpose((2,0,1))
    
    return {'image':torch.from_numpy(image),
            'landmarks':torch.from_numpy(landmarks)}

transform = transforms.Compose([Rescale(256),
                                RandomCrop(224)])   # output_size = 256,256  # output_size = 224,224

fig = plt.figure()
sample = face_dataset[60]
for i in range(1):
  image_transforms = transform(sample)

  ax = plt.subplot(1,3,i+1)
  plt.tight_layout()
  ax.set_title(type(transform).__name__)
  show_landmarks(**image_transforms)

plt.show()

transformed_dataset = FaceLandmarksDataset(csv_file = '/content/drive/MyDrive/data/faces/face_landmarks.csv',
                                          root_dir = '/content/drive/MyDrive/data/faces/',
                                          transform=transforms.Compose([
                                               Rescale(256),
                                               RandomCrop(224),
                                               ToTensor()
                                           ]))

Q. landmark에다 new_w/w,new_h를 왜 곱하는걸까

landmarks data가 각 keypoint의 좌표를 찍어놓은 것이기 때문에 landmark 데이터에도 비율만큼 곱해줘야 맞는 위치를 찍을 수 있다.

Q. crop 할 때 왜 224가 output_size가 되야 할텐데 top+new_w를 하면 224가 무조건 나오는 건 아니지 않나. 이름만 output_size인가

일단 음,, 실행이 되지 않음

Dataloader 클래스

torch.utils.data.DataLoader를 사용하면 반복자(iterator)로서 많은 기능을 해준다.

dataloader = DataLoader(transformed_dataset, batch_size=4, shuffle=True, num_workers=0)

# 배치하는 과정을 보여주는 함수입니다.
def show_landmarks_batch(sample_batched):
    """Show image with landmarks for a batch of samples."""
    images_batch, landmarks_batch = \
            sample_batched['image'], sample_batched['landmarks']
    batch_size = len(images_batch)
    im_size = images_batch.size(2)
    grid_border_size = 2

    grid = utils.make_grid(images_batch)
    plt.imshow(grid.numpy().transpose((1, 2, 0)))

    for i in range(batch_size):
        plt.scatter(landmarks_batch[i, :, 0].numpy() + i * im_size + (i + 1) * grid_border_size,
                    landmarks_batch[i, :, 1].numpy() + grid_border_size,
                    s=10, marker='.', c='r')

        plt.title('Batch from dataloader')

for i, batched_sample in enumerate(dataloader):
  print(i, batched_sample['image'].size(), batched_sample['landmarks'].size()) 

  if i == 3:
    plt.figure()
    show_landmarks_batch(batched_sample)
    plt.axis('off')
    plt.ioff()
    plt.show()
    break

colab 실행 코드

2. Fashion-MNIST를 활용한 Custom Dataset

dataset은 샘플(image)과 정답(label)을 저장하고, dataloader은 dataset을 샘플에 쉽게 접근할 수 있도록 순회 가능한 객체(iterable)로 감싼다.

fashionMNIST를 통해 benchmark 하고자한다.

Custom Dataset

사용자 정의 dataset 클래스는 반드시 3개 함수(init, len, getitem)를 구현해야 하므로, img_dir을 통해 이미지가 있는 디렉토리를, annotations_file을 통해 label이 있는 csv을 불러온다.

import os
import pandas as pd

import torch
from torch.utils.data import Dataset
from torchvision import datasets
from torchvision.transforms import ToTensor
from torchvision.io import read_image

import matplotlib.pyplot as plt

'''
대부분의 데이터 label.csv는 

image1.jpg , 0
image2.jpg , 1
...
image9.jpg , 8
'''

class CustomImageDataset(Dataset):
  def __init__(self, annotations_file, img_dir, transform = None, target_transform=None):
    '''
    __init__함수는 dataset객체가 생성될 때 한번만 실행된다. 
    여기에서 이미지와 label 파일이 포함된 디렉토리와 transform을 초기화한다.
    transform - image, target-transform - label
    '''
    self.img_labels = pd.read_csv(annotations_file) # init에서는 read_csv만 한다.
    self.img_dir = img_dir
    self.transform = transform
    self.target_transform = trarget_transform
  
  def __len__(self):
    '''
    __len__함수는 데이터셋의 샘플 개수를 반환
    '''
    return len(self.img_labels)                     # label의 행 길이(갯수) row

  def __getitem__(self,idx):
    '''
    __getitem__함수는 주어진 인덱스 idx에 해당하는 샘플을 데이터셋에서 불러오고 반환한다.
    인덱스를 기반으로, 디스크에서 이미지의 위치를 식별하고, read_image를 사용하여 이미지를 텐서로 변환, 
    self.img_labels의 csv 데이터로부터 해당하는 label(정답)을 가져오고 변형함수를 호출하고, image와 label을 리턴
    '''
    img_path = os.path.join(self.img_dir, self.img_labels.iloc[idx, 0]) # image 경로들
    image = read_image(img_path)
    label = self.img.labels.iloc[idx, 1]                                # image에 맞는 label 지정
    if self.transform:
      image = self.transform(image)
    if self.target_transform:
      label = self.target_transform(label)
    return image, label

dataloader

from torch.utils.data import DataLoader

training_data = datasets.FashionMNIST(
    root="data", 
    train=True, download=True,
    transform = ToTensor()
)

validation_data = datasets.FashionMNIST(
    root="data", 
    train=False, download=True,
    transform = ToTensor()
)

train_dataloader = DataLoader(training_data, batch_size=64, shuffle=True)
val_dataloader = DataLoader(validation_data, batch_size=64, shuffle=True)

Data 시각화

train_features, train_labels = next(iter(train_dataloader))
print(f"Feature batch shape: {train_features.size()}")
print(f"Labels batch shape: {train_labels.size()}")
img = train_features[0].squeeze() # [64,1,28,28] 을 [28,28]로 펼쳐야 plt 할 수 있다.
print(f"img shape: {img.size()}")
label = train_labels[0]
plt.imshow(img, cmap="gray")
plt.show

Transform

fashionMNIST 특징(image)들은 PIL image 형식이며 정답(label)은 int이다. 학습을 위해 정규화를 통한 이미지와 원핫으로 부호화된 텐서 형태의 label이 필요하다.

ToTensor는 PIL image나 Numpy ndarray를 floatTensor로 변환하고, 이미지의 픽셀의 크기 값을 [0.,1.]범위로 비례하여 조정한다.

Lambda 를 통해 정수를 원-핫으로 부호화된 텐서로 바꾸는 함수를 정의한다. 이 함수는 먼저(데이터셋 정답의 개수인)크기 10짜리 zero tensor를 만들고, scatter_를 호출하여 주어진 정답 y에 해당하는 인덱스에 value=1을 할당한다.

import torch
from torchvision import datasets

transform = ToTensor()
target_transform = Lambda(lambda y: 
                          torch.zeros(10, dytype=torch.float).scatter_(0, torch.tensor(y), value=1))

Instance segmentation dataset에서의 Custom Dataset

객체 검출, 인스턴스 분할 및 키포인트 검출을 학습하기 위한 새로운 사용자 정의 데이터셋을 추가해보고자 한다.

__getitem__ 메소드가 다음을 반환해야 한다.

• image: PIL이미지의 크기(H,W)
• target: 다음의 필드를 포함하는 사전 타입
  • boxes (FloatTensor[N,4]): N개의 bounding box 좌표가 [x0,y0,x1,y1] 형태를 가진다. x와 관련된 값의 범위는 0부터 w이고, y와 관련된 값의 범위는 0부터 H까지다.
  • labels (int64Tensor[N]): 바운딩 박스마다의 라벨 정보, 0 은 배경
  • image_id (int64Tensor[1]): 이미지 구분자이다. 데이터셋의 모든 이미지 간에 고유한 값이어야 하며 평가 중에도 사용된다.
  • area (Tensor[N]): 바운딩 박스의 면적, 면적은 평가 시 작음, 중간, 큰 박스 간의 점수를 내기 위한 기준이고, COCO 평가를 기준으로 한다.
  • iscrowd (Uint8Tensor[N]): 이 값이 팜일 경우 평가에서 제외한다.
  • (선택)masks (Uint8Tensor(N, H, W]): N개의 객체 마다의 분할 마스크 정보
  • (선택)keypoints (FloatTensor[N, K, 3]): N개의 객체마다의 키포인트 정보. 키포인트는 [x,y,visibility] 형태의 값이다. visibility 값이 0인 경우 키포인트는 보이지 않음을 의미한다. data augmentation의 경우 키포인트 좌우 반전의 개념은 데이터 표현에 따라 달라지며, 새로운 키포인트 표현에 대해 코드 부분을 수정해야 할 수도 있다.

모델을 위의 방법대로 리턴하면 학습과 평가 둘 다에 대해 동작한다.

배경은 무조건 0이어야 하기 때문에, 클래스 분류가 개, 고양이를 분류하고자 할때는 0이 아닌 1과 2로 정의해야 한다.

추가로 학습 중 가로 세로 비율 그룹화를 사용하려는 경우, 이미지의 넓이, 높이를 리턴할 수 있도록 get_height_and_width 메소드를 구현하기를 추천한다. 이를 구현하지 않은 경우 모든 데이터셋은 __getitem__를 통해 메모리에 이미지가 로드되어 사용자 정의 메소드를 제공하는 것보다 느릴 수 있다.

폴더 구조는 다음과 같다.

- PennFudanPed/
  - PedMasks/
    - FudanPend00001_mask.png
    - FudanPend00002_mask.png
    - ...
  - PNGImages/
    - FudanPed00001.png
    - FudanPed00002.png
    - ...

image - masks 예시

<img src = https://tutorials.pytorch.kr/_static/img/tv_tutorial/tv_image01.png width=”40%”> <img src = https://tutorials.pytorch.kr/_static/img/tv_tutorial/tv_image02.png width=”40%”>

각 이미지에 해당하는 분할 마스크가 존재한다.

import os
import numpy as np
import torch
from PIL import Image


class PennFudanDataset(torch.utils.data.Dataset):
    def __init__(self, csv_file=None, root_dir, transforms=None):
        self.root_dir = root_dir
        self.transforms = transforms

        # 모든 이미지와 분할 마스크 정렬
        self.imgs = list(sorted(os.listdir(os.path.join(root_dir, "PNGImages")))) # image셋을 불러옴
        self.masks = list(sorted(os.listdir(os.path.join(root_dir, "PedMasks")))) # mask셋을 불러옴

        # detection과 segmentation이므로 label에 대한 값은 가져오지 않는다. 정답에 대한 셋이 mask이다.


    def __getitem__(self, idx): 
        # 이미지와 마스크를 읽어옵니다
        img_path = os.path.join(self.root_dir, "PNGImages", self.imgs[idx]) # 이미지 각각의 데이터를 가져옴
        mask_path = os.path.join(self.root_dir, "PedMasks", self.masks[idx])
        img = Image.open(img_path).convert("RGB")
        
        # 분할 마스크는 RGB로 변환하지 않음을 유의하세요
        # 왜냐하면 각 색상은 다른 인스턴스에 해당하며, 0은 배경에 해당합니다
        mask = Image.open(mask_path)
        
        # PIL 이미지를 numpy 배열로 변환합니다
        mask = np.array(mask)
        
        # 인스턴스들은 각기 다른 색들로 인코딩 되어야 한다.
        obj_ids = np.unique(mask)
        
        # 첫번째 id 는 배경이라 제거합니다
        obj_ids = obj_ids[1:]

        # 컬러 인코딩된 마스크를 바이너리 마스크 세트로 나눕니다. ???
        masks = mask == obj_ids[:, None, None]

        # 각 마스크의 바운딩 박스 좌표를 얻습니다
        num_objs = len(obj_ids)
        boxes = []
        for i in range(num_objs):
            pos = np.where(masks[i])  # 각 masks에 대한 값들
            xmin = np.min(pos[1])     # 중 각각의 x,y 좌표를 지정
            xmax = np.max(pos[1])
            ymin = np.min(pos[0])
            ymax = np.max(pos[0])
            boxes.append([xmin, ymin, xmax, ymax])

        # 모든 것을 torch.Tensor 타입으로 변환합니다
        boxes = torch.as_tensor(boxes, dtype=torch.float32) # 32와 64는 메모리 사용량을 나타내는 것

        # 예제에서는 사람만을 분류하므로 객체는 1종류이다.
        labels = torch.ones((num_objs,), dtype=torch.int64)
        masks = torch.as_tensor(masks, dtype=torch.uint8)   # as_tensor = tensor의 데이터타입이나 device를 바꾸는 방법, 
                                                            # .to()와 똑같은 기능, x.to(device, dtype=torch.float64)
                                                            # device는 .cuda()로, 타입은 .half(), .float() 등으로 바꿀 수도 있음

        image_id = torch.tensor([idx])                      # 이미지 구분자
        area = (boxes[:, 3] - boxes[:, 1]) * (boxes[:, 2] - boxes[:, 0]) # 바운딩 박스의 면적, 평가 시 점수내는 기준
        
        # 모든 인스턴스는 군중(crowd) 상태가 아님을 가정합니다
        iscrowd = torch.zeros((num_objs,), dtype=torch.int64) # 이 값이 참일 경우 평가에서 제외

        target = {}
        target["boxes"] = boxes
        target["labels"] = labels
        target["masks"] = masks
        target["image_id"] = image_id
        target["area"] = area
        target["iscrowd"] = iscrowd

        if self.transforms is not None:
            img, target = self.transforms(img, target)

        return img, target

    def __len__(self):
        return len(self.imgs)

이 장에서는 custom dataset을 위주로 공부할 것이기에 모델 학습과 평가는 하지 않을 것이다. 더 궁금하다면 참고 사이트를 참고하길 바란다.

Reference

• landmark dataset: https://tutorials.pytorch.kr/beginner/data_loading_tutorial.html
• Fashion-MNIST datset: https://tutorials.pytorch.kr/beginner/basics/data_tutorial.html#id7
• Instance segmentation: https://tutorials.pytorch.kr/intermediate/torchvision_tutorial.html