
import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available




def  __magic_name__ (   *__lowerCAmelCase	:						int							)			->   str:
				if not isinstance(__lowerCAmelCase		,  __lowerCAmelCase							):
								__lowerCamelCase    					=	list(__lowerCAmelCase							)
				for i in range(len(__lowerCAmelCase							)							):
								__lowerCamelCase    					=	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  __magic_name__ (   __lowerCAmelCase	:						Exception							)			->   bool:
				__lowerCamelCase    					=	[
				    '''CUDA out of memory.''',  # CUDA OOM
				    '''cuDNN error: CUDNN_STATUS_NOT_SUPPORTED.''',  # CUDNN SNAFU
				    '''DefaultCPUAllocator: can\'t allocate memory''',  # CPU OOM
				]
				if isinstance(__lowerCAmelCase		,  __lowerCAmelCase							) and len(exception.args							) == 1:
								return any(err in exception.args[0] for err in _statements							)
				return False







def  __magic_name__ (   __lowerCAmelCase	:						callable = None		,  __lowerCAmelCase	:						int = 128							)			->   Tuple:
				if function is None:
								return functools.partial(__lowerCAmelCase		,  starting_batch_size=__lowerCAmelCase							)

				__lowerCamelCase    					=	starting_batch_size

				def decorator(*__lowerCAmelCase	:						Dict		,  **__lowerCAmelCase	:						List[str]							):
								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()
								__lowerCamelCase    					=	list(inspect.signature(__lowerCAmelCase							).parameters.keys()							)
								# Guard against user error
								if len(__lowerCAmelCase							) < (len(__lowerCAmelCase							) + 1):
												__lowerCamelCase    					=	''', '''.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(__lowerCAmelCase		,  *__lowerCAmelCase		,  **__lowerCAmelCase							)
												except Exception as e:
																if should_reduce_batch_size(__lowerCAmelCase							):
																				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 ...utils import (
    OptionalDependencyNotAvailable,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)


try:
					if not (is_transformers_available() and is_torch_available() and is_transformers_version(">=", "4.25.0")):
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					from ...utils.dummy_torch_and_transformers_objects import UnCLIPImageVariationPipeline, UnCLIPPipeline
else:
					from .pipeline_unclip import UnCLIPPipeline
					from .pipeline_unclip_image_variation import UnCLIPImageVariationPipeline
					from .text_proj import UnCLIPTextProjModel



