data.py

"""Data provider"""

import torch
import torch.utils.data as data
import os
import nltk
import numpy as np


class PrecompDataset(data.Dataset):
    """
    Load precomputed captions and image features
    Possible options: f30k_precomp, coco_precomp
    """

    def __init__(self, data_path, data_split, vocab, opt):
        self.vocab = vocab
        loc = data_path + '/'

        # Captions
        self.captions = []
        with open(loc+'%s_caps.txt' % data_split, 'r') as f:
            for line in f:
                self.captions.append(line.strip())

        # Image features
        self.loc = loc
        self.images = np.load(loc + '%s_ims.npy' % data_split)

        self.image_embs = np.load(opt.image_emb_path)
        self.text_embs = np.load(opt.text_emb_path)

        self.length = len(self.captions)
        self.im_div = 5
        # the development set for coco is large and so validation would be slow
        if data_split == 'dev':
            self.length = 5000

    def __getitem__(self, caption_id):
        # handle the image redundancy
        image_id = caption_id//self.im_div
        image = torch.Tensor(self.images[image_id])

        caption = self.captions[caption_id]
        vocab = self.vocab

        # Convert caption (string) to word ids.
        tokens = nltk.tokenize.word_tokenize(str(caption).lower())
        caption = []
        caption.append(vocab('<start>'))
        caption.extend([vocab(token) for token in tokens])
        caption.append(vocab('<end>'))
        target = torch.Tensor(caption)

        image_emb = self.image_embs[image_id]
        image_emb = torch.Tensor(image_emb)

        t_text_emb = self.text_embs[caption_id]
        t_text_emb = torch.Tensor(t_text_emb)

        start_id = 5 * image_id
        v_text_emb = self.text_embs[start_id: start_id + 5]
        v_text_emb = torch.Tensor(v_text_emb)

        return image, target, image_id, caption_id, image_emb, v_text_emb, t_text_emb

    def __len__(self):
        return self.length


def collate_fn(data):
    """Build mini-batch tensors from a list of (image, caption) tuples.
    Args:
        data: list of (image, caption) tuple.
            - image: torch tensor of shape (3, 256, 256).
            - caption: torch tensor of shape (?); variable length.

    Returns:
        images: torch tensor of shape (batch_size, 3, 256, 256).
        targets: torch tensor of shape (batch_size, padded_length).
        lengths: list; valid length for each padded caption.
    """
    # Sort a data list by caption length
    data.sort(key=lambda x: len(x[1]), reverse=True)
    images, captions, image_ids, caption_ids, image_emb, v_text_emb, t_text_emb = zip(*data)

    # Merge images (convert tuple of 3D tensor to 4D tensor)
    images = torch.stack(images, 0)

    # Merget captions (convert tuple of 1D tensor to 2D tensor)
    lengths = [len(cap) for cap in captions]
    targets = torch.zeros(len(captions), max(lengths)).long()
    for i, cap in enumerate(captions):
        end = lengths[i]
        targets[i, :end] = cap[:end]

    image_emb = torch.stack(image_emb, 0)
    v_text_emb = torch.stack(v_text_emb, 0)
    t_text_emb = torch.stack(t_text_emb, 0)

    return images, targets, lengths, image_ids, caption_ids, image_emb, v_text_emb, t_text_emb


def get_precomp_loader(data_path, data_split, vocab, opt, batch_size=100,
                       shuffle=True, num_workers=2):
    """Returns torch.utils.data.DataLoader for custom coco dataset."""
    dset = PrecompDataset(data_path, data_split, vocab, opt)

    data_loader = torch.utils.data.DataLoader(dataset=dset,
                                              batch_size=batch_size,
                                              shuffle=shuffle,
                                              pin_memory=True,
                                              collate_fn=collate_fn)
    return data_loader


def get_loaders(data_name, vocab, batch_size, workers, opt):
    dpath = os.path.join(opt.data_path, data_name)
    train_loader = get_precomp_loader(dpath, 'train', vocab, opt,
                                      batch_size, True, workers)
    val_loader = get_precomp_loader(dpath, 'dev', vocab, opt,
                                    batch_size, False, workers)
    return train_loader, val_loader


def get_test_loader(split_name, data_name, vocab, batch_size,
                    workers, opt):
    dpath = os.path.join(opt.data_path, data_name)
    test_loader = get_precomp_loader(dpath, split_name, vocab, opt,
                                     batch_size, False, workers)
    return test_loader