autocatalog/data/dataset.py

import random
from pathlib import Path
import numpy as np
import pandas as pd
import torch
from autocatalog.data.preprocessing import extract_color_features
from autocatalog.utils.logger import get_logger
from datasets import load_dataset
from PIL import Image, ImageOps
from sklearn.model_selection import train_test_split
from torch.utils.data import DataLoader, Dataset
from tqdm.auto import tqdm
logger = get_logger(__name__)

def load_clean_dataset(dataset_name, tasks, label_maps):
    dataset = load_dataset(dataset_name, split="train")
    if "image" not in dataset.column_names:
        raise ValueError("Dataset must contain an image column")

    missing = [task for task in tasks if task not in dataset.column_names]
    if missing:
        raise ValueError(f"Missing task columns: {missing}")

    def valid(row):
        if row.get("image") is None:
            return False

        for task in tasks:
            value = str(row.get(task, "")).strip()
            if not value:
                return False
            if value not in label_maps[task]["label2id"]:
                return False

        return True

    clean_dataset = dataset.filter(valid)
    logger.info("Dataset loaded | raw=%d | clean=%d",len(dataset),len(clean_dataset),)
    return clean_dataset

def _stratify_labels(series):
    counts = series.value_counts()
    return series.apply(lambda value: value if counts[value] >= 2 else "__rare__")

def create_splits(dataset, tasks, output_dir, seed, train_ratio, validation_ratio, test_ratio,):
    total_ratio = train_ratio + validation_ratio + test_ratio
    if round(total_ratio, 6) != 1.0:
        raise ValueError("Split ratios must sum to 1.0")

    metadata = {
        task: [str(value).strip() for value in dataset[task]]
        for task in tasks
    }

    if "id" in dataset.column_names:
        metadata["id"] = dataset["id"]

    if "productDisplayName" in dataset.column_names:
        metadata["productDisplayName"] = dataset["productDisplayName"]

    dataframe = pd.DataFrame(metadata)
    dataframe["dataset_idx"] = np.arange(len(dataset))
    all_indices = dataframe.index.to_numpy()
    temporary_ratio = validation_ratio + test_ratio

    stratify_task = "articleType" if "articleType" in tasks else tasks[-1]
    try:
        train_idx, temporary_idx = train_test_split(
            all_indices,
            test_size=temporary_ratio,
            random_state=seed,
            stratify=_stratify_labels(dataframe[stratify_task]),
        )
    except ValueError:
        train_idx, temporary_idx = train_test_split(
            all_indices,
            test_size=temporary_ratio,
            random_state=seed,
        )

    temporary_dataframe = dataframe.loc[temporary_idx]
    test_share = test_ratio / temporary_ratio

    try:
        validation_idx, test_idx = train_test_split(
            temporary_idx,
            test_size=test_share,
            random_state=seed,
            stratify=_stratify_labels(temporary_dataframe[stratify_task]),
        )
    except ValueError:
        validation_idx, test_idx = train_test_split(
            temporary_idx,
            test_size=test_share,
            random_state=seed,
        )

    train_df = dataframe.loc[train_idx].copy()
    validation_df = dataframe.loc[validation_idx].copy()
    test_df = dataframe.loc[test_idx].copy()

    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    train_df.to_csv(output_dir / "train.csv", index=False)
    validation_df.to_csv(output_dir / "val.csv", index=False)
    test_df.to_csv(output_dir / "test.csv", index=False)

    logger.info("Dataset split | train=%d | validation=%d | test=%d",len(train_df),len(validation_df),len(test_df),)
    return train_df, validation_df, test_df

def load_or_create_color_cache(dataset, cache_path, image_size, feature_dim):
    cache_path = Path(cache_path)
    cache_path.parent.mkdir(parents=True, exist_ok=True)

    if cache_path.exists():
        features = np.load(cache_path)
        expected_shape = (len(dataset), feature_dim)

        if features.shape == expected_shape:
            logger.info("Color cache loaded | shape=%s", features.shape)
            return features

        logger.warning("Color cache shape mismatch; rebuilding")

    features = np.zeros((len(dataset), feature_dim), dtype=np.float32)
    for index in tqdm(range(len(dataset)), desc="Extracting color features"):
        features[index] = extract_color_features(
            dataset[index]["image"],
            image_size=image_size,
        )

    np.save(cache_path, features)
    logger.info("Color cache saved | path=%s", cache_path)
    return features


class FashionMultiTaskDataset(Dataset):
    def __init__(self, source_dataset, indices, color_features, processor, label_maps, tasks, training=False):
        self.source_dataset = source_dataset
        self.indices = list(map(int, indices))
        self.color_features = color_features
        self.processor = processor
        self.label_maps = label_maps
        self.tasks = tasks
        self.training = training

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

    def __getitem__(self, index):
        global_index = self.indices[index]
        item = self.source_dataset[global_index]
        image = item["image"]

        if not isinstance(image, Image.Image):
            image = Image.open(image)

        image = image.convert("RGB")
        if self.training and random.random() < 0.5:
            image = ImageOps.mirror(image)

        pixel_values = self.processor(
            images=image,
            return_tensors="pt",
        )["pixel_values"].squeeze(0)

        labels = {
            task: torch.tensor(
                self.label_maps[task]["label2id"][str(item[task]).strip()],
                dtype=torch.long,
            )
            for task in self.tasks
        }

        color_features = torch.tensor(
            self.color_features[global_index],
            dtype=torch.float32,
        )

        return {
            "pixel_values": pixel_values,
            "color_features": color_features,
            "labels": labels,
            "global_index": global_index,
        }


class MultiTaskCollator:
    def __init__(self, tasks):
        self.tasks = tasks

    def __call__(self, batch):
        return {
            "pixel_values": torch.stack(
                [item["pixel_values"] for item in batch]
            ),
            "color_features": torch.stack(
                [item["color_features"] for item in batch]
            ),
            "labels": {
                task: torch.stack([item["labels"][task] for item in batch])
                for task in self.tasks
            },
            "global_indices": [item["global_index"] for item in batch],
        }


def build_dataloaders( dataset, train_df, validation_df, test_df, color_features, processor, label_maps, tasks, batch_size, num_workers):
    collator = MultiTaskCollator(tasks)
    train_dataset = FashionMultiTaskDataset(
        dataset,
        train_df["dataset_idx"],
        color_features,
        processor,
        label_maps,
        tasks,
        training=True,
    )

    validation_dataset = FashionMultiTaskDataset(
        dataset,
        validation_df["dataset_idx"],
        color_features,
        processor,
        label_maps,
        tasks,
    )

    test_dataset = FashionMultiTaskDataset(
        dataset,
        test_df["dataset_idx"],
        color_features,
        processor,
        label_maps,
        tasks,
    )

    loader_arguments = {
        "batch_size": batch_size,
        "num_workers": num_workers,
        "pin_memory": torch.cuda.is_available(),
        "collate_fn": collator,
    }

    return {
        "train_dataset": train_dataset,
        "validation_dataset": validation_dataset,
        "test_dataset": test_dataset,
        "train_loader": DataLoader(
            train_dataset,
            shuffle=True,
            **loader_arguments,
        ),
        "validation_loader": DataLoader(
            validation_dataset,
            shuffle=False,
            **loader_arguments,
        ),
        "test_loader": DataLoader(
            test_dataset,
            shuffle=False,
            **loader_arguments,
        ),
    }