o NeS@sddlmZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZdd lmZdd lmZmZmZmZmZmZdd lmZdd lmZddlZdd lmZmZddlm Z ddlm!Z!m"Z"m#Z#gdZ$e dddej%j&j'dfddZ(Gddde Z)Gddde*Z+d3ddZ,d3ddZ-ddZ.dd Z/d!d"Z0    d4d#d$Z1   d5d%d&Z2  d3d'd(Z3d)d*Z4d6d+d,Z5d7d-d.Z6d/d0Z7d6d1d2Z8dS)8)print_functionN) DataFeeder) BlockGuard) templatedoc)core) global_scope)convert_np_dtype_to_dtype_default_main_programdefault_startup_program program_guardProgramVariable) LayerHelper)generate) check_dtype check_type) static_only)_get_paddle_place_current_expected_place_set_expected_place)data read_file double_buffer py_readercreate_py_reader_by_dataloadTfloat32c Cstd it}t|dtjtjfdt|dttfdt|}tj t |D]}||dur7d||<d}q(||dkr?d}q(|rGdg|}|j ||||||dd } | S) a **Data Layer** This operator creates the global variable. The global variables can be accessed by all the following operators in the graph. Note: :code:`paddle.fluid.layers.data` is deprecated as it will be removed in a later version. Please use :code:`paddle.fluid.data` . This :code:`paddle.fluid.layers.data` set shape and dtype at compile time but does NOT check the shape or the dtype of fed data, the :code:`paddle.fluid.data` checks the shape and the dtype of data fed by Executor or ParallelExecutor during run time. To feed variable size inputs, users can feed variable size inputs directly to this :code:`paddle.fluid.layers.data` and PaddlePaddle will fit the size accordingly. Or set -1 on the variable dimension when using :code:`paddle.fluid.data` . The default :code:`stop_gradient` attribute of the Variable created by this API is true, which means the gradient won't be passed backward through the data Varaible. Set :code:`var.stop_gradient = False` If user would like to pass backward gradient. Args: name(str): The name/alias of the variable, see :ref:`api_guide_Name` for more details. shape(list|tuple): Tuple declaring the shape. If :code:`append_batch_size` is True and there is no -1 inside :code:`shape`, it should be considered as the shape of the each sample. Otherwise, it should be considered as the shape of the batched data. append_batch_size(bool): 1. If true, it prepends -1 to the shape. For example if shape=[1], the resulting shape is [-1, 1]. This will be useful to set different batch size at run time. 2. If shape contains -1, such as shape=[1, -1]. append_batch_size will be enforced to be be False (ineffective) because PaddlePaddle cannot set more than 1 unknown number on the shape. dtype(np.dtype|VarType|str): The type of the data. Supported dtype: bool, float16, float32, float64, int8, int16, int32, int64, uint8. type(VarType): The output type. Supported dtype: VarType.LOD_TENSOR, VarType.SELECTED_ROWS, VarType.NCCL_ID. Default: VarType.LOD_TENSOR. lod_level(int): The LoD Level. 0 means the input data is not a sequence. Default: 0. stop_gradient(bool): A boolean that mentions whether gradient should flow. Default: True. Returns: The global variable that gives access to the data. Return Type: Variable Examples: .. code-block:: python import paddle.fluid as fluid data = fluid.layers.data(name='x', shape=[784], dtype='float32') rnameshapeNFrT)rr dtypetype stop_gradient lod_levelZis_data)r) rlocalsrsix binary_type text_typelisttuplemovesrangelenZcreate_global_variable) rr Zappend_batch_sizer"r%r#r$helperiZdata_varr1FD:\Projects\ConvertPro\env\Lib\site-packages\paddle/fluid/layers/io.pyr+s,E   rcs,eZdZdZfddZfddZZS)BlockGuardServzl BlockGuardServ class. BlockGuardServ class is used to create an op with a block in a program. cs0t|ts tdtt||jj||_dS)Nz$BlockGuardServ takes a ListenAndServ) isinstance ListenAndServ TypeErrorsuperr3__init__r/ main_programserver)selfr: __class__r1r2r8s  zBlockGuardServ.__init__cs*|durdS|jtt||||S)NF)r: complete_opr7r3__exit__)r;exc_typeexc_valexc_tbr<r1r2r?s zBlockGuardServ.__exit__)__name__ __module__ __qualname____doc__r8r? __classcell__r1r1r<r2r3s r3c@s:eZdZdZdddZddZdd Zd d Zd d ZdS)r5a **ListenAndServ Layer** ListenAndServ is used to create a rpc server bind and listen on specific TCP port, this server will run the sub-block when received variables from clients. Args: endpoint(string): IP:port string which the server will listen on. inputs(list): a list of variables that the server will get from clients. fan_in(int): how many client are expected to report to this server, default: 1. optimizer_mode(bool): whether to run the server as a parameter server, default: True. Examples: .. code-block:: python import paddle.fluid as fluid with fluid.program_guard(main): serv = layers.ListenAndServ( "127.0.0.1:6170", ["X"], optimizer_mode=False) with serv.do(): x = layers.data( shape=[32, 32], dtype='float32', name="X", append_batch_size=False) fluid.initializer.Constant(value=1.0)(x, main.global_block()) layers.scale(x=x, scale=10.0, out=out_var) exe = fluid.Executor(place) exe.run(main) rTcCs,td|_||_g|_||_||_||_dS)Nlisten_and_serv)rr/inputsoutputsendpointfan_inoptimizer_mode)r;rKrIrLrMr1r1r2r8s  zListenAndServ.__init__cCst|SN)r3)r;r1r1r2doszListenAndServ.doc Cs|jj}|}|}t}t}|jD]@}|jr7d|jvr6d|jvr6||jdj ||jdj q|j D]}| |D]}|| ||| |qAq:q||fS)NZGradParam) r/r9 current_block parent_blockr*opsrMrIappendr input_namesinputvar) r;r9rQrRparamsZgradsopZinameZ in_var_namer1r1r2get_params_and_gradss$  z"ListenAndServ.get_params_and_gradscCs,|jj}|j}|dksJ||}|S)Nr)r/r9rQ parent_idxblock)r;progr[rRr1r1r2rRs    zListenAndServ.parent_blockc CsVddlm}|jj}|}|}|jdd|jii|j|j |g|j dgdddS)Nr)DistributedModerHX)rKZFaninZoptimize_blocksZdistributed_modeZgrad_to_block_idr#rIrJattrs) Z$incubate.fleet.parameter_server.moder^r/r9rQrR append_oprIrKrLZSYNC)r;r^r9rQrRr1r1r2r>s  zListenAndServ.complete_opN)rT) rCrDrErFr8rOrZrRr>r1r1r1r2r5s !  r5c Cst|tksJ|durg}nt|tr|g}t|tksJ|d}tt|}td it}tj }|j dd|id|id|d||tj j j id |r`|j d d|idgid|id dSdS) ak Send variables to the server side, and get vars from server side when server have finished running server side program. Args: endpoints (str): comma separated IP:PORT pairs in the order of send_vars to send send_vars (list): variables to send to server sync (bool): whether to wait the request finish N,Sendsendr_Out endpointsepmapraZ send_barrier)re)r#r*r4rsplitsetrr&rZop_proto_and_checker_makerZkOpRoleAttrNamercZOpRoleZRPC)rhZ send_varsZ dummy_outputsyncrir/Zrpc_op_role_namer1r1r2res8       recCst|tksJ|durg}nt|tr|g}t|tksJ|d}tt|}td it}|jdd|id|i||dd|rP|jd d|id |id |S) aY Receive variables from server side Args: endpoints (str): comma separated IP:PORT pairs in the order of send_vars to send get_vars (list): vars to get from server after send completes. sync (bool): whether to wait the request finish Returns: list: list of received variables NrdRecvrecvr_rg)rhriraZ fetch_barrierrh)r#rJrb)rm) r#r*r4rrjrkrr&rc)rhZget_varsZ dummy_inputrlrir/r1r1r2rm/s,    rmcs.fddfdd}|_d_d_S)Ncst}|j}|SrN)r Zfind_varrZ get_reader)scoperWreaderr1r2__get_reader__Ys z3monkey_patch_reader_methods..__get_reader__cs SrN)resetr1)rrr1r2rs^ z*monkey_patch_reader_methods..resetT)rsr$ persistable)rqrsr1)rrrqr2monkey_patch_reader_methodsWs  rvcCsV|j|jtjjjd}|j|j|j |j |j |j d|_ |S)N)rr#T) create_varrrVarDescVarTypeZREADERdescZ set_shapesshapes set_dtypesdtypesZset_lod_levels lod_levelsru)r\rWZnew_varr1r1r2_copy_reader_var_gs rc Cs|j}i}|D]}g||<||}|D] }||||qq|j}i}|D]}g||<||}|D] }||||q6q)|j|j|||d} | S)Nra) rUrVrTrWZ output_namesoutputrcr#Z all_attrs) r\rYZinput_param_namesZ new_input_map param_name arg_namesZarg_nameZoutput_param_namesZnew_output_mapZnew_opr1r1r2_copy_reader_create_op_ps,  rc sdurOttstdttggg}g}gg} D],} | j|| j|t | j| j| j | t | j q!n.ddDddD} g}g}D]} || |t | qcdur}dgt ddD} |durtd} td}td }nd |d g} d |d g}d |d g}t| }t||d t}|j|d}|jdd| gid|gi|| | |dd|j d|_tt|}t||rt|d}j|_|_jd_ d_!d _"fddfddfdd}fdd}fdd}|__#|_$_%|_&|_'S)Nz2feed_list should be a list of Variable instead of cSg|]}t|qSr1)r .0dtr1r1r2 z_py_reader..cSsg|]}dqS)rr1rr1r1r2rsrcSrr1)int)rtr1r1r2rrZlod_tensor_blocking_queueZcreate_py_readerr_queuerqFrZblocking_queuerg) shape_concatr~r}need_check_feedranksraTcs:fdd}tj|tfd_dj_jdS)Nc szBt|D]3}t}|D]}t|tjs&t}||t|}||qjr1n |jr;nq WdSt ye}z t dtjtWYd}~dSd}~ww)Nz.Your decorated reader has raised an exception!)rrZLoDTensorArrayr4Z LoDTensorrkCPUPlacerTexitedpushclose Exceptionkillloggingwarnr'reraisesysexc_info)Zlegacy_expected_placeZtensorsarrayitemtmpex) feed_queuefuncrqr1r2__provider_thread__s.     zE_py_reader..start_provide_thread..__provider_thread__)targetargsT) threadingThreadrthreaddaemonstart)rr)rrqrr2start_provide_threads z(_py_reader..start_provide_threadcs |_dSrNtensor_providerrrpr1r2__set_tensor_provider__rtz+_py_reader..__set_tensor_provider__c stttG}|dur1g}d}tD]\}}}t|}|t||||d|d7}qdd|Dt|td}|j ddWdn1sQwYfd d }|dS) Nr)rr"r r%rcSsg|]}|jqSr1r)r feed_datar1r1r2rsz=_py_reader..__set_paddle_reader__..) feed_listplaceF)Z multi_devicesc3s&D] fddDVqdS)Ncsg|]}|qSr1r1)rZ data_nameslotsr1r2rrzZ_py_reader..__set_paddle_reader__..__tensor_provider__..r1r1Z data_names paddle_readerrr2__tensor_provider__s zF_py_reader..__set_paddle_reader__..__tensor_provider__) r rzipstrrTrrrrZdecorate_reader) rZactual_feed_listcounterr"r r%rZfeederr)rr}rr~r{rr2__set_paddle_reader__s2  z)_py_reader..__set_paddle_reader__cs<jdurjdurd_jd_dSdSdS)NTF)rrrjoinr1)current_reset_methodrqr1r2 __reset__s   z_py_reader..__reset__csjdSrNrr1)rqrr1r2 __start__"sz_py_reader..__start__)(r4r*r6rr#rTr"extendr r.r%rrzr unique_namerr rWrZinit_lod_tensor_blocking_queuer rQrwrcr|rurr rvrrsrrrrZdecorate_tensor_providerZdecorate_paddle_readerZdecorate_batch_generatorZdecorate_sample_list_generatorr)capacityr{r}r~ruse_double_bufferrrrrrr Z dtype_intZ queue_nameZ reader_nameZdouble_buffer_namerW startup_blk startup_var main_prog_varZdouble_buffer_readerrrrr1) rrr}rrr~rqr{rr2 _py_readers               !rcCstdt||||||dS)a[ :api_attr: Static Graph Create a Python reader for data feeding in Python This operator returns a Reader Variable. The Reader provides :code:`decorate_paddle_reader()` and :code:`decorate_tensor_provider()` to set a Python generator as the data source and feed the data from the data source to the Reader Variable. When :code:`Executor::Run()` is invoked in C++ side, the data from the generator would be read automatically. Unlike :code:`DataFeeder.feed()`, the data reading process and :code:`Executor::Run()` process can run in parallel using :code:`py_reader`. The :code:`start()` method of the Reader should be called when each pass begins, while the :code:`reset()` method should be called when the pass ends and :code:`fluid.core.EOFException` raises. Note: :code:`Program.clone()` method cannot clone :code:`py_reader`. You can refer to :ref:`api_fluid_Program` for more details. The :code:`read_file` call needs to be in the program block of :code:`py_reader`. You can refer to :ref:`api_fluid_layers_read_file` for more details. Args: capacity(int): The buffer capacity maintained by :code:`py_reader`. shapes(list|tuple): List of tuples which declaring data shapes. shapes[i] represents the i-th data shape. dtypes(list|tuple): List of strings which declaring data type. Supported dtype: bool, float16, float32, float64, int8, int16, int32, int64, uint8. lod_levels(list|tuple): List of ints which declaring data lod_level. name(basestring): The default value is None. Normally there is no need for user to set this property. For more information, please refer to :ref:`api_guide_Name`. use_double_buffer(bool): Whether use double buffer or not. The double buffer is for pre-reading the data of the next batch and copy the data asynchronously from CPU to GPU. Default is True. Returns: A Reader from which we can get feeding data. Return Type: Variable Examples: 1. The basic usage of :code:`py_reader` is as follows: .. code-block:: python import paddle import paddle.fluid as fluid import paddle.dataset.mnist as mnist def network(image, label): # user defined network, here a softmax regession example predict = fluid.layers.fc(input=image, size=10, act='softmax') return fluid.layers.cross_entropy(input=predict, label=label) reader = fluid.layers.py_reader(capacity=64, shapes=[(-1, 1, 28, 28), (-1, 1)], dtypes=['float32', 'int64']) reader.decorate_paddle_reader( paddle.reader.shuffle(paddle.batch(mnist.train(), batch_size=5), buf_size=1000)) img, label = fluid.layers.read_file(reader) loss = network(img, label) fluid.Executor(fluid.CUDAPlace(0)).run(fluid.default_startup_program()) exe = fluid.ParallelExecutor(use_cuda=True) for epoch_id in range(10): reader.start() try: while True: exe.run(fetch_list=[loss.name]) except fluid.core.EOFException: reader.reset() fluid.io.save_inference_model(dirname='./model', feeded_var_names=[img.name, label.name], target_vars=[loss], executor=fluid.Executor(fluid.CUDAPlace(0))) 2. When training and testing are both performed, two different :code:`py_reader` should be created with different names, e.g.: .. code-block:: python import paddle import paddle.fluid as fluid import paddle.dataset.mnist as mnist def network(reader): img, label = fluid.layers.read_file(reader) # User defined network. Here a simple regression as example predict = fluid.layers.fc(input=img, size=10, act='softmax') loss = fluid.layers.cross_entropy(input=predict, label=label) return fluid.layers.mean(loss) # Create train_main_prog and train_startup_prog train_main_prog = fluid.Program() train_startup_prog = fluid.Program() with fluid.program_guard(train_main_prog, train_startup_prog): # Use fluid.unique_name.guard() to share parameters with test program with fluid.unique_name.guard(): train_reader = fluid.layers.py_reader(capacity=64, shapes=[(-1, 1, 28, 28), (-1, 1)], dtypes=['float32', 'int64'], name='train_reader') train_reader.decorate_paddle_reader( paddle.reader.shuffle(paddle.batch(mnist.train(), batch_size=5), buf_size=500)) train_loss = network(train_reader) # some network definition adam = fluid.optimizer.Adam(learning_rate=0.01) adam.minimize(train_loss) # Create test_main_prog and test_startup_prog test_main_prog = fluid.Program() test_startup_prog = fluid.Program() with fluid.program_guard(test_main_prog, test_startup_prog): # Use fluid.unique_name.guard() to share parameters with train program with fluid.unique_name.guard(): test_reader = fluid.layers.py_reader(capacity=32, shapes=[(-1, 1, 28, 28), (-1, 1)], dtypes=['float32', 'int64'], name='test_reader') test_reader.decorate_paddle_reader(paddle.batch(mnist.test(), 512)) test_loss = network(test_reader) fluid.Executor(fluid.CUDAPlace(0)).run(train_startup_prog) fluid.Executor(fluid.CUDAPlace(0)).run(test_startup_prog) train_exe = fluid.ParallelExecutor(use_cuda=True, loss_name=train_loss.name, main_program=train_main_prog) test_exe = fluid.ParallelExecutor(use_cuda=True, loss_name=test_loss.name, main_program=test_main_prog) for epoch_id in range(10): train_reader.start() try: while True: train_exe.run(fetch_list=[train_loss.name]) except fluid.core.EOFException: train_reader.reset() test_reader.start() try: while True: test_exe.run(fetch_list=[test_loss.name]) except fluid.core.EOFException: test_reader.reset() zpaddle.fluid.layers.py_reader() may be deprecated in the near future. Please use paddle.fluid.io.DataLoader.from_generator() instead.rr{r}r~rrrrrrr1r1r2r0s rc Cs tdt|ddd|||dS)ax :api_attr: Static Graph The OP creates a Python reader for data feeding in Python, it is similar to :ref:`api_fluid_layers_py_reader` except that it can read data from the list of feed variables. Parameters: capacity (int): The buffer capacity maintained by :code:`py_reader`. Its unit is batch number. Set larger :attr:`capacity` if the reader is fast. feed_list (list(Variable)): The feed variables, are usually created by :code:`fluid.data()`. name (str, optional): Normally there is no need for user to set this property. For more information, please refer to :ref:`api_guide_Name`. Default: None. use_double_buffer (bool, optional): Whether use double buffer. If it's True, the OP would prefetch next batch data asynchronously. Default: True. Returns: Reader: A Reader for data feeding. The data types of read data are the same as the data types of variables of :attr:`feed_list`. Examples: .. code-block:: python import paddle import paddle.fluid as fluid import paddle.dataset.mnist as mnist def network(img, label): # User defined network. Here a simple regression as example predict = fluid.layers.fc(input=img, size=10, act='softmax') loss = fluid.layers.cross_entropy(input=predict, label=label) return fluid.layers.mean(loss) MEMORY_OPT = False USE_CUDA = False image = fluid.data(name='image', shape=[None, 1, 28, 28], dtype='float32') label = fluid.data(name='label', shape=[None, 1], dtype='int64') reader = fluid.layers.create_py_reader_by_data(capacity=64, feed_list=[image, label]) reader.decorate_paddle_reader( paddle.reader.shuffle(paddle.batch(mnist.train(), batch_size=5), buf_size=500)) img, label = fluid.layers.read_file(reader) loss = network(img, label) # The definition of custom network and the loss function place = fluid.CUDAPlace(0) if USE_CUDA else fluid.CPUPlace() exe = fluid.Executor(place) exe.run(fluid.default_startup_program()) build_strategy = fluid.BuildStrategy() build_strategy.memory_optimize = True if MEMORY_OPT else False exec_strategy = fluid.ExecutionStrategy() compiled_prog = fluid.compiler.CompiledProgram( fluid.default_main_program()).with_data_parallel( loss_name=loss.name, build_strategy=build_strategy, exec_strategy=exec_strategy) for epoch_id in range(2): reader.start() try: while True: exe.run(compiled_prog, fetch_list=[loss.name]) except fluid.core.EOFException: reader.reset() zpaddle.fluid.layers.create_py_reader_by_data() may be deprecated in the near future. Please use paddle.fluid.io.DataLoader.from_generator() instead.N)rr{r}r~rrrr)rrrrr1r1r2rsFrc Csft|}t}|j|d}|j|d|id|gi|d}d|_t}t||}t||t |S)NrUnderlyingReaderrgraT) rr rQrwrcrur rrrv) op_typerqrbvar_namerrZ startop_opZmain_prog_blockrr1r1r2"__create_shared_decorated_reader__,s     rcCsN|dur|nt|}t}|j|d}|j|d|id|gi|dt|S)Nrrrgra)rr rQrwrcrv)rrqrbrZnew_reader_nameZmain_blkZ new_readerr1r1r2$__create_unshared_decorated_reader__;s  rcCs2t}|durtt||d<td|||dS)a Wrap a double buffer reader. The class Reader contains DecoratedReader and FileReader. Moreover, the DecoratedReader is inherited by CustomReader and BufferedReader. This function is related to BufferedReader. The data will copy to target place with a double buffer queue. If the target place is None, the place that executor perform on will be used. Args: reader (Variable): The Reader Variable need to be wrapped. place (Place|str, optional): The place of target data, such as CPU, GPU, and if use GPU, it's necessary to point out which card is involved. Default is the sample place of executor perform. if ``place`` is string, It can be ``cpu``, ``gpu:x``, where ``x`` is the ndex of the GPUs. name (str, optional): Variable name. Normally there is no need for user to set this property. For more information, please refer to :ref:`api_guide_Name`. Default is None. Returns: Variable(Reader): wrapped reader with double buffer. Examples: .. code-block:: python import paddle.fluid as fluid reader = fluid.layers.py_reader(capacity=64, shapes=[(-1, 1, 28, 28), (-1, 1)], dtypes=['float32', 'int64'], use_double_buffer=False) reader = fluid.layers.double_buffer(reader) image, label = fluid.layers.read_file(reader) NrZcreate_double_buffer_readerr)dictrrupperr)rqrrrbr1r1r2rFsrcsZtdfddtt|jD}jdd|gid|idt|dkr+|d S|S) a :api_attr: Static Graph Execute the given reader and get data via it. A reader is also a Variable. It can be a raw reader generated by `fluid.layers.open_files()` or a decorated one generated by `fluid.layers.double_buffer()` . Args: reader(Variable): The reader to execute. Returns: Tuple[Variable]: Data read from the given reader. Examples: .. code-block:: python import paddle.fluid as fluid reader = fluid.layers.py_reader(capacity=64, shapes=[(-1, 1, 28, 28), (-1, 1)], dtypes=['float32', 'int64']) image, label = fluid.layers.read_file(reader) rcsg|] }jdddqS)Tr)r$r")Z"create_variable_for_type_inference)rrr/r1r2rs zread_file..readReaderrg)r#rIrJrr)rr-r.rzr{rc)rqoutr1rr2ris  rcCsBtdit}d|i}|dur||d<|jdid|i|ddS)a Load operator will load a LoDTensor / SelectedRows variable from disk file. Args: out(Variable): The LoDTensor / SelectedRows need to be loaded.. file_path(STRING): Variable will be loaded from "file_path". load_as_fp16(BOOLEAN): If true, the tensor will be first loaded and then converted to float16 data type. Otherwise, the tensor will be directly loaded without data type conversion. Default is false.. Returns: None Examples: .. code-block:: python import paddle.fluid as fluid tmp_tensor = fluid.layers.create_tensor(dtype='float32') fluid.layers.load(tmp_tensor, "./tmp_tensor.bin") r file_pathN load_as_fp16rgra)r)rr&rc)rrrr/rbr1r1r2rs r)NT)NNTN)NNTrN)NN)9 __future__rmultiprocessingosr'rrZ data_feederrZ control_flowrZlayer_function_generatorrr`rexecutorr Z frameworkr r r r rrZ layer_helperrrrrrrZpaddle.fluid.frameworkrrrr__all__rxryZ LOD_TENSORrr3objectr5rermrvrrrrrrrrrrr1r1r1r2sh           ^ c -(  , - R  #)