o 0 hB0@sddlZddlmZddlmZmZddlmZmZm Z m Z ddl Z ddl mZddlmZmZddlmZgdZe d d d Zd efd dZedGdddeeZedGdddeZedGdddeeZdS)N) defaultdict)IteratorSized)AnyCallableOptionalTypeVar)functional_datapipe) DataChunk IterDataPipe)_check_unpickable_fn)BatcherIterDataPipeGrouperIterDataPipeUnBatcherIterDataPipe_T_coT) covariantnamecCsN|dvrtjd|d|dtddttjjjjj |St dt d|) N)SHARDING_PRIORITIESShardingFilterIterDataPipe`zc` from `torch.utils.data.datapipes.iter.grouping` is going to be removed in PyTorch 2.1Please use `z5` from the `torch.utils.data.datapipes.iter.sharding`)category stacklevelzmodule z has no attribute ) warningswarn FutureWarninggetattrtorchutilsdata datapipesitershardingAttributeError__name__)rr%\/var/www/vscode/kcb/lib/python3.10/site-packages/torch/utils/data/datapipes/iter/grouping.py __getattr__sr'batchc szeZdZUdZeed<eed<eed<defdededede eddf fd d Z de efd d Z defd dZ ZS)r a2 Creates mini-batches of data (functional name: ``batch``). An outer dimension will be added as ``batch_size`` if ``drop_last`` is set to ``True``, or ``length % batch_size`` for the last batch if ``drop_last`` is set to ``False``. Args: datapipe: Iterable DataPipe being batched batch_size: The size of each batch drop_last: Option to drop the last batch if it's not full wrapper_class: wrapper to apply onto each batch (type ``List``) before yielding, defaults to ``DataChunk`` Example: >>> # xdoctest: +SKIP >>> from torchdata.datapipes.iter import IterableWrapper >>> dp = IterableWrapper(range(10)) >>> dp = dp.batch(batch_size=3, drop_last=True) >>> list(dp) [[0, 1, 2], [3, 4, 5], [6, 7, 8]] datapipe batch_size drop_lastF wrapper_classreturnNcs6|dksJdt||_||_||_||_dS)Nrz+Batch size is required to be larger than 0!)super__init__r)r*r+r,)selfr)r*r+r, __class__r%r&r/As   zBatcherIterDataPipe.__init__ccsdg}|jD]}||t||jkr||Vg}qt|dkr.|js0||VdSdSdSNr)r)appendlenr*r,r+)r0r(xr%r%r&__iter__Os    zBatcherIterDataPipe.__iter__cCsPt|jtr|jrt|j|jSt|j|jd|jStt|jd)Nz# instance doesn't have valid length) isinstancer)rr+r5r* TypeErrortyper$r0r%r%r&__len__Zs zBatcherIterDataPipe.__len__)r$ __module__ __qualname____doc__r __annotations__intboolr r;r/rr7r= __classcell__r%r%r1r&r %s(  r unbatchc@s4eZdZdZd dedefddZddZd d Zd S) ra  Undos batching of data (functional name: ``unbatch``). In other words, it flattens the data up to the specified level within a batched DataPipe. Args: datapipe: Iterable DataPipe being un-batched unbatch_level: Defaults to ``1`` (only flattening the top level). If set to ``2``, it will flatten the top two levels, and ``-1`` will flatten the entire DataPipe. Example: >>> # xdoctest: +SKIP >>> from torchdata.datapipes.iter import IterableWrapper >>> source_dp = IterableWrapper([[[0, 1], [2]], [[3, 4], [5]], [[6]]]) >>> dp1 = source_dp.unbatch() >>> list(dp1) [[0, 1], [2], [3, 4], [5], [6]] >>> dp2 = source_dp.unbatch(unbatch_level=2) >>> list(dp2) [0, 1, 2, 3, 4, 5, 6] r8r) unbatch_levelcCs||_||_dSN)r)rF)r0r)rFr%r%r&r/|s zUnBatcherIterDataPipe.__init__ccs(|jD] }|j||jdEdHqdS)NrF)r)_diverF)r0elementr%r%r&r7s zUnBatcherIterDataPipe.__iter__ccs|dkr td|dkr*t|ttfr%|D] }|j|ddEdHqdS|VdS|dkr3|VdSt|ttfrM|D]}|j||ddEdHq= 0rHrr8zunbatch_level z" exceeds the depth of the DataPipe) ValueErrorr9listr rI IndexErrorrF)r0rJrFitemr%r%r&rIs$   zUnBatcherIterDataPipe._diveN)r8) r$r>r?r@r rBr/r7rIr%r%r%r&rds  rgroupbyc@seZdZdZdddddddeedeegefded e d e e d e e d efd dZ ddZ ddZ dddZddZddZddZdS)ra! Groups data from IterDataPipe by keys from ``group_key_fn``, yielding a ``DataChunk`` with batch size up to ``group_size``. (functional name: ``groupby``). The samples are read sequentially from the source ``datapipe``, and a batch of samples belonging to the same group will be yielded as soon as the size of the batch reaches ``group_size``. When the buffer is full, the DataPipe will yield the largest batch with the same key, provided that its size is larger than ``guaranteed_group_size``. If its size is smaller, it will be dropped if ``drop_remaining=True``. After iterating through the entirety of source ``datapipe``, everything not dropped due to the buffer capacity will be yielded from the buffer, even if the group sizes are smaller than ``guaranteed_group_size``. Args: datapipe: Iterable datapipe to be grouped group_key_fn: Function used to generate group key from the data of the source datapipe keep_key: Option to yield the matching key along with the items in a tuple, resulting in `(key, [items])` otherwise returning [items] buffer_size: The size of buffer for ungrouped data group_size: The max size of each group, a batch is yielded as soon as it reaches this size guaranteed_group_size: The guaranteed minimum group size to be yielded in case the buffer is full drop_remaining: Specifies if the group smaller than ``guaranteed_group_size`` will be dropped from buffer when the buffer is full Example: >>> import os >>> # xdoctest: +SKIP >>> from torchdata.datapipes.iter import IterableWrapper >>> def group_fn(file): ... return os.path.basename(file).split(".")[0] >>> source_dp = IterableWrapper(["a.png", "b.png", "a.json", "b.json", "a.jpg", "c.json"]) >>> dp0 = source_dp.groupby(group_key_fn=group_fn) >>> list(dp0) [['a.png', 'a.json', 'a.jpg'], ['b.png', 'b.json'], ['c.json']] >>> # A group is yielded as soon as its size equals to `group_size` >>> dp1 = source_dp.groupby(group_key_fn=group_fn, group_size=2) >>> list(dp1) [['a.png', 'a.json'], ['b.png', 'b.json'], ['a.jpg'], ['c.json']] >>> # Scenario where `buffer` is full, and group 'a' needs to be yielded since its size > `guaranteed_group_size` >>> dp2 = source_dp.groupby(group_key_fn=group_fn, buffer_size=3, group_size=3, guaranteed_group_size=2) >>> list(dp2) [['a.png', 'a.json'], ['b.png', 'b.json'], ['a.jpg'], ['c.json']] Fi'N)keep_key buffer_size group_sizeguaranteed_group_sizedrop_remainingr) group_key_fnrQrRrSrTrUcCst|||_||_||_||_tt|_d|_||_ d|_ |dur7|dur7d|kr1|ks4JJ||_ |durP|durKd|krJ|ksMJJ||_ ||_ t |_ dSr3)r r)rVrQmax_buffer_sizerrMbuffer_elementscurr_buffer_sizerSrTrUr r,)r0r)rVrQrRrSrTrUr%r%r&r/s"  $ zGrouperIterDataPipe.__init__cCsd}d}d}|jD]}t|j||krt|j|}|}q |jdur7||jkr7|js7tdt|j||jdusA||jkrF|j|}|j|8_|j|=|S)NrzFailed to group items)rXkeysr5rTrU RuntimeErrorstrrY)r0 biggest_key biggest_sizeresult_to_yieldfindkeyr%r%r&_remove_biggest_keys*     z'GrouperIterDataPipe._remove_biggest_keyccs"|jD]d}||}|j|||jd7_|jdurK|jt|j|krK||j|}|jr8||fn|V|jt|j|8_|j|=|j|j krh| }|durh||}|jre||fn|Vqt |j D]}||j |}|jt|8_|jr||fn|VqpdS)Nr8)r)rVrXr4rYrSr5r,rQrWratuplerZpop)r0r6keyresultr_r%r%r&r7s.    zGrouperIterDataPipe.__iter__r-cCsd|_tt|_dSr3)rYrrMrXr<r%r%r&resetszGrouperIterDataPipe.resetc CsD|j|j|j|j|j|j|j|j|j|j f }t j dur t |S|SrG) r)rVrQrWrSrTrUr,_valid_iterator_id_number_of_samples_yieldedr getstate_hookr0stater%r%r& __getstate__ s z GrouperIterDataPipe.__getstate__c Cs@|\ |_|_|_|_|_|_|_|_|_|_ d|_ t t |_ dSr3)r)rVrQrWrSrTrUr,rgrhrYrrMrXrjr%r%r& __setstate__1s z GrouperIterDataPipe.__setstate__cCs|jdSrG)rXclearr<r%r%r&__del__AszGrouperIterDataPipe.__del__)r-N)r$r>r?r@r rrrrCrBrr/rar7rfrlrmror%r%r%r&rs81    r)r collectionsrcollections.abcrrtypingrrrr(torch.utils.data.datapipes.iter.shardingr%torch.utils.data.datapipes._decoratorr #torch.utils.data.datapipes.datapiper r 'torch.utils.data.datapipes.utils.commonr __all__rr\r'r rrr%r%r%r&s"    >4