# Data Scientist Day 5
github: https://github.com/Study-boy-dot/Image-Processing-3/tree/main
kaggle: https://www.kaggle.com/dataminingee/cifar10-with-mycnn
## Environment
|Environment|Version|
|-----------|----------|
|python |3.8|
|jupyter notebook|6.5.2|
|cuda|12.0|
## Packages
|Package|Version|
|-----------|----------|
|torch | 2.1.0+cu118|
|torchaudio | 2.1.0+cu118|
|torchsummary | 1.5.1|
|torchvision | 0.16.0+cu118|
|numpy|1.23.4|
|matplotlib|3.3.1|
## Code Review
### Load Cifar10 Dataset
Using `torchvision.dataset.CIFAR10` to load cifar10 dataset
* specify loaded dataset is for training or testing
* specify images transform method(will explain it later)
```
traindata = torchvision.datasets.CIFAR10(root='./data', train=True, transform=transform)
trainloader = torch.utils.data.DataLoader(traindata, batch_size, shuffle=True)
```
### Images Transform
Doing iamges augmentation, apply it to load dataset process
```
transform = transforms.Compose([transforms.Resize(32, 32),
transforms.ToTensor(),
...])
```
### Model Design
self.feature1: CONV => RELU => CONV => RELU => MAXPOOLING => DROPOUT
self.feature2: CONV => RELU => CONV => RELU => MAXPOOLING => DROPOUT
self.classifier: FLATTEN => LINEAR => RELU => DROPOUT => LINEAR(num classes)
**Do not add softmax layer to classifier because loss function has internally apply softmax activation to model's output and use it to calculate cross-entropy loss**
```
self.feature1 = nn.Sequential(nn.Conv2D(input_channel, output_channel, kernel_size, padding),
nn.ReLU(),
...
)
self.feature2 = nn.Sequential(nn.Conv2D(input_channel, output_channel, kernel_size, padding),
...
)
self.classifier = nn.Sequential(nn.Linear(input_channel * width * height, output_size),
...
)
```
**In the class forward(self, x) function will specify how each block transfer feature**
```
def forward(self, x):
x = self.feature1(x)
x = self.feature2(x)
x = self.classifier(x)
return x
```
<font size=20 color=#c4706a weight=bold>Complete Code:</font>
```
import torch
import torch.nn as nn
class ConvNet(nn.Module):
def __init__(self, num_classes):
super(ConvNet, self).__init__()
# Feature extraction layers
self.features1 = nn.Sequential(
nn.Conv2d(3, 32, kernel_size=3, padding=1), # 1st Convolutional layer
nn.ReLU(),
nn.Conv2d(32, 32, kernel_size=3, padding=1), # 2nd Convolutional layer
nn.ReLU(),
nn.MaxPool2d(2), # Max Pooling
nn.Dropout(0.25)
)
self.features2 = nn.Sequential(
nn.Conv2d(32, 64, kernel_size=3, padding=1), # 3rd Convolutional layer
nn.ReLU(),
nn.Conv2d(64, 64, kernel_size=3, padding=1), # 4th Convolutional layer
nn.ReLU(),
nn.MaxPool2d(2), # Max Pooling
nn.Dropout(0.25)
)
# Classification layers
self.classifier = nn.Sequential(
nn.Flatten(), # Flatten the tensor
nn.Linear(64 * 8 * 8, 512), # Fully connected layer
nn.ReLU(),
nn.Dropout(0.5),
nn.Linear(512, num_classes) # Output layer
)
def forward(self, x):
x = self.features1(x)
x = self.features2(x)
x = self.classifier(x)
return x
```
### Model Training
Using Pytorch training model, training process need to specify by user
* Create loss function
* Create optimizer
```
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(model.parameters(), lr, momentum)
```
Start training model, example has not show moving dataset and model to GPU
`model = model.to(device)`
```
for e in range(epochs):
model.train() # turn model to train mode
for inputs, labels in trainloader:
optimizer.zero_grad()
outputs = model(inputs)
# Calculate loss
loss = criterion(outputs, labels)
# Backpropagation & Optimizer step
loss.backward()
optimizer.step()
```
<font size=20 color=#c4706a weight=bold>Complete Code:</font>
```
import copy
def train(model, criterion, optimizer, num_epochs, dataloader_train, dataloader_val=None):
history = {
'loss':[],
'acc': [],
'val_loss' : [],
'val_acc': [],
'best_train_acc' : (0, 0)
}
best_acc = 0.0
best_model_weight = copy.deepcopy(model.state_dict())
train_dataset_size = len(dataloader_train.dataset)
test_dataset_size = len(dataloader_val.dataset)
for e in range(num_epochs):
print(f'\033[1;35mEpoch:{e}\033[0m')
model.train() # change model to training mode
each_epoch_loss = 0.0
each_epoch_acc = 0.0
each_epoch_val_loss = 0.0
each_epoch_val_acc = 0.0
for inputs, labels in dataloader_train:
inputs = inputs.to(device)
labels = labels.to(device)
outputs = model(inputs)
# Calculate loss
loss = criterion(outputs, labels)
# Backpropagation and Optimization
optimizer.zero_grad()
loss.backward()
optimizer.step()
#values, indexes
_, pred = torch.max(outputs, 1) # choose the first biggest value
# Store result
each_epoch_loss += loss.item()
each_epoch_acc += (pred == labels).sum().item()
# Update history result
each_epoch_loss /= train_dataset_size
each_epoch_acc /= train_dataset_size
each_epoch_acc *= 100
history['loss'].append(each_epoch_loss)
history['acc'].append(each_epoch_acc)
print(f'\033[1;34mTrainning: Epoch:{e} Loss:{each_epoch_loss:.2f} Accuracy:{each_epoch_acc:.2f}\033[0m')
# Store best result
if each_epoch_acc > best_acc:
best_acc = each_epoch_acc
best_model_weight = copy.deepcopy(model.state_dict())
history['best_train_acc'] = (e, best_acc)
if dataloader_val:
model.eval() # change model to evaluate mode
with torch.no_grad():
for inputs, labels in dataloader_val:
inputs = inputs.to(device)
labels = labels.to(device)
# Evaluate
outputs = model(inputs)
val_loss = criterion(outputs, labels)
# values, indexes
_, pred = torch.max(outputs, 1)
# Store result
each_epoch_val_loss += val_loss.item()
each_epoch_val_acc += (pred == labels).sum().item()
each_epoch_val_loss /= test_dataset_size
each_epoch_val_acc /= test_dataset_size
each_epoch_val_acc *= 100
history['val_loss'].append(each_epoch_val_loss)
history['val_acc'].append(each_epoch_val_acc)
print(f'\033[1;34mTesting: Epoch:{e} Loss:{each_epoch_val_loss:.2f} Accuracy:{each_epoch_val_acc:.2f}\033[0m')
print(f"Best Result in \033[1;35mEpoch:{history['best_train_acc'][0]}\033[0m \033[1;33mAccuracy:{history['best_train_acc'][1]:.2f}\033[0m")
model.load_state_dict(best_model_weight)
return model, history
```
### Save Model
```
model_path='model.pth'
torch.save(model, model_path)
```
### Append Layer to model
```
model.classfier.append(nn.Linear(4096, 512))
...
```
### Freeze Layer
In transfer learning process, model is pretrained, and only need to train the classfier layer to match with dataset provided.
Freeze the features' layer can lower down training model time, increase effieciency
```
for param in model.parameters():
param.requires_grad = False
```
### Early Stop in Training Process
Specify patience level, when there is continuous validation loss not fewer than the best validation loss over patience level times. Stop the training process
```
def train(model, criterion, optimzer, num_epochs, trainloader, testloader=None):
# Early Stop
best_val_loss = float('inf')
patience = 3
no_improvement_count = 0
for e in range(num_epochs):
'''Training Process code here'''
# Evaluate model
if testloader:
'''Evaluate Model code here'''
if val_loss < best_val_loss:
best_val_loss = val_loss
no_improvement_count = 0
else:
no_improvement_count += 1
# Early Stop Condition
if no_improvement_count >= patience:
print(f"Early stopping after {e} epochs with no improvement.")
break
return model, history
```
Sign in with Wallet
Connect another wallet