U d[@sdZddlmZddlmZmZmZddlZddl Z Gddde Z Gddde Z Gd d d e Z Gd d d e ZGd dde ZGddde ZGddde ZGdddeZddZGdddeZGddde ZGddde ZGddde ZdS)a Defines the base interface for reading and writing operations. Readers/Writers are objects that produce operations that read/write sequences of data. Each operation reads or writes a list of BlobReferences. Readers and Writers must be implemented such that read and write operations are atomic and thread safe. Examples of possible Readers and Writers: QueueReader, QueueWriter, DatasetReader, DatasetWriter, See `dataset.py` for an example of implementation. )core)FieldStructfrom_blob_listNc@sdeZdZdZdddZddZddZd d Zd d Zd dZ ddZ ddZ ddZ dddZ dS)Readera Reader is an abstract class to be implemented in order to provide operations capable of iterating through a dataset or stream of data. A Reader must implement at least one operation, `read`, which adds operations to a net that read the next batch of data. Readers can optionally support the `reset` operation, which is useful when multiple passes over the data are required. NcCs |dk rt|tst||_dSN) isinstancerAssertionError_schemaselfschemar8/tmp/pip-unpacked-wheel-ua33x9lu/caffe2/python/dataio.py__init__'szReader.__init__cCs|jdk std|jS)Nz$Schema not provided for this reader.)r r r rrrr ,sz Reader.schemacCs ||_dSrr r rrr _set_schema0szReader._set_schemacCsdS)zSetup nets to run at task initialization and cleanup time. Args: global_init_net: A net invoked at task init time. global_finish_net: A net invoked at task cleanup time. Nrr init_net finish_netrrrsetup_ex3szReader.setup_excCstd}|gf||S)N reader_body)rNetread)r local_init_netlocal_finish_netread_netrrrread_ex<s zReader.read_excCs.|||\}}}|jr$t|j|}|||fSr)rr r)r rrnets should_stopfieldsrrrread_record_ex@s  zReader.read_record_excCs tddS)a!Append operations to read_net that will read a batch from the underlying data soruce. Operations added to `read_net` must be thread safe and atomic, that is, it should be possible to clone `read_net` and run multiple instances of it in parallel. Args: read_net: the net that will be appended with read operations Returns: A tuple (should_stop, fields), with: should_stop: BlobReference pointing to a boolean scalar blob that indicates whether the read operation was succesfull or whether the end of data has been reached. fields: A tuple of BlobReference containing the latest batch of data that was read. zReaders must implement `read`.NNotImplementedError)r rrrrrGsz Reader.readcCs tddS)zAppend operations to `net` that will reset the reader. This can be used to read the data multiple times. Not all readers support this operation. zThis reader cannot be resetted.Nr#)r netrrrreset]sz Reader.resetcCs(||\}}|jr t|j|}||fSr)rr r)r rr r!rrr read_recordes zReader.read_recordcCsPt|p d}||\}}|dk r2|||g}tjd|||d}||fS)aQCreate an execution step with a net containing read operators. The execution step will contain a `stop_blob` that knows how to stop the execution loop when end of data was reached. E.g.: read_step, fields = reader.execution_step() consume_net = core.Net('consume') consume_net.Print(fields[0], []) p = core.Plan('reader') p.AddStep(read_step.AddNet(consume_net)) core.RunPlan(p) Args: reader_net_name: (optional) the name of the reader_net to be created. The execution step will be named accordingly. Returns: A tuple (read_step, fields), with: read_step: A newly created execution step containing a net with read operations. The step will have `stop_blob` set, in order to stop the loop on end of data. fields: A tuple of BlobReference containing the latest batch of data that was read. readerNz{}_step)Zshould_stop_blob)rrr'Orexecution_stepformat)r Zreader_net_nameZexternal_should_stopZ reader_netr r!Z read_steprrrr*kszReader.execution_step)N)NN)__name__ __module__ __qualname____doc__rr rrrr"rr&r'r*rrrrrs  rc@sNeZdZdZdZddZddZddZd d Zd d Z dd dZ ddZ dS)Writeraq Writer is an abstract class to be implemented in order to provide operations capable of feeding a data stream or a dataset. A Writer must implement 2 operations: `write`, which adds operations to a net that write the write batch of data, and `commit`, which adds operations to a net in order to indicate that no more data will be written. NcCs|jSrrrrrrr sz Writer.schemacCs tddS)aZAdd operations to `writer_net` that write the next batch of data. Operations added to the net must be thread-safe and unique, that is: multiple writers must be able to write to the dataset in parallel. Args: fields: a tuple of BlobReference containing the batch of data to write. zWriters must implement write.Nr#r Z writer_netr!rrrwrites z Writer.writecCs(t|tr||_|}|||dSr)rrr field_blobsr2r1rrr write_records zWriter.write_recordcCs||dS)zExperimental, don't use yetN)commitrrrrrszWriter.setup_excCstd}||||gS)z6Experimental extension to the interface. Don't use yet write_net)rrr2)r r!rr stop_blobr6rrrwrite_exs  zWriter.write_excCsBt|tr||_|}|dkr*|d}|||||}||fS)z7Experimental extension to the interface. Don't use yet.NZdequeue_status)rrr r3NextNamer8)r r!rrr7Z write_netsrrrwrite_record_exs  zWriter.write_record_excCsdS)zAdd operations to `finish_net` that signal end of data. This must be implemented by all Writers, but may be no-op for some of them. Nr)r rrrrr5sz Writer.commit)N) r,r-r.r/r r r2r4rr8r:r5rrrrr0s   r0c@s(eZdZdZddZddZddZdS) ReaderBuilderz1 Allow usage of a reader in distributed fashion. cCs tdSrr#rrrrr szReaderBuilder.schemacKs tdS)z Optionally, perform one-time setup before calling new_reader(). Subclass should make sure this function is only called once. Nr#r kwargsrrrsetupszReaderBuilder.setupcKs tdSrr#r<rrr new_readerszReaderBuilder.new_readerN)r,r-r.r/r r>r?rrrrr;sr;c@s0eZdZdZddZddZddZdd Zd S) PipedReaderBuilderaReaderBuilder that modifies underlying builder by calling `piper` function on each new reader produced, and return the result of the function. This way, it is possible to append data processing pipelines that will be replicated for each reader that gets created. E.g.: PipedReaderBuilder( ReaderBuilder(...), lambda reader: pipe(reader, processor=my_proc)) cCs||_||_dSr)_builder_piper)r ZbuilderZpiperrrrrszPipedReaderBuilder.__init__cCs |jSr)rAr rrrrr szPipedReaderBuilder.schemacKs|jjf|Sr)rAr>r<rrrr>szPipedReaderBuilder.setupcKs4|jfd|jjf|i|}t|tr,|S|S)Nr()rBrAr?rrr()r r=outputrrrr?s  zPipedReaderBuilder.new_readerN)r,r-r.r/rr r>r?rrrrr@s  r@c@sVeZdZdddZddZddZdd Zd d Zd d ZddZ ddZ ddZ dS)PipeNcCsd|_d|_||_||_dS)Nr) _num_writers _num_readersr _obj_key)r r Zobj_keyrrrrsz Pipe.__init__cCs|jSrrrrrrr sz Pipe.schemacCsdSrr)r global_init_netrrrr> sz Pipe.setupcCs tdSrr#rrrrr(sz Pipe.readercCs tdSrr#rrrrwritersz Pipe.writercCs|jSr)rFrrrr num_readersszPipe.num_readerscCs|jSr)rErrrr num_writersszPipe.num_writerscCsB|dk r|jdkr||_|jd7_|jdk r>||j|dSN)r rErG add_attribute)r Z writer_schemaZwriter_init_netrrr _new_writers  zPipe._new_writercCs*|jd7_|jdk r&||j|dSrL)rFrGrN)r Zreader_init_netrrr _new_reader#s zPipe._new_reader)NN) r,r-r.rr r>r(rIrJrKrOrPrrrrrDs rDc@s(eZdZdZddZddZddZdS) CounterReaderz+ Reader that produces increasing integers. cCs(tj|tdtjfdd|_d|_dS)Niterr )rrrnpZint64counterr rrrrr+szCounterReader.__init__cCs6|jdkr2|jgdd|_|jggtjjdd|_dS)NrZ init_countF)shapedtypevalue)rU CreateCounter ConstantFillrDataTypeBOOLr r rHglobal_finish_netrrrr0s zCounterReader.setup_excCs*td}||jgd}|g|j|gfS)NZlimited_reader_counterrM)rrZCountUprUr )r rrZ count_netrYrrrr6s zCounterReader.read_exNr,r-r.r/rrrrrrrrQ)srQc@s@eZdZdZddZddZddZdd Zd d Zd d Z dS)ReaderWithLimitBaseapAbstract Reader constrained by certain conditions. Base class for Reader classes which check for certain conditions to stop further processing (e.g. max number of iterations or time limit). Also produces a boolean blob (data_finished) that can be used to see if the reader exausted all input data (true) or stopped for another reason (false). cCsBtj||jd||_td|_|j|jd|_ d|_ dS)NrSZreader_with_limit data_finished) rrr r(rrr%AddExternalInputr9_data_finishedr )r r(rrrrFs  zReaderWithLimitBase.__init__cCs:|jg|jggdtjjd|j|||||dS)NFrWrYrX)r[rdrr\r]r(r setup_limiterr^rrrrNszReaderWithLimitBase.setup_exc Csxtd}||}|j||\}}}||jjtd}|||||j |g|j g|g||g||fS)aReads from an underlying Reader class, but may stop due to additional constraints. Build and return network(s) to read data from a Reader with additional constraints, depending on which derived class is used. Derived classes implement setup_limited and check_limiter_condition which determine the nature of the constraint imposed on the reader, e.g. iteration limits or time limit. Args: local_init_net: A net invoked at task instance init time (Once per parallel thread). local_finish_net: A net invoked at task instance cleanup time (Once per parallel thread). Zlimited_reader_conditionZlimited_reader_post) rrcheck_limiter_conditionr(rrr ZCopyr)rd) r rrstop_condition_netr rZlocal_data_finishedr!Zcheck_done_netrrrrUs      zReaderWithLimitBase.read_excCs tddS)aConfigure task level init/cleanup nets required to implement limit condition. Must be implemented by subclass. Args: global_init_net: A net invoked at task init time. global_finish_net: A net invoked at task cleanup time. z'Subclass must implement `setup_limiter`Nr#r^rrrrf|sz!ReaderWithLimitBase.setup_limitercCs tddS)aConfigure a net that is invoked between reading batches to see if limit condition is met. Must be implemented by subclass. Args: stop_condition_net: A net invoked to evaluate an early termination condition. z0Subclass must implement `check_limiter_conditionNr#r rhrrrrgsz+ReaderWithLimitBase.check_limiter_conditioncCs|jS)a Return a blob that can be checked after the end of the reading task, which will contain a scalar float indicating whether the underlying reader has been exhausted (True) or whether we stopped because reached the limit of iterations (False). )rdrrrrrbsz!ReaderWithLimitBase.data_finishedN) r,r-r.r/rrrrfrgrbrrrrra<s '  racs2eZdZdZd fdd ZddZddZZS) ReaderWithLimitzReader that stops after `num_iter` batches. If `num_iter` <= 0 or is None, reverts to an unconstrained reader that exports a boolean blob indicating that the reader has exhausted the data steam. rMcs@tt||d|_||_|jdk r<|j|jd|_dS)ayClass initializer. Args: reader: The underlying reader object doing the actual read. num_iter: Number of batches to read. If `None`, the class reverts to a normal reader except that it also produces a data_finished blob as a side effect to indicate whether the input stream is exhausted. NrU)superrjrrUnum_iterr%rcr9)r r(rl __class__rrrs   zReaderWithLimit.__init__cCs$|jr |jg|jgt|jddS)NrV)rUrZintrlr^rrrrfs zReaderWithLimit.setup_limitercCs2|jr||jgdS|jgdgdtjjdSdSNrMFre)rUZ CountDownr[rr\r]rirrrrgsz'ReaderWithLimit.check_limiter_condition)rMr,r-r.r/rrfrg __classcell__rrrmrrjsrjcCs tt|Sr)rjrQ)rlrrr CountUntilsrscs2eZdZdZd fdd ZddZddZZS) ReaderWithTimeLimitzReader that stops after `duration` seconds. If `duration` <= 0 or is None, reverts to an unconstrained reader that exports a boolean blob indicating that the reader has exhausted the data steam. rcs&tt||d|_||_d|_dS)aClass initializer. Args: reader: The underlying reader object doing the actual read. duration: Number of seconds to read. If un-specified, None, or <= 0, the class reverts to a normal reader except that it also produces a data_finished blob as a side effect to indicate whether the input stream is exhausted. N)rkrtrtimerdurationduration_ns_blob)r r(rvrmrrrs zReaderWithTimeLimit.__init__cCsd|jdk r`|jdkr`t|jd}|jgdd|_||j}|j|g|d|_||jggdS)Nriʚ;Z epoch_timer)Z counter_name)rY)rvroZ TimerBeginruTimerGetr[rwZTimerEnd)r rHr_Z duration_ns start_timerrrrfs z!ReaderWithTimeLimit.setup_limitercCsF|jr*||j}|||jgt|jS|jgdgdtj j dSdSrp) rvrxruZGErwstrr r[rr\r])r rhZ time_elapsedrrrrgs z+ReaderWithTimeLimit.check_limiter_condition)rrqrrrmrrts rtc@s(eZdZdZddZddZddZdS) ReaderWithDelayz?Test reader class that inserts a delay between reading batches.cCs tj||jd||_||_dS)NrS)rrr r(delay)r r(r|rrrrszReaderWithDelay.__init__cCs|j||dSr)r(rr^rrrrszReaderWithDelay.setup_excs:td}fdd}||gg|gfj|S)NrcstjdSr)timesleepr|)argsZargdrrrsleep_opsz)ReaderWithDelay.read_ex..sleep_op)rrPythonr(r)r rrrrrrrrs  zReaderWithDelay.read_exNr`rrrrr{sr{cs8eZdZdZfddZddZddZdd ZZS) CompositeReaderzu Base class for a reader that wrap multiple readers, e.g., reading from multiple sources simultaneously. csJt|t|ksttt|jtddt||Dd||_||_dS)z Args: names: list[str] names of readers; used as schema keys readers: list[Reader] Reader instances, must have schema cSsg|]\}}||fqSrrS).0namer(rrr sz,CompositeReader.__init__..rSN) lenr rkrrrzip_names_readers)r namesreadersrmrrr s  zCompositeReader.__init__cCs|jD]}|||qdSr)rr)r rrr(rrrrs zCompositeReader.setup_excCsg}g}g}t|j|jD]<\}}|||\}} } || |||| qg} |d} t|j||D]H\}}} | || | krqrtd |}| | | g| | |qr| | |fS)z7 Stops when one of the reader finished z{}_stop) rrrr"appendextendr3rrr+r))r rrr!Z stop_blobsZall_sub_read_netsrr(Z sub_read_netsr recordZ read_netsZlocal_should_stopr7Zstop_netrrrrs*     zCompositeReader.read_excCs|jD]}||qdSr)rr&)r r%r(rrrr&;s zCompositeReader.reset) r,r-r.r/rrrr&rrrrrmrrs  rcs8eZdZdZfddZddZddZdd ZZS) CompositeReaderBuilderz. A reader builder for CompositeReader cs8tt|||_||_tddt||D|_dS)z Args: names: list[str] names of readers; used as schema keys reader_builders: list[ReaderBuilder] ReaderBuilder instances; must have schema cSsg|]\}}||fqSrrS)rrreader_builderrrrrNsz3CompositeReaderBuilder.__init__..N)rkrrr_reader_buildersrrr )r rZreader_buildersrmrrrDs zCompositeReaderBuilder.__init__cCs|jSrrrrrrr SszCompositeReaderBuilder.schemac Ksi}d|kr|d}nd}t|jD]\}}|t|jdkrP|dk rP||d<|jf|}t|t|@}t|t|@}|tkstd ||tkstd || |q&|S)NlimiterrMzOverlapping keys: {}zOverlapping values: {}) pop enumeraterrr>setkeysvaluesr r+update) r r=Zdata_finished_blobsrirZsub_reader_data_finished_blobsZoverlapping_keysZoverlapping_valuesrrrr>Vs   zCompositeReaderBuilder.setupcKspg}|jD]B}|jf|}t|tr&nt|dr:|}ntd||q t|j |}| |j kslt |S)Nr(z,reader must be an instance of Reader or Pipe) rr?rrhasattrr( ValueErrorrrrr r r )r r=rrr(Z multi_readerrrrr?ms       z!CompositeReaderBuilder.new_reader) r,r-r.r/rr r>r?rrrrrmrr@s  r)r/Z caffe2.pythonrZcaffe2.python.schemarrrZnumpyrTr}objectrr0r;r@rDrQrarjrsrtr{rrrrrrs" u@!&^&.9