&`i,^JddlZddlmZddlmZmZmZmZmZm Z m Z ddl Z ddl ZddlZddlZddlmZddlmZddlmZddlmZmZmZmZmZmZmZedd Zd eej fd Z! d,d ej"d ee ee#eee#fdee ej$eej$fde%d e ej&eej&ff dZ' d-de j(deej$dee e#dfde%d ej&f dZ) d-de e j(ee#e j(ffdee ej$ee#ej$ffdee e#dfde%d e ej&ee#ej&fff dZ* d.de ej+j,efdeej+j,d ej+j,fdZ-dZ.dee#efd e#d ee#effd!Z/ d-dee#ee j(fdee ej$ee#ej$ffdee e#dfde%d ee#eej&ff d"Z0 d/d#ej1dee ej$ee#ej$ffd$e%de%d%eed e ee#ej&fee#eej&fff d&Z2ej3defd'eej&dee e#dfd(e%d ej&fd)Z4d#ed e#fd*Z5ej3defd#edee e#dfd(e%d efd+Z6dS)0N)ThreadPoolExecutor)AnyDictListOptionalSequenceTupleUnion)env_bool)#get_torch_device_manager_by_context)%_unwrap_ndarray_object_type_if_needed)TensorBatchReturnTypeTensorBatchType_is_nested_tensor_sequence _is_tensor_is_tensor_mapping_is_tensor_sequence_is_tensor_sequence_mapping,RAY_AIR_DEFAULT_TENSOR_NON_BLOCKING_TRANSFERTreturncBtS)aGets the correct torch device list configured for this process. Returns a list of torch accelerator (GPU, HPU, NPU...) devices allocated for the current worker. If no accelerators are assigned, then it returns a list with a single CPU device. )r get_devicesq/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/air/_internal/torch_utils.pyrrs / 0 0 < < > >>r data_batchcolumns column_dtypes unsqueezec|o!t|dttf}|s<|r:t|tjs t dt |d|r|ng}fdfd|rLt |ttfvr|gt|z}fdt||DS||S)a.Converts a Pandas dataframe to a torch Tensor or list of torch Tensors. The format of the return type will match the format of ``columns``. If a list of columns is provided, the return type will be a single tensor. If ``columns`` is a list of lists, then the return type will be a list of tensors. Args: data_batch: The pandas dataframe to convert to a torch tensor. columns: The names of the columns in the dataframe to include in the torch tensor. If this arg is a List[List[str]], then the return type will be a List of tensors. This is useful for multi-input models. If None, then use all columns in the ``data_batch``. column_dtype: The torch dtype to use for the tensor. If set to None, then automatically infer the dtype. unsqueeze: If set to True, the tensors will be unsqueezed (reshaped to (N, 1)) before being concatenated into the final tensor. Otherwise, they will be left as is, that is (N, ). Defaults to True. Returns: Either a torch tensor of size (N, len(columns)) where N is the number of rows in the ``data_batch`` Dataframe, or a list of tensors, where the size of item i is (N, len(columns[i])). rz^If `columns` is a list of strings, `column_dtypes` must be None or a single `torch.dtype`.Got z instead.cbt|tjjjr|}|jjtj urHfd|D} tj |S#t$rtj |cYSwxYwtj|S)zcThis recursive function allows to convert pyarrow List dtypes to multi-dimensional tensors.c(g|]}|Srr).0xdtype tensorizes r zEconvert_pandas_to_torch_tensor..tensorize..ds%999qyyE**999rr%) isinstancepdapi extensionsExtensionArrayto_numpyr%typenpobject_torchstack RuntimeError nested_tensor as_tensor)valsr%tensorsr&s ` rr&z1convert_pandas_to_torch_tensor..tensorizeWs dBF-< = = # ==??D :?bj ( (:9999D999G 4{7+++ 4 4 4*733333  4 ?4u555 5s#A77BBc g}|r |}n }|jD]s}||j}  ||}n)#t$r}td|d|d|d}~wwxYw r|d}||tt |dkrtj|d}n|d}|S)Nr(zFailed to convert column z to a Torch Tensor of dtype z2. See above exception chain for the exact failure.)dimr) rvalues Exception ValueErrorrappendlenr2cat) rr%feature_tensorsbatchcolcol_valstefeature_tensorrr&rs rget_tensor_for_columnsz>convert_pandas_to_torch_tensor..get_tensor_for_columnsos!  w'EEE= & &CSz(H Ihe444    QQQQQQ   #KKNN  " "1 % % % %   ! # #"YA>>>NN,Q/Ns 6 AAAc0g|]\}}||S)rr%r)r# subcolumnsr%rIs rr'z2convert_pandas_to_torch_tensor..s=   ! E # ":U C C C   rrK) r)listtupler2r% TypeErrorr/r@zip)rrrr multi_inputrIr&s` ` @@rconvert_pandas_to_torch_tensorrR(sEHEz'!*tUmDDK  = M5;1W1W  2 && 2 2 2   !(ggbG6666606L   tUm 3 3*Oc'll:M    %(-%@%@    &%g]KKKKrFndarrayr%devicez torch.device pin_memoryct|}|jjtjurt dt j5t jdtj |||}dddn #1swxYwY|r:|j jdksJd|j d|d| }|S) aConvert a NumPy ndarray to a Torch Tensor. Args: ndarray: A NumPy ndarray that we wish to convert to a Torch Tensor. dtype: A Torch dtype for the created tensor; if None, the dtype will be inferred from the NumPy ndarray data. device: The device on which the tensor(s) should be placed; if None, the Torch tensor(s) will be constructed on the CPU. pin_memory: Whether to pin the memory of the created tensors. Returns: A Torch Tensor. a<Numpy array of object dtype cannot be converted to a Torch Tensor. This may because the numpy array is a ragged tensor--it contains items of different sizes. If using `iter_torch_batches()` API, you can pass in a `collate_fn` argument to specify custom logic to convert the Numpy array batch to a Torch tensor batch.ignorer%rTNcpuz:Pin memory is only supported for CPU tensors. Got device: z and pin_memory: .) r r%r/r0r1r4warningscatch_warnings simplefilterr2r6rTrU)rSr%rTrUresults rconvert_ndarray_to_torch_tensorr_sF$4G<z?convert_ndarray_batch_to_torch_tensor_batch..sf   &+ 5*4VT*B*BNfX&&%    r) r)r0rSrer@r>nextiterr<r_items)r`rarTrUrCs ``` r+convert_ndarray_batch_to_torch_tensor_batchrls$(BJ'' fd # # 16{{a >5;>>$v}}//00F/ !          *2)9)9    Lr saved_modelmodel_definitionct|tjjr|St|tr(|st d|||St dt|d)zLoads a PyTorch model from the provided ``saved_model``. ``model_definition`` is only used when ``saved_model`` is a torch state dict, which will be loaded into ``model_definition``. Otherwise, ``model_definition`` is discarded. zYAttempting to load torch model from a state_dict, but no `model_definition` was provided.zSaved model is of type zt. The model saved in the checkpoint is expected to be of type `torch.nn.Module`, or a model state dict of type dict.)r)r2nnModulerer>load_state_dictr/)rmrns rload_torch_modelrss+ux//  K & &    ((555 (d;&7&7 ( ( (   rcLt|tjrdSt|tr?|D])\}}t |rdSt |rdS*n3t|t tfr|D]}t |rdSdS)NTF)r)r2Tensorrerkcontains_tensorrMrN)objkvs rrvrvs#u|$$ t C   IIKK  DAqq!! ttq!! tt   C$ ' '  Aq!! tt  5r state_dictprefixcd}|D]^}||rG|t|d}|s|}d}||||<_d|vrm|d|d<|d}|D]E}t|dkr|t|d}||||<F|S)aSStrip the prefix in state_dict, if any and return a new dict. Adapted from https://github.com/pytorch/pytorch/blob/c18da597e0bb1c1aecc97c77a73fed1849057fa4/torch/nn/modules/utils.py The original method modified the dict in-place. Args: state_dict: a state-dict to be loaded to the model. prefix: prefix. FNT _metadatar) startswithr@copypop)rzr{copiedkeynewkeymetadatas r4consume_prefix_in_state_dict_if_present_not_in_placer'sF 5 5 >>& ! ! 5V'F (__.. !+!4!4Jv j  ",["9">">"@"@ ;k* 1 1C3xx1}}V'F'||C00HV   rcLfd|DS)a~Convert a dict mapping column names to lists of ndarrays to Torch Tensors. Args: ndarrays: A dict mapping column names to lists of ndarrays that we wish to convert to Torch Tensors. dtypes: A (dict of) Torch dtype(s) for the created tensors; if None, the dtype will be inferred from the NumPy ndarray data. device: The device on which the tensor(s) should be placed; if None, the Torch tensor(s) will be constructed on the CPU. pin_memory: Whether to pin the memory of the created tensors. Returns: A dict mapping column names to lists of Tensors. c<i|]\}fd|DS)c pg|]2}t|ttrn3S))rarTrUrlr)re)r#rSrfrTrarUs rr'zHconvert_ndarray_list_to_torch_tensor_list...cs_    8+5fd+C+COvh''%       rr)r# col_ndarraysrfrTrarUs @rrhz=convert_ndarray_list_to_torch_tensor_list..bsg    #Hl        (      r)rk)r`rarTrUs ```r)convert_ndarray_list_to_torch_tensor_listrNsF(      '/nn&6&6    rrCcombine_chunks threadpoolcddlm}ddlm}|r||d}t |}|dkr||zdt ttj fdt ttj fffd } | | | } t| Sfd | DS||} t!d | D} t | }| dkr|d t tt$tj fdt tt$tj fffd } d| D}t'| | |}d| D}|D]\}}}||||<|St)| S)aConvert PyArrow batch to PyTorch tensors. Args: batch: PyArrow batch to convert dtypes: A (dict of) Torch dtype(s) for the created tensors; if None, the dtype will be inferred from the NumPy ndarray data. combine_chunks: If True, combine chunks in Arrow batch before converting to tensors. pin_memory: Whether to pin the memory of the created tensors. threadpool: Optional ThreadPoolExecutor for parallel processing. If provided, columns/arrays will be processed in parallel. If None, processing is sequential. Returns: When combine_chunks=True: A dictionary of column name to single tensor. When combine_chunks=False: A dictionary of column name to list of tensors. r)ArrowBlockAccessor)transform_pyarrownumpyr:Ncol_name_col_arrayrcp|\}}|t|ttr|nfSNrarUr)rrf col_arrayrarUs rprocess_columnz.arrow_batch_to_tensors..process_columnsP'9#)!L/9&$/G/GS6(++V)"""rc vi|]5\}}|t|ttr|n6S)rr)r#rfrrarUs rrhz*arrow_batch_to_tensors..sc (Hi E/9&$/G/GS6(++V)rc34K|]}t|VdSNr@)r#arrayss r z)arrow_batch_to_tensors..s(II63v;;IIIIIIr array_itemc t|\}}}||t|ttr|nfSrr)rrf array_indexarrayrarUs r process_arrayz-arrow_batch_to_tensors..process_arraysV0:,+u?3=fd3K3KWvh//QW#-rcHg|]\}}t|D] \}}|||f Sr) enumerate)r#rfridxrs rr'z*arrow_batch_to_tensors..sV$Hf"+F"3"3C3&rc<i|]\}}|dgt|zSrr)r#rfrs rrhz*arrow_batch_to_tensors..s;555$Hf4&3v;;.555rr)ray.data._internal.arrow_blockrray.data._internal.arrow_opsrto_batch_formatr@r strr0rSr2rumaprkretable_to_numpy_dict_chunkedsumr<intrMr)rCrarrUrrr numpy_batch num_columnsrprocessed_cols numpy_list total_arraysr array_itemsprocessed_arraysr^rfrtensors ` ` rarrow_batch_to_tensorsrps0BAAAAA>>>>>>Q((//??HH +&& ??z5 $)#rz/$: sEL()       (^^NK+>  'BB   IIZ5F5F5H5HIIIII *oo !   6 !#sBJ"67 sC-.       (2(8(8(:(:K $JNN=+$N$NOO 55(2(8(8(:(:555F 2B 7 7-+v06x --M=% rtensor_sequence non_blockingcN |s Jd|td|Ds$Jdtd|Dzt|dkr-|#|djt j|kr|dS|dj t fd|DsJd d d |D|djdd t fd |DsJd d d|D|d}|j}|jdd}td|D}t j|g|R||}d}|D]3} || jdz} |||  | || }4|S)aqStack sequence of tensors into a contiguous GPU tensor. Args: tensor_sequence: Sequence of tensors to stack device: The device to move tensors to. If None, tensors are not moved. non_blocking: If True, perform device transfer without forcing a synchronization. Returns: A contiguous tensor on the target device z2Cannot stack empty sequence of tensors. Received: c3JK|]}t|tjVdSr)r)r2rur#rFs rrz+concat_tensors_to_device..s?() 1el##rz5All items must be torch.Tensor. Found invalid types: c`g|]+}t|tjt|,Sr)r)r2rur/rs rr'z,concat_tensors_to_device..s/MMMQAu|1L1LMaMMMrr:Nrc3.K|]}|jkVdSrr()r#rF first_dtypes rrz+concat_tensors_to_device..s*??!qw+%??????rz0All tensors must have the same dtype. Expected: z, got: cg|] }|j Srr(rs rr'z,concat_tensors_to_device..s)K)K)Ka!')K)K)Krc3>K|]}|jddkVdSr:Nshape)r#rF first_shapes rrz+concat_tensors_to_device..s2CCaqwqrr{k)CCCCCCrz4All tensors must have the same shape[1:]. Expected: c,g|]}|jddSrrrs rr'z,concat_tensors_to_device.. s")O)O)O!!'!""+)O)O)Orc30K|]}|jdVdS)rNrrs rrz+concat_tensors_to_device..s(99AQWQZ999999rrXr) allrr@rTr2r%rremptycopy_) rrTrfirstr% shape_tail total_rowsr^ row_startrFrow_endrrs @@rconcat_tensors_to_devicers& NNMOMMNN -<  >MM/MMMBB  ?q  /!,3u|F7K7KKKq!!!!$*K ??????? ? ? N  N N)K)K?)K)K)K N N ? "!$*122.K CCCC?CCC C C R  R R)O)O)O)O)O R R C A E KEQRRJ9999999J[*2z22% O O OFI agaj(y !'' 'EEE Mrct|j}t|ttfr5dt d|D}|d|d}n^t|trGdt d|D}|d|d}n|}|S)aGet a string representation of the possibly nested type of the batch. >>> import torch >>> _get_type_str([1, 2, "???"]) 'list[int | str]' >>> _get_type_str({"a": [1, 2, 3], "b": 4}) 'dict[str, int | list[int]]' >>> _get_type_str({"a": torch.tensor(1), "b": [torch.tensor(2)]}) 'dict[str, Tensor | list[Tensor]]' >>> _get_type_str({"a": torch.tensor(1), "b": {"c": torch.tensor(2)}}) 'dict[str, Tensor | dict[str, Tensor]]' z | c,h|]}t|Sr _get_type_strr#rys r z _get_type_str..-s &G&G&GA}Q'7'7&G&G&Gr[]c,h|]}t|Srrrs rrz _get_type_str..0s &P&P&PA}Q'7'7&P&P&Prz[str, ) r/__name__r)rMrNjoinsortedrer<)rC curr_type val_typesinvalid_type_strs rrrsU $I%$''%JJv&G&G&G&G&GHHII '66)666 E4 %JJv&P&P&P&P&PQQRR ';;y;;;$ rc*|St|r|St|r%t|fd|DSt |r%t|fd|DSt |r!fd|DSt|r!fd|DStdt|d) a Move tensors to the specified device. Concatenate nested lists/tuples of tensors along the first (batch) dimension. For example, for the input ((feature_0_chunk_0,), (feature_1_chunk_0, feature_1_chunk_1)) the output will be (feature_0_chunk_0, feature_1_chunk_0+1) where each feature is concatenated along the batch dimension. Args: batch: A tensor or collection of tensors to move to device. Can be: - A single tensor - A sequence of tensors - A sequence of sequences of tensors. The inner sequence of tensors is combined during GPU transfer. - A mapping (e.g., dict) of keys to tensors or sequences of tensors. The sequence of tensors is combined during GPU transfer. device: The device to move tensors to. If None, tensors are not moved. non_blocking: If True, perform device transfer without forcing a synchronization. Returns: The input tensors moved to the specified device Nrc>g|]}|Srtor#rFrTrs rr'z*move_tensors_to_device..Zs)SSSADDlDCCSSSrc2g|]}t|Srrrs rr'z*move_tensors_to_device..]s& N N N1 %a > > N N NrcFi|]\}}||Srr)r#rxrFrTrs rrhz*move_tensors_to_device..`s/UUUtq!144\4::UUUrc:i|]\}}|t|Srr)r#rxryrTrs rrhz*move_tensors_to_device..bs<   1 '6<@@   rzInvalid input type: z. Expected one of the following: torch.Tensor, List/Tuple[torch.Tensor], Dict[str, torch.Tensor], Mapping[str, List/Tuple[torch.Tensor]]) rrrr/rrrkrr>r)rCrTrs ``rmove_tensors_to_devicer7si:~ % xx\x::: U # # tE{{SSSSSUSSSTTT #E * * tE{{ N N N N N N N N    E " " UUUUUu{{}}UUUU $U + +             5=#7#7 5 5 5   r)NNT)NNFr)NFFN)7r[concurrent.futuresrtypingrrrrrr r rr0pandasr*pyarrowr2ray._private.ray_constantsr ray.air._internal.device_managerr "ray.air.util.data_batch_conversionr ray.data.collate_fnrrrrrrr$DEFAULT_TENSOR_NON_BLOCKING_TRANSFERrTr DataFramerr%boolrurRrSr_rlrprqrsrvrrTablerno_gradrrrrrrrs111111DDDDDDDDDDDDDDDDDD //////PPPPPPTTTTTT(0x2(($ ?T%,'????<@EI jLjL jL eDItDI67 8jLE%+tEK/@"@ABjL jL  5<el+ +, jLjLjLjL^$(37 .. Z. EK . U3./ 0. .  \ ....fDH37 --BJS"*_ 556- U5;S%+-=(>>? @- U3./ 0- -  5<c5</0 01 ----d37  ux,- ux/  X_    <   &$S#X$(+$ #s(^$$$$RDH37 3RZ(() U5;S%+-=(>>? @ U3./ 0   #tEL! !" HDH /3 ll =l U5;S%+-=(>>? @ll l +, l  4U\! "Dd5<.@)@$A AB llll^48=;;el+; U3./ 0;; \ ;;;;|248=6 6 6 U3./ 06 6  6 6 6 6 6 6 r