154 lines
5.0 KiB
Python
154 lines
5.0 KiB
Python
|
|
from abc import ABC, abstractmethod
|
|
import torch
|
|
import torch.nn as nn
|
|
import torch.optim as optim
|
|
import time
|
|
import numpy as np
|
|
from sklearn.metrics import classification_report
|
|
from pathlib import Path
|
|
|
|
class Model(ABC):
|
|
def __init__(self, device, size):
|
|
self.device = device
|
|
self.size = size
|
|
self.model = self.get().to(self.device)
|
|
|
|
@abstractmethod
|
|
def get(self):
|
|
# return the model
|
|
pass
|
|
|
|
def train(self, epochs, loader, rate):
|
|
criterion = nn.CrossEntropyLoss()
|
|
optimizer = optim.Adam(filter(lambda p: p.requires_grad, self.model.parameters()), lr=rate)
|
|
|
|
print(f"Starting training on {self.device}...")
|
|
start_time = time.time()
|
|
self.model.train()
|
|
|
|
for epoch in range(epochs):
|
|
total_loss = 0.0
|
|
for inputs, labels in loader:
|
|
inputs, labels = inputs.to(self.device), labels.to(self.device)
|
|
optimizer.zero_grad()
|
|
outputs = self.model(inputs)
|
|
loss = criterion(outputs, labels)
|
|
loss.backward()
|
|
optimizer.step()
|
|
total_loss += loss.item()
|
|
|
|
print(f"Epoch {epoch+1}/{epochs} | Loss: {total_loss / len(loader):.4f}")
|
|
|
|
if self.device.type == 'cuda': torch.cuda.synchronize()
|
|
print(f"Training completed in: {time.time() - start_time:.2f}s")
|
|
|
|
def evaluate(self, loader):
|
|
self.model.eval()
|
|
all_preds, all_labels = [], []
|
|
print("\nEvaluating...")
|
|
|
|
with torch.no_grad():
|
|
for inputs, labels in loader:
|
|
inputs, labels = inputs.to(self.device), labels.to(self.device)
|
|
outputs = self.model(inputs)
|
|
_, predicted = torch.max(outputs, 1)
|
|
all_preds.extend(predicted.cpu().numpy())
|
|
all_labels.extend(labels.cpu().numpy())
|
|
|
|
accuracy = 100 * (np.array(all_preds) == np.array(all_labels)).sum() / len(all_labels)
|
|
print(f"Test Accuracy: {accuracy:.2f}%")
|
|
print(classification_report(all_labels, all_preds, zero_division=0))
|
|
|
|
|
|
def save(self, filename=None):
|
|
|
|
save_dir = Path("trained_models")
|
|
save_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 2. Determine filename (Default to class name if not provided)
|
|
if filename is None:
|
|
filename = f"{self.__class__.__name__.lower()}.pth"
|
|
|
|
if not filename.endswith('.pth'):
|
|
filename += '.pth'
|
|
|
|
save_path = save_dir / filename
|
|
torch.save(self.model.state_dict(), save_path)
|
|
print(f'Model saved to {save_path}')
|
|
|
|
|
|
|
|
def load(self, arch):
|
|
|
|
file_path = Path("trained_models") / f'{arch.name.lower()}.pth'
|
|
# does file exist
|
|
if not file_path.exists():
|
|
raise FileNotFoundError(f'No checkpoint found at: {file_path}')
|
|
|
|
# Load the weights
|
|
state_dict = torch.load(file_path, map_location=self.device, weights_only=True)
|
|
|
|
self.model.load_state_dict(state_dict)
|
|
self.model.to(self.device)
|
|
print(f'Model loaded from {file_path}')
|
|
|
|
|
|
|
|
# Using the factory patern here
|
|
@staticmethod
|
|
def create(arch, device, size):
|
|
print(f'>> MODEL ARCHITECTURE >> {arch.name}.')
|
|
|
|
match arch:
|
|
|
|
# ResNet18
|
|
case Architecture.RESNET18:
|
|
from architectures.ResNet18 import ResNet18
|
|
return ResNet18(device, size)
|
|
|
|
# ResNet50
|
|
case Architecture.RESNET50:
|
|
from architectures.ResNet50 import ResNet50
|
|
return ResNet50(device, size)
|
|
|
|
# INCEPTION
|
|
case Architecture.INCEPTION:
|
|
from architectures.Inception import Inception
|
|
return Inception(device, size)
|
|
|
|
# DENSENET121
|
|
case Architecture.DENSENET121:
|
|
from architectures.DenseNet121 import DenseNet121
|
|
return DenseNet121(device, size)
|
|
# googleNet
|
|
case Architecture.GOOGLENET:
|
|
from architectures.GoogleNet import GoogleNet
|
|
return GoogleNet(device, size)
|
|
# EfficientNet
|
|
case Architecture.EFFICIENTNET:
|
|
from architectures.EfficentNet import EfficientNet
|
|
return EfficientNet(device, size)
|
|
#ShuffleNet
|
|
case Architecture.SHUFFLENET:
|
|
from architectures.ShuffleNet import ShuffleNet
|
|
return ShuffleNet(device, size)
|
|
# wide ResNet
|
|
case Architecture.WIDE_RESNET:
|
|
from architectures.WideResNet import WideResNet
|
|
return WideResNet(device, size)
|
|
case _:
|
|
raise ValueError(f"Unknown model: {arch}")
|
|
|
|
# model architectures
|
|
from enum import Enum, auto
|
|
|
|
class Architecture(Enum):
|
|
RESNET18 = auto()
|
|
RESNET50 = auto()
|
|
INCEPTION = auto()
|
|
DENSENET121 = auto()
|
|
GOOGLENET = auto()
|
|
EFFICIENTNET = auto()
|
|
SHUFFLENET = auto()
|
|
WIDE_RESNET = auto() |