o biP@sZddlZddlmZmZmZmZmZmZddlZ ddl Z ddl Z ddl Z ddlmZddlmZddlmZmZmZmZmZmZmZddlmZeddZd ee jfd d Z   d6d e jd eeee eee fdeee j!ee j!fde"d ee j#ee j#ff ddZ$   d7de j%dee j!deee dfde"d e j#f ddZ&   d7dee j%ee e j%ffdeee j!ee e j!ffdeee dfde"d ee j#ee e j#fff ddZ' d8dee j(j)efdee j(j)d e j(j)fd d!Z*d"d#Z+d$ee efd%e d ee effd&d'Z,   d7dee ee j%fdeee j!ee e j!ffdeee dfde"d ee ee j#ff d(d)Z-   d9d*e j.deee j!ee e j!ffd+e"de"d ee ee j#ff d,d-Z/e 0defd.ee j#deee dfd/e"d e j#fd0d1Z1d*ed e fd2d3Z2e 0defd*edeee dfd/e"d efd4d5Z3dS):N)AnyDictListOptionalUnionSequence)#get_torch_device_manager_by_context)%_unwrap_ndarray_object_type_if_needed)TensorBatchTypeTensorBatchReturnType _is_tensor_is_tensor_sequence_is_nested_tensor_sequence_is_tensor_mapping_is_tensor_sequence_mapping)env_bool,RAY_AIR_DEFAULT_TENSOR_NON_BLOCKING_TRANSFERTreturncCs tS)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_devicesrrQ/home/ubuntu/.local/lib/python3.10/site-packages/ray/air/_internal/torch_utils.pyrs r data_batchcolumns column_dtypes unsqueezecs|o t|dttf}|s|rt|tjstdt|d|r#|ng}fddfdd|rPt|ttfvrD|gt|}fdd t||DS||d 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.csrt|tjjjr |}|jjtj ur2fdd|D}zt |WSt y1t |YSwt j|dS)zcThis recursive function allows to convert pyarrow List dtypes to multi-dimensional tensors.csg|]}|qSrr).0x)dtype tensorizerr dszEconvert_pandas_to_torch_tensor..tensorize..r) isinstancepdapi extensionsExtensionArrayto_numpyrtypenpobject_torchstack RuntimeError nested_tensor as_tensor)valsrtensors)rr rrWs  z1convert_pandas_to_torch_tensor..tensorizec sg}|r |}n}|jD]4}||j}z||d}Wnty5}z td|d|d|d}~wwr=|d}||qt|dkrRtj|dd}|S|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 ValueErrorrappendlenr*cat) rrfeature_tensorsbatchcolcol_valstefeature_tensor)rrrrrget_tensor_for_columnsos4      z>convert_pandas_to_torch_tensor..get_tensor_for_columnscsg|] \}}||dqS)rrr)r subcolumnsr)r@rrrs z2convert_pandas_to_torch_tensor..rA) r!listtupler*r TypeErrorr'r7zip)rrrr multi_inputr)rr@rrrconvert_pandas_to_torch_tensor(s"$    rHFndarrayrdevicez torch.device pin_memorycCst|}|jjtjurtdttdt j |||d}Wdn1s+wY|rH|j jdksDJd|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.ignorerrJNcpuz:Pin memory is only supported for CPU tensors. Got device: z and pin_memory: .) r rr'r(r)r,warningscatch_warnings simplefilterr*r.rJrK)rIrrJrKresultrrrconvert_ndarray_to_torch_tensors& rTndarraysdtypescsrt|tjr*ttr tdkrtdttt |d}|Sfdd| D}|S)a,Convert a NumPy ndarray batch to a Torch Tensor batch. Args: ndarray: A (dict of) NumPy ndarray(s) that we wish to convert to a Torch Tensor. dtype: A (dict of) Torch dtype(s) 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 (dict of) Torch Tensor(s). r1z[When constructing a single-tensor batch, only a single dtype should be given, instead got: rrJrKc s4i|]\}}|t|ttr|ndqS)rW)rTr!dict)rcol_name col_ndarrayrJrVrKrr sz?convert_ndarray_batch_to_torch_tensor_batch..) r!r(rIrXr7r5nextiterr3rTitems)rUrVrJrKr:rr[r+convert_ndarray_batch_to_torch_tensor_batchs(    r` saved_modelmodel_definitioncCsJt|tjjr |St|tr|std|||Stdt|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!r*nnModulerXr5load_state_dictr')rarbrrrload_torch_models   rfcCsvt|tjrdSt|tr&|D]\}}t|rdSt|r#dSqdSt|ttfr9|D] }t|r8dSq/dS)NTF)r!r*TensorrXr_contains_tensorrCrD)objkvrrrrhs  rh state_dictprefixcCsd}|D]}||r"|t|d}|s|}d}||||<qd|vrN|d|d<|d}|D]}t|dkr>q5|t|d}||||<q5|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) startswithr7copypop)rlrmcopiedkeynewkeymetadatarrr4consume_prefix_in_state_dict_if_present_not_in_place's$  rvcsfdd|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. cs*i|]\}fdd|DqS)cs.g|]}t|ttrndqS))rVrJrKr`r!rX)rrI)rYrJrVrKrrrcszHconvert_ndarray_list_to_torch_tensor_list...r)r col_ndarraysr[)rYrr\bs  z=convert_ndarray_list_to_torch_tensor_list..)r_)rUrVrJrKrr[r)convert_ndarray_list_to_torch_tensor_listNs ryr:combine_chunkscsZddlm}ddlm}|r!||d}fdd|DS||}t|dS)a/Convert 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. Returns: A dictionary of column name to list of tensors. For non-chunked columns, the list will contain a single tensor. r)transform_pyarrow)ArrowBlockAccessornumpycs2i|]\}}|t|ttr|ndqS)rVrKrw)rrY col_arrayr~rrr\sz*arrow_batch_to_tensors..r~)ray.data._internal.arrow_opsr{ray.data._internal.arrow_blockr|to_batch_formatr_table_to_numpy_dict_chunkedry)r:rVrzrKr{r| numpy_batch numpy_listrr~rarrow_batch_to_tensorsps    rtensor_sequence non_blockingc sB|s Jd|tdd|DsJdtdd|D|djtfdd|Ds>Jd d d d|D|djd d tfdd|DsaJdd dd|D|d}|j}|jd d }tdd|D}tj|g|R||d}d}|D]} || jd} ||| j| |d| }q|S)aPStack sequence of tensors into a contiguous GPU tensor. Args: tensor_sequence: Sequence of tensors to stack device: The device to move tensors to 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: css|] }t|tjVqdSN)r!r*rgrr=rrr s  z+concat_tensors_to_device..z5All items must be torch.Tensor. Found invalid types: cSs g|] }t|tjst|qSr)r!r*rgr'rrrrr z,concat_tensors_to_device..rc3s|]}|jkVqdSrr r) first_dtyperrrz0All tensors must have the same dtype. Expected: z, got: cSsg|]}|jqSrr rrrrrsr1Nc3s"|] }|jddkVqdSr1Nshaper) first_shaperrrs z4All tensors must have the same shape[1:]. Expected: cSsg|] }|jddqSrrrrrrrcss|]}|jdVqdS)rNrrrrrrrrMr)allstrrrsumr*emptycopy_) rrJrfirstr shape_tail total_rowsrS row_startr=row_endr)rrrconcat_tensors_to_devicesL     rcCst|j}t|ttfr"dtdd|D}|d|d}|St|tr?dtdd|D}|d|d}|S|}|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 | cSh|]}t|qSr _get_type_strrrkrrr z _get_type_str..[]cSrrrrrrrrrz[str, ) r'__name__r!rCrDjoinsortedrXr3)r: curr_type val_typesinvalid_type_strrrrrs  rcsdur|St|r|jdSt|r#t|fdd|DSt|r5t|fdd|DSt|rEfdd|DSt|rUfdd|DStd t |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 Nrcsg|] }|jdqSrtorrJrrrrrz*move_tensors_to_device..csg|]}t|qSrrrrrrrscs i|] \}}||jdqSrr)rrjr=rrrr\rz*move_tensors_to_device..csi|] \}}|t|qSrr)rrjrkrrrr\s zInvalid input type: z. Expected one of the following: torch.Tensor, List/Tuple[torch.Tensor], Dict[str, torch.Tensor], Mapping[str, List/Tuple[torch.Tensor]]) r rr r'rrr_rr5r)r:rJrrrrmove_tensors_to_devices& r)NNT)NNFr)NFF)4rPtypingrrrrrrr}r(pandasr"r*pyarrow ray.air._internal.device_managerr"ray.air.util.data_batch_conversionr ray.data.collate_fnr r r r rrrray._private.ray_constantsr$DEFAULT_TENSOR_NON_BLOCKING_TRANSFERrJr DataFramerrboolrgrHrIrTr`rcrdrfrhrvryTablerno_gradrrrrrrrs   $   o 3 2     ) $ , 8