data.py

import torch
from datasets import load_dataset
from torch.utils.data import Dataset


class BeansDataset(Dataset):
    def __init__(self, image_processor, tp_size=1, split="train"):
        super().__init__()
        self.image_processor = image_processor
        self.ds = load_dataset("beans")[split]
        self.label_names = self.ds.features["labels"].names
        while len(self.label_names) % tp_size != 0:
            # ensure that the number of labels is multiple of tp_size
            self.label_names.append(f"pad_label_{len(self.label_names)}")
        self.num_labels = len(self.label_names)
        self.inputs = []
        for example in self.ds:
            self.inputs.append(self.process_example(example))

    def __len__(self):
        return len(self.inputs)

    def __getitem__(self, idx):
        return self.inputs[idx]

    def process_example(self, example):
        input = self.image_processor(example["image"], return_tensors="pt")
        input["labels"] = example["labels"]
        return input


def beans_collator(batch):
    return {
        "pixel_values": torch.cat([data["pixel_values"] for data in batch], dim=0),
        "labels": torch.tensor([data["labels"] for data in batch], dtype=torch.int64),
    }