
from importlib import import_module

from .logging import get_logger


__a         =   get_logger(__name__)
class 					lowercase__:



	"""simple docstring"""




	def __init__(			self    :     List[str]			,				SCREAMING_SNAKE_CASE_    :     str			,				SCREAMING_SNAKE_CASE_    :     int=None			)			->	Tuple:
					lowercase_	    =   attrs or []
					if module is not None:
									for key in module.__dict__:
													if key in attrs or not key.startswith('''__'''			):
																	setattr(self			,				SCREAMING_SNAKE_CASE_			,				getattr(SCREAMING_SNAKE_CASE_			,				SCREAMING_SNAKE_CASE_			)			)
					lowercase_	    =   module._original_module if isinstance(SCREAMING_SNAKE_CASE_			,				_PatchedModuleObj			) else module

class 					lowercase__:



	"""simple docstring"""




	a     :str       	=      []
	def __init__(			self    :     List[str]			,				SCREAMING_SNAKE_CASE_    :     Dict			,				SCREAMING_SNAKE_CASE_    :     str			,				SCREAMING_SNAKE_CASE_    :     int			,				SCREAMING_SNAKE_CASE_    :     int=None			)			->	Optional[Any]:
					lowercase_	    =   obj
					lowercase_	    =   target
					lowercase_	    =   new
					lowercase_	    =   target.split('''.'''			)[0]
					lowercase_	    =   {}
					lowercase_	    =   attrs or []
	def __enter__(			self    :     int			)			->	Dict:
					*lowercase_    ,      lowercase_	    =   self.target.split('''.'''			)

					# Patch modules:
					# it's used to patch attributes of submodules like "os.path.join";
					# in this case we need to patch "os" and "os.path"

					for i in range(len(SCREAMING_SNAKE_CASE_			)			):
									try:
													lowercase_	    =   import_module('''.'''.join(submodules[: i + 1]			)			)
									except ModuleNotFoundError:
													continue
									# We iterate over all the globals in self.obj in case we find "os" or "os.path"
									for attr in self.obj.__dir__():
													lowercase_	    =   getattr(self.obj			,				SCREAMING_SNAKE_CASE_			)
													# We don't check for the name of the global, but rather if its value *is* "os" or "os.path".
													# This allows to patch renamed modules like "from os import path as ospath".
													if obj_attr is submodule or (
													    (isinstance(SCREAMING_SNAKE_CASE_			,				_PatchedModuleObj			) and obj_attr._original_module is submodule)
													):
																	lowercase_	    =   obj_attr
																	# patch at top level
																	setattr(self.obj			,				SCREAMING_SNAKE_CASE_			,				_PatchedModuleObj(SCREAMING_SNAKE_CASE_			,				attrs=self.attrs			)			)
																	lowercase_	    =   getattr(self.obj			,				SCREAMING_SNAKE_CASE_			)
																	# construct lower levels patches
																	for key in submodules[i + 1 :]:
																					setattr(SCREAMING_SNAKE_CASE_			,				SCREAMING_SNAKE_CASE_			,				_PatchedModuleObj(getattr(SCREAMING_SNAKE_CASE_			,				SCREAMING_SNAKE_CASE_			,				SCREAMING_SNAKE_CASE_			)			,				attrs=self.attrs			)			)
																					lowercase_	    =   getattr(SCREAMING_SNAKE_CASE_			,				SCREAMING_SNAKE_CASE_			)
																	# finally set the target attribute
																	setattr(SCREAMING_SNAKE_CASE_			,				SCREAMING_SNAKE_CASE_			,				self.new			)

        # Patch attribute itself:
        # it's used for builtins like "open",
        # and also to patch "os.path.join" we may also need to patch "join"
        # itself if it was imported as "from os.path import join".

					if submodules:  # if it's an attribute of a submodule like "os.path.join"
									try:
													lowercase_	    =   getattr(import_module('''.'''.join(SCREAMING_SNAKE_CASE_			)			)			,				SCREAMING_SNAKE_CASE_			)
									except (AttributeError, ModuleNotFoundError):
													return
									# We iterate over all the globals in self.obj in case we find "os.path.join"
									for attr in self.obj.__dir__():
													# We don't check for the name of the global, but rather if its value *is* "os.path.join".
													# This allows to patch renamed attributes like "from os.path import join as pjoin".
													if getattr(self.obj			,				SCREAMING_SNAKE_CASE_			) is attr_value:
																	lowercase_	    =   getattr(self.obj			,				SCREAMING_SNAKE_CASE_			)
																	setattr(self.obj			,				SCREAMING_SNAKE_CASE_			,				self.new			)
					elif target_attr in globals()["__builtins__"]:  # if it'a s builtin like "open"
									lowercase_	    =   globals()['''__builtins__'''][target_attr]
									setattr(self.obj			,				SCREAMING_SNAKE_CASE_			,				self.new			)
					else:
									raise RuntimeError(f'''Tried to patch attribute {target_attr} instead of a submodule.'''			)
	def __exit__(			self    :     Any			,				*SCREAMING_SNAKE_CASE_    :     Dict			)			->	List[Any]:
					for attr in list(self.original			):
									setattr(self.obj			,				SCREAMING_SNAKE_CASE_			,				self.original.pop(SCREAMING_SNAKE_CASE_			)			)
	def 		_lowercase    (			self    :     Any			)			->	str:
					self.__enter__()
					self._active_patches.append(self			)




	def 		_lowercase    (			self    :     Optional[int]			)			->	int:
					try:
									self._active_patches.remove(self			)
					except ValueError:
									# If the patch hasn't been started this will fail
									return None

					return self.__exit__()












'''simple docstring'''







import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available


def       __UpperCamelCase					(      *lowercase_       :  Union[str, Any]   ):
       """simple docstring"""




       if not isinstance(lowercase_   ,       lowercase_   ):
              a_						     =			list(lowercase_   )
       for i in range(len(lowercase_   )   ):
              a_						     =			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       __UpperCamelCase					(      lowercase_       :  Exception   ):
       """simple docstring"""




       a_						     =			[
           '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       __UpperCamelCase					(      lowercase_       :  callable = None   ,       lowercase_       :  int = 128   ):
       """simple docstring"""




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

       a_						     =			starting_batch_size

       def decorator(*lowercase_       :  str   ,       **lowercase_       :  Dict   ):
              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()
              a_						     =			list(inspect.signature(lowercase_   ).parameters.keys()   )
              # Guard against user error
              if len(lowercase_   ) < (len(lowercase_   ) + 1):
                     a_						     =			', '.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




