Repository: mila-udem/fuel
Pull request #68 (2015-04-08 05:57:59)

WIP: ILSVRC 2010 (a.k.a. ImageNet)
Putting this up so that the interested can have a look. In particular looking for some feedback from @vdumoulin, as well as @bartvm for the right place we could split this up in order to speed it up with multiple processes.

dwf: @lamblin @dmitriy-serdyuk This is ready for a preliminary review, though it requires some more tests before merge.
vdumoulin: @dwf I did a quick first pass. I'm not very familiar with ZMQ, so my review for this part of the PR will focus on annoying docstring remarks.
rodrigob: this seems to be moving well. looking forward for giving it a trial.
dwf: Unfortunately storing the data like this comes at a cost, in the sense that
you do incur the cost of the JPEG decoding. You can pipeline it away with
ServerDataStream of course, to an extent. Libjpeg-turbo could help with
this, but it's not super easy to just substitute it in to PIL. Ideally
someone would go and write a batteries included Cython wrapper around the
non-legacy API for libjpeg-turbo (much nicer/less stupid than the legacy
libjpeg API), but that should probably be a separate package.

rodrigob: From [what I read](https://github.com/python-pillow/Pillow/issues/1037) PIL will use libjpeg-turbo if it is installed in the system. So that feature is already in there.
Otherwise there is [jpeg4py](https://pypi.python.org/pypi/jpeg4py/) which also seems to implement what you are saying.
Do you know if "PIL using jpeg-turbo" is faster/slower at decoding than jpeg4py ? 
(maybe [this would help](https://github.com/ajkxyz/jpeg4py/blob/master/tests/benchmark.py))

dwf: @rodrigob jpeg4py is indeed exactly what I was thinking should exist. Thank you for pointing that out and thank you Samsung!

Patch:
diff --git a/.travis-data.sh b/.travis-data.sh
index 18a4ec9..b93cf48 100755
--- a/.travis-data.sh
+++ b/.travis-data.sh
@@ -14,7 +14,14 @@ function download {
 cd $FUEL_DATA_PATH
 
 for dataset in "$@"; do
-  download $dataset
+  if [ "$dataset" == "ilsvrc2010" ]; then
+    wget "http://www.image-net.org/challenges/LSVRC/2010/download/ILSVRC2010_devkit-1.0.tar.gz"
+    wget "http://www.image-net.org/challenges/LSVRC/2010/ILSVRC2010_test_ground_truth.txt"
+  else
+    download $dataset
+  fi
 done
 
+
+
 cd -
diff --git a/.travis.yml b/.travis.yml
index b2db4fd..b3633f6 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -38,7 +38,7 @@ install:
         python setup.py build_ext --inplace
       fi
 script:
-  - ./.travis-data.sh adult mnist binarized_mnist "caltech101_silhouettes 16" cifar10 cifar100 iris
+  - ./.travis-data.sh adult mnist binarized_mnist "caltech101_silhouettes 16" cifar10 cifar100 iris ilsvrc2010
   - function fail { export FAILED=1; }
   - |
       if [[ $TESTS == 'blocks' ]]; then
diff --git a/fuel/converters/__init__.py b/fuel/converters/__init__.py
index 4071459..41c1cd6 100644
--- a/fuel/converters/__init__.py
+++ b/fuel/converters/__init__.py
@@ -18,6 +18,7 @@ from fuel.converters import cifar100
 from fuel.converters import iris
 from fuel.converters import mnist
 from fuel.converters import svhn
+from fuel.converters import ilsvrc2010
 
 __version__ = '0.2'
 all_converters = (
@@ -28,4 +29,5 @@ all_converters = (
     ('cifar100', cifar100.fill_subparser),
     ('iris', iris.fill_subparser),
     ('mnist', mnist.fill_subparser),
-    ('svhn', svhn.fill_subparser))
+    ('svhn', svhn.fill_subparser),
+    ('ilsvrc2010', ilsvrc2010.fill_subparser))
diff --git a/fuel/converters/ilsvrc2010.py b/fuel/converters/ilsvrc2010.py
new file mode 100644
index 0000000..e524551
--- /dev/null
+++ b/fuel/converters/ilsvrc2010.py
@@ -0,0 +1,573 @@
+from __future__ import division
+from collections import OrderedDict
+from functools import partial
+import gzip
+import io
+import os
+import logging
+import os.path
+
+import h5py
+import numpy
+from picklable_itertools.extras import equizip
+from progressbar import ProgressBar
+from PIL import Image
+from scipy.io.matlab import loadmat
+from six.moves import zip, xrange
+import zmq
+
+from fuel.converters.base import check_exists
+from fuel.datasets import H5PYDataset
+from fuel.utils.formats import tar_open
+from fuel.utils.parallel import producer_consumer
+from fuel import config
+
+log = logging.getLogger(__name__)
+
+DEVKIT_ARCHIVE = 'ILSVRC2010_devkit-1.0.tar.gz'
+DEVKIT_META_PATH = 'devkit-1.0/data/meta.mat'
+DEVKIT_VALID_GROUNDTRUTH_PATH = ('devkit-1.0/data/'
+                                 'ILSVRC2010_validation_ground_truth.txt')
+PATCH_IMAGES_TAR = 'patch_images.tar'
+TEST_GROUNDTRUTH = 'ILSVRC2010_test_ground_truth.txt'
+TRAIN_IMAGES_TAR = 'ILSVRC2010_images_train.tar'
+VALID_IMAGES_TAR = 'ILSVRC2010_images_val.tar'
+TEST_IMAGES_TAR = 'ILSVRC2010_images_test.tar'
+IMAGE_TARS = (TRAIN_IMAGES_TAR, VALID_IMAGES_TAR, TEST_IMAGES_TAR,
+              PATCH_IMAGES_TAR)
+PUBLIC_FILES = TEST_GROUNDTRUTH, DEVKIT_ARCHIVE
+ALL_FILES = PUBLIC_FILES + IMAGE_TARS
+
+
+@check_exists(required_files=ALL_FILES)
+def convert_ilsvrc2010(directory, output_directory,
+                       output_filename='ilsvrc2010.hdf5',
+                       shuffle_seed=config.default_seed):
+    """Converter for data from the ILSVRC 2010 competition.
+
+    Source files for this dataset can be obtained by registering at
+    [ILSVRC2010WEB].
+
+    Parameters
+    ----------
+    input_directory : str
+        Path from which to read raw data files.
+    output_directory : str
+        Path to which to save the HDF5 file.
+    output_filename : str, optional
+        The output filename for the HDF5 file. Default: 'ilsvrc2010.hdf5'.
+    shuffle_seed : int or sequence, optional
+        Seed for a random number generator used to shuffle the order
+        of the training set on disk, so that sequential reads will not
+        be ordered by class.
+
+    .. [ILSVRC2010WEB] http://image-net.org/challenges/LSVRC/2010/index
+
+    """
+    devkit_path = os.path.join(directory, DEVKIT_ARCHIVE)
+    test_groundtruth_path = os.path.join(directory, TEST_GROUNDTRUTH)
+    train, valid, test, patch = [os.path.join(directory, fn)
+                                 for fn in IMAGE_TARS]
+    n_train, valid_groundtruth, test_groundtruth, wnid_map = \
+        prepare_metadata(devkit_path, test_groundtruth_path)
+    n_valid, n_test = len(valid_groundtruth), len(test_groundtruth)
+    output_path = os.path.join(output_directory, output_filename)
+
+    with h5py.File(output_path, 'w') as f:
+        log.info('Creating HDF5 datasets...')
+        prepare_hdf5_file(f, n_train, n_valid, n_test)
+        log.info('Processing training set...')
+        process_train_set(f, train, patch, n_train, wnid_map, shuffle_seed)
+        log.info('Processing validation set...')
+        process_other_set(f, 'valid', valid, patch, valid_groundtruth, n_train)
+        log.info('Processing test set...')
+        process_other_set(f, 'test', test, patch, test_groundtruth,
+                          n_train + n_valid)
+        log.info('Done.')
+
+    return (output_path,)
+
+
+def fill_subparser(subparser):
+    """Sets up a subparser to convert the ILSVRC2010 dataset files.
+
+    Parameters
+    ----------
+    subparser : :class:`argparse.ArgumentParser`
+        Subparser handling the `ilsvrc2010` command.
+
+    """
+    subparser.add_argument(
+        "--shuffle-seed", help="Seed to use for randomizing order of the "
+                               "training set on disk.",
+        default=config.default_seed, type=int, required=False)
+    return convert_ilsvrc2010
+
+
+def prepare_metadata(devkit_archive, test_groundtruth_path):
+    """Extract dataset metadata required for HDF5 file setup.
+
+    Parameters
+    ----------
+    devkit_archive : str or file-like object
+        The filename or file-handle for the gzipped TAR archive
+        containing the ILSVRC2010 development kit.
+    test_groundtruth_path : str or file-like object
+        The filename or file-handle for the text file containing
+        the ILSVRC2010 test set ground truth.
+
+    Returns
+    -------
+    n_train : int
+        The number of examples in the training set.
+    valid_groundtruth : ndarray, 1-dimensional
+        An ndarray containing the validation set groundtruth in terms of
+        0-based class indices.
+    test_groundtruth : ndarray, 1-dimensional
+        An ndarray containing the test groundtruth in terms of 0-based
+        class indices.
+    wnid_map : dict
+        A dictionary that maps WordNet IDs to 0-based class indices.
+
+    """
+    # Read what's necessary from the development kit.
+    synsets, cost_matrix, raw_valid_groundtruth = read_devkit(devkit_archive)
+
+    # Mapping to take WordNet IDs to our internal 0-999 encoding.
+    wnid_map = dict(zip((s.decode('utf8') for s in synsets['WNID']),
+                        xrange(1000)))
+
+    # Map the 'ILSVRC2010 ID' to our zero-based ID.
+    ilsvrc_id_to_zero_based = dict(zip(synsets['ILSVRC2010_ID'],
+                                   xrange(len(synsets))))
+
+    # Map the validation set groundtruth to 0-999 labels.
+    valid_groundtruth = [ilsvrc_id_to_zero_based[id_]
+                         for id_ in raw_valid_groundtruth]
+
+    # Raw test data groundtruth, ILSVRC2010 IDs.
+    raw_test_groundtruth = numpy.loadtxt(test_groundtruth_path,
+                                         dtype=numpy.int16)
+
+    # Map the test set groundtruth to 0-999 labels.
+    test_groundtruth = [ilsvrc_id_to_zero_based[id_]
+                        for id_ in raw_test_groundtruth]
+
+    # Ascertain the number of filenames to prepare appropriate sized
+    # arrays.
+    n_train = int(synsets['num_train_images'].sum())
+    log.info('Training set: {} images'.format(n_train))
+    log.info('Validation set: {} images'.format(len(valid_groundtruth)))
+    log.info('Test set: {} images'.format(len(test_groundtruth)))
+    n_total = n_train + len(valid_groundtruth) + len(test_groundtruth)
+    log.info('Total (train/valid/test): {} images'.format(n_total))
+    return n_train, valid_groundtruth, test_groundtruth, wnid_map
+
+
+def create_splits(n_train, n_valid, n_test):
+    n_total = n_train + n_valid + n_test
+    tuples = {}
+    tuples['train'] = (0, n_train)
+    tuples['valid'] = (n_train, n_train + n_valid)
+    tuples['test'] = (n_train + n_valid, n_total)
+    sources = ['encoded_images', 'targets', 'filenames']
+    return OrderedDict(
+        (split, OrderedDict((source, tuples[split]) for source in sources))
+        for split in ('train', 'valid', 'test')
+    )
+
+
+def prepare_hdf5_file(hdf5_file, n_train, n_valid, n_test):
+    """Create datasets within a given HDF5 file.
+
+    Parameters
+    ----------
+    hdf5_file : :class:`h5py.File` instance
+        HDF5 file handle to which to write.
+    n_train : int
+        The number of training set examples.
+    n_valid : int
+        The number of validation set examples.
+    n_test : int
+        The number of test set examples.
+
+    """
+    n_total = n_train + n_valid + n_test
+    splits = create_splits(n_train, n_valid, n_test)
+    hdf5_file.attrs['split'] = H5PYDataset.create_split_array(splits)
+    vlen_dtype = h5py.special_dtype(vlen=numpy.dtype('uint8'))
+    hdf5_file.create_dataset('encoded_images', shape=(n_total,),
+                             dtype=vlen_dtype)
+    hdf5_file.create_dataset('targets', shape=(n_total, 1), dtype=numpy.int16)
+    hdf5_file.create_dataset('filenames', shape=(n_total, 1), dtype='S32')
+
+
+def process_train_set(hdf5_file, train_archive, patch_archive, n_train,
+                      wnid_map, shuffle_seed=None):
+    """Process the ILSVRC2010 training set.
+
+    Parameters
+    ----------
+    hdf5_file : :class:`h5py.File` instance
+        HDF5 file handle to which to write. Assumes `features`, `targets`
+        and `filenames` already exist and have first dimension larger than
+        `n_train`.
+    train_archive :  str or file-like object
+        Filename or file handle for the TAR archive of training images.
+    patch_archive :  str or file-like object
+        Filename or file handle for the TAR archive of patch images.
+    n_train : int
+        The number of items in the training set.
+    wnid_map : dict
+        A dictionary mapping WordNet IDs to class indices.
+    shuffle_seed : int or sequence, optional
+        Seed for a NumPy random number generator that permutes the
+        training set on disk. If `None`, no permutation is performed
+        (this is the default).
+
+    """
+    producer = partial(train_set_producer, train_archive=train_archive,
+                       patch_archive=patch_archive, wnid_map=wnid_map)
+    consumer = partial(image_consumer, hdf5_file=hdf5_file,
+                       num_expected=n_train, shuffle_seed=shuffle_seed)
+    producer_consumer(producer, consumer)
+
+
+def _write_to_hdf5(hdf5_file, index, image_filename, image_data,
+                   class_index):
+    hdf5_file['filenames'][index] = image_filename.encode('ascii')
+    hdf5_file['encoded_images'][index] = image_data
+    hdf5_file['targets'][index] = class_index
+
+
+def train_set_producer(socket, train_archive, patch_archive, wnid_map):
+    """Load/send images from the training set TAR file or patch images.
+
+    Parameters
+    ----------
+    socket : :class:`zmq.Socket`
+        PUSH socket on which to send loaded images.
+    train_archive :  str or file-like object
+        Filename or file handle for the TAR archive of training images.
+    patch_archive :  str or file-like object
+        Filename or file handle for the TAR archive of patch images.
+    wnid_map : dict
+        A dictionary that maps WordNet IDs to 0-based class indices.
+        Used to decode the filenames of the inner TAR files.
+
+    """
+    patch_images = extract_patch_images(patch_archive, 'train')
+    num_patched = 0
+    with tar_open(train_archive) as tar:
+        for inner_tar_info in tar:
+            with tar_open(tar.extractfile(inner_tar_info.name)) as inner:
+                wnid = inner_tar_info.name.split('.')[0]
+                class_index = wnid_map[wnid]
+                filenames = sorted(info.name for info in inner
+                                   if info.isfile())
+                images_gen = (load_from_tar_or_patch(inner, filename,
+                                                     patch_images)
+                              for filename in filenames)
+                pathless_filenames = (os.path.split(fn)[-1]
+                                      for fn in filenames)
+                stream = equizip(pathless_filenames, images_gen)
+                for image_fn, (image_data, patched) in stream:
+                    if patched:
+                        num_patched += 1
+                    socket.send_pyobj((image_fn, class_index), zmq.SNDMORE)
+                    socket.send(image_data)
+    if num_patched != len(patch_images):
+        raise ValueError('not all patch images were used')
+
+
+def image_consumer(socket, hdf5_file, num_expected, shuffle_seed=None,
+                   offset=0):
+    """Fill an HDF5 file with incoming images from a socket.
+
+    Parameters
+    ----------
+    socket : :class:`zmq.Socket`
+        PULL socket on which to receive images.
+    hdf5_file : :class:`h5py.File` instance
+        HDF5 file handle to which to write. Assumes `features`, `targets`
+        and `filenames` already exist and have first dimension larger than
+        `sum(images_per_class)`.
+    num_expected : int
+        The number of items we expect to be sent over the socket.
+    shuffle_seed : int or sequence, optional
+        Seed for a NumPy random number generator that permutes the
+        images on disk.
+    offset : int, optional
+        The offset in the HDF5 datasets at which to start writing
+        received examples. Defaults to 0.
+
+    """
+    with ProgressBar(maxval=num_expected) as pb:
+        if shuffle_seed is None:
+            index_gen = iter(xrange(num_expected))
+        else:
+            rng = numpy.random.RandomState(shuffle_seed)
+            index_gen = iter(rng.permutation(num_expected))
+        for i, num in enumerate(index_gen):
+            image_filename, class_index = socket.recv_pyobj(zmq.SNDMORE)
+            image_data = numpy.fromstring(socket.recv(), dtype='uint8')
+            _write_to_hdf5(hdf5_file, num + offset, image_filename,
+                           image_data, class_index)
+            pb.update(i + 1)
+
+
+def process_other_set(hdf5_file, which_set, image_archive, patch_archive,
+                      groundtruth, offset):
+    """Process the validation or test set.
+
+    Parameters
+    ----------
+    hdf5_file : :class:`h5py.File` instance
+        HDF5 file handle to which to write. Assumes `features`, `targets`
+        and `filenames` already exist and have first dimension larger than
+        `sum(images_per_class)`.
+    which_set : str
+        Which set of images is being processed. One of 'train', 'valid',
+        'test'.  Used for extracting the appropriate images from the patch
+        archive.
+    image_archive : str or file-like object
+        The filename or file-handle for the TAR archive containing images.
+    patch_archive : str or file-like object
+        Filename or file handle for the TAR archive of patch images.
+    groundtruth : iterable
+        Iterable container containing scalar 0-based class index for each
+        image, sorted by filename.
+    offset : int
+        The offset in the HDF5 datasets at which to start writing.
+
+    """
+    producer = partial(other_set_producer, image_archive=image_archive,
+                       patch_archive=patch_archive,
+                       groundtruth=groundtruth, which_set=which_set)
+    consumer = partial(image_consumer, hdf5_file=hdf5_file,
+                       num_expected=len(groundtruth), offset=offset)
+    producer_consumer(producer, consumer)
+
+
+def other_set_producer(socket, which_set, image_archive, patch_archive,
+                       groundtruth):
+    """Push image files read from the valid/test set TAR to a socket.
+
+    Parameters
+    ----------
+    socket : :class:`zmq.Socket`
+        PUSH socket on which to send images.
+    which_set : str
+        Which set of images is being processed. One of 'train', 'valid',
+        'test'.  Used for extracting the appropriate images from the patch
+        archive.
+    image_archive : str or file-like object
+        The filename or file-handle for the TAR archive containing images.
+    patch_archive : str or file-like object
+        Filename or file handle for the TAR archive of patch images.
+    groundtruth : iterable
+        Iterable container containing scalar 0-based class index for each
+        image, sorted by filename.
+
+    """
+    patch_images = extract_patch_images(patch_archive, which_set)
+    num_patched = 0
+    with tar_open(image_archive) as tar:
+        filenames = sorted(info.name for info in tar if info.isfile())
+        images = (load_from_tar_or_patch(tar, filename, patch_images)
+                  for filename in filenames)
+        pathless_filenames = (os.path.split(fn)[-1] for fn in filenames)
+        image_iterator = equizip(images, pathless_filenames, groundtruth)
+        for (image_data, patched), filename, class_index in image_iterator:
+            if patched:
+                num_patched += 1
+            socket.send_pyobj((filename, class_index), zmq.SNDMORE)
+            socket.send(image_data, copy=False)
+    if num_patched != len(patch_images):
+        raise Exception
+
+
+def load_from_tar_or_patch(tar, image_filename, patch_images):
+    """Do everything necessary to process an image inside a TAR.
+
+    Parameters
+    ----------
+    tar : `TarFile` instance
+        The tar from which to read `image_filename`.
+    image_filename : str
+        Fully-qualified path inside of `tar` from which to read an
+        image file.
+    patch_images : dict
+        A dictionary containing filenames (without path) of replacements
+        to be substituted in place of the version of the same file found
+        in `tar`.
+
+    Returns
+    -------
+    image_data : bytes
+        The JPEG bytes representing either the image from the TAR archive
+        or its replacement from the patch dictionary.
+    patched : bool
+        True if the image was retrieved from the patch dictionary. False
+        if it was retrieved from the TAR file.
+
+    """
+    patched = True
+    image_bytes = patch_images.get(os.path.basename(image_filename), None)
+    if image_bytes is None:
+        patched = False
+        try:
+            image_bytes = tar.extractfile(image_filename).read()
+            numpy.array(Image.open(io.BytesIO(image_bytes)))
+        except (IOError, OSError):
+            with gzip.GzipFile(fileobj=tar.extractfile(image_filename)) as gz:
+                image_bytes = gz.read()
+                numpy.array(Image.open(io.BytesIO(image_bytes)))
+    return image_bytes, patched
+
+
+def read_devkit(f):
+    """Read relevant information from the development kit archive.
+
+    Parameters
+    ----------
+    f : str or file-like object
+        The filename or file-handle for the gzipped TAR archive
+        containing the ILSVRC2010 development kit.
+
+    Returns
+    -------
+    synsets : ndarray, 1-dimensional, compound dtype
+        See :func:`read_metadata_mat_file` for details.
+    cost_matrix : ndarray, 2-dimensional, uint8
+        See :func:`read_metadata_mat_file` for details.
+    raw_valid_groundtruth : ndarray, 1-dimensional, int16
+        The labels for the ILSVRC2010 validation set,
+        distributed with the development kit code.
+
+    """
+    with tar_open(f) as tar:
+        # Metadata table containing class hierarchy, textual descriptions, etc.
+        meta_mat = tar.extractfile(DEVKIT_META_PATH)
+        synsets, cost_matrix = read_metadata_mat_file(meta_mat)
+
+        # Raw validation data groundtruth, ILSVRC2010 IDs. Confusingly
+        # distributed inside the development kit archive.
+        raw_valid_groundtruth = numpy.loadtxt(tar.extractfile(
+            DEVKIT_VALID_GROUNDTRUTH_PATH), dtype=numpy.int16)
+    return synsets, cost_matrix, raw_valid_groundtruth
+
+
+def read_metadata_mat_file(meta_mat):
+    """Read ILSVRC2010 metadata from the distributed MAT file.
+
+    Parameters
+    ----------
+    meta_mat : str or file-like object
+        The filename or file-handle for `meta.mat` from the
+        ILSVRC2010 development kit.
+
+    Returns
+    -------
+    synsets : ndarray, 1-dimensional, compound dtype
+        A table containing ILSVRC2010 metadata for the "synonym sets"
+        or "synsets" that comprise the classes and superclasses,
+        including the following fields:
+         * `ILSVRC2010_ID`: the integer ID used in the original
+           competition data.
+         * `WNID`: A string identifier that uniquely identifies
+           a synset in ImageNet and WordNet.
+         * `wordnet_height`: The length of the longest path to
+           a leaf node in the FULL ImageNet/WordNet hierarchy
+           (leaf nodes in the FULL ImageNet/WordNet hierarchy
+           have `wordnet_height` 0).
+         * `gloss`: A string representation of an English
+           textual description of the concept represented by
+           this synset.
+         * `num_children`: The number of children in the hierarchy
+           for this synset.
+         * `words`: A string representation, comma separated,
+           of different synoym words or phrases for the concept
+           represented by this synset.
+         * `children`: A vector of `ILSVRC2010_ID`s of children
+           of this synset, padded with -1. Note that these refer
+           to `ILSVRC2010_ID`s from the original data and *not*
+           the zero-based index in the table.
+         * `num_train_images`: The number of training images for
+           this synset.
+    cost_matrix : ndarray, 2-dimensional, uint8
+        A 1000x1000 matrix containing the precomputed pairwise
+        cost (based on distance in the hierarchy) for all
+        low-level synsets (i.e. the thousand possible output
+        classes with training data associated).
+
+    """
+    mat = loadmat(meta_mat, squeeze_me=True)
+    synsets = mat['synsets']
+    cost_matrix = mat['cost_matrix']
+    new_dtype = numpy.dtype([
+        ('ILSVRC2010_ID', numpy.int16),
+        ('WNID', ('S', max(map(len, synsets['WNID'])))),
+        ('wordnet_height', numpy.int8),
+        ('gloss', ('S', max(map(len, synsets['gloss'])))),
+        ('num_children', numpy.int8),
+        ('words', ('S', max(map(len, synsets['words'])))),
+        ('children', (numpy.int8, max(synsets['num_children']))),
+        ('num_train_images', numpy.uint16)
+    ])
+    new_synsets = numpy.empty(synsets.shape, dtype=new_dtype)
+    for attr in ['ILSVRC2010_ID', 'WNID', 'wordnet_height', 'gloss',
+                 'num_children', 'words', 'num_train_images']:
+        new_synsets[attr] = synsets[attr]
+    children = [numpy.atleast_1d(ch) for ch in synsets['children']]
+    padded_children = [
+        numpy.concatenate((c,
+                           -numpy.ones(new_dtype['children'].shape[0] - len(c),
+                                       dtype=numpy.int16)))
+        for c in children
+    ]
+    new_synsets['children'] = padded_children
+    return new_synsets, cost_matrix
+
+
+def extract_patch_images(f, which_set):
+    """Extracts a dict of the "patch images" for ILSVRC2010.
+
+    Parameters
+    ----------
+    f : str or file-like object
+        The filename or file-handle to the patch images TAR file.
+    which_set : str
+        Which set of images to extract. One of 'train', 'valid', 'test'.
+
+    Returns
+    -------
+    dict
+        A dictionary contains a mapping of filenames (without path) to a
+        bytes object containing the replacement image.
+
+    Notes
+    -----
+    Certain images in the distributed archives are blank, or display
+    an "image not available" banner. A separate TAR file of
+    "patch images" is distributed with the corrected versions of
+    these. It is this archive that this function is intended to read.
+
+    """
+    if which_set not in ('train', 'valid', 'test'):
+        raise ValueError('which_set must be one of train, valid, or test')
+    which_set = 'val' if which_set == 'valid' else which_set
+    patch_images = {}
+    with tar_open(f) as tar:
+        for info_obj in tar:
+            if not info_obj.name.endswith('.JPEG'):
+                continue
+            # Pretty sure that '/' is used for tarfile regardless of
+            # os.path.sep, but I officially don't care about Windows.
+            tokens = info_obj.name.split('/')
+            file_which_set = tokens[-2]
+            if file_which_set != which_set:
+                continue
+            filename = tokens[-1]
+            patch_images[filename] = tar.extractfile(info_obj.name).read()
+    return patch_images
diff --git a/fuel/datasets/imagenet.py b/fuel/datasets/imagenet.py
new file mode 100644
index 0000000..2cd85df
--- /dev/null
+++ b/fuel/datasets/imagenet.py
@@ -0,0 +1,41 @@
+# -*- coding: utf-8 -*-
+from fuel.datasets import H5PYDataset
+from fuel.transformers.defaults import rgb_images_from_encoded_bytes
+from fuel.utils import find_in_data_path
+
+
+class ILSVRC2010(H5PYDataset):
+    u"""The ILSVRC2010 Dataset.
+
+    The ImageNet Large-Scale Visual Recognition Challenge [ILSVRC]
+    is an annual computer vision competition testing object classification
+    and detection at large-scale. This is a wrapper around the data for
+    the 2010 competition, which is (as of 2015) the only year for which
+    test data groundtruth is available.
+
+    Note that the download site for the images is not publicly
+    accessible. To downlaod the images, you may sign up for an account
+    at [SIGNUP].
+
+    .. [ILSVRC] Olga Russakovsky, Jia Deng, Hao Su, Jonathan Krause,
+       Sanjeev Satheesh, Sean Ma, Zhiheng Huang, Andrej Karpathy, Aditya
+       Khosla, Michael Bernstein, Alexander C. Berg and Li Fei-Fei.
+       *ImageNet Large Scale Visual Recognition Challenge*. IJCV, 2015.
+
+    .. [SIGNUP] http://www.image-net.org/signup
+
+    Parameters
+    ----------
+    which_sets : tuple of str
+        Which split to load. Valid values are 'train' (1.2M examples)
+        'valid' (150,000 examples), and 'test' (50,000 examples).
+
+    """
+    filename = 'ilsvrc2010.hdf5'
+    default_transformers = rgb_images_from_encoded_bytes(('encoded_images',))
+
+    def __init__(self, which_sets, **kwargs):
+        kwargs.setdefault('load_in_memory', False)
+        super(ILSVRC2010, self).__init__(
+            file_or_path=find_in_data_path(self.filename),
+            which_sets=which_sets, **kwargs)
diff --git a/fuel/downloaders/__init__.py b/fuel/downloaders/__init__.py
index 259192a..e04b27d 100644
--- a/fuel/downloaders/__init__.py
+++ b/fuel/downloaders/__init__.py
@@ -14,6 +14,7 @@ from fuel.downloaders import cifar100
 from fuel.downloaders import iris
 from fuel.downloaders import mnist
 from fuel.downloaders import svhn
+from fuel.downloaders import ilsvrc2010
 
 all_downloaders = (
     ('adult', adult.fill_subparser),
@@ -23,4 +24,5 @@ all_downloaders = (
     ('cifar100', cifar100.fill_subparser),
     ('iris', iris.fill_subparser),
     ('mnist', mnist.fill_subparser),
-    ('svhn', svhn.fill_subparser))
+    ('svhn', svhn.fill_subparser),
+    ('ilsvrc2010', ilsvrc2010.fill_subparser))
diff --git a/fuel/downloaders/ilsvrc2010.py b/fuel/downloaders/ilsvrc2010.py
new file mode 100644
index 0000000..3c25a8e
--- /dev/null
+++ b/fuel/downloaders/ilsvrc2010.py
@@ -0,0 +1,37 @@
+from fuel.converters.ilsvrc2010 import IMAGE_TARS
+from fuel.downloaders.base import default_downloader
+
+
+def fill_subparser(subparser):
+    """Sets up a subparser to download the ILSVRC2010 dataset files.
+
+    Note that you will need to use `--url-prefix` to download the
+    non-public files (namely, the TARs of images). This is a single
+    prefix that is common to all distributed files, which you can
+    obtain by registering at the ImageNet website [DOWNLOAD].
+
+    Note that these files are quite large and you may be better off
+    simply downloading them separately and running ``fuel-convert``.
+
+    .. [DOWNLOAD] http://www.image-net.org/download-images
+
+
+    Parameters
+    ----------
+    subparser : :class:`argparse.ArgumentParser`
+        Subparser handling the `ilsvrc2010` command.
+
+    """
+    urls = [
+        ('http://www.image-net.org/challenges/LSVRC/2010/'
+         'ILSVRC2010_test_ground_truth.txt'),
+        ('http://www.image-net.org/challenges/LSVRC/2010/'
+         'download/ILSVRC2010_devkit-1.0.tar.gz'),
+    ] + ([None] * len(IMAGE_TARS))
+    filenames = [None, None] + list(IMAGE_TARS)
+    subparser.set_defaults(urls=urls, filenames=filenames)
+    subparser.add_argument('-P', '--url-prefix', type=str, default=None,
+                           help="URL prefix to prepend to the filenames of "
+                                "non-public files, in order to download them. "
+                                "Be sure to include the trailing slash.")
+    return default_downloader
diff --git a/fuel/transformers/defaults.py b/fuel/transformers/defaults.py
index 767d1af..95d2ed5 100644
--- a/fuel/transformers/defaults.py
+++ b/fuel/transformers/defaults.py
@@ -1,8 +1,42 @@
 """Commonly-used default transformers."""
-from fuel.transformers import ScaleAndShift, Cast
+from fuel.transformers import ScaleAndShift, Cast, SourcewiseTransformer
+from fuel.transformers.image import ImagesFromBytes
 
 
 def uint8_pixels_to_floatX(which_sources):
     return (
         (ScaleAndShift, [1 / 255.0, 0], {'which_sources': which_sources}),
         (Cast, ['floatX'], {'which_sources': which_sources}))
+
+
+class ToBytes(SourcewiseTransformer):
+    """Transform a stream of ndarray examples to bytes.
+
+    Notes
+    -----
+    Used for retrieving variable-length byte data stored as, e.g. a uint8
+    ragged array.
+
+    """
+    def __init__(self, stream, **kwargs):
+        kwargs.setdefault('produces_examples', stream.produces_examples)
+        axis_labels = (stream.axis_labels
+                       if stream.axis_labels is not None
+                       else {})
+        for source in kwargs.get('which_sources', stream.sources):
+            axis_labels[source] = (('batch', 'bytes')
+                                   if 'batch' in axis_labels.get(source, ())
+                                   else ('bytes',))
+        kwargs.setdefault('axis_labels', axis_labels)
+        super(ToBytes, self).__init__(stream, **kwargs)
+
+    def transform_source_example(self, example, _):
+        return example.tostring()
+
+    def transform_source_batch(self, batch, _):
+        return [example.tostring() for example in batch]
+
+
+def rgb_images_from_encoded_bytes(which_sources):
+    return ((ToBytes, [], {'which_sources': ('encoded_images',)}),
+            (ImagesFromBytes, [], {'which_sources': ('encoded_images',)}))
diff --git a/fuel/utils.py b/fuel/utils/__init__.py
similarity index 100%
rename from fuel/utils.py
rename to fuel/utils/__init__.py
diff --git a/fuel/utils/formats.py b/fuel/utils/formats.py
new file mode 100644
index 0000000..3c0dc29
--- /dev/null
+++ b/fuel/utils/formats.py
@@ -0,0 +1,23 @@
+"""Low-level utilities for reading a variety of source formats."""
+import tarfile
+import six
+
+
+def tar_open(f):
+    """Open either a filename or a file-like object as a TarFile.
+
+    Parameters
+    ----------
+    f : str or file-like object
+        The filename or file-like object from which to read.
+
+    Returns
+    -------
+    TarFile
+        A `TarFile` instance.
+
+    """
+    if isinstance(f, six.string_types):
+        return tarfile.open(name=f)
+    else:
+        return tarfile.open(fileobj=f)
diff --git a/fuel/utils/parallel.py b/fuel/utils/parallel.py
new file mode 100644
index 0000000..aedea73
--- /dev/null
+++ b/fuel/utils/parallel.py
@@ -0,0 +1,113 @@
+"""Utilities for speeding things up through parallelism.
+
+Currently including:
+
+* A very simple PUSH-PULL reusable producer-consumer pattern
+  using a ZeroMQ socket instead of the (slow, unnecessarily
+  copying) multiprocessing.Queue. See :func:`producer_consumer`.
+
+"""
+from multiprocessing import Process
+import zmq
+
+
+def _producer_wrapper(f, port, addr='tcp://127.0.0.1'):
+    """A shim that sets up a socket and starts the producer callable.
+
+    Parameters
+    ----------
+    f : callable
+        Callable that takes a single argument, a handle
+        for a ZeroMQ PUSH socket. Must be picklable.
+    port : int
+        The port on which the socket should connect.
+    addr : str, optional
+        Address to which the socket should connect. Defaults
+        to localhost ('tcp://127.0.0.1').
+
+    """
+    try:
+        context = zmq.Context()
+        socket = context.socket(zmq.PUSH)
+        socket.connect(':'.join([addr, str(port)]))
+        f(socket)
+    finally:
+        # Works around a Python 3.x bug.
+        context.destroy()
+
+
+def _spawn_producer(f, port, addr='tcp://127.0.0.1'):
+    """Start a process that sends results on a PUSH socket.
+
+    Parameters
+    ----------
+    f : callable
+        Callable that takes a single argument, a handle
+        for a ZeroMQ PUSH socket. Must be picklable.
+
+    Returns
+    -------
+    process : multiprocessing.Process
+        The process handle of the created producer process.
+
+    """
+    process = Process(target=_producer_wrapper, args=(f, port, addr))
+    process.start()
+    return process
+
+
+def producer_consumer(producer, consumer, addr='tcp://127.0.0.1',
+                      port=None, context=None):
+    """A producer-consumer pattern.
+
+    Parameters
+    ----------
+    producer : callable
+        Callable that takes a single argument, a handle
+        for a ZeroMQ PUSH socket. Must be picklable.
+    consumer : callable
+        Callable that takes a single argument, a handle
+        for a ZeroMQ PULL socket.
+    addr : str, optional
+        Address to which the socket should connect. Defaults
+        to localhost ('tcp://127.0.0.1').
+    port : int, optional
+        The port on which the consumer should listen.
+    context : zmq.Context, optional
+        The ZeroMQ Context to use. One will be created otherwise.
+
+    Returns
+    -------
+    result
+        Passes along whatever `consumer` returns.
+
+    Notes
+    -----
+    This sets up a PULL socket in the calling process and forks
+    a process that calls `producer` on a PUSH socket. When the
+    consumer returns, the producer process is terminated.
+
+    Wrap `consumer` or `producer` in a `functools.partial` object
+    in order to send additional arguments; the callables passed in
+    should expect only one required, positional argument, the socket
+    handle.
+
+    """
+    context_created = False
+    if context is None:
+        context_created = True
+        context = zmq.Context()
+    try:
+        consumer_socket = context.socket(zmq.PULL)
+        if port is None:
+            port = consumer_socket.bind_to_random_port(addr)
+        try:
+            process = _spawn_producer(producer, port)
+            result = consumer(consumer_socket)
+        finally:
+            process.terminate()
+        return result
+    finally:
+        # Works around a Python 3.x bug.
+        if context_created:
+            context.destroy()


Test patch:
diff --git a/tests/converters/__init__.py b/tests/converters/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/converters/test_convert_ilsvrc2010.py b/tests/converters/test_convert_ilsvrc2010.py
new file mode 100644
index 0000000..11bf4e7
--- /dev/null
+++ b/tests/converters/test_convert_ilsvrc2010.py
@@ -0,0 +1,670 @@
+from collections import deque
+import hashlib
+import io
+import os
+import tarfile
+from tempfile import NamedTemporaryFile
+import gzip
+
+import numpy
+from numpy.testing import assert_equal
+
+from PIL import Image
+import six
+from six.moves import xrange
+
+import zmq
+
+# from fuel.server import recv_arrays, send_arrays
+from fuel.converters.ilsvrc2010 import (extract_patch_images,
+                                        image_consumer,
+                                        load_from_tar_or_patch,
+                                        other_set_producer,
+                                        prepare_hdf5_file,
+                                        prepare_metadata,
+                                        process_train_set,
+                                        process_other_set,
+                                        read_devkit,
+                                        read_metadata_mat_file,
+                                        train_set_producer,
+                                        DEVKIT_META_PATH,
+                                        DEVKIT_ARCHIVE,
+                                        TEST_GROUNDTRUTH)
+from fuel.utils import find_in_data_path
+from tests import skip_if_not_available
+
+
+class MockSocket(object):
+    """Mock of a ZeroMQ PUSH or PULL socket."""
+    def __init__(self, socket_type, to_recv=()):
+        self.socket_type = socket_type
+        if self.socket_type not in (zmq.PUSH, zmq.PULL):
+            raise NotImplementedError('only PUSH and PULL currently supported')
+        self.sent = deque()
+        self.to_recv = deque(to_recv)
+
+    def send(self, data, flags=0, copy=True, track=False):
+        assert self.socket_type == zmq.PUSH
+        if track:
+            # We don't emulate the behaviour required by this flag.
+            raise NotImplementedError
+        message = {'type': 'send', 'data': data, 'flags': flags, 'copy': copy}
+        self.sent.append(message)
+
+    def send_pyobj(self, obj, flags=0, protocol=2):
+        assert self.socket_type == zmq.PUSH
+        message = {'type': 'send_pyobj', 'obj': obj, 'flags': flags,
+                   'protocol': protocol}
+        self.sent.append(message)
+
+    def recv(self, flags=0, copy=True, track=False):
+        if track:
+            # We don't emulate the behaviour required by this flag.
+            raise NotImplementedError
+        message = self.to_recv.popleft()
+        assert message['type'] == 'recv'
+        if 'flags' in message:
+            assert message['flags'] == flags, 'flags did not match expected'
+        if 'copy' in message:
+            assert message['copy'] == copy, 'copy did not match expected'
+        return message['data']
+
+    def recv_pyobj(self, flags=0):
+        message = self.to_recv.popleft()
+        assert message['type'] == 'recv_pyobj'
+        if 'flags' in message:
+            assert flags == message['flags']
+        return message['obj']
+
+
+class MockH5PYData(object):
+    def __init__(self, shape, dtype):
+        self.data = numpy.empty(shape, dtype)
+        self.dims = MockH5PYDims(len(shape))
+
+    def __setitem__(self, where, what):
+        self.data[where] = what
+
+    def __getitem__(self, index):
+        return self.data[index]
+
+    @property
+    def shape(self):
+        return self.data.shape
+
+    @property
+    def dtype(self):
+        return self.data.dtype
+
+
+class MockH5PYFile(dict):
+    filename = 'NOT_A_REAL_FILE.hdf5'
+
+    def __init__(self):
+        self.attrs = {}
+        self.flushed = 0
+        self.opened = False
+        self.closed = False
+
+    def create_dataset(self, name, shape, dtype):
+        self[name] = MockH5PYData(shape, dtype)
+
+    def flush(self):
+        self.flushed += 1
+
+    def __enter__(self):
+        self.opened = True
+
+    def __exit__(self, type, value, traceback):
+        self.closed = True
+
+
+class MockH5PYDim(object):
+    def __init__(self, dims):
+        self.dims = dims
+        self.scales = []
+
+    def attach_scale(self, dataset):
+        # I think this is necessary for it to be valid?
+        assert dataset in self.dims.scales.values()
+        self.scales.append(dataset)
+
+
+class MockH5PYDims(object):
+    def __init__(self, ndim):
+        self._dims = [MockH5PYDim(self) for _ in xrange(ndim)]
+        self.scales = {}
+
+    def create_scale(self, dataset, name):
+        self.scales[name] = dataset
+
+    def __getitem__(self, index):
+        return self._dims[index]
+
+
+def create_jpeg_data(image):
+    """Create a JPEG in memory.
+
+    Parameters
+    ----------
+    image : ndarray, 3-dimensional
+        Array data representing the image to save. Mode ('L', 'RGB',
+        'CMYK') will be determined from the last (third) axis.
+
+    Returns
+    -------
+    jpeg_data : bytes
+        The image encoded as a JPEG, returned as raw bytes.
+
+    """
+    if image.shape[-1] == 1:
+        mode = 'L'
+    elif image.shape[-1] == 3:
+        mode = 'RGB'
+    elif image.shape[-1] == 4:
+        mode = 'CMYK'
+    else:
+        raise ValueError("invalid shape")
+    pil_image = Image.frombytes(mode=mode, size=image.shape[:2],
+                                data=image.tobytes())
+    jpeg_data = io.BytesIO()
+    pil_image.save(jpeg_data, format='JPEG')
+    return jpeg_data.getvalue()
+
+
+def create_fake_jpeg_tar(seed, min_num_images=5, max_num_images=50,
+                         min_size=20, size_range=30, filenames=None,
+                         random=True, gzip_probability=0.5, offset=0):
+    """Create a TAR file of randomly generated JPEG files.
+
+    Parameters
+    ----------
+    seed : int or sequence
+        Seed for a `numpy.random.RandomState`.
+    min_num_images : int, optional
+        The minimum number of images to put in the TAR file.
+    max_num_images : int, optional
+        The maximum number of images to put in the TAR file.
+    min_size : int, optional
+        The minimum width and minimum height of each image.
+    size_range : int, optional
+        Maximum number of pixels added to `min_size` for image
+        dimensions.
+    filenames : list, optional
+        If provided, use these filenames. Otherwise generate them
+        randomly. Must be at least `max_num_images` long.
+    random : bool, optional
+        If `False`, substitute an image full of a single number,
+        the order of that image in processing.
+    gzip_probability : float, optional
+        With this probability, randomly gzip the JPEG file without
+        appending a gzip suffix.
+    offset : int, optional
+        Where to start the hashes for filenames. Default: 0.
+
+    Returns
+    -------
+    tar_data : bytes
+        A TAR file represented as raw bytes, containing between
+        `min_num_images` and `max_num_images` JPEG files (inclusive).
+
+    Notes
+    -----
+    Randomly choose between RGB, L and CMYK mode images. Also randomly
+    gzips JPEGs to simulate the idiotic distribution format of
+    ILSVRC2010.
+
+    """
+    rng = numpy.random.RandomState(seed)
+    images = []
+    if filenames is None:
+        files = []
+    else:
+        if len(filenames) < max_num_images:
+            raise ValueError('need at least max_num_images = %d filenames' %
+                             max_num_images)
+        files = filenames
+    for i in xrange(rng.random_integers(min_num_images, max_num_images)):
+        if filenames is None:
+            max_len = 27  # so that with suffix, 32 characters
+            files.append('%s.JPEG' %
+                         hashlib.sha1(bytes(i + offset)).hexdigest()[:max_len])
+        im = rng.random_integers(0, 255,
+                                 size=(rng.random_integers(min_size,
+                                                           min_size +
+                                                           size_range),
+                                       rng.random_integers(min_size,
+                                                           min_size +
+                                                           size_range),
+                                       rng.random_integers(1, 4)))
+        if not random:
+            im *= 0
+            assert (im == 0).all()
+            im += i
+            assert numpy.isscalar(i)
+            assert (im == i).all()
+        if im.shape[-1] == 2:
+            im = im[:, :, :1]
+        images.append(im)
+    files = sorted(files)
+    temp_tar = io.BytesIO()
+    with tarfile.open(fileobj=temp_tar, mode='w') as tar:
+        for fn, image in zip(files, images):
+            try:
+                with NamedTemporaryFile(mode='wb', suffix='.JPEG',
+                                        delete=False) as f:
+                    if rng.uniform() < gzip_probability:
+                        gzip_data = io.BytesIO()
+                        with gzip.GzipFile(mode='wb', fileobj=gzip_data) as gz:
+                            gz.write(create_jpeg_data(image))
+                        f.write(gzip_data.getvalue())
+                    else:
+                        f.write(create_jpeg_data(image))
+                tar.add(f.name, arcname=fn)
+            finally:
+                os.remove(f.name)
+    ordered_files = []
+    with tarfile.open(fileobj=io.BytesIO(temp_tar.getvalue()),
+                      mode='r') as tar:
+        for info in tar.getmembers():
+            ordered_files.append(info.name)
+    return temp_tar.getvalue(), ordered_files
+
+
+def create_fake_tar_of_tars(seed, num_inner_tars, *args, **kwargs):
+    """Create a nested TAR of TARs of JPEGs.
+
+    Parameters
+    ----------
+    seed : int or sequence
+        Seed for a `numpy.random.RandomState`.
+    num_inner_tars : int
+        Number of TAR files to place inside.
+
+    Returns
+    -------
+    tar_data : bytes
+        A TAR file represented as raw bytes, TAR files of generated
+        JPEGs.
+    names : list
+        Names of the inner TAR files.
+    jpeg_names : list of lists
+        A list of lists containing the names of JPEGs in each inner TAR.
+
+
+    Notes
+    -----
+    Remainder of positional and keyword arguments are passed on to
+    :func:`create_fake_jpeg_tars`.
+
+    """
+    seeds = numpy.arange(num_inner_tars) + seed
+    tars, fns = [], []
+    offset = 0
+    for s in seeds:
+        tar, fn = create_fake_jpeg_tar(s, *args, offset=offset, **kwargs)
+        tars.append(tar)
+        fns.append(fn)
+        offset += len(fn)
+    names = sorted(str(abs(hash(str(-i - 1)))) + '.tar'
+                   for i, t in enumerate(tars))
+    data = io.BytesIO()
+    with tarfile.open(fileobj=data, mode='w') as outer:
+        for tar, name in zip(tars, names):
+            try:
+                with NamedTemporaryFile(mode='wb', suffix='.tar',
+                                        delete=False) as f:
+                    f.write(tar)
+                outer.add(f.name, arcname=name)
+            finally:
+                os.remove(f.name)
+    return data.getvalue(), names, fns
+
+
+def create_fake_patch_images(filenames=None, num_train=14, num_valid=15,
+                             num_test=21):
+    if filenames is None:
+        num = num_train + num_valid + num_test
+        filenames = ['%x' % abs(hash(str(i))) + '.JPEG' for i in xrange(num)]
+    else:
+        filenames = list(filenames)  # Copy, so list not modified in-place.
+    filenames[:num_train] = ['train/' + f
+                             for f in filenames[:num_train]]
+    filenames[num_train:num_train + num_valid] = [
+        'val/' + f for f in filenames[num_train:num_train + num_valid]
+    ]
+    filenames[num_train + num_valid:] = [
+        'test/' + f for f in filenames[num_train + num_valid:]
+    ]
+    tar = create_fake_jpeg_tar(1, min_num_images=len(filenames),
+                               max_num_images=len(filenames),
+                               filenames=filenames, random=False,
+                               gzip_probability=0.0)[0]
+    return tar
+
+
+def test_prepare_metadata():
+    skip_if_not_available(datasets=[DEVKIT_ARCHIVE, TEST_GROUNDTRUTH])
+    devkit_path = find_in_data_path(DEVKIT_ARCHIVE)
+    test_gt_path = find_in_data_path(TEST_GROUNDTRUTH)
+    n_train, v_gt, t_gt, wnid_map = prepare_metadata(devkit_path,
+                                                     test_gt_path)
+    assert n_train == 1261406
+    assert len(v_gt) == 50000
+    assert len(t_gt) == 150000
+    assert sorted(wnid_map.values()) == list(range(1000))
+    assert all(isinstance(k, six.string_types) and len(k) == 9
+               for k in wnid_map)
+
+
+def test_prepare_hdf5_file():
+    hdf5_file = MockH5PYFile()
+    prepare_hdf5_file(hdf5_file, 10, 5, 2)
+
+    def get_start_stop(hdf5_file, split):
+        rows = [r for r in hdf5_file.attrs['split'] if
+                (r['split'].decode('utf8') == split)]
+        return dict([(r['source'].decode('utf8'), (r['start'], r['stop']))
+                     for r in rows])
+
+    # Verify properties of the train split.
+    train_splits = get_start_stop(hdf5_file, 'train')
+    assert all(v == (0, 10) for v in train_splits.values())
+    assert set(train_splits.keys()) == set([u'encoded_images', u'targets',
+                                            u'filenames'])
+
+    # Verify properties of the valid split.
+    valid_splits = get_start_stop(hdf5_file, 'valid')
+    assert all(v == (10, 15) for v in valid_splits.values())
+    assert set(valid_splits.keys()) == set([u'encoded_images', u'targets',
+                                            u'filenames'])
+
+    # Verify properties of the test split.
+    test_splits = get_start_stop(hdf5_file, 'test')
+    assert all(v == (15, 17) for v in test_splits.values())
+    assert set(test_splits.keys()) == set([u'encoded_images', u'targets',
+                                           u'filenames'])
+
+    from numpy import dtype
+
+    # Verify properties of the encoded_images HDF5 dataset.
+    assert hdf5_file['encoded_images'].shape[0] == 17
+    assert len(hdf5_file['encoded_images'].shape) == 1
+    assert hdf5_file['encoded_images'].dtype.kind == 'O'
+    assert hdf5_file['encoded_images'].dtype.metadata['vlen'] == dtype('uint8')
+
+    # Verify properties of the filenames dataset.
+    assert hdf5_file['filenames'].shape[0] == 17
+    assert len(hdf5_file['filenames'].shape) == 2
+    assert hdf5_file['filenames'].dtype == dtype('S32')
+
+    # Verify properties of the targets dataset.
+    assert hdf5_file['targets'].shape[0] == 17
+    assert hdf5_file['targets'].shape[1] == 1
+    assert len(hdf5_file['targets'].shape) == 2
+    assert hdf5_file['targets'].dtype == dtype('int16')
+
+
+def test_process_train_set():
+    tar_data, names, jpeg_names = create_fake_tar_of_tars(20150925, 5,
+                                                          min_num_images=45,
+                                                          max_num_images=55)
+    all_jpegs = numpy.array(sum(jpeg_names, []))
+    numpy.random.RandomState(20150925).shuffle(all_jpegs)
+    patched_files = all_jpegs[:10]
+    patches_data = create_fake_patch_images(filenames=patched_files,
+                                            num_train=10, num_valid=0,
+                                            num_test=0)
+    hdf5_file = MockH5PYFile()
+    prepare_hdf5_file(hdf5_file, len(all_jpegs), 0, 0)
+    wnid_map = dict(zip((n.split('.')[0] for n in names), range(len(names))))
+
+    process_train_set(hdf5_file, io.BytesIO(tar_data),
+                      io.BytesIO(patches_data), len(all_jpegs),
+                      wnid_map)
+
+    # Other tests cover that the actual images are what they should be.
+    # Just do a basic verification of the filenames and targets.
+
+    assert set(all_jpegs) == set(s.decode('ascii')
+                                 for s in hdf5_file['filenames'][:, 0])
+    assert len(hdf5_file['encoded_images'][:]) == len(all_jpegs)
+    assert len(hdf5_file['targets'][:]) == len(all_jpegs)
+
+
+def test_process_other_set():
+    images, all_filenames = create_fake_jpeg_tar(3, min_num_images=30,
+                                                 max_num_images=40,
+                                                 gzip_probability=0.0)
+    all_filenames_shuffle = numpy.array(all_filenames)
+    numpy.random.RandomState(20151202).shuffle(all_filenames_shuffle)
+    patched_files = all_filenames_shuffle[:15]
+    patches_data = create_fake_patch_images(filenames=patched_files,
+                                            num_train=0, num_valid=15,
+                                            num_test=0)
+    hdf5_file = MockH5PYFile()
+    OFFSET = 50
+    prepare_hdf5_file(hdf5_file, OFFSET, len(all_filenames), 0)
+    groundtruth = [i % 10 for i in range(len(all_filenames))]
+    process_other_set(hdf5_file, 'valid', io.BytesIO(images),
+                      io.BytesIO(patches_data), groundtruth, OFFSET)
+
+    # Other tests cover that the actual images are what they should be.
+    # Just do a basic verification of the filenames.
+
+    assert all(hdf5_file['targets'][OFFSET:, 0] == groundtruth)
+    assert all(a.decode('ascii') == b
+               for a, b in zip(hdf5_file['filenames'][OFFSET:, 0],
+                               all_filenames))
+
+
+def test_train_set_producer():
+    tar_data, names, jpeg_names = create_fake_tar_of_tars(20150923, 5,
+                                                          min_num_images=45,
+                                                          max_num_images=55)
+    all_jpegs = numpy.array(sum(jpeg_names, []))
+    numpy.random.RandomState(20150923).shuffle(all_jpegs)
+    patched_files = all_jpegs[:10]
+    patches_data = create_fake_patch_images(filenames=patched_files,
+                                            num_train=10, num_valid=0,
+                                            num_test=0)
+    train_patches = extract_patch_images(io.BytesIO(patches_data), 'train')
+    socket = MockSocket(zmq.PUSH)
+    wnid_map = dict(zip((n.split('.')[0] for n in names), range(len(names))))
+
+    train_set_producer(socket, io.BytesIO(tar_data), io.BytesIO(patches_data),
+                       wnid_map)
+    tar_data, names, jpeg_names = create_fake_tar_of_tars(20150923, 5,
+                                                          min_num_images=45,
+                                                          max_num_images=55)
+    for tar_name in names:
+        with tarfile.open(fileobj=io.BytesIO(tar_data)) as outer_tar:
+            with tarfile.open(fileobj=outer_tar.extractfile(tar_name)) as tar:
+                for record in tar:
+                    jpeg = record.name
+                    metadata_msg = socket.sent.popleft()
+                    assert metadata_msg['type'] == 'send_pyobj'
+                    assert metadata_msg['flags'] == zmq.SNDMORE
+                    key = tar_name.split('.')[0]
+                    assert metadata_msg['obj'] == (jpeg, wnid_map[key])
+
+                    image_msg = socket.sent.popleft()
+                    assert image_msg['type'] == 'send'
+                    assert image_msg['flags'] == 0
+                    if jpeg in train_patches:
+                        assert image_msg['data'] == train_patches[jpeg]
+                    else:
+                        image_data, _ = load_from_tar_or_patch(tar, jpeg,
+                                                               train_patches)
+                        assert image_msg['data'] == image_data
+
+
+MOCK_CONSUMER_MESSAGES = [
+    {'type': 'recv_pyobj', 'flags': zmq.SNDMORE, 'obj': ('foo.jpeg', 2)},
+    {'type': 'recv', 'flags': 0, 'data': numpy.cast['uint8']([6, 6, 6])},
+    {'type': 'recv_pyobj', 'flags': zmq.SNDMORE, 'obj': ('bar.jpeg', 3)},
+    {'type': 'recv', 'flags': 0, 'data': numpy.cast['uint8']([1, 8, 1, 2, 0])},
+    {'type': 'recv_pyobj', 'flags': zmq.SNDMORE, 'obj': ('baz.jpeg', 5)},
+    {'type': 'recv', 'flags': 0, 'data': numpy.cast['uint8']([1, 9, 7, 9])},
+    {'type': 'recv_pyobj', 'flags': zmq.SNDMORE, 'obj': ('bur.jpeg', 7)},
+    {'type': 'recv', 'flags': 0, 'data': numpy.cast['uint8']([1, 8, 6, 7])},
+]
+
+
+def test_image_consumer():
+    mock_messages = MOCK_CONSUMER_MESSAGES
+    hdf5_file = MockH5PYFile()
+    prepare_hdf5_file(hdf5_file, 4, 5, 8)
+    socket = MockSocket(zmq.PULL, to_recv=mock_messages)
+    image_consumer(socket, hdf5_file, 4)
+
+    assert_equal(hdf5_file['encoded_images'][0], [6, 6, 6])
+    assert_equal(hdf5_file['encoded_images'][1], [1, 8, 1, 2, 0])
+    assert_equal(hdf5_file['encoded_images'][2], [1, 9, 7, 9])
+    assert_equal(hdf5_file['encoded_images'][3], [1, 8, 6, 7])
+    assert_equal(hdf5_file['filenames'][:4], [[b'foo.jpeg'], [b'bar.jpeg'],
+                                              [b'baz.jpeg'], [b'bur.jpeg']])
+    assert_equal(hdf5_file['targets'][:4], [[2], [3], [5], [7]])
+
+
+def test_images_consumer_randomized():
+    mock_messages = MOCK_CONSUMER_MESSAGES + [
+        {'type': 'recv_pyobj', 'flags': zmq.SNDMORE, 'obj': ('jenny.jpeg', 1)},
+        {'type': 'recv', 'flags': 0,
+         'data': numpy.cast['uint8']([8, 6, 7, 5, 3, 0, 9])}
+    ]
+    hdf5_file = MockH5PYFile()
+    prepare_hdf5_file(hdf5_file, 4, 5, 8)
+    socket = MockSocket(zmq.PULL, to_recv=mock_messages)
+    image_consumer(socket, hdf5_file, 5, offset=4, shuffle_seed=0)
+    written_data = set(tuple(s) for s in hdf5_file['encoded_images'][4:9])
+    expected_data = set(tuple(s['data']) for s in mock_messages[1::2])
+    assert written_data == expected_data
+
+    written_targets = set(hdf5_file['targets'][4:9].flatten())
+    expected_targets = set(s['obj'][1] for s in mock_messages[::2])
+    assert written_targets == expected_targets
+
+    written_filenames = set(hdf5_file['filenames'][4:9].flatten())
+    expected_filenames = set(s['obj'][0].encode('ascii')
+                             for s in mock_messages[::2])
+    assert written_filenames == expected_filenames
+
+
+def test_other_set_producer():
+    # Create some fake data.
+    num = 21
+    image_archive, filenames = create_fake_jpeg_tar(seed=1979,
+                                                    min_num_images=num,
+                                                    max_num_images=num)
+    patches = create_fake_patch_images(filenames=filenames,
+                                       num_train=7, num_valid=7, num_test=7)
+
+    valid_patches = extract_patch_images(io.BytesIO(patches), 'valid')
+    test_patches = extract_patch_images(io.BytesIO(patches), 'test')
+    assert len(valid_patches) == 7
+    assert len(test_patches) == 7
+
+    groundtruth = numpy.random.RandomState(1979).random_integers(0, 50,
+                                                                 size=num)
+    assert len(groundtruth) == 21
+    gt_lookup = dict(zip(sorted(filenames), groundtruth))
+    assert len(gt_lookup) == 21
+
+    def check(which_set, set_patches):
+        # Run other_set_producer and push to a fake socket.
+        socket = MockSocket(zmq.PUSH)
+        other_set_producer(socket, which_set, io.BytesIO(image_archive),
+                           io.BytesIO(patches), groundtruth)
+
+        # Now verify the data that socket received.
+        with tarfile.open(fileobj=io.BytesIO(image_archive)) as tar:
+            num_patched = 0
+            for im_fn in filenames:
+                # Verify the label and flags of the first (metadata)
+                # message.
+                label = gt_lookup[im_fn]
+                metadata_msg = socket.sent.popleft()
+                assert metadata_msg['type'] == 'send_pyobj'
+                assert metadata_msg['flags'] == zmq.SNDMORE
+                assert metadata_msg['obj'] == (im_fn, label)
+                # Verify that the second (data) message came from
+                # the right place, either a patch file or a TAR.
+                data_msg = socket.sent.popleft()
+                assert data_msg['type'] == 'send'
+                assert data_msg['flags'] == 0
+                expected, patched = load_from_tar_or_patch(tar, im_fn,
+                                                           set_patches)
+                num_patched += int(patched)
+                assert data_msg['data'] == expected
+            assert num_patched == len(set_patches)
+
+    check('valid', valid_patches)
+    check('test', test_patches)
+
+
+def test_load_from_tar_or_patch():
+    # Setup fake tar files.
+    images, all_filenames = create_fake_jpeg_tar(3, min_num_images=200,
+                                                 max_num_images=200,
+                                                 gzip_probability=0.0)
+    patch_data = create_fake_patch_images(all_filenames[::4], num_train=50,
+                                          num_valid=0, num_test=0)
+
+    patches = extract_patch_images(io.BytesIO(patch_data), 'train')
+
+    assert len(patches) == 50
+    with tarfile.open(fileobj=io.BytesIO(images)) as tar:
+        for fn in all_filenames:
+            image, patched = load_from_tar_or_patch(tar, fn, patches)
+            if fn in patches:
+                assert image == patches[fn]
+                assert patched
+            else:
+                tar_image = tar.extractfile(fn).read()
+                assert image == tar_image
+                assert not patched
+
+
+def test_read_devkit():
+    skip_if_not_available(datasets=[DEVKIT_ARCHIVE])
+    synsets, cost_mat, raw_valid_gt = read_devkit(
+        find_in_data_path(DEVKIT_ARCHIVE))
+    # synset and cost_matrix sanity tests appear in test_read_metadata_mat_file
+    assert raw_valid_gt.min() == 1
+    assert raw_valid_gt.max() == 1000
+    assert raw_valid_gt.dtype.kind == 'i'
+    assert raw_valid_gt.shape == (50000,)
+
+
+def test_read_metadata_mat_file():
+    skip_if_not_available(datasets=[DEVKIT_ARCHIVE])
+    with tarfile.open(find_in_data_path(DEVKIT_ARCHIVE)) as tar:
+        meta_mat = tar.extractfile(DEVKIT_META_PATH)
+        synsets, cost_mat = read_metadata_mat_file(meta_mat)
+    assert (synsets['ILSVRC2010_ID'] ==
+            numpy.arange(1, len(synsets) + 1)).all()
+    assert synsets['num_train_images'][1000:].sum() == 0
+    assert (synsets['num_train_images'][:1000] > 0).all()
+    assert synsets.ndim == 1
+    assert synsets['wordnet_height'].min() == 0
+    assert synsets['wordnet_height'].max() == 19
+    assert synsets['WNID'].dtype == numpy.dtype('S9')
+    assert (synsets['num_children'][:1000] == 0).all()
+    assert (synsets['children'][:1000] == -1).all()
+
+    # Assert the basics about the cost matrix.
+    assert cost_mat.shape == (1000, 1000)
+    assert cost_mat.dtype == 'uint8'
+    assert cost_mat.min() == 0
+    assert cost_mat.max() == 18
+    assert (cost_mat == cost_mat.T).all()
+    # Assert that the diagonal is 0.
+    assert (cost_mat.flat[::1001] == 0).all()
+
+
+def test_extract_patch_images():
+    tar = create_fake_patch_images()
+    assert len(extract_patch_images(io.BytesIO(tar), 'train')) == 14
+    assert len(extract_patch_images(io.BytesIO(tar), 'valid')) == 15
+    assert len(extract_patch_images(io.BytesIO(tar), 'test')) == 21
diff --git a/tests/test_utils.py b/tests/test_utils.py
index fe1d7f0..fef61bc 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -1,7 +1,9 @@
+from functools import partial
 import operator
 import os
 import shutil
 import tempfile
+import time
 
 import numpy
 from numpy.testing import assert_raises, assert_equal
@@ -10,6 +12,7 @@ from six.moves import range, cPickle
 from fuel import config
 from fuel.iterator import DataIterator
 from fuel.utils import do_not_pickle_attributes, find_in_data_path, Subset
+from fuel.utils.parallel import producer_consumer
 
 
 class TestSubset(object):
@@ -257,3 +260,29 @@ class TestDoNotPickleAttributes(object):
 
     def test_value_error_attribute_non_loaded(self):
         assert_raises(ValueError, getattr, NonLoadingClass(), 'attribute')
+
+
+def send_integers(socket, n):
+    socket.send_pyobj(n)
+    for i in range(n):
+        socket.send_pyobj(i ** 2)
+        # This works around strange bug in (probably) libzmq on
+        # OS X 10.9 which one of the pyzmq developers couldn't reproduce
+        # with all the same library versions (albeit OS X 10.10)... real
+        # workers will never be this trivial.
+        time.sleep(1e-6)
+
+
+def receive_integers(socket):
+    num = socket.recv_pyobj()
+    total = 0
+    for i in range(num):
+        recv = socket.recv_pyobj()
+        total += recv
+    return total
+
+
+def test_producer_consumer():
+    assert (producer_consumer(partial(send_integers, n=2000),
+                              receive_integers)
+            == sum(i ** 2 for i in range(2000)))
diff --git a/tests/transformers/test_transformers.py b/tests/transformers/test_transformers.py
index 1550149..c26c80b 100644
--- a/tests/transformers/test_transformers.py
+++ b/tests/transformers/test_transformers.py
@@ -15,6 +15,7 @@ from fuel.transformers import (
     ExpectsAxisLabels, Transformer, Mapping, SortMapping, ForceFloatX, Filter,
     Cache, Batch, Padding, MultiProcessing, Unpack, Merge,
     SourcewiseTransformer, Flatten, ScaleAndShift, Cast, Rename, FilterSources)
+from fuel.transformers.defaults import ToBytes
 
 
 class FlagDataStream(DataStream):
@@ -723,3 +724,26 @@ class TestExpectsAxisLabels(object):
     def test_exception(self):
         assert_raises(ValueError, self.obj.verify_axis_labels, ('a', 'b', 'c'),
                       ('b', 'c', 'd'), 'foo')
+
+
+class TestToBytes(object):
+    def setUp(self):
+        self.string_data = [b'Hello', b'World!']
+        self.dataset = IndexableDataset(
+            indexables={'words': [numpy.fromstring(s, dtype='uint8')
+                                  for s in self.string_data]},
+            axis_labels={'words': ('batch', 'bytes')})
+
+    def test_examplewise(self):
+        stream = DataStream(
+            dataset=self.dataset, iteration_scheme=SequentialExampleScheme(2))
+        decoded_stream = ToBytes(stream)
+        assert_equal(self.string_data,
+                     [s for s, in decoded_stream.get_epoch_iterator()])
+
+    def test_batchwise(self):
+        stream = DataStream(
+            dataset=self.dataset, iteration_scheme=SequentialScheme(2, 2))
+        decoded_stream = ToBytes(stream)
+        assert_equal([self.string_data],
+                     [s for s, in decoded_stream.get_epoch_iterator()])