Z^(hjddlZddlZddlmZmZddlmZddlm Z m Z ddl m Z m Z mZmZmZddlmZddlmZddlmZdd lmZmZgd Zed d ZedGddeZedGddeZGddeZdZGddeeZ GddeZ!edGddeZ"GddeeZ#ed Gd!d"eZ$ed#Gd$d%ee%eZ&y)&N)ABCabstractmethod)deque)IteratorSized)AnyCallableLiteralOptionalTypeVar)functional_datapipe)_SnapshotState) IterDataPipe)_check_unpickable_fn StreamWrapper)ConcaterIterDataPipeDemultiplexerIterDataPipeForkerIterDataPipeMultiplexerIterDataPipeZipperIterDataPipe_T_coT) covariantconcatcFeZdZUdZeeed<defdZdefdZ de fdZ y)raN Concatenates multiple Iterable DataPipes (functional name: ``concat``). The resulting DataPipe will yield all the elements from the first input DataPipe, before yielding from the subsequent ones. Args: datapipes: Iterable DataPipes being concatenated Example: >>> # xdoctest: +REQUIRES(module:torchdata) >>> import random >>> from torchdata.datapipes.iter import IterableWrapper >>> dp1 = IterableWrapper(range(3)) >>> dp2 = IterableWrapper(range(5)) >>> list(dp1.concat(dp2)) [0, 1, 2, 0, 1, 2, 3, 4] datapipesc~t|dk(r tdtd|Ds td||_y)Nrz/Expected at least one DataPipe, but got nothingc3<K|]}t|tywN isinstancer.0dps g/var/www/skyplay_api_hub/venv/lib/python3.12/site-packages/torch/utils/data/datapipes/iter/combining.py z0ConcaterIterDataPipe.__init__..4DB:b,/Dz(Expected all inputs to be `IterDataPipe`)len ValueErrorall TypeErrorrselfrs r$__init__zConcaterIterDataPipe.__init__1s; y>Q NO OD)DDFG G"returnc#FK|jD] }|Ed{y7wr)r)r-r#s r$__iter__zConcaterIterDataPipe.__iter__8s#.. BMM  s !!ctd|jDrtd|jDStt |j d)Nc3<K|]}t|tywrr rr!s r$r%z/ConcaterIterDataPipe.__len__..=>z"e$>r'c32K|]}t|ywrr(r!s r$r%z/ConcaterIterDataPipe.__len__..>82s2w8# instance doesn't have valid length)r*rsumr+type__name__r-s r$__len__zConcaterIterDataPipe.__len__<G >t~~> >888 8tDz2233VWX Xr/N) r> __module__ __qualname____doc__tupler__annotations__r.rr2intr@r/r$rrs:$\""#<#(YYr/rforkc :eZdZdZ d dedededeedfdZy) ra Creates multiple instances of the same Iterable DataPipe (functional name: ``fork``). Args: datapipe: Iterable DataPipe being copied num_instances: number of instances of the datapipe to create buffer_size: this restricts how far ahead the leading child DataPipe can read relative to the slowest child DataPipe. Defaults to ``1000``. Use ``-1`` for the unlimited buffer. copy: copy strategy to use for items yielded by each branch. Supported options are ``None`` for no copying, ``"shallow"`` for shallow object copies, and ``"deep"`` for deep object copies. Defaults to ``None``. Note: All branches of the forked pipeline return the identical object unless the copy parameter is supplied. If the object is mutable or contains mutable objects, changing them in one branch will affect all others. Example: >>> # xdoctest: +REQUIRES(module:torchdata) >>> from torchdata.datapipes.iter import IterableWrapper >>> source_dp = IterableWrapper(range(5)) >>> dp1, dp2 = source_dp.fork(num_instances=2) >>> list(dp1) [0, 1, 2, 3, 4] >>> list(dp2) [0, 1, 2, 3, 4] Ndatapipe num_instances buffer_sizecopyshallowdeepc|dkrtd|d|dk(r|St||||}t|Dcgc]}t||c}Scc}wNz,Expected `num_instances` larger than 0, but z is found)r)_ForkerIterDataPiperange_ChildDataPipe)clsrKrLrMrN containeris r$__new__zForkerIterDataPipe.__new__bsd 1 >}oYW  A O'-dS 6;M6JKy!,KKKsA N) r>rBrCrDrrGr r r[rHr/r$rrCsKB 59 LLL L w012 Lr/rcdeZdZdZedefdZedefdZed dZ edefdZ y) _ContainerTemplatezVAbstract class for container ``DataPipes``. The followings are three required methods. instance_idcyrrHr-r`s r$get_next_element_by_instancez/_ContainerTemplate.get_next_element_by_instancev r/r0cyrrHr?s r$is_every_instance_exhaustedz._ContainerTemplate.is_every_instance_exhaustedzrdr/NcyrrHr?s r$resetz_ContainerTemplate.reset~rdr/cy)zSRaise TypeError if it's not supposed to be implemented to support `list(datapipe)`.NrHrbs r$get_length_by_instancez)_ContainerTemplate.get_length_by_instancesr/r0N) r>rBrCrDrrGrcboolrfrhrjrHr/r$r_r_sska    T    c#ccr/r_c|SrrH)xs r$_no_opros Hr/c eZdZdZ ddedededeedfdZd Z d efd Z d e fd Z d ed efdZ ddZdZdZdZdZy)rUz Container to hold instance-specific information on behalf of ForkerIterDataPipe. It tracks the state of its child DataPipes, maintains the buffer, and yields the next value as requested by the child DataPipes. NrKrLrMrNrOc||_d|_||_t|_||_|j dkrt jdt| t|_ nE|dk(rtj|_ n*|dk(rtj|_ ntd|ddg|z|_d|_d|_d|_t'|Dcgc]}dc}|_ycc}w)NrzPUnlimited buffer size is set for `fork`, please be aware of OOM at random placesrPrQzUnknown copy method `z5` requested, choose one of None, `shallow` or `deep`.T) main_datapipe_datapipe_iteratorrLrbufferrMwarningswarn UserWarningrocopy_fn copymodulerNdeepcopyr)child_pointers slowest_ptr leading_ptrend_ptrrV _child_stop)r-rKrLrMrN_s r$r.z_ForkerIterDataPipe.__init__s&;?*"W &   a  MM:  <!DL Y %??DL V^%..DL'v-bc  * *&* 6;M6J'K'K'Ks C'c,t|jSrr(rrr?s r$r@z_ForkerIterDataPipe.__len__4%%&&r/r`c#K|jg|j|rXt|j|_tj |_t|jD]}d|j|< |j|s|j|xxdz cc<|j-|j||jk(rd|j|<nu|jrK|j||jkr/|j||jz dz }|j|}nE|j||_ t|j}|jj||j||jdzk(rEt#|j}|j|kr!||_ |jj%|j&dk\rB|j|j&|jzkDrt)dd|j&dz|j+||j|sd|j|<t-|jrd|_|j/yy#t $r+d|j|<d|_|j|_ YEwxYw#d|j|<t-|jrd|_|j/wwxYww)NFrTTrz#ForkerIterDataPipe buffer overflow,z buffer size  is insufficient.)rsriterrrr Iterating_snapshot_staterVrLr{r~rtr}r|nextappend StopIterationminpopleftrM BufferErrorrxr*_cleanup)r-r`rZidx return_valnew_mins r$rcz0_ForkerIterDataPipe.get_next_element_by_instances  " " *t/?/? /L&*4+=+=&>D ##1#;#;D 4--. ,&+  # ,- &&{3##K0A50LL,++K8DLLH48D$$[1;;4#6#6{#CtGWGW#W--k:T=M=MMPQQC!%S!1J'+':':;'GD$!%)$*A*A%B  **:6 &&{3t7G7G!7KK!++G'''1+2( ++-$$)((4+;+;d>N>N+NN%=()9)9(::KLM ll:..M&&{3P-1D  [ )4##$*.' %3)!8<((526/'+'7'7  !.-1D  [ )4##$*.' %sDA4L7C K0J 3CK =L 0J=9K<J==K>K>>Lr0cL|jduxrt|jSr)r~r*rr?s r$rfz/_ForkerIterDataPipe.is_every_instance_exhausteds!||4'AC0@0@,AAr/c,t|jSrrrbs r$rjz*_ForkerIterDataPipe.get_length_by_instancerr/cd|_t|_dg|jz|_d|_d|_d|_t|jDcgc]}dc}|_ ycc}wNrT) rsrrtrLr{r|r}r~rVrr-rs r$rhz_ForkerIterDataPipe.resets`"&g  cD$6$66 */0B0B*CDQDDDs A*c|j|j|j|j|j|j f}t jt j|S|Sr)rrrLrMrx_valid_iterator_id_number_of_samples_yieldedr getstate_hookr-states r$ __getstate__z _ForkerIterDataPipe.__getstate__sa          LL  # #  + +    % % 1--e4 4 r/c,|\|_|_|_|_|_|_d|_t|_dg|jz|_ d|_ d|_ d|_ t|jDcgc]}dc}|_ycc}wr)rrrLrMrxrrrsrrtr{r|r}r~rVrr-rrs r$ __setstate__z _ForkerIterDataPipe.__setstate__ s        L  #  +"&g  cD$6$66 */0B0B*CDQDDDs? Bc|jr=|jj}tj||jrrBrCrDrrGr r r.r@rcrlrfrjrhrrrrrHr/r$rUrUs 59 #L#L#L #L w012 #LJ'3 3 jBTB'#'#'E E"+ r/rUcTeZdZUdZdZeed<dedefdZ dZ dZ d efd Z d efd Z y ) rWa Iterable Datapipe that is a child of a main DataPipe. The instance of this class will pass its instance_id to get the next value from its main DataPipe. Note: ChildDataPipe, like all other IterDataPipe, follows the single iterator per IterDataPipe constraint. Since ChildDataPipes share a common buffer, when an iterator is created for one of the ChildDataPipes, the previous iterators for all ChildDataPipes must be invalidated, with the exception when a ChildDataPipe hasn't had an iterator created from it since the last invalidation. See the example below. Example: >>> # xdoctest: +REQUIRES(module:torchdata) >>> # Singler Iterator per IteraDataPipe Invalidation >>> from torchdata.datapipes.iter import IterableWrapper >>> source_dp = IterableWrapper(range(10)) >>> cdp1, cdp2 = source_dp.fork(num_instances=2) >>> it1, it2 = iter(cdp1), iter(cdp2) >>> it3 = iter(cdp1) >>> # The line above invalidates `it1` and `it2`, and resets `ForkerIterDataPipe`. >>> it4 = iter(cdp2) >>> # The line above doesn't invalidate `it3`, because an iterator for `cdp2` hasn't been created since >>> # the last invalidation. Args: main_datapipe: Main DataPipe with a method 'get_next_element_by_instance(instance_id)' instance_id: integer identifier of this instance T_is_child_datapiperrr`cDt|tsJ||_||_yr)r r_rrr`)r-rrr`s r$r.z_ChildDataPipe.__init__Fs#-);<<<+8&r/cL|jj|jSr)rrrcr`r?s r$r2z_ChildDataPipe.__iter__Ls!!!>>t?O?OPPr/cL|jj|jSr)rrrjr`r?s r$r@z_ChildDataPipe.__len__Qs!!889I9IJJr/r0c|jjd|j_n|jj|jk(rm|jxjdz c_|jjstjdt |jj |jj|_|jS)z Update the valid iterator ID for both this DataPipe object and `main_datapipe`. `main_datapipe.reset()` is called when the ID is incremented to a new generation. rrTzSome child DataPipes are not exhausted when __iter__ is called. We are resetting the buffer and each child DataPipe will read from the start again.)rrrrfrurvrwrhr?s r$$_set_main_datapipe_valid_iterator_idz3_ChildDataPipe._set_main_datapipe_valid_iterator_idUs    0 0 845D   1   2 2d6M6M M    1 1Q 6 1%%AAC Y    $ $ &#'"4"4"G"G&&&r/cV||jk(xr||jjk(S)zXCheck the valid iterator ID against that of DataPipe object and that of `main_datapipe`.)rrr)r- iterator_ids r$_check_valid_iterator_idz'_ChildDataPipe._check_valid_iterator_idps1 422 2 Et11DDD r/N)r>rBrCrDrrlrFrrGr.r2r@rrrHr/r$rWrW&sK: $#'l'' Q K'c'6 t r/rWdemuxc DeZdZdZ d dededeegeefde def dZ y) ra+ Splits the input DataPipe into multiple child DataPipes, using the given classification function (functional name: ``demux``). A list of the child DataPipes is returned from this operation. Args: datapipe: Iterable DataPipe being filtered num_instances: number of instances of the DataPipe to create classifier_fn: a function that maps values to an integer within the range ``[0, num_instances - 1]`` or ``None`` drop_none: defaults to ``False``, if ``True``, the function will skip over elements classified as ``None`` buffer_size: this defines the maximum number of inputs that the buffer can hold across all child DataPipes while waiting for their values to be yielded. Defaults to ``1000``. Use ``-1`` for the unlimited buffer. Examples: >>> # xdoctest: +REQUIRES(module:torchdata) >>> from torchdata.datapipes.iter import IterableWrapper >>> def odd_or_even(n): ... return n % 2 >>> source_dp = IterableWrapper(range(5)) >>> dp1, dp2 = source_dp.demux(num_instances=2, classifier_fn=odd_or_even) >>> list(dp1) [0, 2, 4] >>> list(dp2) [1, 3] >>> # It can also filter out any element that gets `None` from the `classifier_fn` >>> def odd_or_even_no_zero(n): ... return n % 2 if n != 0 else None >>> dp1, dp2 = source_dp.demux(num_instances=2, classifier_fn=odd_or_even_no_zero, drop_none=True) >>> list(dp1) [2, 4] >>> list(dp2) [1, 3] rKrL classifier_fn drop_nonerMc|dkrtd|dt|t|||||}t|Dcgc]}t ||c}Scc}wrS)r)r_DemultiplexerIterDataPiperVrW)rXrKrLrrrMrYrZs r$r[z!DemultiplexerIterDataPipe.__new__se 1 >}oYW  ]+ /x W`bmn 6;M6JKy!,KKKsAN)Fr]) r>rBrCrDrrGr rr rlr[rHr/r$rrxs[!P  LLL # 67 L  L  Lr/rc eZdZdZdeededeegeefde def dZ ded efd Z defd Z d e fd Z ded efd ZddZdZdZddeefdZdZy)rz Container to hold instance-specific information on behalf of DemultiplexerIterDataPipe. It tracks the state of its child DataPipes, maintains the buffer, classifies and yields the next correct value as requested by the child DataPipes. rKrLrrrMcb||_d|_||_||_|jdkrt j dt d|_t|Dcgc] }tc}|_ ||_ ||_ d|_ t|Dcgc]}dc}|_ycc}wcc}w)NrzQUnlimited buffer size is set for `demux`, please be aware of OOM at random placesFT)rrrsrLrMrurvrwcurrent_buffer_usagerVr child_buffersrrmain_datapipe_exhaustedr)r-rKrLrrrMrs r$r.z#_DemultiplexerIterDataPipe.__init__s&;?*&   a  MM:  %&!CHCW1Xa%'1X*"',$6;M6J'K'K 2Y(Ls B' B,r`r0cr |js|j|rt|j t dt |j}|j |}|"|jrtj||||jk\s|dkr"t d|jdz d|dz||k(r|S|j|j||xjdz c_ |jdk\r2|j|jkDrtd|jd7) Nz_datapipe_iterator has not been set, likely because this private method is called directly without invoking get_next_element_by_instance() first.rz8Output of the classification fn should be between 0 and rTz. z is returned.z7DemultiplexerIterDataPipe buffer overflow, buffer size r)rrrrsr)rrrrrrLrrrrMr)r-r`valueclassifications r$ _find_nextz%_DemultiplexerIterDataPipe._find_nextsR++t/?/? /L##&&. M001E!//6N%$..++E2&!T%7%77!A% NtOaOadeOeNffhi'( 67,   ~ . 5 5e <  % % * %1$)B)BTEUEU)U!MdN^N^M__pq7r/c#K|jn|j|r_t|j|_tj |_d|_t|jD]}d|j|< |j|sh|j|r5|xjdzc_ |j|jn |j||j|shd|j|<t|jrd|_|j|r|j!|yy#t$r d|j|<d|_d|_YwxYw#d|j|<t|jrd|_|j|r|j!|wwxYww)NFrTT)rsrrrrrrrrrVrLrrrrrr*r)r-r`rZs r$rcz7_DemultiplexerIterDataPipe.get_next_element_by_instances  " " *t/?/? /L&*4+=+=&>D #((  ,1D (4--. ,&+  # , +&&{3%%k2--2-,,[9AACC7"ook:: &&{3-1D  [ )4##$*.'!!+. k*/)78<((57;426/7 -1D  [ )4##$*.'!!+. k*/sEA;F?>AE.E%E.5A F?&E+(E.*E++E..AF<<F?cH|jxrt|jSr)rr*rr?s r$rfz6_DemultiplexerIterDataPipe.is_every_instance_exhausteds++ED4D4D0EEr/ctr)r+rbs r$rjz1_DemultiplexerIterDataPipe.get_length_by_instancesr/Ncd|_d|_t|jDcgc] }t c}|_t|jDcgc]}dc}|_d|_ycc}wcc}wNrTF)rsrrVrLrrrrrs r$rhz _DemultiplexerIterDataPipe.resetsa"&$%!/4T5G5G/HI!egI*/0B0B*CDQDD',$JDs A/ A4c|j|j|j|j|j|j |j f}tjtj|S|Sr) rrrLrMrrrrrrrs r$rz'_DemultiplexerIterDataPipe.__getstate__sj             NN  # #  + +   % % 1--e4 4 r/cL|\|_|_|_|_|_|_|_d|_d|_t|jDcgc] }tc}|_ t|jDcgc]}dc}|_ d|_ ycc}wcc}wr)rrrLrMrrrrrsrrVrrrrrs r$rz'_DemultiplexerIterDataPipe.__setstate__-s          N  #  +"&$%!/4T5G5G/HI!egI*/0B0B*CDQDD',$JDs B B!c|t|jn|g}|D]<}|j|}|s|j}t j ||r(>yr)rVrLrrrr)r-r`idsrZqrs r$rz#_DemultiplexerIterDataPipe._cleanup=sk" $$$ %  /A""1%AIIK++A. /r/c$|jyrrr?s r$rz"_DemultiplexerIterDataPipe.__del__Krr/rkr)r>rBrCrDrrrGr r rlr.rrcrfrjrhrrrrrHr/r$rrsLu%LL # 67 L  L  L2ce@++<FTF##- - /HSM /r/rmuxc<eZdZdZdZdZdZd dZdZdZ d Z y) ra Yields one element at a time from each of the input Iterable DataPipes (functional name: ``mux``). As in, one element from the 1st input DataPipe, then one element from the 2nd DataPipe in the next iteration, and so on. It ends when the shortest input DataPipe is exhausted. Args: datapipes: Iterable DataPipes that will take turn to yield their elements, until the shortest DataPipe is exhausted Example: >>> # xdoctest: +REQUIRES(module:torchdata) >>> from torchdata.datapipes.iter import IterableWrapper >>> dp1, dp2, dp3 = IterableWrapper(range(3)), IterableWrapper(range(10, 15)), IterableWrapper(range(20, 25)) >>> list(dp1.mux(dp2, dp3)) [0, 10, 20, 1, 11, 21, 2, 12, 22] c ||_g|_yr)rrtr,s r$r.z MultiplexerIterDataPipe.__init__bs"  r/c#K|jDcgc] }t|}}t|ri|D])} t|}|jj |+|jEd{|jjt|rhyycc}w#t $r|jjYywxYw7[wr)rrr(rrtrrclear)r-rn iteratorsitrs r$r2z MultiplexerIterDataPipe.__iter__hs&*nn5T!W5 5)n  HEKK&&u- {{ " " KK   )n6 %KK%%' #sDC BC &BC -C .)C C #CC CC ctd|jDr2td|jDt|jzSt t |j d)Nc3<K|]}t|tywrr5r!s r$r%z2MultiplexerIterDataPipe.__len__..vr6r'c32K|]}t|ywrr8r!s r$r%z2MultiplexerIterDataPipe.__len__..wr9r:r;)r*rrr(r+r=r>r?s r$r@zMultiplexerIterDataPipe.__len__usU >t~~> >8883t~~;NN NtDz2233VWX Xr/Ncg|_yr)rtr?s r$rhzMultiplexerIterDataPipe.reset{s  r/c|j|j|jf}tjtj|S|Sr)rrrrrrs r$rz$MultiplexerIterDataPipe.__getstate__~sF NN  # #  + +   % % 1--e4 4 r/c<|\|_|_|_g|_yr)rrrrtrs r$rz$MultiplexerIterDataPipe.__setstate__s$   N  #  + r/c8|jjyr)rtrr?s r$rzMultiplexerIterDataPipe.__del__s r/rk) r>rBrCrDr.r2r@rhrrrrHr/r$rrOs+"  Y r/rzipc^eZdZUdZeeed<deffd Zdeee fdZ de fdZ xZ S)raa Aggregates elements into a tuple from each of the input DataPipes (functional name: ``zip``). The output is stopped as soon as the shortest input DataPipe is exhausted. Args: *datapipes: Iterable DataPipes being aggregated Example: >>> # xdoctest: +REQUIRES(module:torchdata) >>> from torchdata.datapipes.iter import IterableWrapper >>> dp1, dp2, dp3 = IterableWrapper(range(5)), IterableWrapper(range(10, 15)), IterableWrapper(range(20, 25)) >>> list(dp1.zip(dp2, dp3)) [(0, 10, 20), (1, 11, 21), (2, 12, 22), (3, 13, 23), (4, 14, 24)] rcjtd|Ds tdt| ||_y)Nc3<K|]}t|tywrrr!s r$r%z.ZipperIterDataPipe.__init__..r&r'zCAll inputs are required to be `IterDataPipe` for `ZipIterDataPipe`.)r*r+superr.r)r-r __class__s r$r.zZipperIterDataPipe.__init__s5D)DDU  "r/r0c#~K|jDcgc] }t|}}t|Ed{ycc}w7 wr)rrr)r-rKrs r$r2zZipperIterDataPipe.__iter__s748NNCT(^C C ?""D"s=6 =; =ctd|jDrtd|jDStt |j d)Nc3<K|]}t|tywrr5r!s r$r%z-ZipperIterDataPipe.__len__..r6r'c32K|]}t|ywrr8r!s r$r%z-ZipperIterDataPipe.__len__..r9r:r;)r*rrr+r=r>r?s r$r@zZipperIterDataPipe.__len__rAr/)r>rBrCrDrErrFr.rrr2rGr@ __classcell__)rs@r$rrsD \""#<##(5<0#YYr/r)'rNryruabcrr collectionsrcollections.abcrrtypingrr r r r %torch.utils.data.datapipes._decoratorr )torch.utils.data.datapipes._hook_iteratorr#torch.utils.data.datapipes.datapiper'torch.utils.data.datapipes.utils.commonrr__all__rrrr_rorUrWrrrrErrHr/r$rs=#+<<ED<W  4(X$Y<$Y$YNV,L,L,L^cc( X,(:XvO \O dW7L 7L7LtY/AYxUAlAAHU#YeEl3#Y#Yr/