o wZhi#@sddlZddlmZddlmZmZddlmZmZm Z m Z m Z ddl m Z ddlmZddlmZddlmZdd lmZmZd d gZe d d dZedGdd d eeZddZedGdd d eZdS)N) namedtuple)IteratorSized)AnyCallableOptionalTypeVarUnion)default_collate)functional_datapipe)dataframe_wrapper) IterDataPipe)_check_unpickable_fnvalidate_input_colCollatorIterDataPipeMapperIterDataPipe_T_coT) covariantmapcsneZdZUdZeed<eed<  ddededdffdd Zdd Zde e fd d Z de fd d Z ZS)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)) >>> map_dp_1 = dp.map(add_one) # Invocation via functional form is preferred >>> 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] datapipefnNreturncszt||_t|||_||_|dur|durtdt|tt fr3t |dkr/td|d}||_ t ||dS)Nz3`output_col` must be None when `input_col` is None.z3`output_col` must be a single-element list or tupler) super__init__rrr input_col ValueError isinstancelisttuplelen output_colr)selfrrrr! __class__W/var/www/auris/lib/python3.10/site-packages/torch/utils/data/datapipes/iter/callable.pyrFs  zMapperIterDataPipe.__init__cs|jdur|jdur|S|jdur|}n"t|jttfr4tfdd|jD}|j|}n||j}ttrHd}tnd}|jdurvt|jttfrp||jd<t|jddddD]}|=qin||j<n|jdkr|n||j<|rtSS) Nc3s|]}|VqdSNr%).0coldatar%r& dsz/MapperIterDataPipe._apply_fn..TFrr)reverse)rr!rrrrsortedappend)r"r+resargsZt_flagidxr%r*r& _apply_fn]s.           zMapperIterDataPipe._apply_fnccs|jD]}||VqdSr')rr4)r"r+r%r%r&__iter__s zMapperIterDataPipe.__iter__cCs*t|jtr t|jStt|jd)Nz# instance doesn't have valid length)rrrr TypeErrortype__name__)r"r%r%r&__len__s  zMapperIterDataPipe.__len__)NN)r8 __module__ __qualname____doc__r __annotations__rrr4rrr5intr9 __classcell__r%r%r#r&rs '#c Cst|jdkr td|d}t|}g}g}|D] }||vr&tdq|D]F}||vr>t||s9td||}nz ddlm}|j }Wnt y\} ztd| d} ~ ww| t ||||} | | q)td|} | |} | S)Nrz%Only supports one DataFrame per batchrzConversion keys missmatchz5Collate (DF)DataPipe requires callable as dict valuesz?unable to import default collation function from the TorchArrowZ CollateResult)r items RuntimeError df_wrapperZ get_columnskeyscallableZtorcharrow.pytorchZpytorchZrecZDefault Exceptionr0strr) conversionitemZdfZ columns_nameZ tuple_namesZ tuple_valuesnameZ collation_fntapevalueZtpl_clsrr%r%r&_collate_helpersD        rMZcollatec sfeZdZdZedfdedeedefe ee efeeeffdfde eddffdd Z Z S) raf 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 code 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.)] NrrG. collate_fnrcsX|durtj||ddSt|rtj||ddStt|}tj||ddS)N)r)rrrD functoolspartialrM)r"rrGrNr#r%r&rs   zCollatorIterDataPipe.__init__)r8r:r;r<r r r rrdictrFrrr?r%r%r#r&rs.()rO collectionsrcollections.abcrrtypingrrrrr Ztorch.utils.data._utils.collater Z%torch.utils.data.datapipes._decoratorr Z$torch.utils.data.datapipes.dataframer rBZ#torch.utils.data.datapipes.datapiper Z'torch.utils.data.datapipes.utils.commonrr__all__rrrMrr%r%r%r&s$      o+