o Me3b@szddlmZddlmZddlZddlZddlZddlm Z m Z m Z ddl Z ddl Z ddlmZmZmZmZmZddlmZmZdgZejjjd ejjjd ejjjd ejjjd ejjjd ejjjdejjjdejjj dejjj!dejjj"dejjj#dejjj$di Z%ddZ& d&ddZ'd&ddZ( d&ddZ)e*e+efe,efdfdd Z-Gd!d"d"e.Z/Gd#d$d$e.Z0Gd%dde.Z1dS)')print_function)coreN)ziprangexrange)Variabledefault_main_program_current_expected_place_non_static_mode_in_eager_without_dygraph_check)_cpu_num _cuda_ids DataFeederboolfloat16uint16float32float64int8int16int32int64uint8 complex64 complex128c Cst|tjjr|tvrt|Sn2t|tr4|ttjtj tj tj tj tj tjtjtjtjtjf vr3|jSn|dvrr1Zexpected_shape_typeZexpected_element_typeZexpected_tensor_dtypeitemr'r'r( check_shapes" rAc@s<eZdZddZddZddZddZd d Zd d Zd S)DataToLoDTensorConvertercCsZ||_||_||_d}|jD]}|dkr|d7}|dkr!d|_nqt||_|dSNrr)place lod_levelr>r)r&_reset)selfrDrEr>r&Z negtive_countsr'r'r(__init__s   z!DataToLoDTensorConverter.__init__cCs$g|_ddtj|jD|_dS)NcSsg|]}gqSr'r'.0_r'r'r( sz3DataToLoDTensorConverter._reset..)datasixmovesrrElodrGr'r'r(rFszDataToLoDTensorConverter._resetcCs|||j|jdSr+) _feed_impl_rQrE)rGrNr'r'r(feedszDataToLoDTensorConverter.feedcCsR|dkr |j|dS|dt||D]}|||dd|dqdSrC)rNappendlenrS)rGrNrQrEZ each_datar'r'r(rSs z$DataToLoDTensorConverter._feed_impl_cCsDt|j|D]\}}||kr|dkr|dkrtd|j|qdS)NrzDShape not match. What is defined in data layer is {}, but receive {})rr> ValueErrorr7)rGr>s1s2r'r'r( _check_shapes z%DataToLoDTensorConverter._check_shapecCstj|j|jd}|jr0t|jt|jkr0z||j}Wnty/td|j|jwt }| ||j |j dkrF||j||S)Nr%zBReshape error. What is defined in data layer is {}, but receive {}r)r!arrayrNr&r>rVr:rWr7rZ LoDTensorsetrDrEZset_recursive_sequence_lengthsrQrF)rGZarrtr'r'r(dones"    zDataToLoDTensorConverter.doneN) r" __module__ __qualname__rIrFrTrSrZr^r'r'r'r(rBs rBc@s$eZdZddZddZddZdS)BatchedTensorProviderc Cs\||_||_||_g|_||_|D]}|jdksJd|jt|jd|j|j dqdS)Nrzlod_level must be 0rDrEr>r&) rD batch_size generator converters drop_lastrErUrBr>r&)rG feed_listrDrcrdrfvarr'r'r(rIszBatchedTensorProvider.__init__cCsdd|jDS)NcSg|]}|qSr')r^rKcr'r'r(rMz/BatchedTensorProvider._done..)rerRr'r'r(_doneszBatchedTensorProvider._doneccsd}|D]%}tj||jD] \}}|j|q|d7}||jkr,d}|Vq|j s;|dkr;|VdSdd|jDdS)NrrcSrir')rFrjr'r'r(rM rlz2BatchedTensorProvider.__call__..) rdrOrPrrerNrUrcrmrf)rGidx each_sample each_sloteach_converterr'r'r(__call__s    zBatchedTensorProvider.__call__N)r"r_r`rIrmrrr'r'r'r(ras rac@sBeZdZdZdddZddZdddZd d Z  dd d ZdS)ra4 :api_attr: Static Graph DataFeeder converts the data that returned by a reader into a data structure that can feed into Executor. The reader is usually a python generator that returns a list of mini-batch data entries. Parameters: feed_list (list): Variables or names of Variables that need to feed. place (:ref:`api_fluid_CPUPlace` | :ref:`api_fluid_CUDAPlace` ): place indicates the device (CPU | GPU) the data will be fed into, if you want to feed data into GPU, please using :code:`fluid.CUDAPlace(i)` (:code:`i` represents the GPU id), or if you want to feed data into CPU, please using :code:`fluid.CPUPlace()`. program (:ref:`api_fluid_Program` , optional): The Program that will feed data into, if program is None, it will use default_main_program(). Default None. Raises: :code:`ValueError` - If some Variables are not in this Program. Example: .. code-block:: python import numpy as np import paddle import paddle.fluid as fluid place = fluid.CPUPlace() def reader(): for _ in range(4): yield np.random.random([4]).astype('float32'), np.random.random([3]).astype('float32'), main_program = fluid.Program() startup_program = fluid.Program() with fluid.program_guard(main_program, startup_program): data_1 = fluid.data(name='data_1', shape=[None, 2, 2], dtype='float32') data_2 = fluid.data(name='data_2', shape=[None, 1, 3], dtype='float32') out = fluid.layers.fc(input=[data_1, data_2], size=2) # ... feeder = fluid.DataFeeder([data_1, data_2], place) exe = fluid.Executor(place) exe.run(startup_program) feed_data = feeder.feed(reader()) # print feed_data to view feed results # print(feed_data['data_1']) # print(feed_data['data_2']) outs = exe.run(program=main_program, feed=feed_data, fetch_list=[out]) print(outs) NcCsg|_g|_g|_g|_|durt}|D]5}t|tjr%|d |}t|t s.t d|j |j |j |j|j |j|j |jq||_dS)Nrz+Feed list should contain a list of variable) feed_dtypes feed_names feed_shapesfeed_lod_levelr rrO string_typesblockrhrr$rUr&namerEr>rD)rGrgrDprogramZeach_varr'r'r(rIJs    zDataFeeder.__init__c Csg}tj|j|j|jD]\}}}|t|j|||dq |D]'}t |t |ks7Jdt |t |ftj||D] \}}| |q>q!i} tj|j |D] \} }| | | <qS| S)a According to :code:`feed_list` of :code:`DataFeeder` and :code:`iterable` , converts the input into a data structure that can feed into Executor. Parameters: iterable (generator): user defined python generator to read the raw input data Returns: :code:`dict`: a :code:`dict` that contains (variable name - converted tensor) pairs Example: .. code-block:: python # In this example, reader - generator will return a list of ndarray of 3 elements # feed API will convert each ndarray input into a tensor # the return result is a dict with keys: data_1, data_2, data_3 # result['data_1'] a LoD-Tensor with shape of [5, 2, 1, 3]. 5 is batch size, and [2, 1, 3] is the real shape of data_1. # result['data_2'], result['data_3'] are similar. import numpy as np import paddle.fluid as fluid def reader(limit=5): for i in range(1, limit + 1): yield np.ones([6]).astype('float32') * i , np.ones([1]).astype('int64') * i, np.random.random([9]).astype('float32') data_1 = fluid.data(name='data_1', shape=[None, 2, 1, 3]) data_2 = fluid.data(name='data_2', shape=[None, 1], dtype='int64') data_3 = fluid.data(name='data_3', shape=[None, 3, 3], dtype='float32') feeder = fluid.DataFeeder(['data_1','data_2', 'data_3'], fluid.CPUPlace()) result = feeder.feed(reader()) print(result['data_1']) print(result['data_2']) print(result['data_3']) rbzDThe number of fields in data (%d) does not match len(feed_list) (%d)) rOrPrrvrursrUrBrDrVrTrtr^) rGiterable converterrEr>r&rorqrpZret_dictZ each_namer'r'r(rT]s8&    zDataFeeder.feedccst|jtjrddtj||D}nddtj||D}t|t|kr1t d|j}tj ||D] \}}||_| |Vq;||_dS)a Similar with feed function, feed_parallel is used with multiple devices (CPU|GPU). Here :code:`iterable` is a list of python generators. The data return by each generator in the list will be fed into a separate device. Parameters: iterable (list|tuple): list of user-defined python generators. The element number should match the :code:`num_places`. num_places (int, optional): the number of devices. If not provided (None), all available devices on the machine will be used. Default None. Returns: :code:`generator`: a :code:`generator` that generate dict which contains (variable name - converted tensor) pairs, the total number of dicts will be generated matches with the :code:`num_places` .. note:: The number of devices - :code:`num_places` should equal to the generator (element of :code:`iterable` ) number Example: .. code-block:: python import numpy as np import paddle.fluid as fluid def generate_reader(batch_size, base=0, factor=1): def _reader(): for i in range(batch_size): yield np.ones([4]) * factor + base, np.ones([4]) * factor + base + 5 return _reader() x = fluid.data(name='x', shape=[None, 2, 2]) y = fluid.data(name='y', shape=[None, 2, 2], dtype='float32') z = fluid.layers.elementwise_add(x, y) feeder = fluid.DataFeeder(['x','y'], fluid.CPUPlace()) place_num = 2 places = [fluid.CPUPlace() for x in range(place_num)] data = [] exe = fluid.Executor(fluid.CPUPlace()) exe.run(fluid.default_startup_program()) program = fluid.CompiledProgram(fluid.default_main_program()).with_data_parallel(places=places) # print sample feed_parallel r result # for item in list(feeder.feed_parallel([generate_reader(5, 0, 1), generate_reader(3, 10, 2)], 2)): # print(item['x']) # print(item['y']) reader_list = [generate_reader(5, 0, 1), generate_reader(3, 10, 2)] res = exe.run(program=program, feed=list(feeder.feed_parallel(reader_list, 2)), fetch_list=[z]) print(res) cSsg|]}t|qSr')r CUDAPlace)rKir'r'r(rMs z,DataFeeder.feed_parallel..cSsg|]}tqSr')rZCPUPlacerJr'r'r(rMszfeed_parallel takes multiple mini-batches. Each mini-batch will be feed on each device. The number of devices and number of mini-batches must be same.N) rrDrr}rOrPr_get_number_of_places_rVrWrrT)rGr{ num_placesZplacesrDpbatchr'r'r( feed_parallels&6 zDataFeeder.feed_parallelcCs.|durt|St|jtjrttStSr+)intrrDrr}rVrr )rGrr'r'r(rs  z!DataFeeder._get_number_of_places_Tcsfdd}|S)a Decorate the reader (generator) to fit multiple devices. The reader generate multiple mini-batches. Each mini-batch will be fed into a single device. Parameters: reader(generator): a user defined python generator used to get :code:`mini-batch` of data. A :code:`mini-batch` can be regarded as a python generator that returns batches of input entities, just like the below :code:`_mini_batch` in the code example. multi_devices(bool): indicate whether to use multiple devices or not. num_places(int, optional): if :code:`multi_devices` is True, you can specify the number of devices(CPU|GPU) to use, if multi_devices is None, the function will use all the devices of the current machine. Default None. drop_last(bool, optional): whether to drop the last round of data if it is not enough to feed all devices. Default True. Returns: :code:`generator`: a new :code:`generator` which return converted dicts that can be fed into Executor Raises: :code:`ValueError`: If drop_last is False and the data cannot fit devices perfectly. Example: .. code-block:: python import numpy as np import paddle import paddle.fluid as fluid import paddle.fluid.compiler as compiler def reader(): def _mini_batch(batch_size): for i in range(batch_size): yield np.random.random([16]).astype('float32'), np.random.randint(10, size=[1]) for _ in range(10): yield _mini_batch(np.random.randint(1, 10)) place_num = 3 places = [fluid.CPUPlace() for _ in range(place_num)] # a simple network sample data = fluid.data(name='data', shape=[None, 4, 4], dtype='float32') label = fluid.data(name='label', shape=[None, 1], dtype='int64') hidden = fluid.layers.fc(input=data, size=10) feeder = fluid.DataFeeder(place=places[0], feed_list=[data, label]) reader = feeder.decorate_reader(reader, multi_devices=True, num_places=3, drop_last=True) exe = fluid.Executor(places[0]) exe.run(fluid.default_startup_program()) compiled_prog = compiler.CompiledProgram( fluid.default_main_program()).with_data_parallel(places=places) for i,data in enumerate(reader()): # print data if you like # print(i, data) ret = exe.run(compiled_prog, feed=data, fetch_list=[hidden]) print(ret) c3ssD]}|VqdS}g}D]}||t||kr3t||Vg}qs@t|dkrBtddSdS)NrzwThe data batch which cannot fit for devices will be dropped is not implementation. Other strategies are not implemented)rTrrUrVlistrrW)r@numrrf multi_devicesrreaderrGr'r(__reader_creator__1s$     z6DataFeeder.decorate_reader..__reader_creator__r')rGrrrrfrr'rr(decorate_readersBzDataFeeder.decorate_readerr+)NT) r"r_r`__doc__rIrTrrrr'r'r'r(r s < =M )r*)2 __future__rr*rnumpyr!osrOZ six.movesrrrmultiprocessingr<Z frameworkrr r r r r r__all__rrBOOLZFP16ZBF16ZFP32ZFP64ZINT8ZINT16ZINT32ZINT64ZUINT8Z COMPLEX64Z COMPLEX128rr)r3r,r-rr6rrAobjectrBrarr'r'r'r(sJ              & +  8&