Skip to article frontmatterSkip to article content
Site not loading correctly?

This may be due to an incorrect BASE_URL configuration. See the MyST Documentation for reference.

댕냥이

1import

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

print('PyTorch', torch.__version__)

2Dataset

from pathlib import Path
import torch
from PIL import Image
from torchvision import transforms

class CatDogDataset(torch.utils.data.Dataset):
    def __init__(self, 폴더경로, 전처리=None, **kwargs):
        super().__init__(**kwargs)
        self.폴더경로 = Path(폴더경로)
        self.파일목록 = list(self.폴더경로.glob('**/*.jpg'))
        self.파일목록.sort()
        self.전처리 = 전처리

    def __len__(self):
        return len(self.파일목록)
    
    def __getitem__(self, 색인번호):
        파일경로 = self.파일목록[색인번호]
        label = 1 if 'dog' in 파일경로.name else 0
        with Image.open(파일경로) as sample:
            sample = self.전처리(sample) if self.전처리 else sample
        return sample, label
전처리 = transforms.Compose([
    transforms.Resize((180, 180)),
    transforms.ToTensor()
])
    
dataset = {}
개냥이폴더 = Path('/workspace/data/cats_dogs_small')
for split in ['train', 'validation', 'test']:
    dataset[split] = CatDogDataset(개냥이폴더 / split, 전처리=전처리)
    print(f"{split:<10}: {len(dataset[split]):,} samples")

sample, label = dataset['train'][0]
print(type(sample), type(label))
print(sample.shape, sample.dtype, f'label={label}')

3PyTorch

3.1모델

import torch
import torch.nn as nn

def build_torch_model():
    model = nn.Sequential(
        # 합성곱 계층
        nn.Conv2d(in_channels=3, out_channels=32, kernel_size=3),
        nn.ReLU(),
        nn.MaxPool2d(kernel_size=2),

        nn.Conv2d(in_channels=32, out_channels=64, kernel_size=3),
        nn.ReLU(),
        nn.MaxPool2d(kernel_size=2),

        nn.Conv2d(in_channels=64, out_channels=128, kernel_size=3),
        nn.ReLU(),
        nn.MaxPool2d(kernel_size=2),

        nn.Conv2d(in_channels=128, out_channels=256, kernel_size=3),
        nn.ReLU(),
        nn.MaxPool2d(kernel_size=2),

        nn.Conv2d(in_channels=256, out_channels=256, kernel_size=3),
        nn.ReLU(),
        # 분류 출력
        nn.Flatten(),
        nn.Linear(in_features=256 * 7 * 7, out_features=1),
    )
    return model
model = build_torch_model()
# print(model)
# torch.tensor.numel(): 텐서의 원소(값) 개수 반환; numel(): number of elements
print(f'매개변수 개수: {sum(p.numel() for p in model.parameters()):,}')
# 중간 출력 수집
activations = []
# 계층별 중간 출력 수집 함수 등록
# 주의! 디버깅 용도로만 사용
# 순전파 시점에서 추가 함수가 호출되기 때문에 성능에 영향이 있을 수 있음
for layer in model:
    layer.register_forward_hook(
        # 순전파 완료 시점에 호출될 함수 정의 등록
        lambda self, 모듈입력, 모듈출력: activations.append(모듈출력.detach()))

test_loader = torch.utils.data.DataLoader(
    dataset['test'], batch_size=32, shuffle=False)

X_batch, y_batch = next(iter(test_loader))
print(X_batch.shape, y_batch.shape)

outputs = model(X_batch)
print(outputs.shape)

print('\nModel Activations:')
for layer, 중간출력 in zip(model, activations):
    # 클래스명칭
    layer_name = layer.__class__.__name__
    # 배치 크기 제외한 출력 형상
    print(f'{layer_name}: {중간출력.shape[1:]}')

3.2훈련

import time

model = build_torch_model()
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model.to(device)
# 매개변수를 최적화 기법에 등록하기 전에 매개변수 연산 장치 설정
최적화 = torch.optim.RMSprop(model.parameters(), lr=0.001)
손실함수 = nn.BCEWithLogitsLoss()

num_workers = 2
train_loader = torch.utils.data.DataLoader(
    dataset['train'], batch_size=32, shuffle=True, num_workers=num_workers)
val_loader = torch.utils.data.DataLoader(
    dataset['validation'], batch_size=32, shuffle=False, num_workers=num_workers)

학습횟수 = 30; 손실변화 = []
for 에폭 in range(학습횟수):
    start = time.time()
    
    훈련손실 = 0.0
    for X_batch, y_batch in train_loader:
        X_batch = X_batch.to(device)
        y_batch = y_batch.to(device)
        
        # 순전파
        outputs = model(X_batch)
        손실 = 손실함수(outputs, y_batch.float().unsqueeze(1))
        # 평균 손실 = 손실합계 / 배치크기
        # 손실합계 = 평균 손실 * 배치크기
        훈련손실 += 손실.item() * len(X_batch)
        # 역전파
        손실.backward()
        # 매개변수 갱신
        최적화.step()
        최적화.zero_grad()

    # 검증 손실
    with torch.no_grad():
        검증손실 = 0.0
        for X_val, y_val in val_loader:
            X_val = X_val.to(device)
            y_val = y_val.to(device)
            val_outputs = model(X_val)
            val_loss = 손실함수(val_outputs, y_val.float().unsqueeze(1))
            검증손실 += val_loss.item() * len(X_val)
        
    손실변화.append({
        '훈련손실': 훈련손실 / len(train_loader.dataset),
        '검증손실': 검증손실 / len(val_loader.dataset)
    })
    print(f'Epoch {에폭+1:2d} ({time.time() - start:.1f}s)')
    print(pd.Series(손실변화[-1]).round(3).to_string())

model_path = 'cat_dog_model.pth'
torch.save(model, model_path)

3.3평가

results = pd.DataFrame(손실변화)
results.index += 1
results.index.name = 'Epoch'
results.columns = ['Train Loss', 'Validation Loss']
results.to_csv('training_log.csv')
ax = results[1:].plot(style='o--')
def 성능측정(model, loader, device='cpu'):
    채점 = 0; 손실 = 0.0
    with torch.no_grad():  # 기울기 계산 비활성화
        for X_batch, y_batch in loader:
            X_batch = X_batch.to(device)
            y_batch = y_batch.to(device)
            outputs = model(X_batch)
            손실 += 손실함수(outputs, y_batch.float().unsqueeze(1)).item() * len(X_batch)
            예측확률 = torch.sigmoid(outputs).squeeze()
            예측 = (예측확률 > 0.5).long()
            채점 += (예측 == y_batch.long()).float().sum().item()
    
    정확도 = 채점 / len(loader.dataset)
    손실 = 손실 / len(loader.dataset)
    return {'loss': 손실, 'accuracy': 정확도}

get_loader = lambda split: torch.utils.data.DataLoader(
    dataset[split], batch_size=32, shuffle=False)

device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = torch.load(model_path, weights_only=False, map_location=device)
평가결과 = pd.DataFrame({
    'train': 성능측정(model, get_loader('train'), device=device),
    'validation': 성능측정(model, get_loader('validation'), device=device),
    'test': 성능측정(model, get_loader('test'), device=device)
})
display(평가결과.T.round(4))

4Keras

import os
import keras

print(os.environ.get('KERAS_BACKEND', 'tensorflow'))
print('Keras', keras.__version__)

4.1모델

import keras
from keras import layers

def build_model(inputs):
    model = keras.Sequential([
        inputs,
        layers.Conv2D(filters=32, kernel_size=3, activation='relu'), #data_format='channels_first'),
        layers.MaxPooling2D(pool_size=2), #data_format='channels_first'),
        layers.Conv2D(filters=64, kernel_size=3, activation='relu'), #data_format='channels_first'),
        layers.MaxPooling2D(pool_size=2), #data_format='channels_first'),
        layers.Conv2D(filters=128, kernel_size=3, activation='relu'), #data_format='channels_first'),
        layers.MaxPooling2D(pool_size=2), #data_format='channels_first'),
        layers.Conv2D(filters=256, kernel_size=3, activation='relu'), #data_format='channels_first'),
        layers.MaxPooling2D(pool_size=2), #data_format='channels_first'),

        layers.Conv2D(filters=256, kernel_size=3, activation='relu'), #data_format='channels_first'),
        layers.Flatten(),
        layers.Dense(units=1, activation='sigmoid')
    ])
    return model

4.2훈련

전처리 = transforms.Compose([
    transforms.Resize((180, 180)),
    transforms.ToTensor(),
    # C, H, W -> H, W, C
    transforms.Lambda(lambda x: x.permute(1, 2, 0))
])

dataset = {}
개냥이폴더 = Path('/workspace/data/cats_dogs_small')
for split in ['train', 'validation', 'test']:
    shuffle = True if split == 'train' else False
    dataset[split] = CatDogDataset(개냥이폴더 / split, 전처리=전처리)
    print(f"{split:<10}: {len(dataset[split]):,} samples")

train_loader = torch.utils.data.DataLoader(
    dataset['train'], batch_size=32, shuffle=True,
    num_workers=2)
val_loader = torch.utils.data.DataLoader(
    dataset['validation'], batch_size=32, shuffle=False,
    num_workers=2)

keras.backend.clear_session()
model = build_model(keras.Input(shape=(180, 180, 3)))
model.summary()
model.compile(optimizer='rmsprop',
              loss='binary_crossentropy',
              metrics=['accuracy'])

history = model.fit(
    train_loader,
    validation_data=val_loader,
    epochs=30,
    callbacks=[
        keras.callbacks.ModelCheckpoint(
            'cat_dog_model.keras', save_best_only=True),
    ]
)

4.3평가

results = pd.DataFrame(history.history)
results.index += 1
results.index.name = 'Epoch'
display(results.tail())
plt.figure(figsize=(12, 5))
results.plot(y=['loss', 'val_loss'], style='o--', ax=plt.subplot(1, 2, 1))
results.plot(y=['accuracy', 'val_accuracy'], style='o--', ax=plt.subplot(1, 2, 2))
plt.show()
test_loader = torch.utils.data.DataLoader(
    dataset['test'], batch_size=32, shuffle=False,)

model = keras.models.load_model('cat_dog_model.keras')
pd.Series(model.evaluate(test_loader, return_dict=True)).round(4)