o #jQ@sddlZddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z gZ ejdkrBejdkrBedZneZd d Zd d Zd dZddZGdddeZddZddZddZGdddZd"ddZd#d d!ZdS)$N) zip_longest)Queue)Thread)QUEUE_GET_TIMEOUT)darwinforkcst|fdd}|S)a Cache the reader data into memory. Be careful that this method may take long time to process, and consume lots of memory. :code:`reader()` would only call once. Args: reader (generator): a reader object which yields data each time. Returns: generator: a decorated reader object which yields data from cached memory. Examples: .. code-block:: python >>> import paddle >>> def reader(): ... for i in range(3): ... yield i ... >>> # All data is cached into memory >>> cached_reader = paddle.base.io.cache(reader) >>> for i in cached_reader(): ... print(i) 0 1 2 c3sEdHdSNr Zall_datar X/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/paddle/reader/decorator.py__impl__Pszcache..__impl__)tuple)readerrr r r cache-s ! rcfdd}|S)a Creates a data reader that outputs return value of function using output of each data reader as arguments. If input readers output the following data entries: 2 3, and the input func is mul(x, y), the output of the resulted reader will be 6. Args: func: a function to read data and compute result, the output of this function will be set as the output of the resulted data reader. readers (Reader|list of Reader): list of readers whose outputs will be used as arguments of func. Returns: the resulted data reader (Reader) Examples: .. code-block:: python >>> import paddle.reader >>> d = {"h": 0, "i": 1} >>> def func(x): ... return d[x] >>> def reader(): ... yield "h" ... yield "i" >>> map_reader_result = paddle.reader.map_readers(func, reader) c3s6g}D]}||qtg|REdHdSr )appendmaprsrfuncreadersr r rvs zmap_readers..readerr )rrrr rr map_readersV rcr)aV This API creates a decorated reader that outputs the shuffled data. The output data from the origin reader will be saved into a buffer, and then shuffle the data. The size of buffer is determined by argument buf_size. Args: reader(callable): the original reader whose data will be shuffled. buf_size(int): the size of shuffled buffer. Returns: callable: a decorated reader. Examples: .. code-block:: python >>> # doctest: +SKIP('outputs are 0~4 unordered arrangement') >>> def reader(): ... for i in range(5): ... yield i >>> shuffled_reader = paddle.reader.decorator.shuffle(reader, 3) >>> for e in shuffled_reader(): ... print(e) >>> # outputs are 0~4 unordered arrangement c3stg}D]}||t|kr"t||D]}|Vqg}qt|dkr6t||D]}|Vq0dSdS)Nr)rlenrandomshuffle)bufebbuf_sizerr r data_readers       zshuffle..data_readerr )rr$r%r r#r rsrcsfdd}|S)aN Use the input data readers to create a chained data reader. The new created reader chains the outputs of input readers together as its output, and it do not change the format of the outputs. **Note**: ``paddle.reader.chain`` is the alias of ``paddle.base.io.chain``, and ``paddle.base.io.chain`` is recommended to use. For example, if three input readers' outputs are as follows: [0, 0, 0], [10, 10, 10], [20, 20, 20]. The chained reader will output: [0, 0, 0], [10, 10, 10], [20, 20, 20]. Args: readers(list): input data readers. Returns: callable: the new chained data reader. Examples: .. code-block:: python >>> import paddle >>> def reader_creator_3(start): ... def reader(): ... for i in range(start, start + 3): ... yield [i, i, i] ... return reader ... >>> c = paddle.reader.chain(reader_creator_3(0), reader_creator_3(10), reader_creator_3(20)) >>> for e in c(): ... print(e) [0, 0, 0] [1, 1, 1] [2, 2, 2] [10, 10, 10] [11, 11, 11] [12, 12, 12] [20, 20, 20] [21, 21, 21] [22, 22, 22] c3s0g}D]}||qtj|EdHdSr )r itertoolschainrrr r rs zchain..readerr )rrr r(r r's 1r'c@ eZdZdS)ComposeNotAlignedN__name__ __module__ __qualname__r r r r r*r*cs(|ddddfdd}|S)a# Creates a data reader whose output is the combination of input readers. If input readers output following data entries: (1, 2) 3 (4, 5) The composed reader will output: (1, 2, 3, 4, 5) Args: readers (Reader|list of Reader): readers that will be composed together. check_alignment(bool, optional): Indicates whether the input readers are checked for alignment. If True, whether input readers are aligned correctly will be checked, else alignment will not be checkout and trailing outputs will be discarded. Defaults to True. Returns: the new data reader (Reader). Examples: .. code-block:: python >>> def reader_creator_10(dur): ... def reader(): ... for i in range(10): ... yield i ... return reader >>> reader = paddle.reader.decorator.compose(reader_creator_10(0), reader_creator_10(0)) check_alignmentTcSst|tr|S|fSr ) isinstancer)xr r r make_tuple s zcompose..make_tuplec3sg}D]}||qs$t|D] }ttt|dVqdSt|D]}|D] }|dur6tdq,ttt|dVq(dS)Nr z#outputs of readers are not aligned.)rzipsumlistrrr*)rrZoutputsor0r3rr r rs"  zcompose..reader)pop)rkwargsrr r8r composes r;cs6Gddd}|fddfdd}|S)ac Creates a buffered data reader. The buffered data reader will read and save data entries into a buffer. Reading from the buffered data reader will proceed as long as the buffer is not empty. Args: reader(generator): the data reader to read from. size(int): max buffer size. Returns: generator: the buffered data reader. Examples: .. code-block:: python >>> import paddle >>> def reader(): ... for i in range(3): ... yield i ... >>> # Create a buffered reader, and the buffer size is 2. >>> buffered_reader = paddle.reader.decorator.buffered(reader, 2) >>> # Output: 0 1 2 >>> for i in buffered_reader(): ... print(i) 0 1 2 c@r))zbuffered..EndSignalNr+r r r r EndSignalGr/r<cs"|D]}||q|dSr put)rqdendr r read_workerLs zbuffered..read_workerc3s^}td}t||fd}d|_||}|kr-|V|}|ks dSdS)N)maxsizetargetargsT)rrdaemonstartget)rr?tr!rBrCrsizer r r%Qs  zbuffered..data_readerr )rrMr<r%r rLr buffered$s # rNcr)a This API creates a decorated reader, and limits the max number of samples that reader could return. Args: reader(callable): the input reader. n(int): the max number of samples in the reader. Returns: callable: the decorated reader. Examples: .. code-block:: python >>> def reader(): ... for i in range(100): ... yield i >>> firstn_reader = paddle.reader.decorator.firstn(reader, 5) >>> for e in firstn_reader(): ... print(e) 0 1 2 3 4 c3s.tD]\}}|krdS|VqdSr ) enumerate)iitemnrr r firstn_readers zfirstn..firstn_readerr )rrSrTr rRr firstnerrUc@r)) XmapEndSignalNr+r r r r rVr/rVFc sVtfddfddfddfdd f d d }|S) a Use multi-threads to map samples from reader by a mapper defined by user. Args: mapper (callable): a function to map the data from reader. reader (callable): a data reader which yields the data. process_num (int): thread number to handle original sample. buffer_size (int): size of the queue to read data in. order (bool): whether to keep the data order from original reader. Default False. Returns: callable: a decorated reader with data mapping. cs$|D]}||q|dSr r=)rin_queuerPrAr r rCs  z!xmap_readers..read_workercs4d}|D] }|||f|d7}q|dSNrr=)rrWZin_orderrPrAr r order_read_workers   z'xmap_readers..order_read_workercsN|}t|ts||}|||}t|tr ||dSr rJr1rVr>)rW out_queuemappersamplerrAr r handle_workers   z#xmap_readers..handle_workercs|}t|ts4|\}}||}||dkr ||dks|||dd7<|}t|tr ||dSrXr[)rWr\r] out_orderZinsorderr^rrAr r order_handle_workers    z)xmap_readers..order_handle_workerc 3st}t}dg}rn}t||fd}d|_|r%n}r/|||fn||f}g}tD]}t||d}d|_||q:|D]} | qM|} t| tsi| V|} t| tr]d} | kr|} t| tr}| d7} n| V| ksodSdS)NrrETrY) rrrHrIrangerrJr1rV) rWr\r`rFrKrGworkersrPZworkerwr^finish) buffer_sizer_r]rarbrZ process_numrCrr r xreaders@          zxmap_readers..xreader)rV)r]rrhrgrarir ) rgrBr_r]rarbrZrhrCrr xmap_readerss    $rjTc stjdkr tdzddlWnty*}ztdddlWYd}~nd}~wwtt t fr8t dks>> import paddle >>> import numpy as np >>> sample_files = ['sample_file_1', 'sample_file_2'] >>> def fake_input_files(): ... with open(sample_files[0], 'wb') as f: ... np.savez(f, a=np.array([1, 2]), b=np.array([3, 4]), c=np.array([5, 6]), d=np.array([7, 8])) ... with open(sample_files[1], 'wb') as f: ... np.savez(f, a=np.array([9, 10]), b=np.array([11, 12]), c=np.array([13, 14])) ... ... >>> def generate_reader(file_name): ... # load data file ... def _impl(): ... data = np.load(file_name) ... for item in sorted(data.files): ... yield data[item], ... return _impl ... >>> if __name__ == '__main__': ... # generate sample input files ... fake_input_files() ... ... with base.program_guard(base.Program(), base.Program()): ... place = base.CPUPlace() ... # the 1st 2 is batch size ... ... image = paddle.static.data(name='image', dtype='int64', shape=[2, 1, 2]) ... paddle.static.Print(image) ... # print detailed tensor info of image variable ... ... reader = base.io.PyReader(feed_list=[image], capacity=2) ... ... decorated_reader = paddle.reader.multiprocess_reader( ... [generate_reader(sample_files[0]), generate_reader(sample_files[1])], False) ... ... reader.decorate_sample_generator(decorated_reader, batch_size=2, places=[place]) ... ... exe = base.Executor(place) ... exe.run(base.default_startup_program()) ... ... for data in reader(): ... res = exe.run(feed=data, fetch_list=[image]) ... print(res[0]) [[[1 2]], [[3 4]]] [[[5 6]], [[7 8]]] [[[9 10]], [[11 12]]] win32z;The multiprocess_reader method is not supported on windows.rNzThe `ujson` module is not found, use the `json` module, `ujson` encodes and decodes faster, you can install `ujson` through `pip install ujson`.z `readers` must be list or tuple.c Ss^z|D]}|durtd||q|dWdSty.}z|d|d}~ww)Nzsample has None) ValueErrorr> Exception)rqueuer^r!r r r _read_into_queueTs   z-multiprocess_reader.._read_into_queuec 3st}D]}tj||fd}|qt}d}||krXz|jtd}Wnty=}zt d|d}~ww|durG|d7}n |dkrOt d|V||ks"dSdS)NrEr)timeoutzFmultiprocess_reader failed to get data from the multiprocessing.Queue.rYrmzFmultiprocess_reader failed to put data into the multiprocessing.Queue.) fork_contextrProcessrIrrJrrologgingerrorrn)rprp reader_num finish_numr^r!)rq queue_sizerr r queue_reader_s6   z)multiprocess_reader..queue_readerc sz%|D]}|durtd||q|d|WdSty?}z|d||d}~ww)Nzsample has None!rm)rnsenddumpsclosero)rconnr^r!)jsonr r _read_into_pipe{s z,multiprocess_reader.._read_into_pipec 3sg}D]}t\}}||tj||fd}|qt}d}g}||krp|D]}||q.g}|D]/}|} | durU|d7}| ||q:| dkrf| ||t d| Vq:||ks,dSdS)NrErrYrmzFmultiprocess_reader failed to send data into the multiprocessing.Pipe.) rsZPiperrtrIrremoveloadsrecvr~rn) ZconnsrZ parent_connZ child_connrwrxryZconn_to_removerr^)rrrr r pipe_readers<      z(multiprocess_reader..pipe_reader) sysplatformNotImplementedErrorZujsonrowarningswarnrr1r6rr)rZuse_piperzr!r{rr )rrqrrzrr multiprocess_readers2 O     r)F)Trk)r&rumultiprocessingrrrrrpr threadingrZpaddle.base.readerr__all__ version_inforZ get_contextrsrrrr'rnr*r;rNrUrVrjrr r r r s2     ))-;9A) `