io.py 46.0 KB
Newer Older
1
#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
D
dzhwinter 已提交
2
#
D
dzhwinter 已提交
3 4 5
# 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
D
dzhwinter 已提交
6
#
D
dzhwinter 已提交
7
#     http://www.apache.org/licenses/LICENSE-2.0
D
dzhwinter 已提交
8
#
D
dzhwinter 已提交
9 10 11 12 13
# 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.
14 15

from __future__ import print_function
S
rename  
sneaxiy 已提交
16
from ..wrapped_decorator import signature_safe_contextmanager
17
import multiprocessing
P
peizhilin 已提交
18
import os
M
minqiyang 已提交
19
import six
Y
yuyang18 已提交
20
import threading
D
dzhwinter 已提交
21

Y
yuyang18 已提交
22
from ..data_feeder import DataFeeder
23 24
from .control_flow import BlockGuard
from .layer_function_generator import templatedoc
Y
yuyang18 已提交
25
from .. import core
Y
Refine  
Yu Yang 已提交
26
from ..executor import global_scope
Y
yuyang18 已提交
27
from ..framework import convert_np_dtype_to_dtype_, default_main_program, \
28
    default_startup_program, program_guard, Program, Variable
Y
yuyang18 已提交
29 30
from ..layer_helper import LayerHelper
from ..unique_name import generate as unique_name
Y
Yu Yang 已提交
31

Y
Yu Yang 已提交
32
__all__ = [
Y
yuyang 已提交
33
    'data', 'open_files', 'read_file', 'shuffle', 'batch', 'double_buffer',
Q
Qiao Longfei 已提交
34 35
    'random_data_generator', 'py_reader', 'create_py_reader_by_data',
    'Preprocessor', 'load'
Y
Yu Yang 已提交
36
]
Y
Yu Yang 已提交
37 38 39 40 41 42 43 44 45 46


def data(name,
         shape,
         append_batch_size=True,
         dtype='float32',
         lod_level=0,
         type=core.VarDesc.VarType.LOD_TENSOR,
         stop_gradient=True):
    """
K
kavyasrinet 已提交
47
    **Data Layer**
Y
Yu Yang 已提交
48

K
kavyasrinet 已提交
49
    This function takes in the input and based on whether data has
C
caoying03 已提交
50
    to be returned back as a minibatch, it creates the global variable by using
Y
Yu Yang 已提交
51
    the helper functions. The global variables can be accessed by all the
C
caoying03 已提交
52
    following operators in the graph.
Y
Yu Yang 已提交
53 54 55 56

    All the input variables of this function are passed in as local variables
    to the LayerHelper constructor.

57 58 59 60 61
    Notice that paddle would only use :code:`shape` to infer the shapes of 
    following variables in the network during compile-time. During run-time, 
    paddle would not check whether the shape of the feeded data matches the 
    :code:`shape` settings in this function. 

K
kavyasrinet 已提交
62 63
    Args:
       name(str): The name/alias of the function
S
sneaxiy 已提交
64 65 66 67
       shape(list): 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.  
X
Xin Pan 已提交
68 69
       append_batch_size(bool):
          1. If true, it prepends -1 to the shape.
70 71 72 73 74 75
            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.
76
       dtype(np.dtype|VarType|str): The type of data : float32, float16, int etc
K
kavyasrinet 已提交
77 78 79 80 81 82 83 84 85 86 87
       type(VarType): The output type. By default it is LOD_TENSOR.
       lod_level(int): The LoD Level. 0 means the input data is not a sequence.
       stop_gradient(bool): A boolean that mentions whether gradient should flow.

    Returns:
        Variable: The global variable that gives access to the data.

    Examples:
        .. code-block:: python

          data = fluid.layers.data(name='x', shape=[784], dtype='float32')
Y
Yu Yang 已提交
88 89 90
    """
    helper = LayerHelper('data', **locals())
    shape = list(shape)
M
minqiyang 已提交
91
    for i in six.moves.range(len(shape)):
Y
Yu Yang 已提交
92 93 94 95 96 97 98 99 100
        if shape[i] is None:
            shape[i] = -1
            append_batch_size = False
        elif shape[i] < 0:
            append_batch_size = False

    if append_batch_size:
        shape = [-1] + shape  # append batch size as -1

Y
Yu Yang 已提交
101
    data_var = helper.create_global_variable(
Y
Yu Yang 已提交
102 103 104 105 106
        name=name,
        shape=shape,
        dtype=dtype,
        type=type,
        stop_gradient=stop_gradient,
F
fengjiayi 已提交
107 108
        lod_level=lod_level,
        is_data=True)
Y
Yu Yang 已提交
109
    return data_var
T
typhoonzero 已提交
110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134


class BlockGuardServ(BlockGuard):
    """
    BlockGuardServ class.

    BlockGuardServ class is used to create an op with a block in a program.
    """

    def __init__(self, server):
        if not (isinstance(server, ListenAndServ)):
            raise TypeError("BlockGuardServ takes a ListenAndServ")
        super(BlockGuardServ, self).__init__(server.helper.main_program)
        self.server = server

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            return False

        self.server.complete_op()
        return super(BlockGuardServ, self).__exit__(exc_type, exc_val, exc_tb)


class ListenAndServ(object):
    """
Y
yi.wu 已提交
135
    **ListenAndServ Layer**
T
typhoonzero 已提交
136

Y
yi.wu 已提交
137 138 139 140 141 142 143 144 145
    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.
Y
update  
yi.wu 已提交
146

Y
yi.wu 已提交
147 148 149 150 151 152 153 154 155 156 157 158 159 160 161
    Examples:
        .. code-block:: python

            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)

Y
yi.wu 已提交
162 163
            exe = fluid.Executor(place)
            exe.run(main)
T
typhoonzero 已提交
164 165
    """

Y
Yancey1989 已提交
166
    def __init__(self, endpoint, inputs, fan_in=1, optimizer_mode=True):
167
        self.helper = LayerHelper("listen_and_serv")
Y
Yancey1989 已提交
168
        self.inputs = inputs
T
typhoonzero 已提交
169 170 171
        self.outputs = []
        self.endpoint = endpoint
        self.fan_in = fan_in
T
typhoonzero 已提交
172 173
        # FIXME(typhoonzero): add optimizer_mode is stupid, should make it more
        # general.
T
WIP  
typhoonzero 已提交
174
        self.optimizer_mode = optimizer_mode
T
typhoonzero 已提交
175 176 177 178 179 180 181 182 183 184 185 186 187

    def do(self):
        return BlockGuardServ(self)

    def get_params_and_grads(self):
        main_program = self.helper.main_program
        current_block = main_program.current_block()
        parent_block = self.parent_block()
        # params and grads in the same order.
        params = list()
        grads = list()
        for op in current_block.ops:
            # FIXME(typhoonzero): op.inputs is None if it's cloned.
T
WIP  
typhoonzero 已提交
188 189 190 191 192 193 194 195
            if self.optimizer_mode:
                if "Grad" in op.inputs and "Param" in op.inputs:
                    params.append(op.inputs["Param"].name)
                    grads.append(op.inputs["Grad"].name)
            else:
                # simple recv mode, recv operators inputs.
                for iname in op.input_names:
                    for in_var_name in op.input(iname):
T
typhoonzero 已提交
196 197
                        params.append(parent_block.var(in_var_name))
                        grads.append(parent_block.var(in_var_name))
T
typhoonzero 已提交
198 199 200

        return params, grads

T
typhoonzero 已提交
201 202 203 204 205 206 207
    def parent_block(self):
        prog = self.helper.main_program
        parent_idx = prog.current_block().parent_idx
        assert parent_idx >= 0
        parent_block = prog.block(parent_idx)
        return parent_block

T
typhoonzero 已提交
208 209 210 211 212 213
    def complete_op(self):
        main_program = self.helper.main_program
        current_block = main_program.current_block()
        parent_block = self.parent_block()

        parent_block.append_op(
214
            type='listen_and_serv',
Y
Yancey1989 已提交
215
            inputs={"X": self.inputs},
T
typhoonzero 已提交
216 217 218 219
            outputs={},
            attrs={
                'endpoint': self.endpoint,
                'Fanin': self.fan_in,
Y
Yancey1989 已提交
220 221 222
                'optimize_blocks': [
                    current_block
                ],  # did not support multiple optimize blocks in layers
223
                'sync_mode': True,  # did not support async now in layers
Q
qiaolongfei 已提交
224
                'grad_to_block_id': [""]
T
typhoonzero 已提交
225 226 227
            })


228
def Send(endpoints, send_vars, dummy_output=None, sync=True):
T
typhoonzero 已提交
229
    """
Y
yi.wu 已提交
230 231
    Send variables to the server side, and get vars from server
    side when server have finished running server side program.
T
typhoonzero 已提交
232 233

    Args:
Y
yi.wu 已提交
234
        endpoints (str): comma seperated IP:PORT pairs in the order
T
typhoonzero 已提交
235
                   of send_vars to send
Y
yi.wu 已提交
236 237
        send_vars (list): variables to send to server
        sync (bool): whether to wait the request finish
T
typhoonzero 已提交
238 239 240 241

    """
    assert (type(send_vars) == list)

242 243 244 245 246 247 248
    if dummy_output is None:
        dummy_output = []
    elif isinstance(dummy_output, Variable):
        dummy_output = [dummy_output]

    assert (type(dummy_output) == list)

T
typhoonzero 已提交
249
    epmap = endpoints.split(",")
T
typhoonzero 已提交
250
    endpoints = list(set(epmap))
T
typhoonzero 已提交
251 252

    helper = LayerHelper("Send", **locals())
Y
Yancey1989 已提交
253
    rpc_op_role_name = core.op_proto_and_checker_maker.kOpRoleAttrName()
Y
Yancey1989 已提交
254

T
typhoonzero 已提交
255 256 257
    helper.append_op(
        type="send",
        inputs={"X": send_vars},
258
        outputs={"Out": dummy_output},
Y
Yancey1989 已提交
259 260 261 262 263
        attrs={
            "endpoints": endpoints,
            "epmap": epmap,
            rpc_op_role_name: core.op_proto_and_checker_maker.OpRole.RPC
        })
Y
yi.wu 已提交
264
    if sync:
W
Wu Yi 已提交
265 266 267 268 269
        helper.append_op(
            type="send_barrier",
            inputs={"X": dummy_output},
            outputs={"Out": []},
            attrs={"endpoints": endpoints})
270 271


272
def Recv(endpoints, get_vars, dummy_input=None, sync=True):
273
    """
Y
yi.wu 已提交
274
    Receive variables from server side
275 276

    Args:
Y
yi.wu 已提交
277
        endpoints (str): comma seperated IP:PORT pairs in the order
278
                   of send_vars to send
Y
yi.wu 已提交
279 280
        get_vars (list): vars to get from server after send completes.
        sync (bool): whether to wait the request finish
281

Y
yi.wu 已提交
282 283
    Returns:
        list: list of received variables
284 285 286
    """
    assert (type(get_vars) == list)

287 288 289 290 291 292 293
    if dummy_input is None:
        dummy_input = []
    elif isinstance(dummy_input, Variable):
        dummy_input = [dummy_input]

    assert (type(dummy_input) == list)

294 295 296 297 298 299
    epmap = endpoints.split(",")
    endpoints = list(set(epmap))

    helper = LayerHelper("Recv", **locals())
    helper.append_op(
        type="recv",
300
        inputs={"X": dummy_input},
301 302 303
        outputs={"Out": get_vars},
        attrs={"endpoints": endpoints,
               "epmap": epmap})
Y
yi.wu 已提交
304
    if sync:
W
Wu Yi 已提交
305 306 307 308
        helper.append_op(
            type="fetch_barrier",
            outputs={"Out": get_vars},
            attrs={"endpoints": endpoints})
Y
yi.wu 已提交
309
    return get_vars
Y
Yu Yang 已提交
310 311


Y
Refine  
Yu Yang 已提交
312 313 314 315 316 317 318 319 320 321
def monkey_patch_reader_methods(reader):
    def __get_reader__():
        scope = global_scope()
        var = scope.find_var(reader.name)
        return var.get_reader()

    def reset():
        return __get_reader__().reset()

    reader.reset = reset
Y
Yu Yang 已提交
322 323
    reader.stop_gradient = True
    reader.persistable = True
Y
Refine  
Yu Yang 已提交
324 325 326
    return reader


Y
Yu Yang 已提交
327 328 329 330
def _copy_reader_var_(block, var):
    new_var = block.create_var(name=var.name, type=core.VarDesc.VarType.READER)
    new_var.desc.set_shapes(var.desc.shapes())
    new_var.desc.set_dtypes(var.desc.dtypes())
S
sneaxiy 已提交
331
    new_var.desc.set_lod_levels(var.desc.lod_levels())
Y
Yu Yang 已提交
332
    new_var.persistable = True
F
fengjiayi 已提交
333 334 335 336
    return new_var


def _copy_reader_create_op_(block, op):
F
fengjiayi 已提交
337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352
    input_param_names = op.input_names
    new_input_map = {}
    for param_name in input_param_names:
        new_input_map[param_name] = []
        arg_names = op.input(param_name)
        for arg_name in arg_names:
            new_input_map[param_name].append(block.var(arg_name))

    output_param_names = op.output_names
    new_output_map = {}
    for param_name in output_param_names:
        new_output_map[param_name] = []
        arg_names = op.output(param_name)
        for arg_name in arg_names:
            new_output_map[param_name].append(block.var(arg_name))

F
fengjiayi 已提交
353
    new_op = block.append_op(
F
fengjiayi 已提交
354 355 356
        type=op.type,
        inputs=new_input_map,
        outputs=new_output_map,
J
JiayiFeng 已提交
357
        attrs=op.all_attrs())
F
fengjiayi 已提交
358
    return new_op
Y
Yu Yang 已提交
359 360


W
wopeizl 已提交
361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387
@templatedoc(op_type='create_recordio_file_reader')
def open_recordio_file(filename,
                       shapes,
                       lod_levels,
                       dtypes,
                       pass_num=1,
                       for_parallel=True):
    """
    ${comment}

    Args:
       filename(${filename_type}): ${filename_comment}.
       shapes(list): List of tuples which declaring data shapes.
       lod_levels(${lod_levels_type}): ${lod_levels_comment}.
       dtypes(list): List of strs which declaring data type.
       pass_num(int): Number of passes to run.
       for_parallel(Bool): Set it as True if you are going to run
            subsequent operators in parallel.

    Returns:
       ${out_comment}.

    Examples:

        >>> import paddle.fluid as fluid
        >>> reader = fluid.layers.io.open_recordio_file(
        >>>                               filename='./data.recordio',
388
        >>>                               shapes=[(3,224,224), (1,)],
W
wopeizl 已提交
389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414
        >>>                               lod_levels=[0, 0],
        >>>                               dtypes=['float32', 'int64'])
        >>> # Via the reader, we can use 'read_file' layer to get data:
        >>> image, label = fluid.layers.io.read_file(reader)
    """
    dtypes = [convert_np_dtype_to_dtype_(dt) for dt in dtypes]
    shape_concat = []
    ranks = []

    for shape in shapes:
        shape_concat.extend(shape)
        ranks.append(len(shape))

    var_name = unique_name('open_recordio_file')

    startup_blk = default_startup_program().current_block()
    startup_var = startup_blk.create_var(name=var_name)
    startup_blk.append_op(
        type='create_recordio_file_reader',
        outputs={'Out': [startup_var]},
        attrs={
            'shape_concat': shape_concat,
            'lod_levels': lod_levels,
            'filename': filename,
            'ranks': ranks
        })
Y
Yu Yang 已提交
415

W
wopeizl 已提交
416 417 418 419
    startup_var.desc.set_dtypes(dtypes)
    startup_var.persistable = True
    main_prog_var = _copy_reader_var_(default_main_program().current_block(),
                                      startup_var)
F
fengjiayi 已提交
420

W
wopeizl 已提交
421 422
    if pass_num > 1:
        main_prog_var = multi_pass(reader=main_prog_var, pass_num=pass_num)
F
fengjiayi 已提交
423

W
wopeizl 已提交
424
    return monkey_patch_reader_methods(main_prog_var)
Y
Yu Yang 已提交
425 426


F
fengjiayi 已提交
427 428 429 430 431
def random_data_generator(low, high, shapes, lod_levels, for_parallel=True):
    """
    Create a uniform random data generator

    This layer returns a Reader Variable.
432 433 434
    Instead of opening a file and reading data from it, this
    Reader Variable generates float uniform random data by itself.
    It can be used as a dummy reader to test a network without
F
fengjiayi 已提交
435 436 437 438 439 440 441 442 443 444 445 446 447 448 449
    opening a real file.

    Args:
       low(float): The lower bound of data's uniform distribution.
       high(float): The upper bound of data's uniform distribution.
       shapes(list): List of tuples which declaring data shapes.
       lod_levels(list): List of ints which declaring data lod_level.
       for_parallel(Bool): Set it as True if you are going to run
            subsequent operators in parallel.

    Returns:
       Variable: A Reader Variable from which we can get random data.

    Examples:

450
        .. code-block:: python
F
fengjiayi 已提交
451

452 453 454 455 456 457 458
            reader = fluid.layers.random_data_generator(
                                             low=0.0,
                                             high=1.0,
                                             shapes=[[3,224,224], [1]],
                                             lod_levels=[0, 0])
            # Via the reader, we can use 'read_file' layer to get data:
            image, label = fluid.layers.read_file(reader)
F
fengjiayi 已提交
459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490
    """
    dtypes = [core.VarDesc.VarType.FP32] * len(shapes)
    shape_concat = []
    ranks = []

    for shape in shapes:
        shape_concat.extend(shape)
        ranks.append(len(shape))

    var_name = unique_name('random_data_generator')

    startup_blk = default_startup_program().current_block()
    startup_var = startup_blk.create_var(name=var_name)
    startup_blk.append_op(
        type='create_random_data_generator',
        outputs={'Out': [startup_var]},
        attrs={
            'low': low,
            'high': high,
            'shape_concat': shape_concat,
            'lod_levels': lod_levels,
            'ranks': ranks
        })

    startup_var.desc.set_dtypes(dtypes)
    startup_var.persistable = True
    main_prog_var = _copy_reader_var_(default_main_program().current_block(),
                                      startup_var)

    return monkey_patch_reader_methods(main_prog_var)


Q
Qiao Longfei 已提交
491 492 493 494 495 496
def _py_reader(capacity,
               shapes,
               dtypes,
               lod_levels=None,
               name=None,
               use_double_buffer=True,
S
sneaxiy 已提交
497
               feed_list=None):
498

Q
Qiao Longfei 已提交
499 500 501 502 503 504 505 506 507 508
    if feed_list is not None:
        if not isinstance(feed_list, list):
            raise TypeError("feed_list should be a list of Variable"
                            " instead of " + str(type(feed_list)))
        lod_levels = []
        dtypes = []
        shape_concat = []
        ranks = []
        shapes = []

Q
Qiao Longfei 已提交
509 510 511 512 513 514
        for feed_data in feed_list:
            dtypes.append(feed_data.dtype)
            shape_concat.extend(feed_data.shape)
            ranks.append(len(feed_data.shape))
            shapes.append(feed_data.shape)
            lod_levels.append(feed_data.lod_level)
Q
Qiao Longfei 已提交
515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536
    else:
        dtypes = [convert_np_dtype_to_dtype_(dt) for dt in dtypes]
        shape_concat = []
        ranks = []

        for shape in shapes:
            shape_concat.extend(shape)
            ranks.append(len(shape))

        if lod_levels is None:
            lod_levels = [0] * len(shapes)

    if name is None:
        queue_name = unique_name('lod_tensor_blocking_queue')
        reader_name = unique_name('create_py_reader')
        double_buffer_name = unique_name('double_buffer')
    else:
        queue_name = "_".join([name, "queue"])
        reader_name = "_".join([name, "reader"])
        double_buffer_name = "_".join([name, "double_buffer"])

    var = global_scope().var(queue_name)
S
sneaxiy 已提交
537
    feed_queue = core.init_lod_tensor_blocking_queue(var, capacity)
Q
Qiao Longfei 已提交
538 539 540 541

    startup_blk = default_startup_program().current_block()
    startup_var = startup_blk.create_var(name=reader_name)
    startup_blk.append_op(
S
add doc  
sneaxiy 已提交
542
        type='create_py_reader',
Q
Qiao Longfei 已提交
543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573
        inputs={'blocking_queue': [queue_name]},
        outputs={'Out': [startup_var]},
        attrs={
            'shape_concat': shape_concat,
            'lod_levels': lod_levels,
            'ranks': ranks
        })

    startup_var.desc.set_dtypes(dtypes)
    startup_var.persistable = True

    main_prog_var = _copy_reader_var_(default_main_program().current_block(),
                                      startup_var)

    reader = monkey_patch_reader_methods(main_prog_var)
    if use_double_buffer:
        double_buffer_reader = double_buffer(reader, name=double_buffer_name)
        # we return a double buffer reader. However, the reset method comes from
        # py_reader.
        double_buffer_reader.reset = reader.reset
        reader = double_buffer_reader

    # monkey patch py_reader special methods
    reader.queue = feed_queue
    current_reset_method = reader.reset
    reader.thread = None
    reader.tensor_provider = None
    reader.exited = False

    def start_provide_thread(func):
        def __provider_thread__():
S
sneaxiy 已提交
574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593
            try:
                for tensors in func():
                    array = core.LoDTensorArray()
                    for item in tensors:
                        if not isinstance(item, core.LoDTensor):
                            tmp = core.LoDTensor()
                            tmp.set(item, core.CPUPlace())
                            item = tmp

                        array.append(item)

                    if reader.exited:
                        break
                    feed_queue.push(array)
                    if reader.exited:
                        break
                feed_queue.close()
            except Exception as ex:
                feed_queue.close()
                raise ex
Q
Qiao Longfei 已提交
594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617

        reader.thread = threading.Thread(target=__provider_thread__)
        reader.thread.daemon = True
        reader.thread.start()

    def __set_tensor_provider__(func):
        reader.tensor_provider = func

    def __set_paddle_reader__(paddle_reader):
        with program_guard(Program(), Program()):
            actual_feed_list = feed_list
            if actual_feed_list is None:
                actual_feed_list = []
                counter = 0
                for dtype, shape, lod_level in zip(dtypes, shapes, lod_levels):
                    name = str(counter)
                    actual_feed_list.append(
                        data(
                            name=name,
                            dtype=dtype,
                            shape=shape,
                            lod_level=lod_level))
                    counter += 1

Q
Qiao Longfei 已提交
618
            data_names = [feed_data.name for feed_data in actual_feed_list]
Q
Qiao Longfei 已提交
619 620 621 622 623 624 625
            feeder = DataFeeder(
                feed_list=actual_feed_list, place=core.CPUPlace())
            paddle_reader = feeder.decorate_reader(
                paddle_reader, multi_devices=False)

        def __tensor_provider__():
            for slots in paddle_reader():
Q
Qiao Longfei 已提交
626
                yield [slots[data_name] for data_name in data_names]
Q
Qiao Longfei 已提交
627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642

        __set_tensor_provider__(__tensor_provider__)

    def __reset__():
        current_reset_method()
        if reader.thread is not None and reader.tensor_provider is not None:
            reader.exited = True
            reader.thread.join()
            reader.exited = False

    def __start__():
        start_provide_thread(reader.tensor_provider)

    reader.reset = __reset__
    reader.decorate_tensor_provider = __set_tensor_provider__
    reader.decorate_paddle_reader = __set_paddle_reader__
S
sneaxiy 已提交
643 644 645

    reader.decorate_batch_generator = __set_tensor_provider__
    reader.decorate_sample_list_generator = __set_paddle_reader__
Q
Qiao Longfei 已提交
646 647 648 649 650
    reader.start = __start__

    return reader


Y
yuyang18 已提交
651 652 653 654 655
def py_reader(capacity,
              shapes,
              dtypes,
              lod_levels=None,
              name=None,
S
sneaxiy 已提交
656
              use_double_buffer=True):
S
sneaxiy 已提交
657
    """
658
    Create a Python reader for data feeding in Python
F
fengjiayi 已提交
659

660
    This layer returns a Reader Variable.
661 662
    The Reader provides :code:`decorate_paddle_reader()` and
    :code:`decorate_tensor_provider()` to set a Python generator as the data
663 664 665 666 667
    source. More details :ref:`user_guide_use_py_reader_en` .  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
668 669 670
    called when each pass begins, while the :code:`reset()` method should be
    called when the pass ends and :code:`fluid.core.EOFException` raises.
    Note that :code:`Program.clone()` method cannot clone :code:`py_reader`.
S
sneaxiy 已提交
671 672

    Args:
673
       capacity(int): The buffer capacity maintained by :code:`py_reader`.
Y
yuyang18 已提交
674 675 676 677 678
       shapes(list|tuple): List of tuples which declaring data shapes.
       dtypes(list|tuple): List of strs which declaring data type.
       lod_levels(list|tuple): List of ints which declaring data lod_level.
       name(basestring): The prefix Python queue name and Reader name. None will
            be generated automatically.
679
       use_double_buffer(bool): Whether use double buffer or not.
S
sneaxiy 已提交
680 681

    Returns:
682
       Variable: A Reader from which we can get feeding data.
S
sneaxiy 已提交
683 684

    Examples:
685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711
       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 regresssion 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()
H
Huihuang Zheng 已提交
712 713 714 715 716
             try:
                 while True:
                     exe.run(fetch_list=[loss.name])
             except fluid.core.EOFException:
                 reader.reset()
717 718 719 720 721 722 723 724

         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.:
S
sneaxiy 已提交
725

726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750
       .. 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(
H
Huihuang Zheng 已提交
751 752
                     paddle.reader.shuffle(paddle.batch(mnist.train(), batch_size=5),
                                           buf_size=500))
753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792
                 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()
S
sneaxiy 已提交
793
    """
Q
Qiao Longfei 已提交
794 795 796 797 798 799
    return _py_reader(
        capacity=capacity,
        shapes=shapes,
        dtypes=dtypes,
        lod_levels=lod_levels,
        name=name,
S
sneaxiy 已提交
800
        use_double_buffer=use_double_buffer)
Q
Qiao Longfei 已提交
801 802


Q
Qiao Longfei 已提交
803 804 805 806 807 808
def create_py_reader_by_data(capacity,
                             feed_list,
                             name=None,
                             use_double_buffer=True):
    """
    Create a Python reader for data feeding in Python
Q
Qiao Longfei 已提交
809

Q
Qiao Longfei 已提交
810
    This layer returns a Reader Variable.
Q
Qiao Longfei 已提交
811

Q
Qiao Longfei 已提交
812 813
    Works much like py_reader except that it's input is feed_list
    instead of shapes, dtypes and lod_levels
Q
Qiao Longfei 已提交
814

Q
Qiao Longfei 已提交
815 816 817 818 819 820
    Args:
       capacity(int): The buffer capacity maintained by :code:`py_reader`.
       feed_list(list(Variable)): The data feed list.
       name(basestring): The prefix Python queue name and Reader name. None will
            be generated automatically.
       use_double_buffer(bool): Whether use double buffer or not.
Q
Qiao Longfei 已提交
821

Q
Qiao Longfei 已提交
822 823
    Returns:
       Variable: A Reader from which we can get feeding data.
Q
Qiao Longfei 已提交
824

Q
Qiao Longfei 已提交
825
    Examples:
826
       .. code-block:: python
Q
Qiao Longfei 已提交
827

828 829 830
         import paddle
         import paddle.fluid as fluid
         import paddle.dataset.mnist as mnist
831
         import paddle.fluid.compiler as compiler
832 833 834 835 836 837 838

         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)

839 840 841
         MEMORY_OPT = False
         USE_CUDA = False

842 843 844 845 846 847 848 849 850 851 852
         image = fluid.layers.data(name='image', shape=[1, 28, 28], dtype='float32')
         label = fluid.layers.data(name='label', shape=[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)  # some network definition

853 854 855
         place = fluid.CUDAPlace(0) if USE_CUDA else fluid.CPUPlace()
         exe = fluid.Executor(place)
         exe.run(fluid.default_startup_program())
856

857 858 859 860 861 862 863 864 865
         build_strategy = fluid.BuildStrategy()
         build_strategy.memory_optimize = True if MEMORY_OPT else False
         compiled_prog = 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):
866 867 868
             reader.start()
             try:
                 while True:
869
                     exe.run(compiled_prog, fetch_list=[loss.name])
870 871
             except fluid.core.EOFException:
                 reader.reset()
Q
Qiao Longfei 已提交
872 873 874 875 876 877 878 879 880
    """
    return _py_reader(
        capacity=capacity,
        shapes=None,
        dtypes=None,
        lod_levels=None,
        name=name,
        use_double_buffer=use_double_buffer,
        feed_list=feed_list)
S
sneaxiy 已提交
881 882


883 884 885 886
def open_files(filenames,
               shapes,
               lod_levels,
               dtypes,
Y
yuyang18 已提交
887
               thread_num=None,
F
fengjiayi 已提交
888 889
               buffer_size=None,
               pass_num=1,
Y
yuyang18 已提交
890
               is_test=None):
F
fengjiayi 已提交
891 892 893
    """
    Open files

894 895 896
    This layer takes a list of files to read from and returns a Reader Variable.
    Via the Reader Variable, we can get data from given files. All files must
    have name suffixs to indicate their formats, e.g., '*.recordio'.
F
fengjiayi 已提交
897 898 899 900 901 902

    Args:
       filenames(list): The list of file names.
       shapes(list): List of tuples which declaring data shapes.
       lod_levels(list): List of ints which declaring data lod_level.
       dtypes(list): List of strs which declaring data type.
Y
yuyang18 已提交
903 904 905
       thread_num(None): The number of thread to read files.
            Default: min(len(filenames), cpu_number).
       buffer_size(None): The buffer size of reader. Default: 3 * thread_num
F
fengjiayi 已提交
906
       pass_num(int): Number of passes to run.
Y
yuyang18 已提交
907 908 909 910
       is_test(bool|None): Whether `open_files` used for testing or not. If it
            is used for testing, the order of data generated is same as the file
            order. Otherwise, it is not guaranteed the order of data is same
            between every epoch. [Default: False].
F
fengjiayi 已提交
911 912 913 914 915 916 917

    Returns:
       Variable: A Reader Variable via which we can get file data.

    Examples:
       .. code-block:: python

918
         import paddle.fluid as fluid
F
fengjiayi 已提交
919
         reader = fluid.layers.io.open_files(filenames=['./data1.recordio',
F
fengjiayi 已提交
920
                                                     './data2.recordio'],
921
                                             shapes=[(3,224,224), (1,)],
F
fengjiayi 已提交
922
                                             lod_levels=[0, 0],
Y
yuyang18 已提交
923
                                             dtypes=['float32', 'int64'])
F
fengjiayi 已提交
924 925

         # Via the reader, we can use 'read_file' layer to get data:
F
fengjiayi 已提交
926
         image, label = fluid.layers.io.read_file(reader)
F
fengjiayi 已提交
927
    """
Y
yuyang18 已提交
928 929 930 931 932 933 934 935 936
    if thread_num is None:
        thread_num = min(len(filenames), multiprocessing.cpu_count())
    else:
        thread_num = int(thread_num)

    if buffer_size is None:
        buffer_size = 3 * thread_num
    else:
        buffer_size = int(buffer_size)
Y
yuyang18 已提交
937

M
minqiyang 已提交
938
    if isinstance(filenames, six.string_types):
F
fengjiayi 已提交
939
        filenames = [filenames]
F
fengjiayi 已提交
940 941 942 943 944 945 946 947
    dtypes = [convert_np_dtype_to_dtype_(dt) for dt in dtypes]
    shape_concat = []
    ranks = []

    for shape in shapes:
        shape_concat.extend(shape)
        ranks.append(len(shape))

F
fengjiayi 已提交
948
    multi_file_reader_name = unique_name('multi_file_reader')
F
fengjiayi 已提交
949
    startup_blk = default_startup_program().current_block()
F
fengjiayi 已提交
950
    startup_reader = startup_blk.create_var(name=multi_file_reader_name)
Y
yuyang18 已提交
951 952 953 954
    attrs = {
        'shape_concat': shape_concat,
        'lod_levels': lod_levels,
        'ranks': ranks,
Y
yuyang18 已提交
955 956 957
        'file_names': filenames,
        'thread_num': thread_num,
        'buffer_size': buffer_size
Y
yuyang18 已提交
958 959 960
    }
    if is_test is not None:
        attrs['is_test'] = is_test
F
fengjiayi 已提交
961
    startup_blk.append_op(
Y
yuyang18 已提交
962
        type='open_files', outputs={'Out': [startup_reader]}, attrs=attrs)
F
fengjiayi 已提交
963

F
fengjiayi 已提交
964 965 966 967 968 969 970
    startup_reader.desc.set_dtypes(dtypes)
    startup_reader.persistable = True
    main_prog_reader = _copy_reader_var_(default_main_program().current_block(),
                                         startup_reader)
    if pass_num > 1:
        main_prog_reader = multi_pass(
            reader=main_prog_reader, pass_num=pass_num)
F
fengjiayi 已提交
971

F
fengjiayi 已提交
972 973 974
    return monkey_patch_reader_methods(main_prog_reader)


J
JiayiFeng 已提交
975
def __create_shared_decorated_reader__(op_type, reader, attrs):
Y
Yu Yang 已提交
976 977 978
    var_name = unique_name(op_type)
    startup_blk = default_startup_program().current_block()
    startup_var = startup_blk.create_var(name=var_name)
F
fengjiayi 已提交
979
    startop_op = startup_blk.append_op(
Y
Yu Yang 已提交
980 981 982 983 984
        type=op_type,
        inputs={'UnderlyingReader': reader},
        outputs={'Out': [startup_var]},
        attrs=attrs)
    startup_var.persistable = True
F
fengjiayi 已提交
985 986 987 988
    main_prog_block = default_main_program().current_block()
    main_prog_var = _copy_reader_var_(main_prog_block, startup_var)
    _copy_reader_create_op_(main_prog_block, startop_op)
    return monkey_patch_reader_methods(main_prog_var)
Y
Yu Yang 已提交
989 990


991 992
def __create_unshared_decorated_reader__(op_type, reader, attrs, name=None):
    new_reader_name = name if name is not None else unique_name(op_type)
993 994 995 996 997 998 999 1000 1001 1002
    main_blk = default_main_program().current_block()
    new_reader = main_blk.create_var(name=new_reader_name)
    main_blk.append_op(
        type=op_type,
        inputs={'UnderlyingReader': reader},
        outputs={'Out': [new_reader]},
        attrs=attrs)
    return monkey_patch_reader_methods(new_reader)


F
fengjiayi 已提交
1003
def shuffle(reader, buffer_size):
1004
    """
T
Tink_Y 已提交
1005 1006 1007 1008 1009 1010
    Creates a data reader whose data output is shuffled.
    Output from the iterator that created by original reader will be
    buffered into shuffle buffer, and then shuffled. The size of shuffle buffer
    is determined by argument buf_size.

    Args:
H
haowang101779990 已提交
1011 1012 1013 1014 1015
        reader(callable): the original reader whose output will be shuffled.
        buf_size(int): shuffle buffer size.

    Returns:
        callable: the new reader whose output is shuffled.
1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028

    Examples:
        .. code-block:: python

            raw_reader = fluid.layers.io.open_files(filenames=['./data1.recordio',
                                                           './data2.recordio'],
                                                    shapes=[(3,224,224), (1,)],
                                                    lod_levels=[0, 0],
                                                    dtypes=['float32', 'int64'],
                                                    thread_num=2,
                                                    buffer_size=2)
            batch_reader = fluid.layers.batch(reader=raw_reader, batch_size=5)
            shuffle_reader = fluid.layers.shuffle(reader=batch_reader, buffer_size=5000)
1029
    """
1030 1031
    return __create_unshared_decorated_reader__(
        'create_shuffle_reader', reader, {'buffer_size': int(buffer_size)})
Y
Yu Yang 已提交
1032 1033


J
JiayiFeng 已提交
1034
def batch(reader, batch_size):
F
fengjiayi 已提交
1035
    """
1036 1037 1038
    This layer is a reader decorator. It takes a reader and adds
    'batching' decoration on it. When reading with the result
    decorated reader, output data will be automatically organized
F
fengjiayi 已提交
1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052
    to the form of batches.

    Args:
        reader(Variable): The reader to be decorated with 'batching'.
        batch_size(int): The batch size.

    Returns:
        Variable: The reader which has been decorated with 'batching'.

    Examples:
        .. code-block:: python

            raw_reader = fluid.layers.io.open_files(filenames=['./data1.recordio',
                                                           './data2.recordio'],
1053
                                                    shapes=[(3,224,224), (1,)],
F
fengjiayi 已提交
1054 1055 1056 1057 1058 1059 1060 1061 1062
                                                    lod_levels=[0, 0],
                                                    dtypes=['float32', 'int64'],
                                                    thread_num=2,
                                                    buffer_size=2)
            batch_reader = fluid.layers.batch(reader=raw_reader, batch_size=5)

            # If we read data with the raw_reader:
            #     data = fluid.layers.read_file(raw_reader)
            # We can only get data instance by instance.
1063
            #
F
fengjiayi 已提交
1064 1065
            # However, if we read data with the batch_reader:
            #     data = fluid.layers.read_file(batch_reader)
1066 1067
            # Each 5 adjacent instances will be automatically combined together
            # to become a batch. So what we get('data') is a batch data instead
F
fengjiayi 已提交
1068 1069
            # of an instance.
    """
J
JiayiFeng 已提交
1070 1071 1072 1073
    return __create_unshared_decorated_reader__(
        'create_batch_reader', reader, {'batch_size': int(batch_size)})


1074
def double_buffer(reader, place=None, name=None):
Y
yuyang18 已提交
1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091
    """
    Wrap a double buffer reader. 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): the place of target data. Default is the sample place of
            executor perform.

        name(str): Variable name. None if the user does not care.

    Returns:
        wrapped reader with double buffer.

    Examples:

F
flame 已提交
1092 1093
        >>> import paddle.fluid as fluid
        >>> reader = fluid.layers.open_files(filenames=['mnist.recordio'],
Y
yuyang18 已提交
1094
        >>>                                  shapes=[[-1, 784], [-1, 1]],
F
flame 已提交
1095
        >>>                                  lod_levels=[0, 0],
Y
yuyang18 已提交
1096 1097 1098 1099
        >>>                                  dtypes=['float32', 'int64'])
        >>> reader = fluid.layers.double_buffer(reader)
        >>> img, label = fluid.layers.read_file(reader)
    """
Y
Yu Yang 已提交
1100 1101 1102
    attrs = dict()
    if place is not None:
        attrs['place'] = str(place).upper()
1103 1104
    return __create_unshared_decorated_reader__(
        'create_double_buffer_reader', reader, attrs, name=name)
Y
Yu Yang 已提交
1105 1106


F
fengjiayi 已提交
1107
def multi_pass(reader, pass_num):
1108 1109
    return __create_shared_decorated_reader__(
        'create_multi_pass_reader', reader, {'pass_num': int(pass_num)})
F
fengjiayi 已提交
1110 1111


F
fengjiayi 已提交
1112
def read_file(reader):
F
fengjiayi 已提交
1113
    """
F
fengjiayi 已提交
1114
    Execute the given reader and get data via it.
F
fengjiayi 已提交
1115

1116 1117
    A reader is also a Variable. It can be a raw reader generated by
    `fluid.layers.open_files()` or a decorated one generated by
F
fengjiayi 已提交
1118 1119 1120 1121
    `fluid.layers.double_buffer()` and so on.

    Args:

F
fengjiayi 已提交
1122
        reader(Variable): The reader to execute.
F
fengjiayi 已提交
1123 1124

    Returns:
F
fengjiayi 已提交
1125
        Tuple[Variable]: Data read via the given reader.
F
fengjiayi 已提交
1126 1127 1128

    Examples:
        .. code-block:: python
1129 1130
          
           import paddle.fluid as fluid
F
fengjiayi 已提交
1131 1132 1133 1134 1135
           data_file = fluid.layers.open_files(
                filenames=['mnist.recordio'],
                shapes=[(-1, 748), (-1, 1)],
                lod_levels=[0, 0],
                dtypes=["float32", "int64"])
1136
           data_file = fluid.layers.double_buffer(
F
fengjiayi 已提交
1137
                fluid.layers.batch(data_file, batch_size=64))
1138
           input, label = fluid.layers.read_file(data_file)
F
fengjiayi 已提交
1139
    """
Y
Yu Yang 已提交
1140 1141
    helper = LayerHelper('read_file')
    out = [
X
Xin Pan 已提交
1142
        helper.create_variable_for_type_inference(
Y
Yu Yang 已提交
1143
            stop_gradient=True, dtype='float32')
F
fengjiayi 已提交
1144
        for _ in range(len(reader.desc.shapes()))
Y
Yu Yang 已提交
1145 1146
    ]
    helper.append_op(
F
fengjiayi 已提交
1147
        type='read', inputs={'Reader': [reader]}, outputs={'Out': out})
Y
Yu Yang 已提交
1148 1149 1150 1151
    if len(out) == 1:
        return out[0]
    else:
        return out
F
fengjiayi 已提交
1152 1153 1154


class Preprocessor(object):
X
Xin Pan 已提交
1155 1156 1157 1158 1159 1160 1161 1162 1163
    """
    A block for data pre-processing in reader.

    Args:
        reader (Variable): A reader variable.
        name (str, default None): The name of the reader.

    Examples:
          .. code-block:: python
X
Xin Pan 已提交
1164

H
Huihuang Zheng 已提交
1165 1166
           import paddle.fluid as fluid

1167 1168 1169 1170 1171 1172
           reader = fluid.layers.io.open_files(
               filenames=['./data1.recordio', './data2.recordio'],
               shapes=[(3, 224, 224), (1, )],
               lod_levels=[0, 0],
               dtypes=['float32', 'int64']) 

X
Xin Pan 已提交
1173 1174 1175 1176 1177 1178 1179 1180 1181 1182
            preprocessor = fluid.layers.io.Preprocessor(reader=reader)
            with preprocessor.block():
                img, lbl = preprocessor.inputs()
                img_out = img / 2
                lbl_out = lbl + 1
                preprocessor.outputs(img_out, lbl_out)

            data_file = fluid.layers.io.double_buffer(preprocessor())

    """
F
fengjiayi 已提交
1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198
    BEFORE_SUB_BLOCK = 0
    IN_SUB_BLOCK = 1
    AFTER_SUB_BLOCK = 2

    def __init__(self, reader, name=None):
        self.underlying_reader = reader
        new_reader_name = name if name is not None else unique_name(
            "create_custom_reader")
        self.main_prog = default_main_program()
        self.reader = self.main_prog.current_block().create_var(
            name=new_reader_name)
        self.sub_block = None
        self.source_var_names = None
        self.sink_var_names = None
        self.status = Preprocessor.BEFORE_SUB_BLOCK

X
Xin Pan 已提交
1199
    def _is_completed(self):
F
fengjiayi 已提交
1200 1201
        return self.sub_block and self.source_var_names and self.sink_var_names

S
rename  
sneaxiy 已提交
1202
    @signature_safe_contextmanager
F
fengjiayi 已提交
1203 1204
    def block(self):
        self.status = Preprocessor.IN_SUB_BLOCK
W
Wu Yi 已提交
1205
        self.sub_block = self.main_prog._create_block()
F
fengjiayi 已提交
1206
        yield
W
Wu Yi 已提交
1207
        self.main_prog._rollback()
F
fengjiayi 已提交
1208
        self.status = Preprocessor.AFTER_SUB_BLOCK
X
Xin Pan 已提交
1209
        if not self._is_completed():
F
fengjiayi 已提交
1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224
            raise RuntimeError(
                "The definition of preprocessor is incompleted! "
                "Please make sure that you have set input and output "
                "variables by invoking 'inputs' and 'outputs' in "
                "Preprocessor's sub-block.")

    def inputs(self):
        if self.status != Preprocessor.IN_SUB_BLOCK:
            raise RuntimeError(
                "Preprocessor.inputs() can only be invoked inside the sub-block."
            )

        source_shapes = self.underlying_reader.desc.shapes()
        source_dtypes = self.underlying_reader.desc.dtypes()
        source_lod_levels = self.underlying_reader.desc.lod_levels()
F
fengjiayi 已提交
1225 1226
        self.source_var_names = [
            unique_name("preprocessor_source")
M
minqiyang 已提交
1227
            for _ in six.moves.range(len(source_shapes))
F
fengjiayi 已提交
1228
        ]
F
fengjiayi 已提交
1229
        source_vars = []
F
fengjiayi 已提交
1230 1231 1232
        for var_name, shape, dtype, lod_level in zip(
                self.source_var_names, source_shapes, source_dtypes,
                source_lod_levels):
F
fengjiayi 已提交
1233
            source_vars.append(self.main_prog.current_block().create_var(
F
fengjiayi 已提交
1234
                name=var_name, shape=shape, dtype=dtype, lod_level=lod_level))
F
fengjiayi 已提交
1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258
        return source_vars

    def outputs(self, *outs):
        if self.status != Preprocessor.IN_SUB_BLOCK:
            raise RuntimeError(
                "Preprocessor.outputs() can only be invoked inside the sub-block."
            )
        self.sink_var_names = [var.name for var in outs]

    def __call__(self, *args, **kwargs):
        if self.status != Preprocessor.AFTER_SUB_BLOCK:
            raise RuntimeError(
                "Preprocessor output can only be retrieved after rnn block.")

        self.main_prog.current_block().append_op(
            type="create_custom_reader",
            inputs={'UnderlyingReader': self.underlying_reader},
            outputs={'Out': [self.reader]},
            attrs={
                "sub_block": self.sub_block,
                "source_var_names": self.source_var_names,
                "sink_var_names": self.sink_var_names
            })
        return monkey_patch_reader_methods(self.reader)
Y
yuyang18 已提交
1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284


@templatedoc()
def load(out, file_path, load_as_fp16=None):
    """
    ${comment}

    >>> import paddle.fluid as fluid
    >>> tmp_tensor = fluid.layers.create_tensor(dtype='float32')
    >>> fluid.layers.load(tmp_tensor, "./tmp_tensor.bin")

    Args:
        out(${out_type}): ${out_comment}.

        file_path(${file_path_type}): ${file_path_comment}.

        load_as_fp16(${load_as_fp16_type}): ${load_as_fp16_comment}.

    Returns:
        None
    """
    helper = LayerHelper("load", **locals())
    attrs = {"file_path": file_path}
    if load_as_fp16 is not None:
        attrs['load_as_fp16'] = load_as_fp16
    helper.append_op(type="load", inputs={}, output={"Out": out}, args=attrs)