CSC 8607 – Introduction au deep learning

data_dir = "/array/data/imagenet/2012"
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
batch_size = 256
learning_rate = 0.0001
num_epochs = 100
n_workers = 16
model_name = "mlp"

transform_features = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(227),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

print("Loading images...")
# Datasets and loaders
train_dataset = ImageNet(root=data_dir, split="train", transform=transform_features)
print("Train loaded.", len(train_dataset), "examples found.")
val_dataset = ImageNet(root=data_dir, split="val", transform=transform_features)
print("Val loaded.", len(val_dataset), "examples found.")
test_dataset = val_dataset  # Using 'val' as ImageNet does not have a separate test set

train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=n_workers, pin_memory=True)
val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=n_workers, pin_memory=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=n_workers, pin_memory=True)
print("Images loaded.")

train_transform_features = transforms.Compose([
    transforms.RandomResizedCrop(227),
    transforms.RandomHorizontalFlip(),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])

train_dataset = ImageNet(root=data_dir, split="train", transform=train_transform_features)

class MLP(nn.Module):
    def __init__(self, input_size=3*227*227, hidden_size=256, num_classes=1000):
        super(MLP, self).__init__()
        self.classifier = nn.Sequential(
            nn.Linear(input_size, hidden_size),
            nn.ReLU(inplace=True),
            nn.Linear(hidden_size, hidden_size),
            nn.ReLU(inplace=True),
            nn.Linear(hidden_size, num_classes),
        )

    def forward(self, x):
        x = torch.flatten(x, 1)
        x = self.classifier(x)
        return x

def calculate_loss_and_accuracy(loader, model, criterion):
    model.eval()
    running_loss = 0.0
    correct = 0
    total = 0
    with torch.no_grad():
        for images, labels in tqdm(loader, desc="Evaluation"):
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            loss = criterion(outputs, labels)
            running_loss += loss.item() * images.size(0)
            _, predicted = outputs.max(1)
            correct += (predicted == labels).sum().item()
            total += labels.size(0)
    avg_loss = running_loss / total
    accuracy = correct / total
    model.train()
    return avg_loss, accuracy

model = MLP(input_size=3*227*227, hidden_size=256, num_classes=1000).to(device)

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate, weight_decay=weight_decay)

# TensorBoard writer
writer = SummaryWriter()

# Training loop
for epoch in range(num_epochs):
    print("Running epoch", epoch)
    model.train()
    running_loss = 0.0
    correct_train = 0
    total_train = 0

    for i, (images, labels) in enumerate(tqdm(train_loader, desc="Training")):
        images, labels = images.to(device), labels.to(device)
        optimizer.zero_grad(set_to_none=True)
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

        running_loss += loss.item()
        _, predicted = outputs.max(1)
        correct_train += (predicted == labels).sum().item()
        total_train += labels.size(0)

    train_loss = running_loss / len(train_loader)
    train_accuracy = correct_train / total_train
    val_loss, val_accuracy = calculate_loss_and_accuracy(val_loader, model, criterion)

    writer.add_scalar('Loss/Train', train_loss, epoch)
    writer.add_scalar('Loss/Validation', val_loss, epoch)
    writer.add_scalar('Accuracy/Train', train_accuracy, epoch)
    writer.add_scalar('Accuracy/Validation', val_accuracy, epoch)

    print(
        f"Epoch [{epoch + 1}/{num_epochs}], Train Loss: {train_loss:.4f}, Train Accuracy: {train_accuracy:.4f}, Val Loss: {val_loss:.4f}, Val Accuracy: {val_accuracy:.4f}")

# Final test loss and accuracy
test_loss, test_accuracy = calculate_loss_and_accuracy(test_loader, model, criterion)
hparams = {
    'batch_size': batch_size,
    'learning_rate': learning_rate,
    'num_epochs': num_epochs,
    "model": model_name
}
metrics = {
    'hparam/test_loss': test_loss,
    'hparam/test_accuracy': test_accuracy
}
writer.add_hparams(hparams, metrics)
print(f"Test Loss: {test_loss:.4f}, Test Accuracy: {test_accuracy:.4f}")

# Save model
torch.save(model.state_dict(), f"{model_name}_imagenet.pth")

writer.close()

early_stop_patience = 5
best_val_loss = float('inf')
early_stop_counter = 0

...
    print(
        f"Epoch [{epoch + 1}/{num_epochs}], Train Loss: {train_loss:.4f}, Train Accuracy: {train_accuracy:.4f}, Val Loss: {val_loss:.4f}, Val Accuracy: {val_accuracy:.4f}")

    # Early stopping check
    if val_loss < best_val_loss:
        best_val_loss = val_loss
        early_stop_counter = 0
        torch.save(model.state_dict(), f"best_{model_name}_imagenet.pth")
    else:
        early_stop_counter += 1
    if early_stop_counter >= early_stop_patience:
        print("Early stopping triggered. Stopping training.")
        break

# Load the best model
model.load_state_dict(torch.load(f"best_{model_name}_imagenet.pth"))

# Final test loss and accuracy
test_loss, test_accuracy = calculate_loss_and_accuracy(test_loader, model, criterion)
hparams = {
    'batch_size': batch_size,
    'learning_rate': learning_rate,
    'num_epochs': num_epochs,
    "model": model_name
}
metrics = {
    'hparam/test_loss': test_loss,
    'hparam/test_accuracy': test_accuracy
}
writer.add_hparams(hparams, metrics)
print(f"Test Loss: {test_loss:.4f}, Test Accuracy: {test_accuracy:.4f}")

writer.close()

class AlexNet(nn.Module):
    def __init__(self, num_classes=1000):
        super(AlexNet, self).__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=11, stride=4, padding=2),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=3, stride=2),
            nn.Conv2d(64, 192, kernel_size=5, padding=2),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=3, stride=2),
            nn.Conv2d(192, 384, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(384, 256, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.Conv2d(256, 256, kernel_size=3, padding=1),
            nn.ReLU(inplace=True),
            nn.MaxPool2d(kernel_size=3, stride=2),
        )
        self.classifier = nn.Sequential(
            nn.Dropout(),
            nn.Linear(256 * 6 * 6, 4096),
            nn.ReLU(inplace=True),
            nn.Dropout(),
            nn.Linear(4096, 4096),
            nn.ReLU(inplace=True),
            nn.Linear(4096, num_classes),
        )

    def forward(self, x):
        x = self.features(x)
        x = torch.flatten(x, 1)
        x = self.classifier(x)
        return x

if model_name == "alexnet":
    model = AlexNet(num_classes=1000).to(device)
elif model_name == "mlp":
    model = MLP(input_size=3*227*227, hidden_size=256, num_classes=1000).to(device)
else:
    raise ValueError("Invalid model_name. Choose 'alexnet' or 'mlp'.")

def calculate_total_parameters(model):
    """Calculate the total number of parameters in a model."""
    return sum(p.numel() for p in model.parameters())

• conv1 = 64*56*56*4/1024 = 784
• pool1 = 182
• conv2 = 547
• pool2 = 127
• conv3 = 254
• conv4 = 169
• conv5 = 169
• pool5 = 36
• flatten = 36
• linear1 = 16
• linear2 = 16
• linear3 = 4