PyDataProviderWrapper.py

# Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
This module provide a wrapper(decorator) to wrap a data process method into a
PyDataProvider. Some examples are shown `here <data_provider/python_case.html>`_.
"""

import struct
import array
import random
import gc
import logging
import pstats
import sys
import numpy
import functools

__all__ = [
    'DenseSlot', 'SlotType', 'SparseNonValueSlot', 'StringSlot',
    'SparseValueSlot', 'IndexSlot', 'PoolSize', 'GeneralPyDataProvider',
    'provider', 'init_hook_wrapper'
]

try:  # Just for profile mode, will try to import cProfile first.
    # Most python will contains cProfile, cProfile/profile are basically same.
    # ref: https://docs.python.org/2/library/profile.html#introduction-to-the-profilers
    import cProfile as profile
except ImportError:
    import profile

try:
    import cPickle as pickle
except ImportError:
    import six.moves.cPickle as pickle

import io


class SlotType(object):  # Just a hint for user.
    pass


class DenseSlot(SlotType):
    """
    Dense Slot Type: Each item is the value of a Dense Vector.

    Its yield format for :code:`provider` is:

    - **NonSeq**: [float, float, ... ]
    - **Seq**: [[float, float, ...], [float, float ....], ... ]
    - **SubSeq**: [[[float, float, ...], [float ....], ...] ,  \
                   [[float, float, ...], [float ....], ...] , ...]
    """

    def __init__(self, dim):
        """
        :param dim: slot dimension
        :type dim: int
        """
        self.dim = dim
        self.type = 0


class SparseNonValueSlot(SlotType):
    """
    Sparse NonValue Slot Type: Each item is the id of a Sparse Vector.

    Its yield format for :code:`provider` is:

    - **NonSeq**: [int, int, ...]
    - **Seq**: [[int, int, ...], [int, int, ...], ... ]
    - **SubSeq**: [[[int, int, ...], [int, ....], ...] ,  \
                   [[int, int, ...], [int, ....], ...] , ...]
    """

    def __init__(self, dim):
        """
        :param dim: slot dimension
        :type dim: int
        """
        self.dim = dim
        self.type = 1


class SparseValueSlot(SlotType):
    """
    Sparse Value Slot Type: Each item is the id and value of a Sparse Vector.

    Its yield format for :code:`provider` is:

    - **NonSeq**: [(int, float), (int, float), ... ]
    - **Seq**: [[(int,float), (int, float), ... ], \
                [(int, float), (int, float), ...], ... ]
    - **SubSeq**: [[[(int,float), ...], [(int, float), ....], ...] ,  \
                   [[(int,float), ...], [(int, float), ....], ...] , ...]
    """

    def __init__(self, dim):
        """
        :param dim: slot dimension.
        :type dim: int
        """
        self.dim = dim
        self.type = 2


class IndexSlot(SlotType):
    """
    Index Value Slot Type: Each item is the id of Label.

    Its yield format for :code:`provider` is:

    - **NonSeq**: int
    - **Seq**:  [int, int, ....]
    - **SubSeq**: [[int, int, ...], [int, int, ...], ... ]
    """

    def __init__(self, dim):
        """
        :param dim: slot dimension
        :type dim: int
        """
        self.dim = dim
        self.type = 3


class StringSlot(SlotType):
    """
    String Value Slot Type: Each item is a string for printout, \
                            can be used in DataLayer too.

    Its yield format for :code:`provider` is:

    - **NonSeq**: string
    - **Seq**: [string, string, ....]
    - **SubSeq**:  [[string, string, ...], [string, string, ...], ... ]
    """

    def __init__(self, dim):
        """
        :param dim: slot dimension
        :type dim: string
        """
        self.dim = dim
        self.type = 6


class SparseNonValueHandler(object):
    """
    Private Class, Use for converting python object to paddle string.
    """

    def __init__(self):
        self.offsets = []
        self.value = []
        self.offset_count = 0

    def __call__(self, ele):
        """
        It will be invoked when scan each sparse data.

        :param ele: list of sparse data, maybe non-value [ idx, ... ] or value.
                    [ (idx, val), ... ]
        :type ele: list
        """
        self.offsets.append(self.offset_count)
        self.offset_count += len(ele)
        self.processElement(ele)

    def processElement(self, ele):
        """
        Process for element list. See __call__ for more document.
        """
        self.value += ele

    def done(self, data_stream, int_packer):
        """
        Dump data to stream.
        :param data_stream: Output Stream.
        :param int_packer:  A struct.Struct("i") object
        """
        data_stream.write(array.array("i", self.offsets).tostring())
        data_stream.write(int_packer.pack(self.offset_count))
        data_stream.write(array.array("i", self.value).tostring())


class SparseValueHandler(SparseNonValueHandler):
    """
    Private class, use for converting python obj to paddle string.
    """

    def __init__(self):
        SparseNonValueHandler.__init__(self)
        self.weight = []

    def processElement(self, ele):
        for idx, w in ele:
            self.value.append(idx)
            self.weight.append(w)

    def done(self, data_stream, int_packer):
        SparseNonValueHandler.done(self, data_stream, int_packer)
        data_stream.write(int_packer.pack(self.offset_count))
        data_stream.write(array.array("f", self.weight).tostring())


class StringHandler(object):
    """
    Private Class, Use for converting python object to paddle string.
    """

    def __init__(self, data_stream, int_packer):
        self.data_stream = data_stream
        self.int_packer = int_packer

    def __call__(self, ele):
        """
        It will be invoked when scan each string data.
        :param ele: string data
        :type ele: str
        """
        self.data_stream.write(self.int_packer.pack(len(ele)))
        self.data_stream.write(array.array("c", ele).tostring())


class GeneralPyDataProvider:
    def __init__(self, *file_list, **kwargs):
        """
        :param file_list: input file_list
        """
        del kwargs  # unused
        gc.disable()
        assert isinstance(self.logger, logging.Logger)
        self.use_seq_flag = hasattr(self, "use_seq_flag") and self.use_seq_flag
        self.slots_num = len(self.getSlots())
        self.file_list = list(file_list)
        self.generators = map(self.generateData, self.file_list)
        self.int_packer = struct.Struct("i")
        self.head_packer = struct.Struct("ii")
        self.float_packer = struct.Struct("f")
        self.shuffler = lambda *args, **kwargs: None
        self.data_pool = []
        self.has_subseq = []
        self.has_checked = False

        self.debug = hasattr(self, "debug") and self.debug

        if hasattr(self, "profile_filename") and isinstance(
                self.profile_filename, str):
            self.profile_count = 0
            self.is_profile = True
        else:
            self.is_profile = False

        if not hasattr(self, "file_count") or not isinstance(self.file_count,
                                                             int):
            self.file_count = sys.maxint

        if not hasattr(self, "can_over_batch_size"):
            self.can_over_batch_size = True
        elif not self.can_over_batch_size:
            self.logger.warn(
                "User should ensure every data size is not larger than batch"
                " size when can_over_batch_size = False")

        self.data_pool_idx = 0

    def reset(self):
        """Reset all data in provider."""

        self.logger.debug("reset dataprovider.")
        self.generators = map(self.generateData, self.file_list)
        self.shuffler = lambda *args, **kwargs: None
        self.data_pool = []
        self.data_pool_idx = 0
        if self.file_count != 0:
            self.max_pool_size = 0

        # When use Profile, each pass will print a profile result.
        if self.is_profile:
            if hasattr(self, "profiler") and isinstance(self.profiler,
                                                        profile.Profile):
                self.profiler.disable()
                fn = "%s_%d" % (self.profile_filename, self.profile_count)
                sortby = "cumulative"
                with open(fn, "w") as f:
                    pstats.Stats(
                        self.profiler,
                        stream=f).sort_stats(sortby).print_stats()
                self.logger.info("saving profile to file %s" % fn)
                self.profile_count += 1
            self.logger.info("resetting profile")
            self.profiler = profile.Profile()
            self.profiler.enable()

    def shuffle(self):
        """ shuffle data"""
        if not self.should_shuffle:
            return
        else:
            self.logger.debug("shuffling data.")
            random.shuffle(self.generators)
            self.shuffler = random.shuffle

    def getSlots(self):
        """
        :return : return a list of SlotType
        :rtype: list
        """
        return []

    def generateData(self, fn):
        """
        :param fn: file name
        :return: a generator to yield data one by one.
        """
        raise NotImplementedError

    def calculateDataBatchSize(self, data):
        """
        :param data: One sample which yield by generateData
        :type data: list
        :return: The batch size that the data contribute.
        :rtype: int
        """
        return 1

    def getHeader(self):
        """return paddle header format"""
        ret = self.head_packer.pack(self.slots_num, self.use_seq_flag)
        for obj in self.getSlots():
            ret += self.head_packer.pack(obj.type, obj.dim)
        return ret

    def getHeaderNative(self):
        return self.use_seq_flag, self.getSlots()

    def getNextBatchNative(self, batch_size):
        ret_list = []
        self.__prepareData(batch_size, ret_list)
        return ret_list

    def getNextBatch(self, batch_size):
        """
        :param batch_size: the batch_size approximately return.
        :return: return paddle pyDataProvider format, just see documents.
        :rtype: str

        NOTE: If can_over_batch_size is True, the return batch_size >= input batch_size.
              Otherwise, the return batch_size < input batch_size, BUT USER MUST ENSURE THAT each data's batch size
              is less than input batch_size.
        """
        ret_list = []
        current_batch_size = self.__prepareData(batch_size, ret_list)
        # create unified format for ret_list with differnt slots_num
        if self.slots_num == 1:
            ret_list = [ret_list]

        if current_batch_size == 0:
            return self.int_packer.pack(current_batch_size)
        data_bytes = io.BytesIO()
        seq_bytes = io.BytesIO()
        subseq_bytes = io.BytesIO()
        data_stream = io.BufferedWriter(data_bytes)
        seq_stream = io.BufferedWriter(seq_bytes)
        subseq_stream = io.BufferedWriter(subseq_bytes)

        def convertDataImpl(idx, data_callback):
            """
            This method will handle sequence in return data. invoke data_callback one by one.
            :param idx: the slot index.
            :param data_callback: a callback, which type is (each sample) => None.
            """
            indices = 0
            slot_sample_num = len(ret_list)
            if self.use_seq_flag:
                slot_sample_num = 0
                if self.has_subseq[idx]:  # has sub-sequence
                    slot_subseq_num = 0
                    for dat in ret_list:
                        dat = dat[idx]
                        slot_subseq_num += len(dat)
                        for sub_dat in dat:
                            slot_sample_num += len(sub_dat)
                    subseq_stream.write(self.int_packer.pack(slot_subseq_num))
                else:
                    for dat in ret_list:
                        dat = dat[idx]
                        slot_sample_num += len(dat)
                seq_stream.write(self.int_packer.pack(len(ret_list)))
            data_stream.write(self.int_packer.pack(slot_sample_num))

            for dat in ret_list:
                dat = dat[idx]
                if self.use_seq_flag:
                    seq_stream.write(self.int_packer.pack(indices))
                    if self.has_subseq[idx]:  # has sub-sequence
                        for sub_dat in dat:
                            writeDataStream(sub_dat, data_callback)
                            subseq_stream.write(self.int_packer.pack(indices))
                            indices += len(sub_dat)
                    else:
                        writeDataStream(dat, data_callback)
                        indices += len(dat)
                else:
                    writeDataStream(dat, data_callback)

        def writeDataStream(dat, data_callback):
            if self.use_seq_flag > 0:
                if data_callback is None:  # Special for index slot
                    data_stream.write(array.array("i", dat).tostring())
                else:
                    for ele in dat:
                        data_callback(ele)
            else:
                if data_callback is None:  # Special for index slot
                    data_stream.write(self.int_packer.pack(dat))
                else:
                    data_callback(dat)

        try:
            for i in range(self.slots_num):
                slot = self.getSlots()[i]
                # According to the data_type, each slot data will be converted to binary
                if isinstance(slot, DenseSlot):
                    convertDataImpl(i, lambda e: data_stream.write(
                        array.array("f", e).tostring()))
                elif isinstance(slot, SparseNonValueSlot):
                    handler = SparseNonValueHandler()
                    convertDataImpl(i, handler)
                    handler.done(data_stream, self.int_packer)
                elif isinstance(slot, SparseValueSlot):
                    handler = SparseValueHandler()
                    convertDataImpl(i, handler)
                    handler.done(data_stream, self.int_packer)
                elif isinstance(slot, IndexSlot):
                    convertDataImpl(i, None)
                elif isinstance(slot, StringSlot):
                    handler = StringHandler(data_stream, self.int_packer)
                    convertDataImpl(i, handler)
                else:
                    raise RuntimeError("The data_type must be 0/1/2/3/6")
            data_stream.flush()
            seq_stream.flush()
            subseq_stream.flush()

            return "".join([
                self.int_packer.pack(current_batch_size), data_bytes.getvalue(),
                seq_bytes.getvalue(), subseq_bytes.getvalue()
            ])

        finally:
            data_stream.close()
            seq_stream.close()
            subseq_stream.close()
            data_bytes.close()
            seq_bytes.close()
            subseq_bytes.close()

    def hasSubseq(self, ret_list):
        # create unified format for ret_list with differnt slots_num
        if self.slots_num == 1:
            ret_list = [ret_list]
        # decide whether slot has sub-sequence using its first sample
        for i in range(self.slots_num):
            slot = self.getSlots()[i]
            dat = ret_list[0][i][0]
            if isinstance(slot, IndexSlot) or isinstance(slot, StringSlot):
                if isinstance(dat, list) or isinstance(dat, numpy.ndarray):
                    self.has_subseq.append(1)  # has_subseq = True
                    continue
            elif isinstance(dat[0], list) or isinstance(dat[0], numpy.ndarray):
                self.has_subseq.append(1)  # has_subseq = True
                continue
            self.has_subseq.append(0)  # has_subseq = False

    def checkOrder(self):
        first_noSubseq_slot = self.slots_num
        last_subseq_slot = -1
        for i in range(self.slots_num):
            if not self.has_subseq[i]:
                first_noSubseq_slot = i
                break
        for i in range(self.slots_num):
            if self.has_subseq[i]:
                last_subseq_slot = i
        if first_noSubseq_slot < last_subseq_slot:
            raise RuntimeError(
                "slot hasSubseq must put before than slot without subseq")
        self.has_checked = True

    def __prepareData(self, batch_size, ret_list):
        current_batch_size = 0
        could_exit = False
        while not could_exit:
            if len(self.data_pool) == 0:
                self.data_pool_idx = 0
                self.fillPool()
            if len(self.data_pool) != 0:
                for idx in xrange(self.data_pool_idx, len(self.data_pool)):
                    current_batch_size += self.calculateDataBatchSize(
                        self.data_pool[idx])
                    if current_batch_size >= batch_size:
                        could_exit = True
                        break
                if current_batch_size > batch_size and not self.can_over_batch_size:  # if cannot over batch size
                    current_batch_size -= self.calculateDataBatchSize(
                        self.data_pool[idx])
                    idx -= 1

                ret_list += self.data_pool[self.data_pool_idx:idx + 1]

                # for speed reason, just shift left index, not delete data actually.
                self.data_pool_idx = idx + 1

                if self.data_pool_idx == len(self.data_pool):
                    self.data_pool = []
            else:
                break
        if self.use_seq_flag and not self.has_checked:  # compute self.has_subseq and checkOrder only at first time
            self.hasSubseq(ret_list)
            self.checkOrder()
        return current_batch_size

    def fillPool(self):
        """
        Fill the pool to max_pool_size. If max_pool_size is None, then read file_count to pool.
        """
        if self.max_pool_size == 0:
            for i in xrange(min(self.file_count, len(self.generators))):
                self.data_pool += list(self.generators[i])
            self.generators = self.generators[min(self.file_count,
                                                  len(self.generators)):]
            self.max_pool_size = len(self.data_pool)
        else:
            while len(self.data_pool) < self.max_pool_size and len(
                    self.generators) != 0:
                try:
                    self.data_pool.append(self.generators[0].next())
                except StopIteration:
                    self.generators.pop(0)
        self.shuffler(self.data_pool)


class PoolSize(object):
    """Max number of sample which contains in provider."""

    def __init__(self, pool_size):
        self.size = pool_size


def default_init_hook(cls, *args, **kwargs):
    """ default hook, do nothing """
    del cls, args, kwargs


def provider(slots=None,
             use_seq=False,
             should_shuffle=True,
             pool_size=1,
             can_over_batch_size=True,
             calc_batch_size=lambda data: 1,
             debug=False,
             init_hook=default_init_hook,
             profile_filename=None):
    """
    The decorator for PyDataProvider. User should use this to create Provider class.
    User should only concern how to read sample from file.

    So the basic usage is:

    ..  code-block:: python

        @provider(some data provider config here...)
        def process(obj, file_name):
            while not at end of file_name:
                sample = readOneSampleFromFile(file_name)
                yield sample.

    The configuration of data provider should be setup by:

    :param init_hook: A callback will be invoked when PyDataProvider instance \
                      created. The parameter is (obj, \*args, \*\*kwargs).

                      - **obj**: actually data provider instance, which \
                                 contains some global objects in obj.xxxxx, \
                                 and is used by process function.

                        1. **obj.slots**: a list of SlotType Object. Can be \
                                          set in init. For example, obj.slots = \
                                          [DenseSlot(9), IndexSlot(2)].
                        2. **obj.logger**: a logger object. User can invoke \
                                          obj.logger.info(), obj.logger.fatal(), etc.

                      - **args** and **kwargs**: the data provider __init__ \
                                                 parameters. For example, load_data_args \
                                                 will be found in \*\*kwargs, \
                                                 and if you want to recieve \
                                                 it from trainer_config, \
                                                 recommand to use init_hook_wrapper
    :type init_hook: callable

    :param pool_size:
                      - **int**: it will read at most pool_size files to memory.
                      - **PoolSize**: it will read at most PoolSize.size samples to memory.
                      - If not set, it will read all the files to memory.
    :type pool_size: int | PoolSize

    :param slots: Specify the SlotTypes, can also be set in init_hook. It has two formats:

                  - A list of SlotType objects. For example, slots = \
                    [DenseSlot(9), IndexSlot(2)].
                  - A method return a list of SlotTypes, and the parameter of \
                    method is (obj, \*file_list, \*\*kwargs).
    :type slots: list | callable

    :param use_seq:  False if use no sequence (Default). True if use sequence:

                     - If sequence has **no sub-sequence**: Each slot will \
                       return a list of data. This list is one sequence. \
                       So the return format likes \
                       [[a0, a1, a2], [b1, b2, b3, b4], [c1]].
                     - If sequence has **sub-sequence**: Each slot will return \
                       a nested-list of data. This list contains several \
                       sub-lists, each sub-list is one sub-sequence. \
                       So the return format likes \
                       [[[a0, a1, a2], [a4, a5]], [[b1, b2, b3, b4], [b5, b6]], [[c1], [c2]]].
    :type use_seq: bool

    :param should_shuffle: True if data should shuffle.
    :type should_shuffle: bool

    :param calc_batch_size: The method calculate each data's batch size.

                            - Default is the batch size of one sample.
                            - User can customize by **lamda** funtion. For example, \
                              :code:`calc_batch_size = lambda data : len(data)` \
                              means calculating the token number of a sequence data.
    :type calc_batch_size: callable

    :param can_over_batch_size: Whether :code:`actual batch size >= input batch size`

                                - **True** (>=): getNextBatch method can return more data (Default).
                                - **False** (<): user must ensure that each data's batch size < input batch size.
    :type can_over_batch_size: bool

    :param debug: True if enable debug logger and some debug check. Default is False.
    :type debug: bool

    :param profile_filename: None if disable profile (Default). Otherwise, \
                             the data provider will dump profile result when \
                             reset. And the dump filename is \
                             **<profile_filename>_<reset_count>**.
    :type profile_filename: None | Str
    """

    def _wrapper(handler):
        class Cls(GeneralPyDataProvider):
            """ Real PyDataProvider Class. """

            def __init__(self, *file_list, **kwargs):
                logging.basicConfig(
                    format="[%(levelname)s %(asctime)s %(filename)s:%(lineno)s]"
                    " %(message)s")

                self.logger = logging.getLogger("")
                if debug:
                    self.logger.setLevel(logging.DEBUG)
                    self.logger.debug("Running pydataprovider in debug mode.")
                else:
                    self.logger.setLevel(logging.INFO)

                init_hook(self, *file_list, **kwargs)
                if callable(slots):
                    self.slots = slots(self, *file_list, **kwargs)
                elif slots is not None:
                    self.slots = slots

                if isinstance(pool_size, int):
                    self.max_pool_size = 0
                    self.file_count = pool_size
                elif isinstance(pool_size, PoolSize):
                    self.max_pool_size = pool_size.size
                    self.file_count = 0
                else:
                    raise RuntimeError
                self.can_over_batch_size = can_over_batch_size
                self.debug = debug
                self.profile_filename = profile_filename
                self.use_seq_flag = use_seq
                self.should_shuffle = should_shuffle
                GeneralPyDataProvider.__init__(self, *file_list, **kwargs)

            def getSlots(self):
                return self.slots

            def generateData(self, f):
                return handler(self, f)

            def calculateDataBatchSize(self, data):
                return calc_batch_size(data)

        return Cls

    return _wrapper


def init_hook_wrapper(func):
    """
    Wrap a method for PyDataProviderWrapper's init_hook. This method can
    receive parameter from trainer_config's load_data_args. The load_data_args
    must pass a pickle.dumps() value, and dump a map as keyword args. The
    wrapped method :code:`func` will receive them as keyword args.

    So an example usage is:

    ..  code-block:: python

        @init_hook_wrapper
        def hook(obj, dictionary, file_list, **kwargs):
            obj.dictionary = dictionary
            obj.slots = [IndexSlot(len(obj.dictionary)),
                         IndexSlot(len(open(file_list[0], "r").readlines()))]

    :param func: init_hook function
    :type func: callable
    :return: wrapped method, can be passed into @provider.
    """

    @functools.wraps(func)
    def wrapper(obj, *file_list, **kwargs):
        args = kwargs.get("load_data_args", dict())
        if isinstance(args, basestring):
            args = pickle.loads(args)
        args['file_list'] = file_list
        func(obj=obj, **args)

    return wrapper