src/utils/base_dataset.py

from abc import ABC, abstractmethod
from typing import Callable, Iterable, List, Any, Tuple, Dict, Union, Optional

import numpy as np
import torch

from torch.nn.utils.rnn import pad_sequence
from torch.utils.data import IterableDataset

from src.utils.collections import chunks, flatten

import logging


logger = logging.getLogger(__name__)


def batchify(tensors: List[torch.Tensor], padding_value: int) -> torch.Tensor:
    return pad_sequence(tensors, batch_first=True, padding_value=padding_value)


def batchify_matrices(tensors: List[torch.Tensor], padding_value: int) -> torch.Tensor:
    x = max([t.shape[0] for t in tensors])
    y = max([t.shape[1] for t in tensors])
    out_matrix = torch.zeros((len(tensors), x, y))
    out_matrix += padding_value
    for i, tensor in enumerate(tensors):
        out_matrix[i][0 : tensor.shape[0], 0 : tensor.shape[1]] = tensor
    return out_matrix


def batchify_matrices(tensors: List[torch.Tensor], padding_value: int) -> torch.Tensor:
    x = max([t.shape[0] for t in tensors])
    y = max([t.shape[1] for t in tensors])
    out_matrix = torch.zeros((len(tensors), x, y))
    out_matrix += padding_value
    for i, tensor in enumerate(tensors):
        out_matrix[i][0 : tensor.shape[0], 0 : tensor.shape[1]] = tensor
    return out_matrix


class BaseDataset(IterableDataset):
    def __init__(
        self,
        dataset_iterator_func: Optional[Callable[[], Iterable[Dict[str, Any]]]],
        tokens_per_batch: int,
        max_batch_size: Optional[int],
        main_field: str,
        fields_batchers: Optional[Dict[str, Union[None, Callable[[list], Any]]]],
        section_size: int,
        prebatch: bool,
        shuffle: bool,
        max_length: int,
    ):
        super().__init__()

        # you can subclass TokenBasedDataset in this way
        if dataset_iterator_func is not None:
            self.dataset_iterator_func = dataset_iterator_func

        self.tokens_per_batch = tokens_per_batch
        self.max_batch_size = max_batch_size
        self.main_field = main_field
        self.fields_batcher = fields_batchers
        self.section_size = section_size
        self.prebatch = prebatch
        self.shuffle = shuffle
        self.max_length = max_length

        if self.shuffle and not self.prebatch:
            logger.warning("If you set prebatch to False the shuffle parameters has no effect")

    def prebatch_elements(self, dataset_elements: list) -> list:
        if self.shuffle:
            dataset_elements = sorted(
                dataset_elements,
                key=lambda de: len(de[self.main_field]) + torch.randint(0, 10, (1,)),
            )
            dataset_elements = list(chunks(dataset_elements, 2048))
            np.random.shuffle(dataset_elements)
            dataset_elements = flatten(dataset_elements)
        else:
            dataset_elements = sorted(dataset_elements, key=lambda de: len(de[self.main_field]))

        return dataset_elements

    def materialize_batches(self, dataset_elements: List[Dict[str, Any]]) -> List[Dict[str, Any]]:

        if self.prebatch:
            dataset_elements = self.prebatch_elements(dataset_elements)

        batches = []
        current_batch = []

        # function that creates a batch from the 'current_batch' list
        def output_batch() -> Dict[str, Any]:

            batch_dict = dict()

            de_values_by_field = {fn: [de[fn] for de in current_batch if fn in de] for fn in self.fields_batcher}

            # in case you provide fields batchers but in the batch there are no elements for that field
            de_values_by_field = {fn: fvs for fn, fvs in de_values_by_field.items() if len(fvs) > 0}

            assert len(set([len(v) for v in de_values_by_field.values()]))

            de_values_by_field = {
                fn: fvs for fn, fvs in de_values_by_field.items() if all([fv is not None for fv in fvs])
            }

            for field_name, field_values in de_values_by_field.items():
                field_batch = (
                    self.fields_batcher[field_name](field_values)
                    if self.fields_batcher[field_name] is not None
                    else field_values
                )

                batch_dict[field_name] = field_batch

            return batch_dict

        for de in dataset_elements:

            if self.max_batch_size is not None and len(current_batch) == self.max_batch_size:
                batches.append(output_batch())
                current_batch = []

            de_main_len = len(de[self.main_field])

            # some callback to filter out samples for example

            if de_main_len > self.max_length:
                logger.warning(f"Discarding element: max length exceeded ({de_main_len} > {self.max_length})")
                continue

            if de_main_len > self.tokens_per_batch:
                logger.warning(
                    f'Discarding element: length greater than "tokens per batch"'
                    f" ({de_main_len} > {self.tokens_per_batch})"
                )
                continue

            future_max_len = max(
                de_main_len,
                max([len(bde[self.main_field]) for bde in current_batch], default=0),
            )

            future_tokens_per_batch = future_max_len * (len(current_batch) + 1)

            if future_tokens_per_batch >= self.tokens_per_batch:
                batches.append(output_batch())
                current_batch = []

            current_batch.append(de)

        if len(current_batch) != 0:
            batches.append(output_batch())

        return batches

    def __iter__(self):

        current_dataset_elements = []

        for i, dataset_elem in enumerate(self.dataset_iterator_func()):

            if len(current_dataset_elements) == self.section_size:
                for batch in self.materialize_batches(current_dataset_elements):
                    yield batch
                current_dataset_elements = []

            current_dataset_elements.append(dataset_elem)

            if i % 10_000 == 0:
                logger.info(f"Processed: {i} number of elements")

        if len(current_dataset_elements) != 0:
            for batch in self.materialize_batches(current_dataset_elements):
                yield batch