upload task2

src/eval/task2_eval.py

@@ -0,0 +1,144 @@
+### EVALUATION CODE
+
+"""
+
+!gdown --id 13R5JajJ4oh_enkRjfs2fThITGsIMlFX8
+
+!unzip /content/test_set.zip -d /content
+
+"""
+import torch
+import torch.nn as nn
+from timm import create_model
+from torch.utils.data import DataLoader, Dataset
+import torchvision.transforms as transforms
+import os
+from PIL import Image
+from utils.task2_utils import *
+
+
+# Transforms
+transform = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5])
+])
+
+# Dataset and DataLoader
+root_dir = "/content/test_set"  # Replace with your dataset path
+dataset = BottleRotationDataset(root_dir=root_dir, transform=transform)
+dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
+
+# Load a saved checkpoint
+model = ViTRegression.load_from_checkpoint("/Users/marcodonnarumma/Desktop/BottleAROL/src/models/vit_tiny_rotation_epoch_epoch=100.ckpt")
+
+import os
+import matplotlib.pyplot as plt
+import torch
+import torchvision.transforms as transforms
+from PIL import Image, ImageDraw, ImageFont
+
+# Ensure the output directory exists
+output_dir = "/content/predictions"
+os.makedirs(output_dir, exist_ok=True)
+
+# Function to make predictions, save images with labels, and return predictions and targets
+def predict_and_save_images(model, dataloader, output_dir):
+    model.eval()  # Set the model to evaluation mode
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    model.to(device)
+
+    all_predictions = []  # Store all predictions
+    all_targets = []      # Store all corresponding targets
+
+    # Iterate through the DataLoader
+    for batch_idx, (images, labels) in enumerate(dataloader):
+        images, labels = images.to(device), labels.to(device)
+        with torch.no_grad():
+            predictions = model(images).squeeze()  # Get model predictions
+
+        # Append predictions and targets to the lists
+        all_predictions.extend(predictions.cpu().tolist())
+        all_targets.extend(labels.cpu().tolist())
+
+        # Process each image in the batch
+        for i in range(len(images)):
+            # Convert the image back to [0, 1] range
+            image = images[i].cpu().permute(1, 2, 0).numpy()  # Convert to (H, W, C)
+            image = (image * 0.5 + 0.5) * 255  # Reverse normalization and scale to 0-255
+            image = Image.fromarray(image.astype("uint8"))
+
+            # Create a new canvas with extra space above the image
+            new_height = image.height + 50  # Add 50px for text
+            canvas = Image.new("RGB", (image.width, new_height), "white")  # White background
+            canvas.paste(image, (0, 50))  # Paste the image below the text area
+
+            # Draw real and predicted labels on the canvas
+            draw = ImageDraw.Draw(canvas)
+            font = ImageFont.load_default()  # Use a default font
+            real_label = f"Real: {labels[i].item():.2f}"
+            predicted_label = f"Pred: {predictions[i].item():.2f}"
+
+            # Write the labels with different colors
+            draw.text((10, 10), real_label, fill="red", font=font)  # Real label in red
+            draw.text((10, 30), predicted_label, fill="blue", font=font)  # Predicted label in blue
+
+            # Save the annotated image
+            canvas.save(os.path.join(output_dir, f"image_{batch_idx}_{i}.png"))
+
+    print(f"Annotated images saved to: {output_dir}")
+    return all_predictions, all_targets  # Return all predictions and targets
+
+# Run predictions, save images, and get predictions/targets
+predictions, targets = predict_and_save_images(model, dataloader, output_dir)
+
+# Print a few predictions and targets for verification
+for i in range(5):  # Display the first 5 predictions and targets
+    print(f"Target: {targets[i]:.2f}, Prediction: {predictions[i]:.2f}")
+
+import numpy as np
+
+def compute_metrics(predictions, targets):
+    # Convert to numpy arrays
+    predictions = np.array(predictions)
+    targets = np.array(targets)
+
+    # MAE
+    mae = np.mean(np.abs(predictions - targets))
+
+    # MSE
+    mse = np.mean((predictions - targets) ** 2)
+
+    # RMSE
+    rmse = np.sqrt(mse)
+
+    # MAPE
+    mape = np.mean(np.abs((targets - predictions) / (targets + 1e-8))) * 100  # Avoid division by zero
+
+    # R-squared
+    target_mean = np.mean(targets)
+    r2 = 1 - np.sum((targets - predictions) ** 2) / np.sum((targets - target_mean) ** 2)
+
+    # Angular Error
+    angular_error = np.mean(np.minimum(
+        np.abs(targets - predictions),
+        360 - np.abs(targets - predictions)
+    ))
+
+    return {
+        "Mean Absolute Error": mae,
+        "Root Mean Squared Error": rmse,
+    }
+
+metrics = compute_metrics(predictions, targets)
+
+for metric, value in metrics.items():
+    print(f"{metric}: {value:.2f}")
+
+
+"""
+
+	•	MAE averages the errors linearly.
+	•	RMSE grows quadratically due to squaring, amplifying the effect of the larger error.
+
+"""

src/training/task2_training.py

@@ -0,0 +1,34 @@
+"""
+!gdown --id 1JIpccI_hh8L5Rr_yf81cSXA3__K0fXVn
+
+!unzip /content/dataset_label.zip -d /content
+
+# !unzip /content/drive/MyDrive/PROJECT\ SDP/dataset_label.zip -d /content
+
+!pip install pytorch_lightning
+
+"""
+
+import pytorch_lightning as pl
+import torch
+import torch.nn as nn
+from timm import create_model
+from torch.utils.data import DataLoader, Dataset
+import torchvision.transforms as transforms
+import os
+from PIL import Image
+from utils.task2_utils import *
+
+
+
+# Trainer configuration
+trainer = pl.Trainer(
+    max_epochs=100,
+    accelerator="gpu" if torch.cuda.is_available() else "cpu",
+    devices=1,
+    callbacks=[checkpoint_callback, CustomProgressBar()],  # Add the custom ProgressBar
+)
+
+# Initialize and Train
+model = ViTRegression.load_from_checkpoint("/content/vit_tiny_rotation_epoch_epoch=100.ckpt")
+trainer.fit(model, dataloader)

src/utils/task2_utils.py

@@ -0,0 +1,140 @@
+import pytorch_lightning as pl
+import torch
+import torch.nn as nn
+from timm import create_model
+from torch.utils.data import DataLoader, Dataset
+import torchvision.transforms as transforms
+import os
+from PIL import Image
+
+
+# Define the dataset
+class BottleRotationDataset(Dataset):
+    def __init__(self, root_dir, transform=None):
+        self.image_paths = []
+        self.labels = []
+        self.transform = transform
+        for degree_folder in os.listdir(root_dir):
+            folder_path = os.path.join(root_dir, degree_folder)
+            if os.path.isdir(folder_path):
+                degree = float(degree_folder)
+                for img_name in os.listdir(folder_path):
+                    self.image_paths.append(os.path.join(folder_path, img_name))
+                    self.labels.append(degree)
+
+    def __len__(self):
+        return len(self.image_paths)
+
+    def __getitem__(self, idx):
+        image = Image.open(self.image_paths[idx]).convert('RGB')
+        label = self.labels[idx]
+        if self.transform:
+            image = self.transform(image)
+        return image, torch.tensor(label, dtype=torch.float32)
+
+# Define the custom PyTorch Lightning model
+class ViTRegression(pl.LightningModule):
+    def __init__(self, learning_rate=1e-4):
+        super().__init__()
+        self.save_hyperparameters()
+        self.vit = create_model('vit_tiny_patch16_224', pretrained=True)
+        num_features = self.vit.head.in_features
+        self.vit.head = nn.Identity()  # Remove classification head
+        self.regressor = nn.Linear(num_features, 1)  # Add regression head
+        self.criterion = nn.MSELoss()
+
+    def forward(self, x):
+        x = self.vit(x)
+        x = self.regressor(x)
+        return x
+
+    def training_step(self, batch, batch_idx):
+        images, targets = batch  # Targets are now in the range [0, 360]
+        outputs = self(images).squeeze()  # Predictions from the model
+        loss = self.criterion(outputs, targets)  # Compute loss without normalization
+
+        # Log the learning rate
+        current_lr = self.optimizers().param_groups[0]['lr']
+        self.log('learning_rate', current_lr, on_step=False, on_epoch=True)
+
+        # Log only at the end of the epoch
+        self.log('train_loss', loss, on_step=False, on_epoch=True)
+
+        return loss
+
+    def configure_optimizers(self):
+        # Define default values
+        default_learning_rate = 1e-4
+        default_scheduler_patience = 5
+        default_scheduler_factor = 0.1
+        default_scheduler_threshold = 100  # Adjust to match your loss scale
+        default_min_lr = 1e-6
+
+        # Retrieve hyperparameters or use defaults
+        learning_rate = self.hparams.get('learning_rate', default_learning_rate)
+        scheduler_patience = self.hparams.get('scheduler_patience', default_scheduler_patience)
+        scheduler_factor = self.hparams.get('scheduler_factor', default_scheduler_factor)
+        scheduler_threshold = self.hparams.get('scheduler_threshold', default_scheduler_threshold)
+        min_lr = self.hparams.get('min_lr', default_min_lr)
+
+        # Define optimizer
+        optimizer = torch.optim.Adam(self.parameters(), lr=learning_rate)
+
+        # Define scheduler
+        scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
+            optimizer,
+            mode='min',                  # Minimize the monitored metric
+            factor=scheduler_factor,     # Reduce learning rate by this factor
+            patience=scheduler_patience, # Wait this many epochs with no improvement
+            threshold=scheduler_threshold,  # Minimum improvement to consider
+            threshold_mode='abs',        # Use absolute threshold for large-scale losses
+            cooldown=0,                  # No cooldown period after reduction
+            min_lr=min_lr                # Minimum learning rate allowed
+        )
+
+        # Return optimizer and scheduler
+        return {
+            'optimizer': optimizer,
+            'lr_scheduler': {
+                'scheduler': scheduler,
+                'monitor': 'train_loss',  # Metric to monitor for learning rate adjustment
+                'interval': 'epoch',      # Check at the end of every epoch
+                'frequency': 1            # Frequency of scheduler updates
+            }
+        }
+
+# Transforms
+transform = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5])
+])
+
+# Dataset and DataLoader
+root_dir = "/content/dataset_label"  # Replace with your dataset path
+dataset = BottleRotationDataset(root_dir=root_dir, transform=transform)
+dataloader = DataLoader(dataset, batch_size=32, shuffle=True)
+
+# Lightning Trainer with Model Checkpoint Callback
+checkpoint_callback = pl.callbacks.ModelCheckpoint(
+    dirpath="checkpoints/",
+    filename="vit_tiny_rotation_epoch_{epoch}",
+    save_top_k=-1,  # Save all checkpoints
+    every_n_epochs=5,  # Save every 5 epochs
+)
+
+from pytorch_lightning.callbacks import ProgressBar
+
+# Custom ProgressBar to ensure logging at the end of the epoch
+class CustomProgressBar(ProgressBar):
+    def __init__(self):
+        super().__init__()
+
+    def on_train_epoch_end(self, trainer, pl_module):
+        train_loss = trainer.callback_metrics.get("train_loss", None)
+        learning_rate = trainer.callback_metrics.get("learning_rate", None)
+        if train_loss:
+            print(f"Epoch {trainer.current_epoch + 1}: Train Loss = {train_loss:.4f}")
+
+        if learning_rate:
+            print(f"Epoch {trainer.current_epoch + 1}: Learning Rate: {learning_rate:.6f}")