U Mh#@sddlZddlmZddlmZmZmZmZmZm Z m Z m Z m Z ddl mZddlmZddlmZddlmZddlmZmZd d gZed d d ZedGdd d eeZddZedGdd d eZdS)N) namedtuple) CallableIteratorSizedTypeVarOptionalUnionAnyDictList)functional_datapipe)default_collate)dataframe_wrapper) IterDataPipe)_check_unpickable_fnvalidate_input_colCollatorIterDataPipeMapperIterDataPipeT_coT) covariantmapcsfeZdZUdZeed<eed<deeddfdd Zdd Ze e d d d Z e d d dZ 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] datapipefnN)rrreturncszt||_t|||_||_|dkr<|dk r`sz/MapperIterDataPipe._apply_fn..TFrr)reverse)rr#rrr r!sortedappend)r$r-resargsZt_flagidxr'r,r( _apply_fnYs,            zMapperIterDataPipe._apply_fn)rccs|jD]}||VqdSr))rr6)r$r-r'r'r(__iter__|s zMapperIterDataPipe.__iter__cCs.t|jtrt|jStt|jddS)Nz# instance doesn't have valid length)rrrr" TypeErrortype__name__)r$r'r'r(__len__s   zMapperIterDataPipe.__len__)NN)r: __module__ __qualname____doc__r__annotations__rrr6rrr7intr; __classcell__r'r'r%r(rs '#c Cst|jdkrtd|d}t|}g}g}|D]}||kr8tdq8|D]}||kr|t||srtd||}nHzddlm}|j }Wn,tk r} ztd| W5d} ~ XYnX| t ||||} | | qRt d|} | |} | 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 Exception df_wrapperZ get_columnskeyscallableZtorcharrow.pytorchZpytorchZrecZDefaultr2strr) conversionitemdfZ columns_nameZ tuple_namesZ tuple_valuesnameZ collation_fntapevalueZtpl_clsr!r'r'r(_collate_helpers2         rOZcollatec sbeZdZdZedfeeeede fe ee e feee fffeeddfdd 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.)] N.)rrH collate_fnrcsT|dk rtj||dn6t|r4tj||dntt|}tj||ddS)N)r)rrrF functoolspartialrO)r$rrHrPr%r'r(rs  zCollatorIterDataPipe.__init__)r:r<r=r>r rrrrr r rGrrAr'r'r%r(rs1 )rQ collectionsrtypingrrrrrrr r r Z%torch.utils.data.datapipes._decoratorr Ztorch.utils.data._utils.collater Z$torch.utils.data.datapipes.dataframerrDZ#torch.utils.data.datapipes.datapiperZ'torch.utils.data.datapipes.utils.commonrr__all__rrrOrr'r'r'r(s  ,     q&