U Mh/@sddlZddlmZddlmZmZmZmZmZm Z m Z m Z ddl Z ddlmZddlmZmZddlmZddd gZe d d d Zed ddZedGdddeeZedGdd d eZedGdddeeZdS)N) defaultdict)AnyCallable DefaultDictIteratorListOptionalSizedTypeVar)functional_datapipe) DataChunk IterDataPipe)_check_unpickable_fnBatcherIterDataPipeGrouperIterDataPipeUnBatcherIterDataPipeT_coT) covariantnamecCsR|dkr:tjd|d|dtddttjjjjj |St dt d|dS) 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$Z/var/www/html/venv/lib/python3.8/site-packages/torch/utils/data/datapipes/iter/grouping.py __getattr__sr&batchcsleZdZUdZeed<eed<eed<defeeeddfdd Z e ed d d Z ed d dZ Z S)ra2 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_lastFN)r(r)r*returncs6|dkstdt||_||_||_||_dS)Nrz+Batch size is required to be larger than 0!)AssertionErrorsuper__init__r(r)r* wrapper_class)selfr(r)r*r/ __class__r$r%r.:s  zBatcherIterDataPipe.__init__r+ccsZg}|jD],}||t||jkr ||Vg}q t|dkrV|jsV||VdSNr)r(appendlenr)r/r*)r0r'xr$r$r%__iter__Gs    zBatcherIterDataPipe.__iter__cCsVt|jtr>|jr"t|j|jSt|j|jd|jSntt|jddS)Nz# instance doesn't have valid length) isinstancer(r r*r6r) TypeErrortyper#r0r$r$r%__len__Rs  zBatcherIterDataPipe.__len__)r# __module__ __qualname____doc__r __annotations__intboolr r.rr8r> __classcell__r$r$r1r%rs   Zunbatchc@s2eZdZdZd eedddZddZdd 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] r9r( unbatch_levelcCs||_||_dSNrF)r0r(rGr$r$r%r.tszUnBatcherIterDataPipe.__init__ccs&|jD]}|j||jdEdHqdS)NrG)r(_diverG)r0elementr$r$r%r8zs zUnBatcherIterDataPipe.__iter__ccs|dkrtd|dkrNt|ttfrF|D]}|j|ddEdHq*q|VnT|dkr^|VnDt|ttfr|D]}|j||ddEdHqpntd|jddS)Nz unbatch_level must be -1 or >= 0rIrr9zunbatch_level z" exceeds the depth of the DataPipe) ValueErrorr:listr rJ IndexErrorrG)r0rKrGitemr$r$r%rJ~szUnBatcherIterDataPipe._diveN)r9) r#r?r@rAr rCr.r8rJr$r$r$r%r\s groupbyc @seZdZdZddddddeeeegefee e e e e edddZ d d Z d d Z dd ddZddZddZddZdS)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_remaining)r( group_key_fnrRrSrTrUrVcCst|||_||_||_||_tt|_d|_||_ d|_ |dk rl|dk rld|kr`|ksfnt ||_ |dk r|dk rd|kr|ksnt ||_ ||_ t |_dSr4)rr(rWrRmax_buffer_sizerrNbuffer_elementscurr_buffer_sizerTrUr,rVr r/)r0r(rWrRrSrTrUrVr$r$r%r.s"  "zGrouperIterDataPipe.__init__cCsd}d}d}|jD](}t|j||krt|j|}|}q|jdk rn||jkrn|jsntdt|j||jdks||jkr|j|}|j|8_|j|=|S)NrzFailed to group items)rYkeysr6rUrV RuntimeErrorstrrZ)r0Z biggest_keyZ biggest_sizeresult_to_yieldZfindkeyr$r$r%_remove_biggest_keys z'GrouperIterDataPipe._remove_biggest_keyccs"|jD]}||}|j|||jd7_|jdk r|jt|j|kr||j|}|jrn||fn|V|jt|j|8_|j|=|j|j kr| }|dk r||}|jr||fn|Vqt |j D]>}||j |}|jt|8_|jr||fn|VqdS)Nr9)r(rWrYr5rZrTr6r/rRrXr_tupler[pop)r0r7keyresultr^r$r$r%r8s$    zGrouperIterDataPipe.__iter__r3cCsd|_tt|_dSr4)rZrrNrYr=r$r$r%resetszGrouperIterDataPipe.resetc CsD|j|j|j|j|j|j|j|j|j|j f }t j dk r@t |S|SrH) r(rWrRrXrTrUrVr/_valid_iterator_id_number_of_samples_yieldedr Z getstate_hookr0stater$r$r% __getstate__ s  z GrouperIterDataPipe.__getstate__c Cs@|\ |_|_|_|_|_|_|_|_|_|_ d|_ t t |_ dSr4)r(rWrRrXrTrUrVr/rerfrZrrNrYrgr$r$r% __setstate__s z GrouperIterDataPipe.__setstate__cCs|jdSrH)rYclearr=r$r$r%__del__,szGrouperIterDataPipe.__del__)r#r?r@rAr rrrrDrCrr.r_r8rdrirjrlr$r$r$r%rs*0  )r collectionsrtypingrrrrrrr r Z(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__rr]r&rrrr$r$r$r%s$ (    =4