o wZhB0@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)ZSHARDING_PRIORITIESZShardingFilterIterDataPipe`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 FutureWarninggetattrtorchutilsdataZ datapipesiterZshardingAttributeError__name__)rr!W/var/www/auris/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')r,r$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'r1r& TypeErrortyper r,r!r!r"__len__Zs zBatcherIterDataPipe.__len__)r __module__ __qualname____doc__r __annotations__intboolr r7r+rr3r9 __classcell__r!r!r-r"r %s(  r Zunbatchc@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] r4r% unbatch_levelcCs||_||_dSN)r%rA)r,r%rAr!r!r"r+|s zUnBatcherIterDataPipe.__init__ccs(|jD] }|j||jdEdHqdS)NrA)r%_diverA)r,elementr!r!r"r3s zUnBatcherIterDataPipe.__iter__ccs|dkr td|dkr*t|ttfr%|D] }|j|ddEdHqdS|VdS|dkr3|VdSt|ttfrM|D]}|j||ddEdHq= 0rCrr4zunbatch_level z" exceeds the depth of the DataPipe) ValueErrorr5listr rD IndexErrorrA)r,rErAitemr!r!r"rDs$   zUnBatcherIterDataPipe._diveN)r4) r r:r;r<r r>r+r3rDr!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_fnrLrMrNrOrPcCst|||_||_||_||_tt|_d|_||_ d|_ |dur7|dur7d|kr1|ks4JJ||_ |durP|durKd|krJ|ksMJJ||_ ||_ t |_ dSr/)r r%rQrLmax_buffer_sizerrHbuffer_elementscurr_buffer_sizerNrOrPr r()r,r%rQrLrMrNrOrPr!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)rSkeysr1rOrP RuntimeErrorstrrT)r,Z biggest_keyZ biggest_sizeresult_to_yieldZfindkeyr!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)Nr4)r%rQrSr0rTrNr1r(rLrRrYtuplerUpop)r,r2keyresultrXr!r!r"r3s.    zGrouperIterDataPipe.__iter__r)cCsd|_tt|_dSr/)rTrrHrSr8r!r!r"resetszGrouperIterDataPipe.resetc CsD|j|j|j|j|j|j|j|j|j|j f }t j dur t |S|SrB) r%rQrLrRrNrOrPr(_valid_iterator_id_number_of_samples_yieldedr Z getstate_hookr,stater!r!r" __getstate__ s z GrouperIterDataPipe.__getstate__c Cs@|\ |_|_|_|_|_|_|_|_|_|_ d|_ t t |_ dSr/)r%rQrLrRrNrOrPr(r_r`rTrrHrSrar!r!r" __setstate__1s z GrouperIterDataPipe.__setstate__cCs|jdSrB)rSclearr8r!r!r"__del__AszGrouperIterDataPipe.__del__)r)N)r r:r;r<r rrrr?r>rr+rYr3r^rcrdrfr!r!r!r"rs81    r)r collectionsrcollections.abcrrtypingrrrrZ(torch.utils.data.datapipes.iter.shardingrZ%torch.utils.data.datapipes._decoratorr Z#torch.utils.data.datapipes.datapiper r Z'torch.utils.data.datapipes.utils.commonr __all__rrWr#r rrr!r!r!r"s"    >4