---
# System prepended metadata

title: Data Scientist Day 5
tags: [image_process, Data Scientist]

---

# 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
```