dataprovider_converter.py 8.7 KB
Newer Older
1
# Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved
Y
yuyang18 已提交
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
#
# 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.

import paddle.trainer.PyDataProvider2 as dp2
import collections
import swig_paddle
Y
Yu Yang 已提交
18
import numpy
Y
Yu Yang 已提交
19
import itertools
Y
yuyang18 已提交
20 21 22 23 24

__all__ = ['DataProviderConverter']


class IScanner(object):
Y
Yu Yang 已提交
25 26 27 28 29 30 31 32 33 34 35 36 37
    """
    The scanner will scan Python object two passes, then convert it to Paddle's
    argument.

    In the first pass, `pre_scan` will be invoked by every data instance, and
    then invoke `finish_pre_scan` to arguments. And the second pass do the same
    thing except the functions changed to `scan`, `finish_scan`.

    During the first pass, a scanner may count the shape of input matrix and
    allocate memory for this argument. Then fill the data into this  argument
    in second pass.
    """

Y
yuyang18 已提交
38 39
    def __init__(self, input_type, pos):
        self.input_type = input_type
D
dangqingqing 已提交
40 41
        if not isinstance(self.input_type, dp2.InputType):
            raise ValueError("input type should be dataprovider2.InputType")
Y
yuyang18 已提交
42
        self.pos = pos
D
dangqingqing 已提交
43 44 45 46 47 48 49
        # data_in_gpu is used to indicate whether to create argument on GPU
        # or not in GPU mode. Now if using one thread (trainer_count=1),
        # trainer uses NeuralNetwork which needs to create argument on GPU
        # before calling forward function. So, set data_in_gpu to True.
        # Otherwise, trainer uses MultiGradientMachine which will transfer
        # data from CPU to GPU in the forward function, set data_in_gpu to
        # False in this case.
D
dangqingqing 已提交
50 51
        self.data_in_gpu = swig_paddle.isUsingGpu(
        ) and swig_paddle.getTrainerCount() == 1
Y
yuyang18 已提交
52

Y
Yu Yang 已提交
53
    def pre_scan(self, dat):
Y
Yu Yang 已提交
54 55 56 57 58 59
        """
        First pass scan method. During this method, the scanner could count the
        data number, and get the total memory size this batch would use.

        :param dat: The python object.
        """
Y
Yu Yang 已提交
60 61 62
        pass

    def finish_pre_scan(self, argument):
Y
Yu Yang 已提交
63 64 65 66 67 68 69
        """
        Finish first scan pass. Allocate the memory.

        :param argument: Output arguments object.
        :type argument: swig_paddle.Arguments
        :return:
        """
Y
Yu Yang 已提交
70 71
        pass

Y
yuyang18 已提交
72
    def scan(self, dat):
Y
Yu Yang 已提交
73 74 75 76 77
        """
        Second pass scan method. Copy the data to arguments.

        :param dat: The python object.
        """
Y
yuyang18 已提交
78 79 80
        pass

    def finish_scan(self, argument):
Y
Yu Yang 已提交
81 82 83 84 85 86
        """
        Finish second pass. Finalize the resources, etc.

        :param argument: Output arguments object.
        :type argument: swig_paddle.Arguments
        """
Y
yuyang18 已提交
87 88 89 90
        pass


class DenseScanner(IScanner):
91 92 93 94
    """
    :type __mat__: numpy.ndarray
    """

Y
yuyang18 已提交
95 96
    def __init__(self, input_type, pos):
        IScanner.__init__(self, input_type, pos)
Y
Yu Yang 已提交
97
        self.__mat__ = None
Y
Yu Yang 已提交
98 99 100 101 102 103 104 105 106
        self.__height__ = 0

    def pre_scan(self, dat):
        self.__height__ += 1

    def finish_pre_scan(self, argument):
        self.__mat__ = numpy.ndarray(
            shape=(self.__height__, self.input_type.dim), dtype=numpy.float32)
        self.__height__ = 0
Y
yuyang18 已提交
107 108

    def scan(self, dat):
Y
Yu Yang 已提交
109 110
        self.__mat__[self.__height__] = dat
        self.__height__ += 1
Y
yuyang18 已提交
111 112 113

    def finish_scan(self, argument):
        assert isinstance(argument, swig_paddle.Arguments)
114 115
        if self.__mat__.dtype != numpy.float32:
            self.__mat__ = self.__mat__.astype(numpy.float32)
116
        m = swig_paddle.Matrix.createDenseFromNumpy(self.__mat__, True,
D
dangqingqing 已提交
117
                                                    self.data_in_gpu)
Y
yuyang18 已提交
118 119 120 121 122 123 124 125 126 127 128 129 130
        argument.setSlotValue(self.pos, m)


class SparseBinaryScanner(IScanner):
    def __init__(self, input_type, pos):
        IScanner.__init__(self, input_type, pos)
        self.__rows__ = [0]
        self.__cols__ = []
        self.__height__ = 0
        self.__value__ = []

    def scan(self, dat):
        self.extend_cols(dat)
E
emailweixu 已提交
131
        self.__rows__.append(len(self.__cols__))
Z
Z-TAO 已提交
132
        self.__height__ += 1
Y
yuyang18 已提交
133 134 135 136 137 138

    def extend_cols(self, dat):
        self.__cols__.extend(dat)

    def finish_scan(self, argument):
        assert isinstance(argument, swig_paddle.Arguments)
139 140 141 142 143 144 145
        m = swig_paddle.Matrix.createSparse(
            self.__height__,
            self.input_type.dim,
            len(self.__cols__),
            len(self.__value__) == 0,
            False,  # trans
            False)  # TODO supoort GPU
Y
yuyang18 已提交
146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162
        assert isinstance(m, swig_paddle.Matrix)
        m.sparseCopyFrom(self.__rows__, self.__cols__, self.__value__)
        argument.setSlotValue(self.pos, m)


class SparseFloatScanner(SparseBinaryScanner):
    def __init__(self, input_type, pos):
        SparseBinaryScanner.__init__(self, input_type, pos)

    def extend_cols(self, dat):
        self.__cols__.extend((x[0] for x in dat))
        self.__value__.extend((x[1] for x in dat))


class IndexScanner(IScanner):
    def __init__(self, input_type, pos):
        IScanner.__init__(self, input_type, pos)
163 164 165 166 167 168 169 170 171
        self.__ids__ = None
        self.__idx__ = 0

    def pre_scan(self, dat):
        self.__idx__ += 1

    def finish_pre_scan(self, argument):
        self.__ids__ = [0] * self.__idx__
        self.__idx__ = 0
Y
yuyang18 已提交
172 173

    def scan(self, dat):
174 175
        self.__ids__[self.__idx__] = dat
        self.__idx__ += 1
Y
yuyang18 已提交
176 177

    def finish_scan(self, argument):
D
dangqingqing 已提交
178
        ids = swig_paddle.IVector.create(self.__ids__, self.data_in_gpu)
Y
yuyang18 已提交
179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219
        assert isinstance(argument, swig_paddle.Arguments)
        argument.setSlotIds(self.pos, ids)


class SequenceScanner(IScanner):
    def __init__(self, input_type, pos, inner_scanner, setter):
        IScanner.__init__(self, input_type, pos)
        self.__seq__ = [0]
        self.__inner_scanner__ = inner_scanner
        self.__setter__ = setter

    def scan(self, dat):
        self.__seq__.append(self.__seq__[-1] + self.get_size(dat))
        for each in dat:
            self.__inner_scanner__.scan(each)

    def finish_scan(self, argument):
        seq = swig_paddle.IVector.create(self.__seq__, False)
        self.__setter__(argument, self.pos, seq)
        self.__inner_scanner__.finish_scan(argument)

    def get_size(self, dat):
        if isinstance(self.__inner_scanner__, SequenceScanner):
            return sum(self.__inner_scanner__.get_size(item) for item in dat)
        else:
            return len(dat)


class DataProviderConverter(object):
    def __init__(self, input_types):
        self.input_types = input_types
        assert isinstance(self.input_types, collections.Sequence)
        for each in self.input_types:
            assert isinstance(each, dp2.InputType)

    def convert(self, dat, argument=None):
        if argument is None:
            argument = swig_paddle.Arguments.createArguments(0)
        assert isinstance(argument, swig_paddle.Arguments)
        argument.resize(len(self.input_types))

220 221 222 223
        scanners = [
            DataProviderConverter.create_scanner(i, each_type)
            for i, each_type in enumerate(self.input_types)
        ]
Y
yuyang18 已提交
224 225

        for each_sample in dat:
Y
Yu Yang 已提交
226 227 228 229 230 231 232 233
            for each_step, scanner in itertools.izip(each_sample, scanners):
                scanner.pre_scan(each_step)

        for scanner in scanners:
            scanner.finish_pre_scan(argument)

        for each_sample in dat:
            for each_step, scanner in itertools.izip(each_sample, scanners):
Y
yuyang18 已提交
234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258
                scanner.scan(each_step)

        for scanner in scanners:
            scanner.finish_scan(argument)

        return argument

    def __call__(self, dat, argument=None):
        return self.convert(dat, argument)

    @staticmethod
    def create_scanner(i, each):
        assert isinstance(each, dp2.InputType)
        retv = None
        if each.type == dp2.DataType.Dense:
            retv = DenseScanner(each, i)
        elif each.type == dp2.DataType.Index:
            retv = IndexScanner(each, i)
        elif each.type == dp2.DataType.SparseNonValue:
            retv = SparseBinaryScanner(each, i)
        elif each.type == dp2.DataType.SparseValue:
            retv = SparseFloatScanner(each, i)
        assert retv is not None

        if each.seq_type == dp2.SequenceType.SUB_SEQUENCE:
259 260 261 262 263 264 265 266 267 268
            retv = SequenceScanner(
                each, i, retv,
                lambda a, p, seq: a.setSlotSubSequenceStartPositions(p, seq))

        if each.seq_type in [
                dp2.SequenceType.SUB_SEQUENCE, dp2.SequenceType.SEQUENCE
        ]:
            retv = SequenceScanner(
                each, i, retv,
                lambda a, p, seq: a.setSlotSequenceStartPositions(p, seq))
Y
yuyang18 已提交
269
        return retv