
import functools
import gc
import inspect

import torch

from .imports import is_npu_available, is_xpu_available


def 						UpperCAmelCase      (     *a_						)		->     Union[str, Any]:







				"""simple docstring"""



				if not isinstance(lowerCamelCase_		,       lowerCamelCase_						):
								__A             =			list(lowerCamelCase_						)
				for i in range(len(lowerCamelCase_						)						):
								__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      (     a_						)		->     Union[str, Any]:







				"""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(lowerCamelCase_		,       lowerCamelCase_						) and len(exception.args						) == 1:
								return any(err in exception.args[0] for err in _statements						)
				return False


def 						UpperCAmelCase      (     a_ = None		,       a_ = 1_2_8						)		->     Optional[int]:







				"""simple docstring"""



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

				__A             =			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()
								__A             =			list(inspect.signature(lowerCamelCase_						).parameters.keys()						)
								# Guard against user error
								if len(lowerCamelCase_						) < (len(lowerCamelCase_						) + 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(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


import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Audio, Features, Value
from .base import TaskTemplate





@dataclass(frozen=__SCREAMING_SNAKE_CASE  )
class 							UpperCAmelCase				(  __SCREAMING_SNAKE_CASE  ):




	'''simple docstring'''





	snake_case_    =					field(default="automatic-speech-recognition"      ,      metadata={"include_in_asdict_even_if_is_default": True}  )
	snake_case_    =					Features({"audio": Audio()}  )
	snake_case_    =					Features({"transcription": Value("string"  )}  )
	snake_case_    =					"audio"
	snake_case_    =					"transcription"






	def     UpperCamelCase_  (	self :			Optional[int]							,A :			Union[str, Any]			):
					if self.audio_column not in features:
									raise ValueError(f'''Column {self.audio_column} is not present in features.'''			)
					if not isinstance(features[self.audio_column]							,A			):
									raise ValueError(f'''Column {self.audio_column} is not an Audio type.'''			)
					__A             =			copy.deepcopy(self			)
					__A             =			self.input_schema.copy()
					__A             =			features[self.audio_column]
					__A             =			input_schema
					return task_template







	@property
	def     UpperCamelCase_  (	self :			int			):
					return {self.audio_column: "audio", self.transcription_column: "transcription"}
