


'''simple docstring'''





# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
    import torch




class 				UpperCAmelCase			(			TensorFormatter[Mapping, "torch.Tensor", Mapping]						):


 def __init__(      self							:	Any			,       __lowerCamelCase							:	Optional[int]=None			,       **__lowerCamelCase							:	Optional[Any] ):
    super().__init__(features=__lowerCamelCase )
    UpperCAmelCase__	:List[str]      =      torch_tensor_kwargs
    import torch  # noqa import torch at initialization


 def        __SCREAMING_SNAKE_CASE					(      self							:	Any			,       __lowerCamelCase							:	Optional[Any] ):
    import torch

    if isinstance(__lowerCamelCase			,       __lowerCamelCase ) and column:
       if all(
           isinstance(__lowerCamelCase			,       torch.Tensor ) and x.shape == column[0].shape and x.dtype == column[0].dtype
           for x in column ):
          return torch.stack(__lowerCamelCase )
    return column


 def        __SCREAMING_SNAKE_CASE					(      self							:	str			,       __lowerCamelCase							:	str ):
    import torch

    if isinstance(__lowerCamelCase			,       (str, bytes, type(__lowerCamelCase )) ):
       return value
    elif isinstance(__lowerCamelCase			,       (np.character, np.ndarray) ) and np.issubdtype(value.dtype			,       np.character ):
       return value.tolist()

    UpperCAmelCase__	:List[Any]      =      {}

    if isinstance(__lowerCamelCase			,       (np.number, np.ndarray) ) and np.issubdtype(value.dtype			,       np.integer ):
       UpperCAmelCase__	:Any      =      {'''dtype''': torch.intaa}
    elif isinstance(__lowerCamelCase			,       (np.number, np.ndarray) ) and np.issubdtype(value.dtype			,       np.floating ):
       UpperCAmelCase__	:Any      =      {'''dtype''': torch.floataa}
    elif config.PIL_AVAILABLE and "PIL" in sys.modules:
       import PIL.Image

       if isinstance(__lowerCamelCase			,       PIL.Image.Image ):
          UpperCAmelCase__	:Any      =      np.asarray(__lowerCamelCase )
    return torch.tensor(__lowerCamelCase			,       **{**default_dtype, **self.torch_tensor_kwargs} )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Tuple			,       __lowerCamelCase							:	List[Any] ):
    import torch

    # support for torch, tf, jax etc.
    if hasattr(__lowerCamelCase			,       '''__array__''' ) and not isinstance(__lowerCamelCase			,       torch.Tensor ):
       UpperCAmelCase__	:List[Any]      =      data_struct.__array__()
    # support for nested types like struct of list of struct
    if isinstance(__lowerCamelCase			,       np.ndarray ):
       if data_struct.dtype == object:  # torch tensors cannot be instantied from an array of objects
          return self._consolidate([self.recursive_tensorize(__lowerCamelCase ) for substruct in data_struct] )
    elif isinstance(__lowerCamelCase			,       (list, tuple) ):
       return self._consolidate([self.recursive_tensorize(__lowerCamelCase ) for substruct in data_struct] )
    return self._tensorize(__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict			,       __lowerCamelCase							:	dict ):
    return map_nested(self._recursive_tensorize			,       __lowerCamelCase			,       map_list=__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any]			,       __lowerCamelCase							:	pa.Table ):
    UpperCAmelCase__	:Any      =      self.numpy_arrow_extractor().extract_row(__lowerCamelCase )
    UpperCAmelCase__	:Optional[Any]      =      self.python_features_decoder.decode_row(__lowerCamelCase )
    return self.recursive_tensorize(__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any]			,       __lowerCamelCase							:	pa.Table ):
    UpperCAmelCase__	:Any      =      self.numpy_arrow_extractor().extract_column(__lowerCamelCase )
    UpperCAmelCase__	:List[Any]      =      self.python_features_decoder.decode_column(__lowerCamelCase			,       pa_table.column_names[0] )
    UpperCAmelCase__	:Union[str, Any]      =      self.recursive_tensorize(__lowerCamelCase )
    UpperCAmelCase__	:str      =      self._consolidate(__lowerCamelCase )
    return column





 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any]			,       __lowerCamelCase							:	pa.Table ):
    UpperCAmelCase__	:Tuple      =      self.numpy_arrow_extractor().extract_batch(__lowerCamelCase )
    UpperCAmelCase__	:str      =      self.python_features_decoder.decode_batch(__lowerCamelCase )
    UpperCAmelCase__	:List[str]      =      self.recursive_tensorize(__lowerCamelCase )
    for column_name in batch:
       UpperCAmelCase__	:List[str]      =      self._consolidate(batch[column_name] )
    return batch







'''simple docstring'''





import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__lowerCamelCase												=						logging.get_logger(__name__)


def        a__     (					):
   # Get the sagemaker specific mp parameters from smp_options variable.
   UpperCAmelCase__	:Any      =      os.getenv('''SM_HP_MP_PARAMETERS''',   '''{}'''						)
   try:
      # Parse it and check the field "partitions" is included, it is required for model parallel.
      UpperCAmelCase__	:Any      =      json.loads(UpperCamelCase_						)
      if "partitions" not in smp_options:
         return False
   except json.JSONDecodeError:
      return False

   # Get the sagemaker specific framework parameters from mpi_options variable.
   UpperCAmelCase__	:Optional[Any]      =      os.getenv('''SM_FRAMEWORK_PARAMS''',   '''{}'''						)
   try:
      # Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
      UpperCAmelCase__	:str      =      json.loads(UpperCamelCase_						)
      if not mpi_options.get('''sagemaker_mpi_enabled''',   UpperCamelCase_						):
         return False
   except json.JSONDecodeError:
      return False
   # Lastly, check if the `smdistributed` module is present.
   return importlib.util.find_spec('''smdistributed'''						) is not None


if is_sagemaker_model_parallel_available():
    import smdistributed.modelparallel.torch as smp

    smp.init()




@dataclass
class 				UpperCAmelCase			(			_snake_case						):
 UpperCAmelCase				=					field(
     default=""						, metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}						, )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    super().__post_init__()
    warnings.warn(
        '''`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use '''
        '''`TrainingArguments` instead.'''			,       __lowerCamelCase			,       )


 @cached_property
 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    logger.info('''PyTorch: setting up devices''' )
    if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
       logger.warning(
           '''torch.distributed process group is initialized, but local_rank == -1. '''
           '''In order to use Torch DDP, launch your script with `python -m torch.distributed.launch''' )
    if self.no_cuda:
       UpperCAmelCase__	:Union[str, Any]      =      torch.device('''cpu''' )
       UpperCAmelCase__	:Dict      =      0
    elif is_sagemaker_model_parallel_available():
       UpperCAmelCase__	:Dict      =      smp.local_rank()
       UpperCAmelCase__	:List[str]      =      torch.device('''cuda'''			,       __lowerCamelCase )
       UpperCAmelCase__	:Optional[int]      =      1
    elif is_sagemaker_dp_enabled():
       import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

       torch.distributed.init_process_group(backend='''smddp'''			,       timeout=self.ddp_timeout_delta )
       UpperCAmelCase__	:int      =      int(os.getenv('''SMDATAPARALLEL_LOCAL_RANK''' ) )
       UpperCAmelCase__	:int      =      torch.device('''cuda'''			,       self.local_rank )
       UpperCAmelCase__	:Union[str, Any]      =      1
    elif self.local_rank == -1:
       # if n_gpu is > 1 we'll use nn.DataParallel.
       # If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
       # Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
       # trigger an error that a device index is missing. Index 0 takes into account the
       # GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
       # will use the first GPU in that env, i.e. GPU#1
       UpperCAmelCase__	:str      =      torch.device('''cuda:0''' if torch.cuda.is_available() else '''cpu''' )
       # Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
       # the default value.
       UpperCAmelCase__	:str      =      torch.cuda.device_count()
    else:
       # Here, we'll use torch.distributed.
       # Initializes the distributed backend which will take care of synchronizing nodes/GPUs
       if not torch.distributed.is_initialized():
          torch.distributed.init_process_group(backend='''nccl'''			,       timeout=self.ddp_timeout_delta )
       UpperCAmelCase__	:str      =      torch.device('''cuda'''			,       self.local_rank )
       UpperCAmelCase__	:Union[str, Any]      =      1

    if device.type == "cuda":
       torch.cuda.set_device(__lowerCamelCase )

    return device


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	List[str] ):
    if is_sagemaker_model_parallel_available():
       return smp.dp_size()

    return super().world_size


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict ):
    return not is_sagemaker_model_parallel_available()





 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	int ):
    return False



