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, 

from huggingface_hub import hf_hub_download

hf_dataset_identifier = "sayakpaul/ucf101-subset"
filename = "UCF101_subset.tar.gz"
file_path = hf_hub_download(
    repo_id=hf_dataset_identifier, filename=filename, repo_type="dataset"
)
!tar xf {file_path}
dataset_root_path = "UCF101_subset"

!find {dataset_root_path} | head -4
import pathlib

dataset_root_path = pathlib.Path(dataset_root_path)
train_count = len(list(dataset_root_path.glob("train/*/*.avi")))
val_count = len(list(dataset_root_path.glob("val/*/*.avi")))
test_count = len(list(dataset_root_path.glob("test/*/*.avi")))
total_video = train_count + val_count + test_count
print(f"Train videos: {train_count}")
print(f"Test videos: {test_count}")
print(f"Validation videos: {val_count}")
print(f"Total videos: {total_video}")
all_video_file_paths = (
    list(dataset_root_path.glob("train/*/*.avi"))
    + list(dataset_root_path.glob("val/*/*.avi"))
    + list(dataset_root_path.glob("test/*/*.avi"))
)
all_video_file_paths[:5]
class_labels = sorted({str(path).split("/")[2] for path in all_video_file_paths})
my_label2id = {label: i for i, label in enumerate(class_labels)}
my_id2label = {i: label for label, i in my_label2id.items()}

print(f"Unique classes: {list(my_label2id.keys())}.")
model_checkpt = "google/vivit-b-16x2-kinetics400" # pre-trained ViViT model
batch_size = 4

from transformers import VivitImageProcessor

image_processor = VivitImageProcessor.from_pretrained(model_checkpt)
from transformers import VivitForVideoClassification

model = VivitForVideoClassification.from_pretrained(
    model_checkpt,
    id2label=my_id2label,
    label2id=my_label2id,
    ignore_mismatched_sizes=True,  # provide this in case you're planning to fine-tune an already fine-tuned checkpoint
)
import pytorchvideo.data

from pytorchvideo.transforms import (
    ApplyTransformToKey,
    RemoveKey,
    Normalize,
    ShortSideScale,
    RandomShortSideScale,
    UniformTemporalSubsample,
)

from torchvision.transforms import (
    RandomCrop,
    RandomHorizontalFlip,
    Resize,
    Compose,
    Lambda,
)
import os

mean = image_processor.image_mean
std = image_processor.image_std
if "shortest_edge" in image_processor.size:
  height = width = image_processor.size["shortest_edge"]
else:
  height = image_processor.size["height"]
  width = image_processor.size["width"]
resize_to = (height, width)

num_frames_to_sample = model.config.num_frames
sample_rate = 4
fps = 30
clip_duration = num_frames_to_sample * sample_rate / fps
# Training dataset transformations
train_transform = Compose(
  [
    ApplyTransformToKey(
      key="video",
      transform=Compose(
        [
          UniformTemporalSubsample(num_frames_to_sample),
          Lambda(lambda x: x / 255.0),
          Normalize(mean, std),
          RandomShortSideScale(min_size=256, max_size=320),
          RandomCrop(resize_to),
          RandomHorizontalFlip(p=0.5),
        ]
      ),
    ),
  ]
)

# Training dataset
train_dataset = pytorchvideo.data.Ucf101(
  data_path=os.path.join(dataset_root_path, "train"),
  clip_sampler=pytorchvideo.data.make_clip_sampler("random", clip_duration),
  decode_audio=False,
  transform=train_transform,
)

# Validation and test dataset transformations
val_transform = Compose(
  [
    ApplyTransformToKey(
      key="video",
      transform=Compose(
        [
          UniformTemporalSubsample(num_frames_to_sample),
          Lambda(lambda x: x / 255.0),
          Normalize(mean, std),
          Resize(resize_to),
        ]
      ),
    ),
  ]
)

# Validation and test datasets
val_dataset = pytorchvideo.data.Ucf101(
  data_path=os.path.join(dataset_root_path, "val"),
  clip_sampler=pytorchvideo.data.make_clip_sampler("uniform", clip_duration),
  decode_audio=False,
  transform=val_transform,
)

test_dataset = pytorchvideo.data.Ucf101(
  data_path=os.path.join(dataset_root_path, "test"),
  clip_sampler=pytorchvideo.data.make_clip_sampler("uniform", clip_duration),
  decode_audio=False,
  transform=val_transform,
)
train_dataset.num_videos, val_dataset.num_videos, test_dataset.num_videos
sample_video = next(iter(train_dataset))
sample_video.keys()

def investigate_video(sample_video):
  """Utility function to investigate the keys present in a video sample."""
  for k in sample_video:
    if k == "video":
      print(k, sample_video["video"].shape)
    else:
      print(k, sample_video[k])

  print(f"Video label: {my_id2label[sample_video[k]]}")


investigate_video(sample_video)
import imageio
import numpy as np
from IPython.display import HTML, Image

def unnormalize_img(img):
  """Un-normalizes the image pixels."""
  img = (img * std) + mean
  img = (img * 255).astype("uint8")
  return img.clip(0, 255)

def create_gif(video_tensor, filename="sample.gif"):
  """Prepares a GIF from a video tensor."""
  frames = []
  for video_frame in video_tensor:
      frame_unnormalized = unnormalize_img(video_frame.permute(1, 2, 0).numpy())
      frames.append(frame_unnormalized)
  kargs = {"duration": 0.4}
  imageio.mimsave(filename, frames, "GIF", **kargs)
  return filename

def display_gif(video_tensor, gif_name="sample.gif"):
  """Prepares and displays a GIF from a video tensor."""
  video_tensor = video_tensor.permute(1, 0, 2, 3)    # re-arranging dimension of video_tensor
  gif_filename = create_gif(video_tensor, gif_name)
  return Image(filename=gif_filename)

video_tensor = sample_video["video"]
display_gif(video_tensor)

from transformers import TrainingArguments, Trainer

model_name = model_checkpt.split("/")[-1]
new_model_name = f"{model_name}-finetuned-ucf101-subset"
num_epochs = 5

args = TrainingArguments(
    new_model_name,
    auto_find_batch_size=True,
    remove_unused_columns=False,
    evaluation_strategy="epoch",
    save_strategy="epoch",
    learning_rate=5e-5,
    per_device_train_batch_size=batch_size,
    per_device_eval_batch_size=batch_size,
    warmup_ratio=0.1,
    logging_steps=10,
    load_best_model_at_end=True,
    metric_for_best_model="accuracy",
    push_to_hub=False,
    max_steps=(train_dataset.num_videos // batch_size) * num_epochs,
    num_train_epochs=num_epochs,
)
import evaluate

metric = evaluate.load("accuracy")

def compute_metrics(eval_pred):
  """Computes accuracy on a batch of predictions."""
  predictions = np.argmax(eval_pred.predictions, axis=1)
  return metric.compute(predictions=predictions, references=eval_pred.label_ids)

import torch

def collate_fn(examples):
    """The collation function to be used by `Trainer` to prepare data batches."""
    # permute to (num_frames, num_channels, height, width)
    pixel_values = torch.stack(
        [example["video"].permute(1, 0, 2, 3) for example in examples]
    )
    labels = torch.tensor([example["label"] for example in examples])
    return {"pixel_values": pixel_values, "labels": labels}
trainer = Trainer(
    model,
    args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset,
    tokenizer=image_processor,
    compute_metrics=compute_metrics,
    data_collator=collate_fn,
)
train_results = trainer.train()
trainer.save_model()
test_results = trainer.evaluate(test_dataset)
trainer.log_metrics("test", test_results)
trainer.save_metrics("test", test_results)
trainer.save_state()
trained_model = VivitForVideoClassification.from_pretrained(new_model_name)
sample_test_video = next(iter(val_dataset))
investigate_video(sample_test_video)

def run_inference(model, video):
    """Utility to run inference given a model and test video.

    The video is assumed to be preprocessed already.
    """
    # (num_frames, num_channels, height, width)
    perumuted_sample_test_video = video.permute(1, 0, 2, 3)

    inputs = {
        "pixel_values": perumuted_sample_test_video.unsqueeze(0),
        "labels": torch.tensor(
            [sample_test_video["label"]]
        ),  # this can be skipped if you don't have labels available.
    }
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    inputs = {k: v.to(device) for k, v in inputs.items()}
    model = model.to(device)

    # forward pass
    with torch.no_grad():
        outputs = model(**inputs)
        logits = outputs.logits

    return logits
logits = run_inference(trained_model, sample_test_video["video"])
display_gif(sample_test_video["video"])
predicted_class_idx = logits.argmax(-1).item()
print("Predicted class:", model.config.id2label[predicted_class_idx])