o bi# @sddlZddlmZmZmZmZmZmZmZm Z m Z m Z ddl Z ddlmZddlmZer@ddlZddlZddlZddlmZe dedZe ede d fZe deeee ed feedfeeeffZd ed efd dZd ed efddZd ed efddZd ed efddZd ed efddZ ed ed efddZ!e de d eedffZ"eGdddeeZ#eGddde#dZ$eGddde#eee j%fZ&eGd d!d!e#d"Z'eGd#d$d$e$Z(dS)%N) TYPE_CHECKINGAnyDictGenericListMappingOptionalTupleTypeVarUnion) DataBatch) DeveloperAPI) CollatedData DataBatchType)bound torch.Tensor)r..batchreturncCsddl}t||jS)z*Check if a batch is a single torch.Tensor.rN)torch isinstanceTensor)rrrG/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/collate_fn.py _is_tensor1s rcC t|ttfotdd|DS)aDCheck if a batch is a sequence of torch.Tensors. >>> import torch >>> _is_tensor_sequence(torch.ones(1)) False >>> _is_tensor_sequence([torch.ones(1), torch.ones(1)]) True >>> _is_tensor_sequence((torch.ones(1), torch.ones(1))) True >>> _is_tensor_sequence([torch.ones(1), 1]) False cs|]}t|VqdSNr.0trrr Ez&_is_tensor_sequence..rlisttupleallrrrr_is_tensor_sequence8s r(cCr)a=Check if a batch is a sequence of sequences of torch.Tensors. Stops at one level of nesting. >>> import torch >>> _is_nested_tensor_sequence([torch.ones(1), torch.ones(1)]) False >>> _is_nested_tensor_sequence( ... ([torch.ones(1), torch.ones(1)], [torch.ones(1)]) ... ) True csrrr(rrrrr!U z-_is_nested_tensor_sequence..r#r'rrr_is_nested_tensor_sequenceHs r+cC t|totdd|DS)aCheck if a batch is a mapping of keys to torch.Tensors. >>> import torch >>> _is_tensor_mapping({"a": torch.ones(1), "b": torch.ones(1)}) True >>> _is_tensor_mapping({"a": torch.ones(1), "b": [torch.ones(1), torch.ones(1)]}) False csrrrrvrrrr!cr"z%_is_tensor_mapping..rrr&valuesr'rrr_is_tensor_mappingZs r1cCr,)aECheck if a batch is a mapping of keys to sequences of torch.Tensors. >>> import torch >>> _is_tensor_sequence_mapping({"a": torch.ones(1), "b": torch.ones(1)}) False >>> _is_tensor_sequence_mapping( ... {"a": (torch.ones(1), torch.ones(1)), "b": [torch.ones(1), torch.ones(1)]} ... ) True csrrr)r-rrrr!qr*z._is_tensor_sequence_mapping..r/r'rrr_is_tensor_sequence_mappingfs r2cCs(t|pt|pt|pt|pt|S)a"Check if a batch matches any of the TensorBatchType variants. This function checks if the input batch is one of the following types: 1. A single torch.Tensor 2. A sequence of torch.Tensors 3. A sequence of sequences of torch.Tensors 4. A mapping (e.g., dict) of keys to torch.Tensors 5. A mapping (e.g., dict) of keys to sequences of torch.Tensors Args: batch: The input batch to check. Can be any type. Returns: bool: True if the batch matches any TensorBatchType variant, False otherwise. )rr(r+r1r2r'rrris_tensor_batch_typevsr3c@s(eZdZdZejdeddfddZdS) CollateFnzAbstract interface for collate_fn for `iter_torch_batches`. See doc-string of `collate_fn` in `iter_torch_batches` API for more details. rrrcCdS)zConvert a batch of data to collated format. Args: batch: The input batch to collate. Returns: The collated data in the format expected by the model. Nrselfrrrr__call__s zCollateFn.__call__N)__name__ __module__ __qualname____doc__abcabstractmethodrr8rrrrr4sr4c@eZdZdZd ddZdS) ArrowBatchCollateFnaCollate function that takes pyarrow.Table as the input batch type. Arrow tables with chunked arrays can be efficiently transferred to GPUs without combining the chunks with the `arrow_batch_to_tensors` utility function. See `DefaultCollateFn` for example. r pyarrow.TablerrcCr5)zConvert a batch of pyarrow.Table to collated format. Args: batch: The input pyarrow.Table batch to collate. Returns: The collated data in the format expected by the model. Nrr6rrrr8 zArrowBatchCollateFn.__call__N)rrArrr9r:r;r<r8rrrrr@sr@rAc@s,eZdZdZdeeejfddfddZdS)NumpyBatchCollateFnzQCollate function that takes a dictionary of numpy arrays as the input batch type.rrrcCr5)zConvert a batch of numpy arrays to collated format. Args: batch: The input dictionary of numpy arrays batch to collate. Returns: The collated data in the format expected by the model. Nrr6rrrr8rBzNumpyBatchCollateFn.__call__N) r9r:r;r<rstrnpndarrayr8rrrrrDs rDc@r?) PandasBatchCollateFnzGCollate function that takes a pandas.DataFrame as the input batch type.rpandas.DataFramerrcCr5)zConvert a batch of pandas.DataFrame to collated format. Args: batch: The input pandas.DataFrame batch to collate. Returns: The collated data in the format expected by the model. Nrr6rrrr8rBzPandasBatchCollateFn.__call__N)rrIrrrCrrrrrHsrHrIcsteZdZdZ   ddeedeedffdeeedfdeffd d Z d d d eee dffddZ Z S)DefaultCollateFnzIDefault collate function for converting Arrow batches to PyTorch tensors.NFdtypesz torch.dtypedevicez torch.device pin_memorycs@ddl}t||_t|tr|||_n||_||_dS)aInitialize the collate function. Args: dtypes: The torch dtype(s) for the created tensor(s); if None, the dtype will be inferred from the tensor data. device: The device on which the tensor should be placed. Can be a string (e.g. "cpu", "cuda:0") or a torch.device object. pin_memory: Whether to pin the memory of the created tensors. rN)rsuper__init__rKrrErLrM)r7rKrLrMr __class__rrrOs   zDefaultCollateFn.__init__rrArrcCs,ddlm}|jjdk}|||j||jdS)zConvert an Arrow batch to PyTorch tensors. Args: batch: PyArrow Table to convert Returns: Dictionary mapping column names to lists of tensors r)arrow_batch_to_tensorscpu)rKcombine_chunksrM)ray.air._internal.torch_utilsrRrLtyperKrM)r7rrRrTrrrr8s zDefaultCollateFn.__call__)NNF) r9r:r;r<rr rrEboolrOrr8 __classcell__rrrPrrJs&rJ))r=typingrrrrrrrr r r numpyrFray.data.blockr ray.util.annotationsr pandaspyarrowrray.data.datasetrrTensorSequenceTyperETensorBatchTyperWrr(r+r1r2r3TensorBatchReturnTyper4r@rGrDrHrJrrrrs`0