/usr/local/lib64/python3.6/site-packages/caffe2/python/__pycache__
Edit: /usr/local/lib64/python3.6/site-packages/caffe2/python/__pycache__/dataio.cpython-36.pyc (23978B)
3
Eg[ @ s d Z ddlmZ ddlmZmZmZ ddlZddl Z G dd de
Z G dd de
Z G d d
d
e
Z
G d d d e
ZG d
d de
ZG dd de ZG dd de ZG dd deZdd ZG dd deZG dd de ZG dd d e ZG d d d e
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 @ sd e Zd ZdZdddZdd Zdd Zd d
Zd d Zd
d Z dd Z
dd Z dd Z 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.
Nc C s |d k rt |tst|| _d S )N)
isinstancer AssertionError_schema)selfschema r @/usr/local/lib64/python3.6/site-packages/caffe2/python/dataio.py__init__' s zReader.__init__c C s | j d k std| j S )Nz$Schema not provided for this reader.)r r )r
r r r
r , s z
Reader.schemac C s
|| _ d S )N)r )r
r r r r
_set_schema0 s zReader._set_schemac C s dS )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.
Nr )r
init_net
finish_netr r r
setup_ex3 s zReader.setup_exc C s t jd}|gf| j| S )N reader_body)r Netread)r
local_init_netlocal_finish_netread_netr r r
read_ex< s
zReader.read_exc C s. | j ||\}}}| jr$t| j|}|||fS )N)r r r )r
r r nets should_stopfieldsr r r
read_record_ex@ s
zReader.read_record_exc C s t ddS )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.
z Readers must implement `read`.N)NotImplementedError)r
r r r r
r G s z Reader.readc C s t ddS )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.
z This reader cannot be resetted.N)r )r
netr r r
reset] s z Reader.resetc C s( | j |\}}| jr t| j|}||fS )N)r r r )r
r r r r r r
read_recorde s zReader.read_recordc C sP t j|p
d}| j|\}}|dk r2|j||g}t jdj|||d}||fS )aQ Create 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)r r r! Orexecution_stepformat)r
Zreader_net_nameZexternal_should_stopZ
reader_netr r Z read_stepr r r
r$ k s zReader.execution_step)N)NN)__name__
__module__ __qualname____doc__r r r r r r r r r! r$ r r r r
r s
r c @ sN e Zd ZdZdZdd Zdd Zdd Zd d
Zd d Z dd
dZ
dd Z 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.
Nc C s | j S )N)r )r
r r r
r s z
Writer.schemac C s t ddS )aZ Add 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.
z Writers must implement write.N)r )r
writer_netr r r r
write s
z Writer.writec C s( t |tr|| _|j }| j|| d S )N)r r r field_blobsr, )r
r+ r r r r
write_record s
zWriter.write_recordc C s | j | dS )zExperimental, don't use yetN)commit)r
r r r r r
r s zWriter.setup_exc C s t jd}| j|| |gS )z6Experimental extension to the interface. Don't use yet write_net)r r r, )r
r r r stop_blobr0 r r r
write_ex s
zWriter.write_exc C sB t |tr|| _|j }|dkr*|jd}| j||||}||fS )z7Experimental extension to the interface. Don't use yet.NZdequeue_status)r r r r- NextNamer2 )r
r r r r1 Z
write_netsr r r
write_record_ex s
zWriter.write_record_exc C s dS )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
r r r r
r/ s z
Writer.commit)N) r& r' r( r) r r r, r. r r2 r4 r/ r r r r
r* s
r* c @ s( e Zd ZdZdd Zdd Zdd ZdS )
ReaderBuilderz1 Allow usage of a reader in distributed fashion. c C s
t d S )N)r )r
r r r
r s zReaderBuilder.schemac K s
t dS )z
Optionally, perform one-time setup before calling new_reader().
Subclass should make sure this function is only called once.
N)r )r
kwargsr r r
setup s zReaderBuilder.setupc K s
t d S )N)r )r
r6 r r r
new_reader s zReaderBuilder.new_readerN)r& r' r( r) r r7 r8 r r r r
r5 s r5 c @ s0 e Zd ZdZdd Zdd Zdd Zdd Zd
S ) PipedReaderBuildera ReaderBuilder 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))
c C s || _ || _d S )N)_builder_piper)r
ZbuilderZpiperr r r
r s zPipedReaderBuilder.__init__c C s
| j j S )N)r: r )r
r r r
r s zPipedReaderBuilder.schemac K s | j jf |S )N)r: r7 )r
r6 r r r
r7 s zPipedReaderBuilder.setupc K s4 | j f d| jjf |i|}t|tr,|S |j S )Nr" )r; r: r8 r r r" )r
r6 outputr r r
r8 s z PipedReaderBuilder.new_readerN)r& r' r( r) r r r7 r8 r r r r
r9 s
r9 c @ sV e Zd ZdddZdd Zdd Zdd Zd
d Zd d
Zdd Z dd Z
dd Z dS )PipeNc C s d| _ d| _|| _|| _d S )Nr ) _num_writers _num_readersr _obj_key)r
r Zobj_keyr r r
r s z
Pipe.__init__c C s | j S )N)r )r
r r r
r
s z Pipe.schemac C s d S )Nr )r
global_init_netr r r
r7
s z
Pipe.setupc C s
t d S )N)r )r
r r r
r" s z Pipe.readerc C s
t d S )N)r )r
r r r
writer s z Pipe.writerc C s | j S )N)r? )r
r r r
num_readers s zPipe.num_readersc C s | j S )N)r> )r
r r r
num_writers s zPipe.num_writersc C sB |d k r| j d kr|| _ | jd7 _| jd k r>|j| j| d S )N )r r> r@
add_attribute)r
Z
writer_schemaZwriter_init_netr r r
_new_writer s
zPipe._new_writerc C s* | j d7 _ | jd k r&|j| j| d S )NrE )r? r@ rF )r
Zreader_init_netr r r
_new_reader# s
zPipe._new_reader)NN) r& r' r( r r r7 r" rB rC rD rG rH r r r r
r= s
r= c @ s( e Zd ZdZdd Zdd Zdd ZdS )
CounterReaderz+ Reader that produces increasing integers. c C s( t j| tdtjfd d | _d | _d S )Niter)r )r r r npZint64counterr )r
r r r
r + s zCounterReader.__init__c C s6 | j d kr2|jg dd| _ |jg g tjjdd| _d S )Nr )
init_countF)shapedtypevalue)rL
CreateCounter ConstantFillr DataTypeBOOLr )r
rA global_finish_netr r r
r 0 s
zCounterReader.setup_exc C s* t jd}|j| jgd}|g| j|gfS )NZlimited_reader_counterrE )r r ZCountUprL r )r
r r Z count_netrP r r r
r 6 s
zCounterReader.read_exN)r& r' r( r) r r r r r r r
rI ) s rI c @ s@ e Zd ZdZdd Zdd Zdd Zdd Zd
d Zd d
Z dS )ReaderWithLimitBaseap Abstract 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).
c C sB t j| |jd || _tjd| _| jj| jjd| _ d | _
d S )N)r Zreader_with_limit
data_finished) r r r r" r r r AddExternalInputr3 _data_finishedr )r
r" r r r
r F s z ReaderWithLimitBase.__init__c C s: |j g | jgg dtjjd | jj|| | j|| d S )NF)rN rP rO )rR rY r rS rT r" r
setup_limiter)r
rA rU r r r
r N s
z ReaderWithLimitBase.setup_exc C sx t jd}| j|}| jj||\}}}| j| jj t jd}|j|| |j| j |g| j g |g| |g ||fS )a Reads 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)
r r check_limiter_conditionr" r r r ZCopyr# rY ) r
r r stop_condition_netr r Zlocal_data_finishedr Zcheck_done_netr r r
r U s
zReaderWithLimitBase.read_exc C s t ddS )a Configure 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`N)r )r
rA rU r r r
rZ | s z!ReaderWithLimitBase.setup_limiterc C s t ddS )a Configure 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_conditionN)r )r
r\ r r r
r[ s z+ReaderWithLimitBase.check_limiter_conditionc C s | j S )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).
)rY )r
r r r
rW s z!ReaderWithLimitBase.data_finishedN)
r& r' r( r) r r r rZ r[ rW r r r r
rV < s '
rV c s2 e Zd ZdZd fdd Zdd Zdd Z ZS )
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.
rE c s@ t t| j| d| _|| _| jdk r<| jj| jjd| _dS )ay Class 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.
NrL )superr] r rL num_iterr rX r3 )r
r" r_ ) __class__r r
r s
zReaderWithLimit.__init__c C s$ | j r |jg | j gt| jd d S )N)rM )rL rQ intr_ )r
rA rU r r r
rZ s z ReaderWithLimit.setup_limiterc C s2 | j r|j| j gdS |jg dg dtjjdS d S )NrE F)rN rP rO )rL Z CountDownrR r rS rT )r
r\ r r r
r[ s
z'ReaderWithLimit.check_limiter_condition)rE )r& r' r( r) r rZ r[
__classcell__r r )r` r
r] s r] c C s t t | S )N)r] rI )r_ r r r
CountUntil s rc c s2 e Zd ZdZd fdd Zdd Zdd Z ZS )
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.
r c s& t t| j| d| _|| _d| _dS )a Class 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)r^ rd r timerdurationduration_ns_blob)r
r" rf )r` r r
r s
z ReaderWithTimeLimit.__init__c C sd | j d k r`| j dkr`t| j d }|jg dd| _|j| j}|j|g|d| _|j| jgg d S )Nr
Z epoch_timer)Z counter_name)rP i ʚ;)rf ra Z
TimerBeginre TimerGetrR rg ZTimerEnd)r
rA rU Z duration_ns
start_timer r r
rZ s z!ReaderWithTimeLimit.setup_limiterc C sF | j r*|j| j}|j|| jgt| jS |jg dg dtj j
dS d S )NrE F)rN rP rO ) rf rj re ZGErg strr rR r rS rT )r
r\ Z time_elapsedr r r
r[ s z+ReaderWithTimeLimit.check_limiter_condition)r )r& r' r( r) r rZ r[ rb r r )r` r
rd s rd c @ s( e Zd ZdZdd Zdd Zdd ZdS ) ReaderWithDelayz?Test reader class that inserts a delay between reading batches.c C s t j| |jd || _|| _d S )N)r )r r r r" delay)r
r" rn r r r
r s zReaderWithDelay.__init__c C s | j j|| d S )N)r" r )r
rA rU r r r
r s zReaderWithDelay.setup_exc s: t jd} fdd}|j|g g |gf jj| S )Nr c s t j j d S )N)timesleeprn )argsZargd)r
r r
sleep_op s z)ReaderWithDelay.read_ex.
.sleep_op)r r Pythonr" r )r
r r r rr r )r
r
r s
zReaderWithDelay.read_exN)r& r' r( r) r r r r r r r
rm s rm c s8 e Zd ZdZ fddZdd Zdd Zdd Z ZS )
CompositeReaderzu
Base class for a reader that wrap multiple readers, e.g., reading from
multiple sources simultaneously.
c sJ t |t |ksttt| jtdd t||D d || _|| _dS )z
Args:
names: list[str] names of readers; used as schema keys
readers: list[Reader] Reader instances, must have schema
c S s g | ]\}}||j fqS r )r ).0namer" r r r
s z,CompositeReader.__init__..)r N) lenr r^ rt r r zip_names_readers)r
namesreaders)r` r r
r s
zCompositeReader.__init__c C s x| j D ]}|j|| qW d S )N)r{ r )r
r r r" r r r
r s zCompositeReader.setup_exc C s g }g }g }xNt | j| jD ]<\}}|j||\}} }
|j| |j| |j|
j q W g } |d } xZt | j||D ]H\}}}
| j| |
| krqxtjdj |}|j
| |
g| | j| qxW | | |fS )z7
Stops when one of the reader finished
rE z{}_stop) ry rz r{ r appendextendr- r r r% r# )r
r r r Z
stop_blobsZall_sub_read_netsrv r" Z
sub_read_netsr recordZ read_netsZlocal_should_stopr1 Zstop_netr r r
r s&
zCompositeReader.read_exc C s x| j D ]}|j| qW d S )N)r{ r )r
r r" r r r
r ; s zCompositeReader.reset) r& r' r( r) r r r r rb r r )r` r
rt s
rt c s8 e Zd ZdZ fddZdd Zdd Zdd Z ZS )
CompositeReaderBuilderz.
A reader builder for CompositeReader
c s8 t t| j || _|| _tdd t||D | _dS )z
Args:
names: list[str] names of readers; used as schema keys
reader_builders: list[ReaderBuilder] ReaderBuilder instances;
must have schema
c S s g | ]\}}||j fqS r )r )ru rv reader_builderr r r
rw O s z3CompositeReaderBuilder.__init__..N)r^ r r rz _reader_buildersr ry r )r
r| Zreader_builders)r` r r
r D s z CompositeReaderBuilder.__init__c C s | j S )N)r )r
r r r
r S s z CompositeReaderBuilder.schemac K s i }d|kr|j d}nd }xt| jD ]\}}|t| jd krR|d k rR||d< |jf |}t|j t|j @ }t|j t|j @ }|t kstdj ||t kstdj ||j
| q(W |S )NlimiterrE zOverlapping keys: {}zOverlapping values: {}) pop enumerater rx r7 setkeysvaluesr r% update) r
r6 Zdata_finished_blobsr ir Z sub_reader_data_finished_blobsZoverlapping_keysZoverlapping_valuesr r r
r7 V s z CompositeReaderBuilder.setupc K st g }xL| j D ]B}|jf |}t|tr(n t|dr<|j }ntd|j| q W t| j |}|j
| j kspt |S )Nr" z,reader must be an instance of Reader or Pipe)
r r8 r r hasattrr"
ValueErrorr rt rz r r r )r
r6 r} r r" Z multi_readerr r r
r8 m s
z!CompositeReaderBuilder.new_reader) r& r' r( r) r r r7 r8 rb r r )r` r
r @ s
r )r) Z
caffe2.pythonr Zcaffe2.python.schemar r r ZnumpyrK ro objectr r* r5 r9 r= rI rV r] rc rd rm rt r r r r r
s" u@!&^&.9