&`i#$ ddlZddlmZddlmZmZmZmZmZm Z m Z m Z m Z m Z ddlZddlmZddlmZddlmZerddlZddlZddlZddlmZddlmZmZe dd Ze ed e d fZe d eeee ed fe ed fe eeffZdede fdZ!dede fdZ"dede fdZ#dede fdZ$dede fdZ%edede fdZ&e d e d eed ffZ'eGddeeZ(eGdde(dZ)eGdde(eeej*fZ+eGdde(dZ,eGd d!e)Z-dS)"N)ThreadPoolExecutor) TYPE_CHECKINGAnyDictGenericListMappingOptionalTupleTypeVarUnion) env_integer) DataBatch) DeveloperAPI) CollatedDataTorchDeviceType DataBatchTyper)bound torch.Tensor)r..batchreturnc4ddl}t||jS)z*Check if a batch is a single torch.Tensor.rN)torch isinstanceTensor)rrs g/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/collate_fn.py _is_tensorr4sLLL eU\ * **clt|ttfotd|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 c34K|]}t|VdSNr.0ts r z&_is_tensor_sequence..Hs(3Q3QaJqMM3Q3Q3Q3Q3Q3Qrrlisttupleallrs r_is_tensor_sequencer,;s4 edE] + + Q3Q3Q53Q3Q3Q0Q0QQrclt|ttfotd|DS)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 c34K|]}t|VdSr!r,r#s rr&z-_is_nested_tensor_sequence..Xs<44#$A444444rr'r+s r_is_nested_tensor_sequencer0KsG edE] + + 44(-44411rct|to*td|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 c34K|]}t|VdSr!r"r$vs rr&z%_is_tensor_mapping..fs(-T-Tjmm-T-T-T-T-T-Trrr r*valuesr+s r_is_tensor_mappingr7]s8 eW % % T#-T-TU\\^^-T-T-T*T*TTrct|to*td|DS)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 c34K|]}t|VdSr!r/r3s rr&z._is_tensor_sequence_mapping..ts<..#$A......rr5r+s r_is_tensor_sequence_mappingr:isK eW % % #..(- ...++rct|p;t|p,t|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,r0r7r:r+s ris_tensor_batch_typer<ysX$ 5 . u % % . %e , , . e $ $ . 'u - - rc<eZdZdZejdeddfdZdS) CollateFnzAbstract interface for collate_fn for `iter_torch_batches`. See doc-string of `collate_fn` in `iter_torch_batches` API for more details. rrrcdS)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. Nselfrs r__call__zCollateFn.__call__s  rN)__name__ __module__ __qualname____doc__abcabstractmethodrrCr@rrr>r>sP   m          rr>ceZdZdZddZdS) 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.TablerrcdS)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. Nr@rAs rrCzArrowBatchCollateFn.__call__  rN)rrLrrrDrErFrGrCr@rrrKrKs2       rrKrLc<eZdZdZdeeejfddfdZdS)NumpyBatchCollateFnzQCollate function that takes a dictionary of numpy arrays as the input batch type.rrrcdS)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. Nr@rAs rrCzNumpyBatchCollateFn.__call__rNrN) rDrErFrGrstrnpndarrayrCr@rrrQrQsG[[  d3 ?3         rrQceZdZdZddZdS) PandasBatchCollateFnzGCollate function that takes a pandas.DataFrame as the input batch type.rpandas.DataFramerrcdS)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. Nr@rAs rrCzPandasBatchCollateFn.__call__rNrN)rrXrrrOr@rrrWrWs.QQ       rrWrXc eZdZdZeddZdddefdeedee dffded d e d e ffd Z d Z dddeee dfee edfffdZxZS)DefaultCollateFnzIDefault collate function for converting Arrow batches to PyTorch tensors.2RAY_DATA_DEFAULT_COLLATE_FN_THREADPOOL_MAX_WORKERSNFdtypesz torch.dtypedevicer pin_memory num_workerscddl}t||_t |t t fr|j||_n||_||_||_ d|_ dS)aHInitialize 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. num_workers: Number of worker threads for parallel tensor conversion. Defaults to `RAY_DATA_DEFAULT_COLLATE_FN_THREADPOOL_MAX_WORKERS`. rN) rsuper__init__r^rrSintr_r`ra _threadpool)rBr^r_r`rar __class__s rrdzDefaultCollateFn.__init__sz$    fsCj ) ) !&%,v..DKK DK$&9=rcbt|ddr|jddSdS)z#Clean up threadpool on destruction.rfNF)wait)getattrrfshutdown)rBs r__del__zDefaultCollateFn.__del__s@ 4 - - 2   % %5 % 1 1 1 1 1 2 2rrrLrrcddlm}|jdkr!|jt |j|_|jduo|jjdk}|||j||j|jS)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_tensorsN) max_workerscpu)r^combine_chunksr` threadpool) ray.air._internal.torch_utilsrnrarfrr_typer^r`)rBrrnrqs rrCzDefaultCollateFn.__call__ s        a  D$4$<1d>NOOOD D0NT[5E5N%% ;)'     r)rDrErFrGr_DEFAULT_NUM_WORKERSr r rrSboolrerdrlrrC __classcell__)rgs@rr[r[s SS&;< LP.2 / >>}d3 3E.FFGH>*+> >  >>>>>><222  $ tC'($sD4H/H*II J        rr[).rHconcurrent.futuresrtypingrrrrrr r r r r numpyrTray._private.ray_constantsrray.data.blockrray.util.annotationsrpandaspyarrowrray.data.datasetrrrTensorSequenceTyperSTensorBatchTypervrr,r0r7r:r<TensorBatchReturnTyper>rKrUrQrWr[r@rrrs 111111                        222222$$$$$$------?MMMNNNLLL((((((>>>>>>>>{;;;      c !" C   C# #$ & +c+d++++ Rs Rt R R R R cd$ Uc Ud U U U U s t     2 n       &   $     )O4   &      )Dbj$9:           9%78     I I I I I *I I I I I r