'''simple docstring'''




import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available

def _lowercase	(      *lowerCamelCase__						)					->       Optional[Any]:





       """simple docstring"""


       if not isinstance(__a							,    __a						):
              __UpperCAmelCase  :				Union[str, Any]       				=		list(__a						)
       for i in range(len(__a						)						):
              __UpperCAmelCase  :				List[str]       				=		None
       gc.collect()
       if is_xpu_available():
              torch.xpu.empty_cache()
       elif is_npu_available():
              torch.npu.empty_cache()
       else:
              torch.cuda.empty_cache()
       return objects

def _lowercase	(      lowerCamelCase__						)					->       str:





       """simple docstring"""


       __UpperCAmelCase  :				Union[str, Any]       				=		[
           "CUDA out of memory.",  # CUDA OOM
           "cuDNN error: CUDNN_STATUS_NOT_SUPPORTED.",  # CUDNN SNAFU
           "DefaultCPUAllocator: can't allocate memory",  # CPU OOM
       ]
       if isinstance(__a							,    __a						) and len(exception.args						) == 1:
              return any(err in exception.args[0] for err in _statements						)
       return False


def _lowercase	(      lowerCamelCase__ = None							,    lowerCamelCase__ = 128						)					->       List[Any]:





       """simple docstring"""


       if function is None:
              return functools.partial(__a							,    starting_batch_size=__a						)

       __UpperCAmelCase  :				Tuple       				=		starting_batch_size

       def decorator(*lowerCamelCase__							,    **lowerCamelCase__						):
              nonlocal batch_size
              gc.collect()
              if is_xpu_available():
                     torch.xpu.empty_cache()
              elif is_npu_available():
                     torch.npu.empty_cache()
              else:
                     torch.cuda.empty_cache()
              __UpperCAmelCase  :				Optional[int]       				=		list(inspect.signature(__a						).parameters.keys()						)
              # Guard against user error
              if len(__a						) < (len(__a						) + 1):
                     __UpperCAmelCase  :				List[Any]       				=		", ".join([f"""{arg}={value}""" for arg, value in zip(params[1:]							,    args[1:]						)]						)
                     raise TypeError(
                         f"""Batch size was passed into `{function.__name__}` as the first argument when called."""
                         f"""Remove this as the decorator already does so: `{function.__name__}({arg_str})`"""						)
              while True:
                     if batch_size == 0:
                            raise RuntimeError("No executable batch size found, reached zero."						)
                     try:
                            return function(__a							,    *__a							,    **__a						)
                     except Exception as e:
                            if should_reduce_batch_size(__a						):
                                   gc.collect()
                                   if is_xpu_available():
                                          torch.xpu.empty_cache()
                                   elif is_npu_available():
                                          torch.npu.empty_cache()
                                   else:
                                          torch.cuda.empty_cache()
                                   batch_size //= 2
                            else:
                                   raise

       return decorator







'''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 		__A       (TensorFormatter[Mapping, "torch.Tensor", Mapping]   ):






       def __init__(      self				,     UpperCamelCase_=None				,     **UpperCamelCase_      ):
              super().__init__(features=UpperCamelCase_      )
              __UpperCAmelCase  :				Union[str, Any]       				=		torch_tensor_kwargs
              import torch  # noqa import torch at initialization






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              import torch

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






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              import torch

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

              __UpperCAmelCase  :				int       				=		{}

              if isinstance(UpperCamelCase_				,     (np.number, np.ndarray)      ) and np.issubdtype(value.dtype				,     np.integer      ):
                     __UpperCAmelCase  :				Optional[int]       				=		{"dtype": torch.intaa}
              elif isinstance(UpperCamelCase_				,     (np.number, np.ndarray)      ) and np.issubdtype(value.dtype				,     np.floating      ):
                     __UpperCAmelCase  :				str       				=		{"dtype": torch.floataa}
              elif config.PIL_AVAILABLE and "PIL" in sys.modules:
                     import PIL.Image

                     if isinstance(UpperCamelCase_				,     PIL.Image.Image      ):
                            __UpperCAmelCase  :				str       				=		np.asarray(UpperCamelCase_      )
              return torch.tensor(UpperCamelCase_				,     **{**default_dtype, **self.torch_tensor_kwargs}      )






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              import torch

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






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              return map_nested(self._recursive_tensorize				,     UpperCamelCase_				,     map_list=UpperCamelCase_      )






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				List[str]       				=		self.numpy_arrow_extractor().extract_row(UpperCamelCase_      )
              __UpperCAmelCase  :				Union[str, Any]       				=		self.python_features_decoder.decode_row(UpperCamelCase_      )
              return self.recursive_tensorize(UpperCamelCase_      )






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				Union[str, Any]       				=		self.numpy_arrow_extractor().extract_column(UpperCamelCase_      )
              __UpperCAmelCase  :				Optional[Any]       				=		self.python_features_decoder.decode_column(UpperCamelCase_				,     pa_table.column_names[0]      )
              __UpperCAmelCase  :				List[Any]       				=		self.recursive_tensorize(UpperCamelCase_      )
              __UpperCAmelCase  :				List[str]       				=		self._consolidate(UpperCamelCase_      )
              return column







       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				int       				=		self.numpy_arrow_extractor().extract_batch(UpperCamelCase_      )
              __UpperCAmelCase  :				Any       				=		self.python_features_decoder.decode_batch(UpperCamelCase_      )
              __UpperCAmelCase  :				Optional[int]       				=		self.recursive_tensorize(UpperCamelCase_      )
              for column_name in batch:
                     __UpperCAmelCase  :				Tuple       				=		self._consolidate(batch[column_name]      )
              return batch






