Following this instruction :
1. Strictly adhere to PEP-8 standards, ensuring code is clean, well-structured, and follows Python's style guide for readability.
2. Use appropriate modules like `typing` for type hints, ensuring code clarity, maintainability, and leveraging Python's standard libraries effectively.
3. Focus on time, space, robustness, scalability; write optimized, efficient code that can handle large datasets and scale seamlessly.
4. Implement comprehensive error handling, catching exceptions gracefully, providing meaningful error messages, and ensuring code robustness under unexpected conditions.
5. Ensure the developer agent writes advanced, maintainable Python code, balancing performance and readability while adhering to best practices.
 t 


TASK: we have generalised below code , best generalised , scalable, robustness,
# General imports
import os
import time
import copy
import numpy as np
import collections

# Plotting imports
from PIL import Image
import matplotlib.pyplot as plt
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import plotly.express as px

# Torch imports
import torch
import torch.nn as nn
import torch.backends.cudnn as cudnn
import torch.optim as optim
from torch.optim.lr_scheduler import CosineAnnealingLR

# Torchvision imports
import torchvision
from torchvision import datasets, models
from torchvision.transforms import v2

# Enable CuDNN benchmark mode for improved performance on compatible GPUs
cudnn.benchmark = True

# Turn on interactive mode for matplotlib
plt.ion()

# selecting the available device
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(device)

# Classes in CIFAR-10 dataset (airplane, automobile, bird, cat, deer, dog, frog, horse, ship, truck)
NUM_CLASSES = 10
# Image size for VGG19 (224x224)
IMAGE_SIZE = 224

mean, std = [0.4914, 0.4822, 0.4465], [0.247, 0.243, 0.261]

# Data transformations for training set
train_transforms = v2.Compose([
    v2.Resize((IMAGE_SIZE, IMAGE_SIZE)),
    v2.AutoAugment(policy=v2.AutoAugmentPolicy.CIFAR10),
    v2.ToTensor(),
    v2.Normalize(mean, std)
])

# Data transformations for validation set
validation_transforms = v2.Compose([
    v2.Resize((IMAGE_SIZE, IMAGE_SIZE)),
    v2.ToTensor(),
    v2.Normalize(mean, std)
])
# Dataset for training objectives
train_set = torchvision.datasets.CIFAR10(
    root='./data',
    train=True,
    download=True,
    transform=train_transforms
)

# Dataset for validation objectives
validation_set = torchvision.datasets.CIFAR10(
    root='./data',
    train=False,
    download=True,
    transform=validation_transforms
)
def plot_class_distribution(dataset, dataset_name):
    """
    Print and plot the class distribution of a dataset.

    Args:
        dataset (torch.utils.data.Dataset): The dataset to analyze.
        dataset_name (str): The name of the dataset.

    Returns:
        None
    """
    # Extract labels from the dataset
    labels = [y for _, y in dataset]

    # Count the number of images per category
    counter = collections.Counter(labels)

    # Print class image counter
    print(f"Class Image Counter for {dataset_name} Data")
    print(counter, "\n")

    # Optionally, plot the class distribution
    plt.bar(counter.keys(), counter.values())
    plt.xlabel("Class")
    plt.ylabel("Number of Images")
    plt.title(f"Class Distribution for {dataset_name} Data")
    plt.show()

num_gpus = torch.cuda.device_count()

# DataLoader for training set
train_loader = torch.utils.data.DataLoader(
    train_set,
    batch_size=batch_size,
    shuffle=True,
    num_workers=2*num_gpus,
    pin_memory=True
)

# DataLoader for validation set
validation_loader = torch.utils.data.DataLoader(
    validation_set,
    batch_size=batch_size,
    shuffle=False,
    num_workers=2*num_gpus,
    pin_memory=True
)

def imshow(tensor, title=None):
    """
    Display a batch of images in a grid.

    Args:
        tensor (torch.Tensor): The input tensor containing the images.
        title (str, optional): The title of the plot. Defaults to None.
    """
    image = torchvision.utils.make_grid(tensor).numpy().transpose((1, 2, 0))

    # Denormalize the image
    mean = np.array([0.4914, 0.4822, 0.4465])
    std = np.array([0.247, 0.243, 0.261])
    image = std * image + mean
    image = np.clip(image, 0, 1)

    # Increase the plot size
    plt.figure(figsize=(10, 10))

    # Plot the image
    plt.imshow(image)
    if title:
        plt.title(title)
    plt.pause(0.001)  # pause a bit so that plots are updated
def get_lr(opt):
    """
    Helper function to get the current learning rate from the optimizer.

    Args:
        opt (Optimizer): The optimizer object.

    Returns:
        float: The current learning rate.
    """
    for param_group in opt.param_groups:
        return param_group["lr"]
def metrics_batch(output, target):
    """
    Helper function to count the number of correct predictions in a batch.

    Args:
        output (torch.Tensor): Model predictions.
        target (torch.Tensor): Target labels.

    Returns:
        int: Number of correct predictions.
    """
    # Get the predicted class for each example in the batch
    pred = output.argmax(dim=1, keepdim=True)
    # Compare predicted class with target class and count correct predictions
    corrects = pred.eq(target.view_as(pred)).sum().item()
    return corrects


def loss_batch(loss_func, model_output, target, optimizer=None):
    """
    Helper function to compute the loss value per batch of data.

    Args:
        loss_func: Loss function.
        model_output (torch.Tensor): Model predictions.
        target (torch.Tensor): Target labels.
        optimizer: Optimizer for backpropagation (optional).

    Returns:
        tuple: Loss value, number of correct predictions.
    """
    # Calculate the loss using the specified loss function
    loss_value = loss_func(model_output, target)

    # Get performance metric (number of correct predictions)
    metric_batch = metrics_batch(model_output, target)

    if optimizer is not None:
        # Backpropagation and optimization step
        optimizer.zero_grad()
        #loss_value.requires_grad = True
        loss_value.backward()
        optimizer.step()

    return loss_value.item(), metric_batch


def loss_epoch(model, loss_func, data_loader, check_id=False, optimizer=None):
    """
    Compute the average loss value and performance metric over an epoch.

    Args:
        model: The neural network model.
        loss_func: Loss function.
        data_loader: DataLoader for the dataset.
        check_id (bool): Flag to check only the first batch.
        optimizer: Optimizer for backpropagation (optional).

    Returns:
        tuple: Average loss value, average performance metric.
    """
    running_loss = 0.0
    running_metric = 0.0
    total_samples = len(data_loader.dataset)

    # Internal loop over batches in the DataLoader
    for inputs, labels in data_loader:
        inputs, labels = inputs.to(device), labels.to(device)
        model_output = model(inputs)
        # Calculate loss and performance metric for the batch
        loss_batch_value, metric_batch_value = loss_batch(
            loss_func, model_output, labels, optimizer
        )
        running_loss += loss_batch_value

        if metric_batch_value is not None:
            running_metric += metric_batch_value

        if check_id:
            break  # Stop if only checking the first batch

    # Compute the average loss and performance metric over the entire epoch
    average_loss = running_loss / float(total_samples)
    average_metric = (
        running_metric / float(total_samples) if total_samples > 0 else None
    )

    return average_loss, average_metric
def train_val(model, params, verbose=False):
    """
    Trains and validates a model using the given parameters.

    Args:
        model (torch.nn.Module): The model to be trained and validated.
        params (dict): A dictionary containing the parameters for training and validation.
            - epochs (int): The number of epochs to train the model.
            - loss_func (torch.nn.Module): The loss function to be used.
            - optimiser (torch.optim.Optimizer): The optimizer to be used for training.
            - train_dl (torch.utils.data.DataLoader): The data loader for the training dataset.
            - val_dl (torch.utils.data.DataLoader): The data loader for the validation dataset.
            - check_id (int): The ID for checking the loss and metric.
            - lr_scheduler (torch.optim.lr_scheduler._LRScheduler): The learning rate scheduler.
            - path (str): The path to save the best model weights.

        verbose (bool, optional): Whether to print additional information during training. Defaults to False.

    Returns:
        tuple: A tuple containing the trained model, loss history, and metric history.
            - model (torch.nn.Module): The trained model.
            - loss_history (dict): A dictionary containing the loss values for each epoch.
            - metric_history (dict): A dictionary containing the metric values for each epoch.
    """

    # extract model parameters
    epochs = params["epochs"]
    loss_func = params["loss_func"]
    opt = params["optimizer"]
    train_dl = params["train_dl"]
    val_dl = params["val_dl"]
    check_id = params["check_id"]
    lr_scheduler = params["lr_scheduler"]
    path = params["path"]

    loss_history = {"train": [], "val": []}  # history of loss values in each epoch
    metric_history = {"train": [], "val": []}  # histroy of metric values in each epoch
    best_model_wts = copy.deepcopy(model.state_dict())  # copy weights for best model
    best_loss = float("inf")  # initialize best loss to a large value

    # main loop
    for epoch in range(epochs):
        current_lr = get_lr(opt)  # get current learning rate
        if verbose:
            print(f"Epoch {epoch}/{epochs-1}, current lr={current_lr}")

        # train model on training dataset
        model.train()
        train_loss, train_metric = loss_epoch(model, loss_func, train_dl, check_id, opt)

        # collect loss and metric for training dataset
        loss_history["train"].append(train_loss)
        metric_history["train"].append(train_metric)

        # evaluate model on validation dataset
        model.eval()
        with torch.no_grad():
            val_loss, val_metric = loss_epoch(model, loss_func, val_dl, check_id)

        # store best model
        if val_loss < best_loss:
            best_loss = val_loss
            best_model_wts = copy.deepcopy(model.state_dict())

            # store weights into a local file
            torch.save(model.state_dict(), path)
            if verbose:
                print("Copied best model weights!")

        # collect loss and metric for validation dataset
        loss_history["val"].append(val_loss)
        metric_history["val"].append(val_metric)

        # learning rate schedule
        lr_scheduler.step()

        if verbose:
            print(
                f"train loss: {train_loss:.6f}, train accuracy: {100*train_metric:.2f}, validation loss: {val_loss:.6f}, validation accuracy: {100*val_metric:.2f}"
            )
            print("")

    # load best model weights
    model.load_state_dict(best_model_wts)

    return model, loss_history, metric_history
def plot_out(loss_hist, metric_hist, epochs=None):
    """
    Plot training and validation loss, and training and validation metrics over epochs.

    Args:
        loss_hist (dict): Dictionary containing 'train' and 'val' keys for loss history.
        metric_hist (dict): Dictionary containing 'train' and 'val' keys for metric history.
        epochs (int): Number of epochs.
    """

    # Create subplot with two columns
    fig = make_subplots(
        rows=1, cols=2, subplot_titles=["Model Loss", "Model Accuracy"]
    )

    # Plot Loss History
    for phase, color in zip(["train", "val"], ["#F1C40F", "#232323"]):
        fig.add_trace(
            go.Scatter(
                x=list(range(1, epochs + 1)),
                y=loss_hist[phase],
                name=phase,
                mode="lines",
                line_color=color,
            ),
            row=1,
            col=1,
        )

    # Plot Metric History
    for phase, color in zip(["train", "val"], ["#F1C40F", "#232323"]):
        fig.add_trace(
            go.Scatter(
                x=list(range(1, epochs + 1)),
                y=metric_hist[phase],
                name=phase,
                mode="lines",
                line_color=color,
            ),
            row=1,
            col=2,
        )

    # Update layout and display the plot
    fig.update_layout(
        template="plotly_white",
        showlegend=False,
        title="Loss & Accuracy History",
        height=400,
    )
    fig.update_layout(yaxis2=dict(range=[0.4, 1]))
    fig.show()
# Create the base model from the pre-trained model VGG19
pre_vgg19 = models.vgg19(weights='IMAGENET1K_V1')
pre_vgg19
# change the number of classes
pre_vgg19.classifier[6].out_features = NUM_CLASSES

# freeze convolution weights
for param in pre_vgg19.features.parameters():
    param.requires_grad = False
# Send earlier defined model to device
device = torch.device("cuda:0")
pre_vgg19 = pre_vgg19.to(device)

# Loss function, optimizer, LR scheduler
loss_func = nn.CrossEntropyLoss(reduction="sum")
optimizer = optim.Adam(pre_vgg19.parameters(), lr=learning_rate)
lr_scheduler = CosineAnnealingLR(optimizer, T_max=5, eta_min=1e-6)

# Set Training Parameters
training_params = {
    "epochs": 10,
    "optimizer": optimizer,
    "loss_func": loss_func,
    "train_dl": train_loader,
    "val_dl": validation_loader,
    "check_id": False,
    "lr_scheduler": lr_scheduler,
    "path": "pre_vgg19.pt",
}

# Train and validate the model
pre_vgg19, loss_history, metric_history = train_val(
    pre_vgg19, training_params, verbose=False
)
# Plot History
plot_out(loss_history, metric_history, epochs=training_params["epochs"])