Classification de galaxies avec Python

import cv2
import os
import glob

# Mon chemin vers le dossier contenant les images originales
input_dir = '/Users/victor/Downloads/images_training_rev1'

# Mon chemin vers le dossier où les images traitées seront sauvegardées 
output_dir = '/Users/victor/Desktop/processed_images'

# On passe a travers toutes les images JPG dans le dossier originale
image_paths = glob.glob(os.path.join(input_dir, '*.jpg'))

for img_path in image_paths:
    img = cv2.imread(img_path)  # Lire l'image
    
    # On calcul le centre et découpe l'image
    center_x, center_y = img.shape[1] // 2, img.shape[0] // 2
    cropped_img = img[center_y - 106:center_y + 106, center_x - 106:center_x + 106]
    
    # On redimensionne l'image à 64x64 pixels
    resized_img = cv2.resize(cropped_img, (64, 64))
    
    # Sauvegarde de l'image
    output_path = os.path.join(output_dir, os.path.basename(img_path))
    cv2.imwrite(output_path, resized_img)
        

import pandas as pd
from sklearn.model_selection import train_test_split

# Chemin vers le fichier CSV contenant les probabilités des classes
csv_file_path = '/Users/victor/Downloads/training_solutions_rev1.csv'

# On import les données
data = pd.read_csv(csv_file_path)
 
# On prend les GalaxyID et on en fait un tableau 1D contenant tout les ID des images
galaxy_ids = data['GalaxyID'].values

# On Sépare les données en sous-ensembles d'entraînement et de test de manière aléatoire, avec 20% de test comme demandé et le reste en entrainement
# C'est pour eviter qu'il s'entraine sur des images dont il sera testé après 
train_ids, test_ids = train_test_split(galaxy_ids, test_size=0.2, random_state=42)

# Affichage des tailles pour vérifier que tout est correct
print(f"Taille de l'ensemble d'entraînement: {len(train_ids)}")
print(f"Taille de l'ensemble de test: {len(test_ids)}")

from torch.utils.data import DataLoader
from torchvision import datasets
from torchvision.transforms import ToTensor

# On crée GalaxyDataset pour l'entraînement avec la fonction donnée GalaxyDataset : 
train_dataset = GalaxyDataset(
    labels='/Users/victor/Downloads/training_solutions_rev1.csv',
    img_dir='/Users/victor/Desktop/processed_images',  
    inds=train_ids.astype(str),
    ext=".jpg",  
    transform=ToTensor(),
    target_transform=None
)

# On crée GalaxyDataset pour le test avec la fonction donnée GalaxyDataset : 
test_dataset = GalaxyDataset(
    labels='/Users/victor/Downloads/training_solutions_rev1.csv',
    img_dir='/Users/victor/Desktop/processed_images',  
    inds=test_ids.astype(str),
    ext=".jpg",  # 
    transform=ToTensor(),
    target_transform=None
)

# On verifie que tout est bon : 
print(train_dataset)
print(len(train_dataset))
print(type(train_dataset[0]))  
print(type(train_dataset[0][0]))
print(train_dataset[0][0].shape)
print(type(train_dataset[0][0]))

        

    
# Création de DataLoader pour l'ensemble d'entraînement
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)

# Création de DataLoader pour l'ensemble de test
test_loader = DataLoader(test_dataset, batch_size=64, shuffle=True)

        

import torch
import torch.nn as nn
import torch.nn.functional as F

class GalaxyCNN(nn.Module):
    def __init__(self):
        super(GalaxyCNN, self).__init__()
        
        # Première couche de convolution
        self.conv1 = nn.Conv2d(3, 6, kernel_size=5, padding=3)
        self.bn1 = nn.BatchNorm2d(6)
        
        # Deuxième couche de convolution
        self.conv2 = nn.Conv2d(6, 16, kernel_size=5, padding=3)
        self.bn2 = nn.BatchNorm2d(16)
        
        # Aplatissement des caractéristiques pour les couches pleinement connectées
        self.flatten = nn.Flatten()

        self.linear_entry_dim = get_output_width(get_output_width(
            get_output_width(get_output_width(64, 5,  padding=3 ), 2, stride=2), 5 ,  padding=3
        ), 2, stride=2)

        
        # Première couche pleinement connectée
        self.fc1 = nn.Linear(16 * self.linear_entry_dim**2, 120)  # La taille doit corresponde a celle la sortie 
                                                                  # de la dernière couche de pooling, on utilise donc get_output_width
        
        # Deuxième couche pleinement connectée
        self.fc2 = nn.Linear(120, 84)
        
        # Couche finale pour les 37 classes
        self.fc3 = nn.Linear(84, 37)
        
    def forward(self, x):
        # Appliquer la première couche de convolution, batch norm, ReLU et pooling
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.max_pool2d(x, 2)
        
        # Appliquer la deuxième couche de convolution, batch norm, ReLU et pooling
        x = F.relu(self.bn2(self.conv2(x)))
        x = F.max_pool2d(x, 2)
        
        # Aplatissement des caractéristiques
        x = self.flatten(x)
        
        # Passer à travers les couches pleinement connectées avec ReLU pour les deux premières
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        
        # Sortie finale sans ReLU mais avec sigmoide pour les probabilités
        x = torch.sigmoid(self.fc3(x))
        return x

        

import torch
import torch.nn as nn
import torch.nn.functional as F

device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # nécessaire pour savoir sur quoi va travailler le réseau (voir tp)

from torch import nn, optim
from torch.utils.data import DataLoader

# Initialisation du model 
model = GalaxyCNN().to(device) 

# Hyper paramètre 

learning_rate = 1e-3
epochs = 6

# Fonction de perte et optimiseur
loss_fn = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

def train_loop(dataloader, model, loss_fn, optimizer):
    size = len(dataloader.dataset)
    train_loss = 0.0
    
    for batch, (X, y, _) in enumerate(dataloader):  # On ajoute d'un _ pour ignorer img_id 
        X, y = X.to(device), y.to(device)
        
        # Calcul de la prédiction et de la perte
        pred = model(X)
        loss = loss_fn(pred, y)
        
        # Rétropropagation
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        loss, current = loss.item(), batch * len(X)
        train_loss += loss * X.size(0)

        # Progrès
        if batch % 100 == 0:
            print(f"Loss: {loss}, [{current}/{size}]")
    
    return train_loss / size

def test_loop(dataloader, model, loss_fn):
    size = len(dataloader.dataset)
    test_loss = 0
    num_batches = len(dataloader)

    with torch.no_grad():
        for X, y, _ in dataloader:  # On ajoute d'un _ pour ignorer img_id
            X, y = X.to(device), y.to(device)
            
            pred = model(X)
            test_loss += loss_fn(pred, y).item()  # calcul de perte

    test_loss /= num_batches
    print(f"Avg loss: {test_loss} \n")
    return test_loss

# Boucles d'entraînement et de test
train_losses = []
test_losses = []
for epoch in range(epochs):
    print(f"Epoch {epoch+1}\n-------------------------------")
    train_loss = train_loop(train_loader, model, loss_fn, optimizer)
    test_loss = test_loop(test_loader, model, loss_fn)
    train_losses.append(train_loss)
    test_losses.append(test_loss)
print("Done!")


        

# Affichage de l'évolution du model 
import matplotlib.pyplot as plt

plt.figure(figsize=(10, 6))
plt.plot(train_losses, label='Training Loss')
plt.plot(test_losses, label='Testing Loss')
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.title('Loss Evolution')
plt.legend()
plt.show()


        

            
from torch.utils.data import DataLoader

# mode évaluation
model.eval()

def calculate_rmse(dataloader, model):
    squared_errors = []
    with torch.no_grad():  # Pour l'évaluation, pas besoin de calculer les gradients
        for X, y, _ in dataloader:  # On ignore img_id
            X, y = X.to(device), y.to(device)
            predictions = model(X)
            errors = predictions - y  # Erreur = prédiction - vérité
            squared_errors.extend((errors ** 2).mean(1).cpu().numpy())  # On calcul simplement l'erreur quadratique moyenne par échantillon
    
    mean_squared_error = np.mean(squared_errors)  # Moyenne des erreurs quadratiques sur tout le dataset
    rmse = np.sqrt(mean_squared_error)  # Racine carrée de la MSE pour obtenir le RMSE
    return rmse

# On calcul le RMSE sur les données de test
rmse = calculate_rmse(test_loader, model)
print(f"RMSE sur les données de test: {rmse}")

# Comparaison avec la méthode simpliste
rmse_simpliste = 0.16194
print(f"RMSE de la méthode simpliste : {rmse_simpliste}")
if rmse < rmse_simpliste:
    print("Le modèle CNN performe mieux que la méthode simpliste.")
else:
    print("La méthode simpliste performe mieux que le modèle CNN.")
        

from torchvision import transforms

train_transforms = transforms.Compose([
    transforms.RandomHorizontalFlip(),  # Retournement horizontal aléatoire
    transforms.RandomRotation(60),  # Rotation aléatoire
    transforms.ColorJitter(brightness=0.4, contrast=0.4),  # Ajustement de luminosité et de contraste aléatoirement
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
])


        

class GalaxyCNN3(nn.Module):
    def __init__(self):
        super(GalaxyCNN3, self).__init__()
        # Couche de convolution 1
        self.conv1 = nn.Conv2d(3, 8, kernel_size=5, padding=3)  # Modification : 6 à 8 filtres
        self.bn1 = nn.BatchNorm2d(8)  # Normalisation par batch ajustée pour 8 filtres
        
        # Couche de convolution 2
        self.conv2 = nn.Conv2d(8, 16, kernel_size=5, padding=3)  # Entrée ajustée pour 8 canaux
        self.bn2 = nn.BatchNorm2d(16)  # Normalisation par batch pour 16 filtres

        # Couche de convolution 3 ajoutée
        self.conv3 = nn.Conv2d(16, 32, kernel_size=5, padding=2)  # Nouvelle couche : 16 à 32 filtres, padding ajusté
        self.bn3 = nn.BatchNorm2d(32)  # Normalisation par batch pour 32 filtres
        
        # Aplatissement des données pour la première couche pleinement connectée
        self.flatten = nn.Flatten()
        
        # Couches pleinement connectées avec ajustement pour le dropout
        self.fc1 = nn.Linear(2048, 256)  # Ajustement suite à l'ajout d'une couche de convolution
        self.fc2 = nn.Linear(256, 128)  # Ajustement de la dimension et ajout d'une autre couche de Dropout après
        self.fc3 = nn.Linear(128, 37)  # Sortie pour 37 classes
        
    def forward(self, x):
        # Passage à travers la couche de convolution 1 -> Normalisation par batch -> ReLU -> Pooling
        x = F.max_pool2d(F.relu(self.bn1(self.conv1(x))), 2)
        
        # Passage à travers la couche de convolution 2 -> Normalisation par batch -> ReLU -> Pooling
        x = F.max_pool2d(F.relu(self.bn2(self.conv2(x))), 2)
        
        # Passage à travers la nouvelle couche de convolution 3 -> Normalisation par batch -> ReLU -> Pooling
        x = F.max_pool2d(F.relu(self.bn3(self.conv3(x))), 2)
        
        # Aplatir les sorties pour les couches pleinement connectées
        x = self.flatten(x)
        #print(x.shape) 
        
        # Première couche pleinement connectée -> ReLU
        x = F.relu(self.fc1(x))
        
        # Deuxième couche pleinement connectée -> ReLU
        x = F.relu(self.fc2(x))
        
        # Troisième couche pleinement connectée, pas de ReLU ici
        x = self.fc3(x)
        
        # Application de la fonction sigmoïde pour obtenir des probabilités
        x = torch.sigmoid(x)
        return x