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 pipeline Training, evaluation,inference( from test data) , save plots training , evaluation,inference in the plots(plotly module and save html)
save the weights of in the folder, 

import os
import random
from functools import partial
from typing import Any
import evaluate
import matplotlib.pyplot as plt
import numpy as np
import torch
import torch.nn as nn
from datasets import Dataset, DatasetDict, load_dataset, load_metric
from peft import LoraConfig, get_peft_model
from PIL import Image
from torch.utils.data import DataLoader
from tqdm.notebook import tqdm
from transformers import (CLIPImageProcessor, CLIPModel, CLIPProcessor,
                          CLIPTokenizerFast, Trainer, TrainingArguments)
from transformers.tokenization_utils_base import BatchEncoding
from datasets.formatting.formatting import LazyBatch

os.environ["CURL_CA_BUNDLE"] = ""
os.environ["TOKENIZERS_PARALLELISM"] = "false"

def seed_all(seed: int):
    random.seed(seed)
    torch.manual_seed(seed)
    np.random.seed(seed)
    
seed_all(69)
dataset = load_dataset("pcuenq/oxford-pets")
dataset_train_val = dataset['train'].train_test_split(test_size=0.3)
dataset_val_test = dataset_train_val['test'].train_test_split(test_size=0.2)

dataset = DatasetDict(
    {
        "train": dataset_train_val['train'],  
        "val": dataset_val_test['test'],
        "test": dataset_val_test['train']
    }
)
def show_images(ds: Dataset):
    fig, axs = plt.subplots(3, 3, figsize=(13, 13))
    for i in range(9):
        axs[i // 3][i % 3].imshow(ds[i]['image'])
        axs[i // 3][i % 3].set_title(ds[i]['label'])

show_images(dataset['train'])
labels = set(dataset['train']['label'])
label2id = {label: i for i, label in enumerate(labels)}
id2label = {i: label for label, i in label2id.items()}

labels = list(label2id)
MODEL_NAME = "openai/clip-vit-base-patch32"
TOKENIZER = CLIPTokenizerFast.from_pretrained(MODEL_NAME)
IMAGE_PROCESSOR = CLIPImageProcessor.from_pretrained(MODEL_NAME)
def transform_class_labels(
    items: LazyBatch, tokenizer: CLIPTokenizerFast, label2id: dict[str, int]
) -> dict[str, Any]:
    """Function used to encode item's label prompt with tokenizer

    Args:
        items (LazyBatch): Input dataset items
        tokenizer (CLIPTokenizerFast): CLIP's tokenizer
        label2id (dict[str, int]): mapping from label to id

    Returns:
        dict[str, Any]: transformed items
    """
    # this prompt is better than prompt consisting only of the class name
    label_prompt = [f"a photo of {label}" for label in items["label"]]
    output = tokenizer(label_prompt, padding=True, return_tensors="pt")
    items["input_ids"] = output["input_ids"]
    items["attention_mask"] = output["attention_mask"]
    items["label_id"] = [label2id[label] for label in items["label"]]
    return items


def transform_image(items: LazyBatch, image_processor: CLIPImageProcessor) -> dict[str, Any]:
    """Function used to preprocess input image with image processor

    Args:
        items (LazyBatch): Input dataset items
        image_processor (CLIPImageProcessor): CLIP's image processor

    Returns:
        dict[str, Any]: transformed items
    """
    output = image_processor(items["image"], return_tensors="pt")
    items["pixel_values"] = output["pixel_values"]
    return items
dataset = dataset.map(partial(transform_class_labels, tokenizer=TOKENIZER, label2id=label2id), batched=True)
dataset.set_transform(partial(transform_image, image_processor=IMAGE_PROCESSOR))
# utilities functions
def get_module_device(module: nn.Module) -> torch.device:
    """Function to get current module device

    Args:
        module (nn.Module): pytorch module

    Returns:
        torch.device: module's device
    """
    return next(module.parameters()).device


def freeze_params(module: nn.Module, freeze_top_percent: float = 1.0) -> None:
    """Function used to freeze module by setting 'requires_grad' to False

    Args:
        module (nn.Module): pytorch module
        freeze_top_percent (float, optional): percentage of layers starting from the top that need to be frozen. Defaults to 1.0.
    """
    all_params_length = len(list(module.parameters()))
    for indx, param in enumerate(module.parameters()):
        if int(all_params_length * freeze_top_percent) <= indx:
            break
        param.requires_grad = False


def print_trainable_parameters(model: nn.Module) -> None:
    """Function prints statistics about trainable parameters

    Args:
        model (nn.Module): pytorch module
    """
    trainable_params = 0
    all_param = 0
    for _, param in model.named_parameters():
        all_param += param.numel()
        if param.requires_grad:
            trainable_params += param.numel()
    print(
        f"Trainable params: {(trainable_params / 10**6):.4f}M || All params: {(all_param / 10**6):.4f}M || Trainable%: {100 * trainable_params / all_param:.2f}%"
    )
class CLIPClassifier(nn.Module):
    def __init__(self, clip_model: CLIPModel, tokenizer: CLIPTokenizerFast, labels: list[str]):
        super().__init__()
        self.model = clip_model
        self.tokenizer = tokenizer
        self.logit_scale = self.model.logit_scale.exp()
        self.label2id = {label: i for i, label in enumerate(labels)}
        self.labels_embeddings = nn.Parameter(self.generate_labels_embeddings(labels))

    def generate_labels_embeddings(self, labels: list[str]) -> torch.Tensor:
        """Function prepares label's prompt embeddings usings language model

        Args:
            labels (list[str]): list with dataset labels

        Returns:
            torch.Tensor: labels embeddings of size (num_labels, dim)
        """
        # tokenize label's prompt
        labels_inputs = self.tokenizer(
            [f"a photo of {label}" for label in labels],
            return_tensors="pt",
            padding=True,
        ).to(get_module_device(self.model))
        # run language model on this prompt and get embeddings
        labels_embeddings = self.model.get_text_features(**labels_inputs)
        # normilize embeddings
        labels_embeddings /= labels_embeddings.norm(p=2, dim=-1, keepdim=True)
        return labels_embeddings

    def forward(self, images: torch.Tensor) -> torch.Tensor:
        """Function to calculate similarity between image and label's promts embeddings

        Args:
            images (torch.Tensor): image tensor

        Returns:
            torch.Tensor: output similarity tensor of size (batch_size,  num_labels)
        """
        image_features = self.model.get_image_features(
            images
        )  # (batch_size, dim)
        image_features /= image_features.norm(p=2, dim=-1, keepdim=True)
        return torch.matmul(image_features, self.labels_embeddings.T) * self.logit_scale

def calculate_accuracy(model: CLIPClassifier, dataloader: DataLoader) -> float:
    """Function calculates accuracy of given model on dataloader

    Args:
        model (CLIPClassifier): CLIP classifier model
        dataloader (DataLoader): evaluation dataloader

    Returns:
        float: model's accuracy
    """
    metric = evaluate.load("accuracy")
    predictions_list = []
    references_list = []
    device = get_module_device(model)
    for batch in tqdm(
        dataloader, total=len(dataloader), desc="Evaluate model on dataset"
    ):
        batch["pixel_values"] = batch["pixel_values"].to(device)
        predictions = model(batch["pixel_values"])
        predictions_list.append(torch.argmax(predictions, dim=1))
        references_list.append(batch["label_id"])
    return metric.compute(
        predictions=torch.concat(predictions_list),
        references=torch.concat(references_list),
    )["accuracy"]

def collate_fn(items: LazyBatch) -> dict[str, Any]:
    return {
        "pixel_values": torch.stack(
            [item["pixel_values"] for item in items]
        ),  # (batch_size, 3, 224, 224)
        "input_ids": torch.tensor(
            [item["input_ids"] for item in items]
        ),  # (batch_size, max_length)
        "attention_mask": torch.tensor(
            [item["attention_mask"] for item in items]
        ),  # (batch_size, max_length)
        "label_id": torch.tensor([item["label_id"] for item in items]),  # (batch_size),
        "return_loss": True,
    }

@torch.no_grad()
def evaluate_clip_classifier(
    model: nn.Module,
    dataset: Dataset,
    tokenizer: CLIPTokenizerFast,
    labels: list[str],
    batch_size: int = 64,
    num_workers: int = 5,
    device: str = "cuda",
) -> None:
    """Function evaluates CLIP model on given dataset

    Args:
        model (nn.Module): CLIP model
        dataset (Dataset): evaluation dataset
        tokenizer (CLIPTokenizerFast): CLIP tokenizer
        labels (list[str]): list with dataset labels
        batch_size (int): batch size. Defaults to 64
        num_workers (int): number of workers for dataloader. Defaults to 5
        device (str, optional): model's device. Defaults to "cuda".
    """
    clip_classifier = CLIPClassifier(model, tokenizer, labels)
    test_dataloader = DataLoader(
        dataset, batch_size=batch_size, num_workers=num_workers, collate_fn=collate_fn
    )
    clip_classifier = clip_classifier.to(device)
    acc = calculate_accuracy(clip_classifier, test_dataloader)
    print(f"Model accuracy: {acc}")
clip_baseline = CLIPModel.from_pretrained(MODEL_NAME)
freeze_params(clip_baseline)
print_trainable_parameters(clip_baseline)
evaluate_clip_classifier(clip_baseline, dataset['test'], TOKENIZER, labels)
def collate_train_fn(items: LazyBatch):
    items = collate_fn(items)
    items.pop("label_id")
    return items


def get_default_training_args(
    experiment_name: str,
    lr: float,
    batch_size: int = 256,
    num_epoch: int = 2,
    num_workers: int = 15,
) -> TrainingArguments:
    """Function gets default training arguments for HF Trainer

    Args:
        experiment_name (str): experiment name
        lr (float): learning rate
        batch_size (int, optional): batch size. Defaults to 256.
        num_epoch (int, optional): number of epoch to train. Defaults to 2.
        num_workers (int, optional): number of workers for dataloader. Defaults to 15.

    Returns:
        TrainingArguments: arguments object for HF Trainer
    """
    return TrainingArguments(
        experiment_name,
        per_device_train_batch_size=batch_size,
        learning_rate=lr,
        num_train_epochs=num_epoch,
        per_device_eval_batch_size=batch_size,
        gradient_accumulation_steps=1,
        logging_steps=10,
        save_total_limit=2,
        evaluation_strategy="epoch",
        save_strategy="epoch",
        fp16=True,
        remove_unused_columns=False,
        load_best_model_at_end=True,
        dataloader_num_workers=num_workers,
    )
clip_full_finetuned = CLIPModel.from_pretrained(MODEL_NAME)
trainer = Trainer(
    model=clip_full_finetuned,
    args=get_default_training_args("clip-all-layers-tuning-oxford-pets", 3e-6),
    data_collator=collate_train_fn,
    train_dataset=dataset["train"],
    eval_dataset=dataset["val"],
)

trainer.train()


print_trainable_parameters(clip_full_finetuned)
evaluate_clip_classifier(clip_full_finetuned, dataset['test'], TOKENIZER, labels)

clip_text_model_tuning = CLIPModel.from_pretrained(MODEL_NAME)
freeze_params(clip_text_model_tuning.vision_model)

trainer = Trainer(
    model=clip_text_model_tuning,
    args=get_default_training_args("clip-text-model-tuning-oxford-pets", 3e-5),
    data_collator=collate_train_fn,
    train_dataset=dataset["train"],
    eval_dataset=dataset["val"],
)

trainer.train()
print_trainable_parameters(clip_text_model_tuning)
evaluate_clip_classifier(clip_text_model_tuning, dataset['test'], TOKENIZER, labels)
clip_vision_model_tuning = CLIPModel.from_pretrained(MODEL_NAME)
freeze_params(clip_vision_model_tuning.text_model)

trainer = Trainer(
    model=clip_vision_model_tuning,
    args=get_default_training_args("clip-vision-model-tuning-oxford-pets", 3e-5),
    data_collator=collate_train_fn,
    train_dataset=dataset["train"],
    eval_dataset=dataset["test"],
)

trainer.train()
print_trainable_parameters(clip_vision_model_tuning)
evaluate_clip_classifier(clip_vision_model_tuning, dataset['test'], TOKENIZER, labels)
clip_partial_tuning = CLIPModel.from_pretrained(MODEL_NAME)
freeze_params(clip_partial_tuning.text_model, freeze_top_percent=0.7)
freeze_params(clip_partial_tuning.vision_model, freeze_top_percent=0.7)
trainer = Trainer(
    model=clip_partial_tuning,
    args=get_default_training_args("clip-partial-model-tuning-oxford-pets", 3e-5),
    data_collator=collate_train_fn,
    train_dataset=dataset["train"],
    eval_dataset=dataset["val"],
)

trainer.train()
print_trainable_parameters(clip_partial_tuning)
evaluate_clip_classifier(clip_partial_tuning, dataset['test'], TOKENIZER, labels)

clip_lora_tuning = CLIPModel.from_pretrained(MODEL_NAME)
config = LoraConfig(
    r=64,
    lora_alpha=64,
    target_modules=['q_proj', 'k_proj', 'v_proj'],
)

lora_model = get_peft_model(clip_lora_tuning, config)
trainer = Trainer(
    model=lora_model,
    args=get_default_training_args("clip-lora-model-tuning-oxford-pets", 3e-4),
    data_collator=collate_train_fn,
    train_dataset=dataset["train"],
    eval_dataset=dataset["val"],
)

trainer.train()
print_trainable_parameters(lora_model)
evaluate_clip_classifier(lora_model, dataset['test'], TOKENIZER, labels)