Xj$6ddlZddlmZddlmZmZmZddlmZm Z ddl Z ddl m Z ddl mZddlmZddlmZdd lmZmZd d gZe d d ZedGdd eeZdZedGdd eZdS)N) namedtuple)CallableIteratorSized)AnyTypeVar)default_collate)functional_datapipe)dataframe_wrapper) IterDataPipe)_check_unpickable_fnvalidate_input_colCollatorIterDataPipeMapperIterDataPipe_T_coT) covariantmapcveZdZUdZeed<eed< d dededdffd ZdZde e fdZ de fd Z xZS) ra Applies a function over each item from the source DataPipe (functional name: ``map``). The function can be any regular Python function or partial object. Lambda function is not recommended as it is not supported by pickle. Args: datapipe: Source Iterable DataPipe fn: Function being applied over each item input_col: Index or indices of data which ``fn`` is applied, such as: - ``None`` as default to apply ``fn`` to the data directly. - Integer(s) is used for list/tuple. - Key(s) is used for dict. output_col: Index of data where result of ``fn`` is placed. ``output_col`` can be specified only when ``input_col`` is not ``None`` - ``None`` as default to replace the index that ``input_col`` specified; For ``input_col`` with multiple indices, the left-most one is used, and other indices will be removed. - Integer is used for list/tuple. ``-1`` represents to append result at the end. - Key is used for dict. New key is acceptable. Example: >>> # xdoctest: +SKIP >>> from torchdata.datapipes.iter import IterableWrapper, Mapper >>> def add_one(x): ... return x + 1 >>> dp = IterableWrapper(range(10)) >>> # Invocation via functional form is preferred ... map_dp_1 = dp.map(add_one) >>> list(map_dp_1) [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] >>> # We discourage the usage of `lambda` functions as they are not serializable with `pickle` >>> # Use `functools.partial` or explicitly define the function instead >>> map_dp_2 = Mapper(dp, lambda x: x + 1) >>> list(map_dp_2) [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] datapipefnNreturnctjdt||_t |||_||_||tdt|ttfr*t|dkrtd|d}||_t||dS)Nzpython.data_pipes.mapz3`output_col` must be None when `input_col` is None.z3`output_col` must be a single-element list or tupler)torch_C_log_api_usage_oncesuper__init__rr r input_col ValueError isinstancelisttuplelen output_colr)selfrrrr% __class__s VC:\Projects\DSCons\.venv\Lib\site-packages\torch\utils\data\datapipes\iter\callable.pyrzMapperIterDataPipe.__init__Hs $$%<===   R   "  !7RSS S j4- 0 0 ':"" !VWWW#AJ$2y)))))c|j|j|S|j|}nlt|jtt fr+t fd|jD}|j|}n ||j}tt rd}t nd}|jat|jtt fr5||jd<t |jdddD]}|=n6||j<n+|jdkr|n ||j<|rt nS)Nc3(K|] }|V dSN).0coldatas r( z/MapperIterDataPipe._apply_fn..gs'==sc======r)TFrr)reverse)rr%rr!r"r#sortedappend)r&r0resargst_flagidxs ` r( _apply_fnzMapperIterDataPipe._apply_fn`s > !do&=774== > !''$--CC u 6 6 0====dn=====D$'4.CC''$t~.//C dE " " F::DDF ? "$.4-88 +*-T^A&'!$."4dCCC""CS "(+T^$$"$$ C    (+T_%%.uT{{{$.r)c#LK|jD]}||VdSr,)rr:)r&r0s r(__iter__zMapperIterDataPipe.__iter__s<M ' 'D..&& & & & & ' 'r)ct|jtrt|jSt t |jd)Nz# instance doesn't have valid length)r!rrr$ TypeErrortype__name__)r&s r(__len__zMapperIterDataPipe.__len__sF dmU + + &t}%% %4::.SSSTTTr))NN)r@ __module__ __qualname____doc__r __annotations__rrr:rrr<intrA __classcell__r's@r(rrs&&PLLL  *** *  ******0!/!/!/F'(5/''''UUUUUUUUUr)cpt|jdkrtd|d}tj|}g}g}|D]}||vrtd|D]}||vr-t ||std||}nC ddlm}|j }n"#t$r} td| d} ~ wwxYw| t||||} | | td|} | |} | S)Nrz%Only supports one DataFrame per batchrzConversion keys mismatchz5Collate (DF)DataPipe requires callable as dict valuesz?unable to import default collation function from the TorchArrow CollateResult)r$items RuntimeError df_wrapper get_columnscallabletorcharrow.pytorchpytorchrecDefault Exceptionr5strr) conversionitemdf columns_name tuple_names tuple_valuesname collation_fntapevaluetpl_clsr#s r(_collate_helperrbs 4:BCCC aB)"--LKL;; | # #9:: : $## :  Jt,-- "K&d+LL 000000"w00    "U  3t99%%% RX&&E""""+66G G\ "E LsB66 CCCcollatec xeZdZdZedfdededefee ezeezfzdzdedzddffd Z xZ S) raI Collates samples from DataPipe to Tensor(s) by a custom collate function (functional name: ``collate``). By default, it uses :func:`torch.utils.data.default_collate`. .. note:: While writing a custom collate function, you can import :func:`torch.utils.data.default_collate` for the default behavior and `functools.partial` to specify any additional arguments. Args: datapipe: Iterable DataPipe being collated collate_fn: Customized collate function to collect and combine data or a batch of data. Default function collates to Tensor(s) based on data type. Example: >>> # xdoctest: +SKIP >>> # Convert integer data to float Tensor >>> class MyIterDataPipe(torch.utils.data.IterDataPipe): ... def __init__(self, start, end): ... super(MyIterDataPipe).__init__() ... assert end > start, "this example only works with end >= start" ... self.start = start ... self.end = end ... ... def __iter__(self): ... return iter(range(self.start, self.end)) ... ... def __len__(self): ... return self.end - self.start >>> ds = MyIterDataPipe(start=3, end=7) >>> print(list(ds)) [3, 4, 5, 6] >>> def collate_fn(batch): ... return torch.tensor(batch, dtype=torch.float) >>> collated_ds = CollateIterDataPipe(ds, collate_fn=collate_fn) >>> print(list(collated_ds)) [tensor(3.), tensor(4.), tensor(5.), tensor(6.)] NrrV. collate_fnrc8|%t||dSt|r%t||dStjt |}t||dS)N)r)rrrO functoolspartialrb)r&rrVrer's r(rzCollatorIterDataPipe.__init__s  ! GG  X*  5 5 5 5 5 ## :  j 99999'. KK   j 99999r)) r@rBrCrDr r rrdictrUrrGrHs@r(rrs%%X!&* :::S#X& sSy(S.( )* : tO : ::::::::::r))rg collectionsrcollections.abcrrrtypingrrrtorch.utils.data._utils.collater %torch.utils.data.datapipes._decoratorr $torch.utils.data.datapipes.dataframer rM#torch.utils.data.datapipes.datapiper 'torch.utils.data.datapipes.utils.commonr r__all__rrrbrr-r)r(rss""""""5555555555 ;;;;;;EEEEEEPPPPPP<<<<<<  4(((UoUoUoUoUoUe,oUoUoUd(((VY::::::::::-:::: ::::::r)