● TensorFlow2 (tf.keras)でニューラルネットワークを構築
# ① import
import tensorflow as tf
from tensorflow.keras.layers import Dense, Dropout, Flatten
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.datasets import mnist
import os
from matplotlib import pyplot as plt
batch_size = 128
epochs = 30
# ②データの取得
(x_train, y_train), (x_test, y_test) = mnist.load_data()
# データの前処理
x_train, x_test = x_train / 255.0, x_test / 255.0
# ③モデルの構築 (Sequential法)
model = tf.keras.models.Sequential([
Flatten(),
Dense(120, activation='relu'),
Dropout(0.2), # 過学習防止
Dense(10, activation='softmax')
])
# # ③モデルの構築 (class法 ※PyTorchに似た書き方)
# class MnistMode(tf.keras.Model):
# def __init__(self):
# super().__init__()
# self.flatten = Flatten()
# self.d1 = Dense(120, activation="relu")
# self.dropout = Dropout(0.2)
# self.d2 = Dense(10, activation="softmax")
#
# def call(self, x, training=False):
# x = self.flatten(x)
# x = self.d1(x)
# if training: # 予測時には適用しない
# x = self.dropout(x, training=training)
# x = self.d2(x)
# return x
#
#
# model = MnistMode()
# ④モデルのコンパイル
model.compile(optimizer=Adam(),
loss=tf.keras.losses.SparseCategoricalCrossentropy(),
metrics=['accuracy'])
checkpoint_save_path = "./checkpoint/mnist.ckpt"
if os.path.exists(checkpoint_save_path + '.index'):
print('-------------重みロード-----------------')
model.load_weights(checkpoint_save_path)
# 訓練中にチェックポイントを保存します。
# (再び訓練を行うことなくモデルを使用することができます）
cp_callback = tf.keras.callbacks.ModelCheckpoint(filepath=checkpoint_save_path,
save_weights_only=True,
save_best_only=True)
# ⑤モデルの学習
history = model.fit(x_train, y_train, batch_size=batch_size, epochs=epochs,
validation_data=(x_test, y_test),
callbacks=[cp_callback])
# ⑥モデルの構造の表示
model.summary()
# ---------- 学習過程（accuracyとloss）をプロット ----------
plt.figure(figsize=(10, 5))
metrics = ['loss', 'accuracy']
for i in range(len(metrics)):
metric = metrics[i]
plt.subplot(1, 2, i + 1)
plt.plot(history.history[metric], label='Training ' + metric)
plt.plot(history.history['val_' + metric], label='Validation ' + metric)
plt.title(metric)
plt.legend()
plt.show()
● PyTorchでニューラルネットワークを構築
# ① import
import torch
from torch import nn
from torch.nn import functional as F
from torch import optim
import torchvision
from matplotlib import pyplot as plt
batch_size = 128
epochs = 30
torch.manual_seed(20)
# ②データの取得、データの前処理
train_loader = torch.utils.data.DataLoader(
torchvision.datasets.MNIST('mnist_data', train=True, download=True,
transform=torchvision.transforms.Compose([
torchvision.transforms.ToTensor(),
torchvision.transforms.Normalize(
(0.5,), (0.5,))
])),
batch_size=batch_size, shuffle=True)
test_loader = torch.utils.data.DataLoader(
torchvision.datasets.MNIST('mnist_data/', train=False, download=True,
transform=torchvision.transforms.Compose([
torchvision.transforms.ToTensor(),
torchvision.transforms.Normalize(
(0.5,), (0.5,))
])),
batch_size=batch_size, shuffle=False)
# ③モデルの構築
class MnistMode(nn.Module):
def __init__(self):
super().__init__()
self.fc1 = nn.Linear(1 * 28 * 28, 120)
self.fc2 = nn.Linear(120, 10)
def forward(self, x):
x = x.view(-1, 1 * 28 * 28)
x = F.relu(self.fc1(x))
if self.training:
x = F.dropout(x, 0.2, training=self.training)
x = self.fc2(x)
return x
device = "cuda" if torch.cuda.is_available() else "cpu"
model = MnistMode().to(device)
# ④ 損失関数、最適化手法
loss_func = nn.CrossEntropyLoss().to(device) # softmax処理含み
optimizer = optim.Adam(model.parameters(), lr=0.0001)
# ⑤モデルの学習
history = {'loss': [], 'val_loss': [], 'accuracy': [], 'val_accuracy': []}
train_loader_len = len(train_loader)
for epoch in range(epochs):
total_train_loss = 0
total_train_acc = 0
model.train() # 学習モードに (training->True)
print('----- Epoch {}/{} ------'.format(epoch + 1, epochs))
for batch_idx, (x, y) in enumerate(train_loader):
x = x.to(device)
y = y.to(device)
out = model(x)
loss = loss_func(out, y)
total_train_loss += loss.item()
optimizer.zero_grad()
loss.backward()
optimizer.step()
pred_train = out.argmax(dim=1)
correct = (pred_train == y).sum().item() / x.shape[0]
total_train_acc += correct
if (batch_idx + 1) % 50 == 0:
print('{:6}/{} loss: {:.5f} accuracy: {:.5f}'.format(batch_idx + 1, train_loader_len,
loss.item(), correct))
cnt = len(train_loader)
history['loss'].append(total_train_loss / cnt)
history['accuracy'].append(total_train_acc / cnt)
total_val_loss = 0
total_val_acc = 0
model.eval() # 推論モードに (training->False)
with torch.no_grad():
for x, y in test_loader:
x = x.to(device)
y = y.to(device)
x = x.view(-1, 1 * 28 * 28)
out = model(x)
loss = loss_func(out, y)
pred = out.argmax(dim=1)
correct = (pred == y).sum().item()
total_val_loss += loss.item()
total_val_acc += correct
history['val_loss'].append(total_val_loss / len(test_loader))
history['val_accuracy'].append(total_val_acc / len(test_loader.dataset))
format_str = 'Epoch {}/{} loss: {:.5f} - accuracy: {:.5f} - val_loss: {:.5f} - val_accuracy: {:.5f}'
print(format_str.format(epoch + 1, epochs,
history['loss'][-1], history['accuracy'][-1],
history['val_loss'][-1], history['val_accuracy'][-1]))
# ⑥モデルの構造の表示
print(model)
# ---------- 学習過程（lossとaccuracy）をプロット ----------
plt.figure(figsize=(10, 5))
metrics = ['loss', 'accuracy']
for i in range(len(metrics)):
metric = metrics[i]
plt.subplot(1, 2, i + 1)
plt.plot(history[metric], label='Training ' + metric)
plt.plot(history['val_' + metric], label='Validation ' + metric)
plt.title(metric)
plt.legend()
plt.show()
●シンプルなCNN (PyTorch)
%%time
#Jupyterで実行用
%matplotlib inline
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np
def fix_seeds(seed=0):
# random.seed(seed)
# np.random.seed(seed)
torch.manual_seed(seed)
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True
# ==============================
# 1. デバイス設定
# ==============================
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("Using device:", device)
# ==============================
# 2. データセットの用意（CIFAR-10）
# ==============================
# 学習用：ランダムクロップ＋左右反転
train_transform = transforms.Compose([
transforms.RandomCrop(32, padding=4),
transforms.RandomHorizontalFlip(),
transforms.ToTensor(),
# transforms.Normalize((0.4914, 0.4822, 0.4465), # CIFAR-10 の平均
# (0.2023, 0.1994, 0.2010)) # CIFAR-10 の標準偏差
transforms.Normalize((0.5, 0.5, 0.5), # mean(R,G,B)
(0.5, 0.5, 0.5)) # std(R,G,B)
])
test_transform = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.5, 0.5, 0.5), # mean(R,G,B)
(0.5, 0.5, 0.5)) # std(R,G,B)
])
train_dataset = torchvision.datasets.CIFAR10(
root="./CIFAR10",
train=True,
download=True,
transform=train_transform
)
test_dataset = torchvision.datasets.CIFAR10(
root="./CIFAR10",
train=False,
download=True,
transform=test_transform
)
train_loader = DataLoader(
train_dataset,
batch_size=128,
shuffle=True
)
test_loader = DataLoader(
test_dataset,
batch_size=128,
shuffle=False
# num_workers=3
)
# CIFAR-10 のクラス名
classes = (
'plane', 'car', 'bird', 'cat', 'deer',
'dog', 'frog', 'horse', 'ship', 'truck'
)
# ==============================
# 3. CNNモデル定義
# ==============================
class SimpleCNN(nn.Module):
"""
CIFAR-10 (3x32x32) 用のシンプルなCNNモデル。
Conv → ReLU → MaxPool を3回繰り返し、
最後に全結合で10クラス分類する。
"""
def __init__(self, num_classes=10):
super(SimpleCNN, self).__init__()
self.features = nn.Sequential(
# (N, 3, 32, 32) → (N, 32, 32, 32) → Poolで (N, 32, 16, 16)
nn.Conv2d(3, 32, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(2, 2),
nn.Dropout2d(0.25),
# (N, 32, 16, 16) → (N, 64, 16, 16) → Poolで (N, 64, 8, 8)
nn.Conv2d(32, 64, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(2, 2),
nn.Dropout2d(0.25),
# (N, 64, 8, 8) → (N, 128, 8, 8) → Poolで (N, 128, 4, 4)
nn.Conv2d(64, 128, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(2, 2),
nn.Dropout2d(0.25),
)
# 128 * 4 * 4 = 2048 次元の特徴ベクトル
self.classifier = nn.Sequential(
nn.Linear(128 * 4 * 4, 256),
nn.ReLU(inplace=True),
nn.Dropout(0.5),
nn.Linear(256, num_classes)
)
def forward(self, x):
x = self.features(x) # (N, 128, 4, 4)
x = x.view(x.size(0), -1) # (N, 2048)
x = self.classifier(x) # (N, 10)
return x
# ==============================
# 4. 学習用／評価用 関数
# ==============================
def train_one_epoch(model, loader, criterion, optimizer, device):
model.train()
running_loss = 0.0
total = 0
correct = 0
for images, labels in loader:
images = images.to(device)
labels = labels.to(device)
optimizer.zero_grad()
outputs = model(images)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
running_loss += loss.item() * images.size(0)
_, predicted = torch.max(outputs, 1)
total += labels.size(0)
correct += (predicted == labels).sum().item()
epoch_loss = running_loss / total
epoch_acc = correct / total
return epoch_loss, epoch_acc
def evaluate(model, loader, criterion, device):
model.eval()
running_loss = 0.0
total = 0
correct = 0
with torch.no_grad():
for images, labels in loader:
images = images.to(device)
labels = labels.to(device)
outputs = model(images)
loss = criterion(outputs, labels)
running_loss += loss.item() * images.size(0)
_, predicted = torch.max(outputs, 1)
total += labels.size(0)
correct += (predicted == labels).sum().item()
epoch_loss = running_loss / total
epoch_acc = correct / total
return epoch_loss, epoch_acc
# ==============================
# 5. 学習履歴の可視化用 関数
# ==============================
def plot_history(train_loss_list, test_loss_list,
train_acc_list, test_acc_list):
epochs = range(1, len(train_loss_list) + 1)
plt.figure(figsize=(12, 5))
# --- Loss ---
plt.subplot(1, 2, 1)
plt.plot(epochs, train_loss_list, label='Train Loss')
plt.plot(epochs, test_loss_list, label='Test Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.title('Train vs Test Loss')
plt.legend()
plt.grid(True)
# --- Accuracy ---
plt.subplot(1, 2, 2)
plt.plot(epochs, train_acc_list, label='Train Acc')
plt.plot(epochs, test_acc_list, label='Test Acc')
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.title('Train vs Test Accuracy')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# ==============================
# 6. モデル読み込み & 予測表示用 関数
# ==============================
def load_model_for_inference(model_path, device, num_classes=10):
model = SimpleCNN(num_classes=num_classes)
state_dict = torch.load(model_path, map_location=device)
model.load_state_dict(state_dict)
model.to(device)
model.eval()
return model
def imshow(img):
img = img.cpu().numpy()
img = np.transpose(img, (1, 2, 0)) # (C,H,W) -> (H,W,C)
# 逆正規化： x_norm = (x - 0.5) / 0.5 → x = x_norm * 0.5 + 0.5
img = img * 0.5 + 0.5
img = np.clip(img, 0, 1)
plt.imshow(img)
plt.axis('off')
def show_sample_predictions(model, dataset, classes, device, num_images=5):
"""
テストデータから数枚取り出し、画像と予測ラベルを Jupyter 上に表示する。
"""
model.eval()
plt.figure(figsize=(12, 3))
# ランダムに画像をサンプリング
indices = torch.randint(0, len(dataset), (num_images,))
for i, idx in enumerate(indices):
img, label = dataset[idx]
# バッチ次元を追加して (1, C, H, W) にしてからモデルに入れる
input_img = img.unsqueeze(0).to(device)
with torch.no_grad():
outputs = model(input_img)
_, predicted = torch.max(outputs, 1)
pred_label = predicted.item()
plt.subplot(1, num_images, i + 1)
imshow(img)
plt.title(f"P:{classes[pred_label]}\nT:{classes[label]}")
plt.axis('off')
plt.tight_layout()
plt.show()
# ==============================
# 7. メイン処理
# ==============================
def main():
fix_seeds(112)
model = SimpleCNN(num_classes=10).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9)
num_epochs =10
# --- 学習履歴を保存するリスト ---
train_loss_list = []
train_acc_list = []
test_loss_list = []
test_acc_list = []
for epoch in range(num_epochs):
train_loss, train_acc = train_one_epoch(
model, train_loader, criterion, optimizer, device
)
test_loss, test_acc = evaluate(
model, test_loader, criterion, device
)
train_loss_list.append(train_loss)
train_acc_list.append(train_acc)
test_loss_list.append(test_loss)
test_acc_list.append(test_acc)
print(f"Epoch [{epoch+1}/{num_epochs}] "
f"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f} "
f"Test Loss: {test_loss:.4f}, Test Acc: {test_acc:.4f}")
# --- 1) 学習履歴のグラフ表示 ---
plot_history(train_loss_list, test_loss_list,
train_acc_list, test_acc_list)
# --- 2) モデルの保存 ---
model_path = "simple_cnn_cifar10.pth"
torch.save(model.state_dict(), model_path)
print(f"Model saved to {model_path}")
# --- 3) 保存したモデルを読み込んで予測を表示 ---
inference_model = load_model_for_inference(model_path, device, num_classes=10)
show_sample_predictions(inference_model, test_dataset, classes, device,
num_images=5)
if __name__ == "__main__":
main()
● 簡易ResNet18 (PyTorch)
# JetsonコンテナでのPython直接実行用
# ============================================================
# 必要ライブラリ
# ============================================================
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision import datasets, transforms
import matplotlib
import matplotlib.pyplot as plt
matplotlib.use("Agg") #JetsonコンテナでのPython 3直接実行用
import numpy as np
import os
import time
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("device:", device)
start_tm = time.time()
def format_duration_hms(seconds_total):
seconds_total = round(seconds_total, 2)
hours = int(seconds_total // 3600)
minutes = int((seconds_total % 3600) // 60)
seconds = seconds_total % 60.0
return f"{hours}:{minutes:02d}:{seconds:05.2f}"
def fix_seeds(seed=0):
torch.manual_seed(seed)
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True
# ============================================================
# BasicBlock（ResNet の基本ブロック）
# ============================================================
class BasicBlock(nn.Module):
"""
ResNet18/34 で使用される基本 Residual Block
"""
expansion = 1 # 出力チャンネル倍率（BasicBlock は 1）
def __init__(self, in_channels, out_channels, stride=1):
super(BasicBlock, self).__init__()
# Main Path: F(x)
self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size=3,
stride=stride, padding=1, bias=False)
self.bn1 = nn.BatchNorm2d(out_channels)
self.conv2 = nn.Conv2d(out_channels, out_channels, kernel_size=3,
stride=1, padding=1, bias=False)
self.bn2 = nn.BatchNorm2d(out_channels)
# Shortcut Path（shape が変わる場合のみ 1x1 Conv）
self.shortcut = nn.Sequential()
if stride != 1 or in_channels != out_channels * self.expansion:
self.shortcut = nn.Sequential(
nn.Conv2d(in_channels, out_channels * self.expansion,
kernel_size=1, stride=stride, bias=False),
nn.BatchNorm2d(out_channels * self.expansion)
)
def forward(self, x):
out = F.relu(self.bn1(self.conv1(x)))
out = self.bn2(self.conv2(out))
out += self.shortcut(x) # 残差接続（F(x) + x）
return F.relu(out)
# ============================================================
# ResNet 本体（ResNet18）
# ============================================================
class ResNet(nn.Module):
def __init__(self, block, num_blocks, num_classes=10):
super(ResNet, self).__init__()
self.in_channels = 64
# CIFAR-10 は 32x32 → Conv1 を 3x3 に調整
self.conv1 = nn.Conv2d(3, 64, kernel_size=3,
stride=1, padding=1, bias=False)
self.bn1 = nn.BatchNorm2d(64)
# 各 Residual layer
self.layer1 = self._make_layer(block, 64, num_blocks[0], stride=1)
self.layer2 = self._make_layer(block, 128, num_blocks[1], stride=2)
self.layer3 = self._make_layer(block, 256, num_blocks[2], stride=2)
self.layer4 = self._make_layer(block, 512, num_blocks[3], stride=2)
self.avgpool = nn.AdaptiveAvgPool2d((1, 1))
self.fc = nn.Linear(512 * block.expansion, num_classes)
def _make_layer(self, block, out_channels, num_blocks, stride):
"""
最初のブロックだけ stride を変える
"""
layers = []
layers.append(block(self.in_channels, out_channels, stride))
self.in_channels = out_channels * block.expansion
# 残りのブロックは stride=1
for _ in range(1, num_blocks):
layers.append(block(self.in_channels, out_channels))
return nn.Sequential(*layers)
def forward(self, x):
out = F.relu(self.bn1(self.conv1(x)))
out = self.layer1(out)
out = self.layer2(out)
out = self.layer3(out)
out = self.layer4(out)
out = self.avgpool(out)
out = torch.flatten(out, 1)
out = self.fc(out)
return out
def ResNet18():
return ResNet(BasicBlock, [2, 2, 2, 2])
fix_seeds(122)
# ============================================================
# CIFAR-10 のロード
# ============================================================
transform = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),
])
train_dataset = datasets.CIFAR10(
root="./CIFAR10", train=True, download=True, transform=transform)
test_dataset = datasets.CIFAR10(
root="./CIFAR10", train=False, download=True, transform=transform)
train_loader = DataLoader(train_dataset, batch_size=128, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=128, shuffle=False)
# ============================================================
# モデル・損失関数・最適化
# ============================================================
model = ResNet18().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# ログ用
train_losses, test_losses = [], []
train_accs, test_accs = [], []
# ============================================================
# 6. 学習ループ
# ============================================================
epochs = 2
for epoch in range(epochs):
# ---- Train ----
model.train()
train_loss, correct, total = 0, 0, 0
for imgs, labels in train_loader:
imgs, labels = imgs.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(imgs)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()
train_loss += loss.item() * imgs.size(0)
_, predicted = outputs.max(1)
total += labels.size(0)
correct += predicted.eq(labels).sum().item()
train_losses.append(train_loss / total)
train_accs.append(correct / total)
# ---- Test ----
model.eval()
test_loss, correct, total = 0, 0, 0
with torch.no_grad():
for imgs, labels in test_loader:
imgs, labels = imgs.to(device), labels.to(device)
outputs = model(imgs)
loss = criterion(outputs, labels)
test_loss += loss.item() * imgs.size(0)
_, predicted = outputs.max(1)
total += labels.size(0)
correct += predicted.eq(labels).sum().item()
test_losses.append(test_loss / total)
test_accs.append(correct / total)
print(f"Epoch [{epoch+1}/{epochs}] "
f"Train Loss: {train_losses[-1]:.4f}, Train Acc: {train_accs[-1]*100:.2f}% | "
f"Test Loss: {test_losses[-1]:.4f}, Test Acc: {test_accs[-1]*100:.2f}%")
# ============================================================
# 学習結果の可視化（Loss・Accuracy）
# ============================================================
plt.figure(figsize=(15, 5))
plt.subplot(1, 2, 1)
plt.plot(train_losses, label="Train Loss")
plt.plot(test_losses, label="Test Loss")
plt.title("Loss")
plt.legend()
plt.subplot(1, 2, 2)
plt.plot(train_accs, label="Train Acc")
plt.plot(test_accs, label="Test Acc")
plt.title("Accuracy")
plt.legend()
plt.savefig("ResNet2.png")
# plt.show()
# ============================================================
# 8. モデル保存
# ============================================================
model_path = "resnet18.pth"
torch.save(model.state_dict(), model_path)
# ============================================================
# モデル読み込み → 推論
# ============================================================
model2 = ResNet18().to(device)
model2.load_state_dict(torch.load(model_path))
model2.eval()
classes = train_dataset.classes
# テスト画像1枚で推論
img, label = test_dataset[2]
# 表示用に unnormalize
# plt.imshow(np.transpose(img.numpy() * 0.5 + 0.5, (1, 2, 0)))
# plt.title(f"True Label: {classes[label]}")
# # plt.show()
# 推論
with torch.no_grad():
x = img.unsqueeze(0).to(device)
pred = model2(x)
predicted_class = pred.argmax(1).item()
print("label:", classes[label])
print("Predicted:", classes[predicted_class])
duration = time.time() - start_tm
print(f"duration: {format_duration_hms(duration)}")