DŽgjddlZddlZddlmZmZddlmZddlm Z m Z m Z m Z m Z mZmZmZmZmZddlmZddlmZddlmZddlmZmZgd Zed d Zed GddeZedGddeZGddeZ dZ!Gddee Z"GddeZ#edGddeZ$Gddee Z%edGd d!eZ&ed"Gd#d$eeeZ'y)%N)ABCabstractmethod)deque) AnyCallableDequeIteratorListLiteralOptionalSizedTupleTypeVar)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 q/home/mcse/projects/flask_80/flask-venv/lib/python3.12/site-packages/torch/utils/data/datapipes/iter/combining.py z0ConcaterIterDataPipe.__init__..>DB:b,/Dz(Expected all inputs to be `IterDataPipe`)len ValueErrorall TypeErrorrselfrs r'__init__zConcaterIterDataPipe.__init__;s; y>Q NO OD)DDFG G"returnc#FK|jD] }|Ed{y7wr!)r)r0r&s r'__iter__zConcaterIterDataPipe.__iter__Bs#.. BMM  s !!ctd|jDrtd|jDStt |j d)Nc3<K|]}t|tywr!r#r r$s r'r(z/ConcaterIterDataPipe.__len__..G>z"e$>r*c32K|]}t|ywr!r+r$s r'r(z/ConcaterIterDataPipe.__len__..H82s2w8# instance doesn't have valid length)r-rsumr.type__name__r0s r'__len__zConcaterIterDataPipe.__len__FG >t~~> >888 8tDz2233VWX Xr2N) rA __module__ __qualname____doc__rr__annotations__r1r r5intrCr2r'rr%s:$\""#<#(YYr2rforkc :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)clsrMrNrOrP containeris r'__new__zForkerIterDataPipe.__new__lsd 1 >}oYW  A O'-dS 6;M6JKy!,KKKsA N) rArErFrGrrIr r r]rJr2r'rrMsKB 59 LLL L w012 Lr2rcdeZdZdZedefdZedefdZed dZ edefdZ y) _ContainerTemplatezVAbstract class for container ``DataPipes``. The followings are three required methods. instance_idcyr!rJr0rbs r'get_next_element_by_instancez/_ContainerTemplate.get_next_element_by_instance r2r3cyr!rJrBs r'is_every_instance_exhaustedz._ContainerTemplate.is_every_instance_exhaustedrfr2Ncyr!rJrBs r'resetz_ContainerTemplate.resetrfr2cy)zSRaise TypeError if it's not supposed to be implemented to support `list(datapipe)`.NrJrds r'get_length_by_instancez)_ContainerTemplate.get_length_by_instancesr2r3N) rArErFrGrrIreboolrhrjrlrJr2r'rara}ska    T    c#ccr2rac|Sr!rJ)xs r'_no_oprqs Hr2c eZdZdZ ddedededeedfdZd Z d efd Z d e fd Z d ed efdZ ddZdZdZdZdZy)rWz 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. NrMrNrOrPrQc||_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 placesrRrSzUnknown copy method `z5` requested, choose one of None, `shallow` or `deep`.T) main_datapipe_datapipe_iteratorrNrbufferrOwarningswarn UserWarningrqcopy_fn copymodulerPdeepcopyr,child_pointers slowest_ptr leading_ptrend_ptrrX _child_stop)r0rMrNrOrP_s r'r1z_ForkerIterDataPipe.__init__s&;?*"W &   a  MM:  <!DL Y %??DL V^%..DL'v-bc  * *&* 6;M6J'K'K'Ks C'c,t|jSr!r+rtrBs r'rCz_ForkerIterDataPipe.__len__4%%&&r2rbc#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)NFrVTrz#ForkerIterDataPipe buffer overflow,z buffer size  is insufficient.)ruriterrtr Iterating_snapshot_staterXrNr}rrvrr~nextappend StopIterationminpopleftrO BufferErrorrzr-_cleanup)r0rbr\idx return_valnew_mins r'rez0_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>>Lr3cL|jduxrt|jSr!)rr-rrBs r'rhz/_ForkerIterDataPipe.is_every_instance_exhausteds!||4'AC0@0@,AAr2c,t|jSr!rrds r'rlz*_ForkerIterDataPipe.get_length_by_instancerr2cd|_t|_dg|jz|_d|_d|_d|_t|jDcgc]}dc}|_ ycc}wNrT) rurrvrNr}r~rrrXrr0rs r'rjz_ForkerIterDataPipe.resets`"&g  cD$6$66 */0B0B*CDQDDDs A*c|j|j|j|j|j|j f}t jt j|S|Sr!)rtrNrOrz_valid_iterator_id_number_of_samples_yieldedr getstate_hookr0states r' __getstate__z _ForkerIterDataPipe.__getstate__ sa          LL  # #  + +    % % 1--e4 4 r2c,|\|_|_|_|_|_|_d|_t|_dg|jz|_ d|_ d|_ d|_ t|jDcgc]}dc}|_ycc}wr)rtrNrOrzrrrurrvr}r~rrrXrr0rrs r' __setstate__z _ForkerIterDataPipe.__setstate__s        L  #  +"&g  cD$6$66 */0B0B*CDQDDDs? Bc|jr=|jj}tj||jr>> # 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_datapipertrbcDt|tsJ||_||_yr!)r#rartrb)r0rtrbs r'r1z_ChildDataPipe.__init__Ps#-);<<<+8&r2cL|jj|jSr!)rtrerbrBs r'r5z_ChildDataPipe.__iter__Vs!!!>>t?O?OPPr2cL|jj|jSr!)rtrlrbrBs r'rCz_ChildDataPipe.__len__[s!!889I9IJJr2r3c|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. rrVzSome child DataPipes are not exhausted when __iter__ is called. We are resetting the buffer and each child DataPipe will read from the start again.)rtrrhrwrxryrjrBs r'$_set_main_datapipe_valid_iterator_idz3_ChildDataPipe._set_main_datapipe_valid_iterator_id_s    0 0 845D   1   2 2d6M6M M    1 1Q 6 1%%AAC Y    $ $ &#'"4"4"G"G&&&r2cV||jk(xr||jjk(S)zXCheck the valid iterator ID against that of DataPipe object and that of `main_datapipe`.)rrt)r0 iterator_ids r'_check_valid_iterator_idz'_ChildDataPipe._check_valid_iterator_idzs1 422 2 Et11DDD r2N)rArErFrGrrnrHrrIr1r5rCrrrJr2r'rYrY0sK: $#'l'' Q K'c'6 t r2rYdemuxc 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] rMrN classifier_fn drop_nonerOc|dkrtd|dt|t|||||}t|Dcgc]}t ||c}Scc}wrU)r,r_DemultiplexerIterDataPiperXrY)rZrMrNrrrOr[r\s r'r]z!DemultiplexerIterDataPipe.__new__se 1 >}oYW  ]+ /x W`bmn 6;M6JKy!,KKKsAN)Fr_) rArErFrGrrIrrr rnr]rJr2r'rrs[!P  LLL # 67 L  L  Lr2rc 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. rMrNrrrOcb||_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)rtrurNrOrwrxrycurrent_buffer_usagerXr child_buffersrrmain_datapipe_exhaustedr)r0rMrNrrrOrs r'r1z#_DemultiplexerIterDataPipe.__init__s&;?*&   a  MM:  %&!CHCW1Xa%'1X*"',$6;M6J'K'K 2Y(Ls B' B,rbr3cr |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 rVz. z is returned.z7DemultiplexerIterDataPipe buffer overflow, buffer size r)rrrrur,rrrrrrNrrrrOr)r0rbvalueclassifications 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__pq7r2c#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)NFrVT)rurrrtrrrrrXrNrrrrrr-r)r0rbr\s r'rez7_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-rrBs r'rhz6_DemultiplexerIterDataPipe.is_every_instance_exhausteds++ED4D4D0EEr2ctr!)r.rds r'rlz1_DemultiplexerIterDataPipe.get_length_by_instancesr2Ncd|_d|_t|jDcgc] }t c}|_t|jDcgc]}dc}|_d|_ycc}wcc}wNrTF)rurrXrNrrrrrs r'rjz _DemultiplexerIterDataPipe.reset"sa"&$%!/4T5G5G/HI!egI*/0B0B*CDQDD',$JDs A/ A4c|j|j|j|j|j|j |j f}tjtj|S|Sr!) rtrNrOrrrrrrrs r'rz'_DemultiplexerIterDataPipe.__getstate__)sj             NN  # #  + +   % % 1--e4 4 r2cL|\|_|_|_|_|_|_|_d|_d|_t|jDcgc] }tc}|_ t|jDcgc]}dc}|_ d|_ ycc}wcc}wr)rtrNrOrrrrrurrXrrrrrs r'rz'_DemultiplexerIterDataPipe.__setstate__7s          N  #  +"&$%!/4T5G5G/HI!egI*/0B0B*CDQDD',$JDs B B!c|t|jn|g}|D]<}|j|}|s|j}t j ||r(>yr!)rXrNrrrr)r0rbidsr\qrs r'rz#_DemultiplexerIterDataPipe._cleanupGsk" $$$ %  /A""1%AIIK++A. /r2c$|jyr!rrBs r'rz"_DemultiplexerIterDataPipe.__del__Urr2rmr!)rArErFrGrrrIrr rnr1rrerhrlrjrrrrrJr2r'rrsLu%LL # 67 L  L  L2ce@++<FTF##- - /HSM /r2rmuxc<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!)rrvr/s r'r1z MultiplexerIterDataPipe.__init__ls"  r2c#K|jDcgc] }t|}}t|ri|D])} t|}|jj |+|jEd{|jjt|rhyycc}w#t $r|jjYywxYw7[wr!)rrr+rrvrrclear)r0rp iteratorsitrs r'r5z MultiplexerIterDataPipe.__iter__rs&*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|tywr!r8r$s r'r(z2MultiplexerIterDataPipe.__len__..r9r*c32K|]}t|ywr!r;r$s r'r(z2MultiplexerIterDataPipe.__len__..r<r=r>)r-rrr+r.r@rArBs r'rCzMultiplexerIterDataPipe.__len__sU >t~~> >8883t~~;NN NtDz2233VWX Xr2Ncg|_yr!)rvrBs r'rjzMultiplexerIterDataPipe.resets  r2c|j|j|jf}tjtj|S|Sr!)rrrrrrs r'rz$MultiplexerIterDataPipe.__getstate__sF NN  # #  + +   % % 1--e4 4 r2c<|\|_|_|_g|_yr!)rrrrvrs r'rz$MultiplexerIterDataPipe.__setstate__s$   N  #  + r2c8|jjyr!)rvrrBs r'rzMultiplexerIterDataPipe.__del__s r2rm) rArErFrGr1r5rCrjrrrrJr2r'rrYs+"  Y r2rzipc^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|tywr!r"r$s r'r(z.ZipperIterDataPipe.__init__..r)r*zCAll inputs are required to be `IterDataPipe` for `ZipIterDataPipe`.)r-r.superr1r)r0r __class__s r'r1zZipperIterDataPipe.__init__s5D)DDX  "r2r3c#~K|jDcgc] }t|}}t|Ed{ycc}w7 wr!)rrr)r0rMrs r'r5zZipperIterDataPipe.__iter__s748NNCT(^C C ?""D"s=6 =; =ctd|jDrtd|jDStt |j d)Nc3<K|]}t|tywr!r8r$s r'r(z-ZipperIterDataPipe.__len__..r9r*c32K|]}t|ywr!r;r$s r'r(z-ZipperIterDataPipe.__len__..r<r=r>)r-rrr.r@rArBs r'rCzZipperIterDataPipe.__len__rDr2)rArErFrGrrrHr1r rr5rIrC __classcell__)rs@r'rrsD \""#<##(5<0#YYr2r)(rPr{rwabcrr collectionsrtypingrrrr r r r r rr%torch.utils.data.datapipes._decoratorr)torch.utils.data.datapipes._hook_iteratorr#torch.utils.data.datapipes.datapiper'torch.utils.data.datapipes.utils.commonrr__all__rrrrarqrWrYrrrrrJr2r'rsF#   FD<W  4(X$Y<$Y$YNV,L,L,L^cc( X,(:XvO \O dW7L 7L7LtY/AYxUAlAAHU#YeEl3#Y#Yr2