3 P1 실습: 다층 퍼셉트론

# TensorFlow 를 사용할 때 기타 정보 표시 제한. Tensorflow를 불러오기 전에 실행해야 함
import os
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"

import tensorflow as tf
import keras
import torch

# TensorFlow
tf_gpus = tf.config.list_physical_devices("GPU")

print("=== TensorFlow ===")
print("Version:", tf.__version__)
print("GPU 사용 가능:", len(tf_gpus) > 0)
if tf_gpus:
    print("GPU:", tf_gpus[0].name)

print()

# PyTorch
print("=== PyTorch ===")
print("Version:", torch.__version__)
print("GPU 사용 가능:", torch.cuda.is_available())
if torch.cuda.is_available():
    print("GPU:", torch.cuda.get_device_name(0))
    print("CUDA version:", torch.version.cuda)

from keras.datasets import mnist

(train_images, train_labels), (test_images, test_labels) = mnist.load_data()

print("train_images:", train_images.shape)
print("train_labels:", train_labels.shape)
print("test_images :", test_images.shape)
print("test_labels :", test_labels.shape)

print("변환 전:", train_images.shape)
print("픽셀값 범위:", train_images.min(), "~", train_images.max())

train_images = train_images.reshape((60000, 28 * 28)).astype("float32") / 255
test_images = test_images.reshape((10000, 28 * 28)).astype("float32") / 255

print("변환 후:", train_images.shape)
print("픽셀값 범위:", train_images.min(), "~", train_images.max())

from keras import layers

model = keras.Sequential([
    layers.Dense(512, activation="relu"),
    layers.Dense(10, activation="softmax"),
])

model.build(input_shape=(None, 28 * 28))
model.summary()

model.compile(
    optimizer="adam",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

history = model.fit(
    train_images,
    train_labels,
    epochs=20,
    batch_size=128,
    validation_split=0.2,
)

plt.figure(figsize=(7, 4))

epochs = range(1, len(history.history["loss"]) + 1)

plt.plot(epochs, history.history["loss"], marker="o", label="train")
plt.plot(epochs, history.history["val_loss"], marker="o", label="validation")

plt.xticks(range(1, len(history.history["loss"]) + 1))
plt.xlabel("epoch")
plt.ylabel("loss")
plt.legend()
plt.show()

from keras import layers

model = keras.Sequential([
    layers.Dense(512, activation="relu"),
    layers.Dense(10, activation="softmax"),
])

model.compile(
    optimizer="adam",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

from keras import callbacks

early_stopping = callbacks.EarlyStopping(
    monitor="val_loss",
    patience=3,
    restore_best_weights=True,
)

checkpoint = callbacks.ModelCheckpoint(
    filepath="best_mlp.keras",
    monitor="val_loss",
    save_best_only=True,
)

history = model.fit(
    train_images,
    train_labels,
    epochs=20,
    batch_size=128,
    validation_split=0.2,
    callbacks=[early_stopping, checkpoint],
)

plt.figure(figsize=(7, 4))

epochs = range(1, len(history.history["loss"]) + 1)

plt.plot(epochs, history.history["loss"], marker="o", label="train")
plt.plot(epochs, history.history["val_loss"], marker="o", label="validation")

plt.xticks(range(1, len(history.history["loss"]) + 1))
plt.xlabel("epoch")
plt.ylabel("val_loss")
plt.legend()
plt.show()

best_model = model

best_model = keras.models.load_model("best_mlp.keras") # best_model = model 이라고 해도 됨.

predictions = best_model.predict(test_images[:10], verbose=0)

i = 0
print("10개 출력:", predictions[i])
print("예측 숫자:", predictions[i].argmax())
print("가장 큰 출력값:", predictions[i].max())
print("실제 정답:", test_labels[i])

# 원래 모양으로 되돌려 시각화
image = test_images[0].reshape(28, 28)

plt.imshow(image, cmap="gray")
plt.title(
    f"prediction = {predictions[0].argmax()}, " # 예측값
    f"label = {test_labels[0]}"                 # 실제 정답 라벨
)
plt.axis("off")
plt.show()

test_loss, test_acc = best_model.evaluate(
    test_images,
    test_labels,
    batch_size=128,
)

print(f"test loss     : {test_loss:.4f}")
print(f"test accuracy : {test_acc:.4f}")

test_loss, test_acc = model.evaluate(
    test_images,
    test_labels,
    batch_size=128,
)

print(f"test loss     : {test_loss:.4f}")
print(f"test accuracy : {test_acc:.4f}")

import torch
from torch import nn

torch_model = nn.Sequential(
    nn.Linear(28 * 28, 512),
    nn.ReLU(),
    nn.Linear(512, 10),
)

torch_model

from torch.utils.data import TensorDataset, DataLoader

x_train_t = torch.tensor(train_images, dtype=torch.float32)
y_train_t = torch.tensor(train_labels, dtype=torch.long)

x_test_t = torch.tensor(test_images, dtype=torch.float32)
y_test_t = torch.tensor(test_labels, dtype=torch.long)

train_dataset = TensorDataset(x_train_t, y_train_t)
test_dataset = TensorDataset(x_test_t, y_test_t)

train_loader = DataLoader(
    train_dataset,
    batch_size=128,
    shuffle=True,
)

test_loader = DataLoader(
    test_dataset,
    batch_size=256,
    shuffle=False,
)

# 훈련 device 지정: GPU 또는 CPU
device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
print("device:", device)

# PyTorch 모델을 device로 이동
torch_model = torch_model.to(device)

# 손실 함수와 옵티마이저 정의
loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(torch_model.parameters())

# 훈련 루프
epochs = 20  # 에포크 수

for epoch in range(epochs):
    torch_model.train()                   # 모델을 훈련 모드로 설정

    running_loss = 0.0                    # 에포크 내 누적 손실 초기화 (에포크 손실값 계산)
    correct = 0                           # 에포크 내 누적 정답 개수 초기화 (에포크 정확도 계산)
    total = 0                             # 에포크 내 누적 샘플 수 초기화

    # 배치 훈련
    for x_batch, y_batch in train_loader:
        x_batch = x_batch.to(device)      # 입력 데이터를 GPU 또는 CPU로 이동
        y_batch = y_batch.to(device)      # 타깃 데이터를 GPU 또는 CPU로 이동

        optimizer.zero_grad()

        outputs = torch_model(x_batch)
        loss = loss_fn(outputs, y_batch)

        loss.backward()
        optimizer.step()

        # 배치 손실 누적
        running_loss += loss.item() * x_batch.size(0)
        # 정답 수 누적
        correct += (outputs.argmax(dim=1) == y_batch).sum().item()
        # 배치 샘플 수 누적 (에포크 내에서 훈련된 샘플 수)
        total += y_batch.size(0)

    epoch_loss = running_loss / total     # 에포크 평균 손실 계산
    epoch_acc = correct / total           # 에포크 정확도 계산

    print(
        f"epoch {epoch + 1}: "
        f"loss={epoch_loss:.4f}, accuracy={epoch_acc:.4f}"
    )

torch_model.eval()

correct = 0
total = 0

with torch.no_grad():
    for x_batch, y_batch in test_loader:
        x_batch = x_batch.to(device)
        y_batch = y_batch.to(device)

        outputs = torch_model(x_batch)
        predictions_t = outputs.argmax(dim=1)

        correct += (predictions_t == y_batch).sum().item()
        total += y_batch.size(0)

torch_test_acc = correct / total

print(f"PyTorch test accuracy: {torch_test_acc:.4f}")

import torch
from torch import nn

torch_model = nn.Sequential(
    nn.Linear(28 * 28, 512),
    nn.ReLU(),
    nn.Linear(512, 10),
)

torch_model

from torch.utils.data import DataLoader, TensorDataset, random_split

x_train_t = torch.tensor(train_images, dtype=torch.float32)
y_train_t = torch.tensor(train_labels, dtype=torch.long)

x_test_t = torch.tensor(test_images, dtype=torch.float32)
y_test_t = torch.tensor(test_labels, dtype=torch.long)

full_train_dataset = TensorDataset(x_train_t, y_train_t)
test_dataset = TensorDataset(x_test_t, y_test_t)

train_size = int(0.8 * len(full_train_dataset))
val_size = len(full_train_dataset) - train_size
train_dataset, val_dataset = random_split(
    full_train_dataset, [train_size, val_size]
)

train_loader = DataLoader(train_dataset, batch_size=128, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=256, shuffle=False)
test_loader = DataLoader(test_dataset, batch_size=256, shuffle=False)

from copy import deepcopy

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
torch_model = torch_model.to(device)

loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(torch_model.parameters())

epochs = 20
patience = 3
best_val_loss = float("inf")
best_model_state = None
wait = 0

for epoch in range(epochs):
    torch_model.train()
    running_loss = 0.0
    correct = 0
    total = 0

    for x_batch, y_batch in train_loader:
        x_batch = x_batch.to(device)
        y_batch = y_batch.to(device)

        optimizer.zero_grad()
        outputs = torch_model(x_batch)
        loss = loss_fn(outputs, y_batch)
        loss.backward()
        optimizer.step()

        running_loss += loss.item() * x_batch.size(0)
        correct += (outputs.argmax(dim=1) == y_batch).sum().item()
        total += y_batch.size(0)

    train_loss = running_loss / total
    train_acc = correct / total

    torch_model.eval()
    val_running_loss = 0.0
    val_correct = 0
    val_total = 0

    with torch.no_grad():
        for x_batch, y_batch in val_loader:
            x_batch = x_batch.to(device)
            y_batch = y_batch.to(device)

            outputs = torch_model(x_batch)
            loss = loss_fn(outputs, y_batch)

            val_running_loss += loss.item() * x_batch.size(0)
            val_correct += (outputs.argmax(dim=1) == y_batch).sum().item()
            val_total += y_batch.size(0)

    val_loss = val_running_loss / val_total
    val_acc = val_correct / val_total

    print(
        f"epoch {epoch + 1}: "
        f"loss={train_loss:.4f}, accuracy={train_acc:.4f}, "
        f"val_loss={val_loss:.4f}, val_accuracy={val_acc:.4f}"
    )

    if val_loss < best_val_loss:
        best_val_loss = val_loss
        best_model_state = deepcopy(torch_model.state_dict())
        torch.save(best_model_state, "best_torch_mlp.pt")
        wait = 0
    else:
        wait += 1
        if wait >= patience:
            print("Early stopping")
            break

torch_model.load_state_dict(best_model_state)