

'''simple docstring'''



import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available







def    a__       (  *a__						):




     """simple docstring"""


     if not isinstance(lowercase__      ,    lowercase__						):
          __SCREAMING_SNAKE_CASE     =      list(lowercase__						)
     for i in range(len(lowercase__						)						):
          __SCREAMING_SNAKE_CASE     =      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    a__       (  a__						):




     """simple docstring"""


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







def    a__       (  a__ = None      ,    a__ = 1_28						):




     """simple docstring"""


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

     __SCREAMING_SNAKE_CASE     =      starting_batch_size

     def decorator(*a__      ,    **a__						):
          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()
          __SCREAMING_SNAKE_CASE     =      list(inspect.signature(lowercase__						).parameters.keys()						)
          # Guard against user error
          if len(lowercase__						) < (len(lowercase__						) + 1):
               __SCREAMING_SNAKE_CASE     =      ''', '''.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(lowercase__      ,    *lowercase__      ,    **lowercase__						)
               except Exception as e:
                    if should_reduce_batch_size(lowercase__						):
                         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










from typing import Optional

from .. import Features, NamedSplit
from ..packaged_modules.text.text import Text
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader
class        _lowercase      (			A__       ):






				'''simple docstring'''



				def __init__(       self			:int			,			lowerCAmelCase__			:NestedDataStructureLike[PathLike]			,			lowerCAmelCase__			:Optional[NamedSplit] = None			,			lowerCAmelCase__			:Optional[Features] = None			,			lowerCAmelCase__			:str = None			,			lowerCAmelCase__			:bool = False			,			lowerCAmelCase__			:bool = False			,			lowerCAmelCase__			:Optional[int] = None			,			**lowerCAmelCase__			:Optional[int]			,			)		->			Tuple:
										super().__init__(
										    lowerCAmelCase__			,			split=lowerCAmelCase__			,			features=lowerCAmelCase__			,			cache_dir=lowerCAmelCase__			,			keep_in_memory=lowerCAmelCase__			,			streaming=lowerCAmelCase__			,			num_proc=lowerCAmelCase__			,			**lowerCAmelCase__			,			)
										__SCREAMING_SNAKE_CASE						:     List[str]					=		path_or_paths if isinstance(lowerCAmelCase__			,			lowerCAmelCase__	) else {self.split: path_or_paths}
										__SCREAMING_SNAKE_CASE						:     int					=		Text(
										    cache_dir=lowerCAmelCase__			,			data_files=lowerCAmelCase__			,			features=lowerCAmelCase__			,			**lowerCAmelCase__			,			)



				def 		__magic_name__(       self			:Dict	)		->			Tuple:
										# Build iterable dataset
										if self.streaming:
																__SCREAMING_SNAKE_CASE						:     int					=		self.builder.as_streaming_dataset(split=self.split	)
										# Build regular (map-style) dataset
										else:
																__SCREAMING_SNAKE_CASE						:     List[str]					=		None
																__SCREAMING_SNAKE_CASE						:     str					=		None
																__SCREAMING_SNAKE_CASE						:     Dict					=		None
																__SCREAMING_SNAKE_CASE						:     Tuple					=		None

																self.builder.download_and_prepare(
																    download_config=lowerCAmelCase__			,			download_mode=lowerCAmelCase__			,			verification_mode=lowerCAmelCase__			,			base_path=lowerCAmelCase__			,			num_proc=self.num_proc			,			)
																__SCREAMING_SNAKE_CASE						:     Optional[int]					=		self.builder.as_dataset(
																    split=self.split			,			verification_mode=lowerCAmelCase__			,			in_memory=self.keep_in_memory	)
										return dataset
