o ̳i@sddlmZmZmZmZmZmZmZmZddl m Z ddl m Z m Z ddlmZddlmZeddd ZGd d d e e Zd eee fd e ede e fddZGddde e ZdS))AnyCallableDictIterableIteratorMappingOptionalTypeVar)Sampler)BaseNodeT)Mapper)StatefulKT) covariantcsveZdZdZdZdZdeeffdd Zdde e e e fffdd Z d efd d Zd e e e ffd dZZS)IterableWrappera Thin Wrapper that converts any Iterable (including torch.utils.data.IterableDataset) in to a BaseNode. If iterable implements the Stateful Protocol, it will be saved and restored with its state_dict/load_state_dict methods. Args: iterable (Iterable[T]): Iterable to convert to BaseNode. IterableWrapper calls iter() on it. :warning: Note the distinction between state_dict/load_state_dict defined on Iterable, vs Iterator. Only the Iterable's state_dict/load_state_dict are used. _num_yieldediterablecst||_d|_dSN)super__init__r_it)selfr __class__L/home/ubuntu/.local/lib/python3.10/site-packages/torchdata/nodes/adapters.pyr&s  zIterableWrapper.__init__N initial_statec sd|_d|_t||durW||j|_t|jtr-|j||j t |j|_dSt |j|_t |jD]}zt |jWq8t yTtd|jd|dwdSt |j|_dS)NrTried to fast-forward / items during init but hit StopIteration after 4 items, this is likely a bug or malformed state_dict)rrrresetNUM_YIELDED_KEY isinstancerrload_state_dict ITERABLE_KEYiterrangenext StopIteration ValueErrorrrirrrr"+s*       zIterableWrapper.resetreturncCst|j}|jd7_|SNr )r)rrritemrrrr)Bs zIterableWrapper.nextcCs,|j|ji}t|jtr|j||j<|Sr)r#rr$rr state_dictr&rr2rrr get_stateGs  zIterableWrapper.get_stater)__name__ __module__ __qualname____doc__r#r&rr rrrstrrr"r)r4 __classcell__rrrrrs  r map_datasetsamplerr.cCst|}t||j}|S)aThin Wrapper that converts any MapDataset in to a torchdata.node If you want parallelism, copy this and replace Mapper with ParallelMapper. Args: map_dataset (Mapping[K, T]): - Apply map_dataset.__getitem__ to the outputs of sampler. sampler (Sampler[K]): )SamplerWrapperr __getitem__)r;r< sampler_node mapper_noderrrMapStyleWrapperNs rAc seZdZdZdZdZdZ  ddeede d e e e ge fffd d Z dd e e eefffd d ZdefddZde eeffddZede de fddZZS)r=a Convert a sampler into a BaseNode. This is nearly identical to IterableWrapper except it includes a hook to call set_epoch on the sampler, if it supports it. Args: sampler (Sampler): Sampler to wrap. initial_epoch (int): initial epoch to set on the sampler epoch_updater (Optional[Callable[[int], int]] = None): callback to update epoch at start of new iteration. It's called at the beginning of each iterator request, except the first one. r_epoch_samplerrNr< initial_epoch epoch_updatercs8t||_||_d|_d|_|p|j|_d|_dS)NrF) rrr<epochr_started_default_epoch_updaterrEr)rr<rDrErrrrks   zSamplerWrapper.__init__rc st||durb||j|_||j|_t|jtr,|j ||j t |j|_ nVt |jdr9|j|jt |j|_ t|jD]}zt|j WqDty`td|jd|dwn d|_|jro||j|_t |jdr||j|jt |j|_ d|_dS)N set_epochrr r!rF)rr"r#r EPOCH_KEYrFr$r<rr% SAMPLER_KEYr'rhasattrrIr(r)r*r+rGrEr,rrrr"ys8            zSamplerWrapper.resetr.cCs"d|_t|j}|jd7_|S)NTr )rGr)rrr0rrrr)s zSamplerWrapper.nextcCs4|j|j|j|ji}t|jtr|j||j<|Sr) r#rrJrFr$r<rr2rKr3rrrr4s  zSamplerWrapper.get_staterFcCs|dSr/r)clsrFrrrrHsz%SamplerWrapper._default_epoch_updater)rNr)r5r6r7r8r#rJrKr r intrrrrr9rr"r)r4 classmethodrHr:rrrrr=[s&   r=N)typingrrrrrrrr torch.utils.datar torchdata.nodes.base_noder r maprtypesrrrrAr=rrrrs(    &9