



from ...processing_utils import ProcessorMixin

class       _SCREAMING_SNAKE_CASE	(UpperCamelCase ):
     lowerCAmelCase					=			"""WhisperFeatureExtractor"""
     lowerCAmelCase					=			"""WhisperTokenizer"""





     def __init__(       self :							Tuple     ,  UpperCamelCase :							str     ,  UpperCamelCase :							List[str]				)->Union[str, Any]:
      super().__init__(UpperCamelCase     ,  UpperCamelCase				)
      __SCREAMING_SNAKE_CASE       :    int       =	self.feature_extractor
      __SCREAMING_SNAKE_CASE       :    List[str]       =	False





     def 	__snake_case    (       self :							int     ,  UpperCamelCase :							List[Any]=None     ,  UpperCamelCase :							int=None     ,  UpperCamelCase :							List[str]=True				)->str:
      return self.tokenizer.get_decoder_prompt_ids(task=UpperCamelCase     ,  language=UpperCamelCase     ,  no_timestamps=UpperCamelCase				)





     def __call__(       self :							Optional[int]     ,  *UpperCamelCase :							Tuple     ,  **UpperCamelCase :							Any				)->Dict:
      # For backward compatibility
      if self._in_target_context_manager:
       return self.current_processor(*UpperCamelCase     ,  **UpperCamelCase				)

      __SCREAMING_SNAKE_CASE       :    Dict       =	kwargs.pop("audio"     ,  UpperCamelCase				)
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	kwargs.pop("sampling_rate"     ,  UpperCamelCase				)
      __SCREAMING_SNAKE_CASE       :    List[str]       =	kwargs.pop("text"     ,  UpperCamelCase				)
      if len(UpperCamelCase				) > 0:
       __SCREAMING_SNAKE_CASE       :    Tuple       =	args[0]
       __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	args[1:]

      if audio is None and text is None:
       raise ValueError("You need to specify either an `audio` or `text` input to process."				)

      if audio is not None:
       __SCREAMING_SNAKE_CASE       :    Tuple       =	self.feature_extractor(UpperCamelCase     ,  *UpperCamelCase     ,  sampling_rate=UpperCamelCase     ,  **UpperCamelCase				)
      if text is not None:
       __SCREAMING_SNAKE_CASE       :    List[str]       =	self.tokenizer(UpperCamelCase     ,  **UpperCamelCase				)

      if text is None:
       return inputs

      elif audio is None:
       return encodings
      else:
       __SCREAMING_SNAKE_CASE       :    str       =	encodings["input_ids"]
       return inputs





     def 	__snake_case    (       self :							Union[str, Any]     ,  *UpperCamelCase :							Tuple     ,  **UpperCamelCase :							str				)->Optional[int]:
      return self.tokenizer.batch_decode(*UpperCamelCase     ,  **UpperCamelCase				)





     def 	__snake_case    (       self :							Tuple     ,  *UpperCamelCase :							List[str]     ,  **UpperCamelCase :							Tuple				)->Any:
      return self.tokenizer.decode(*UpperCamelCase     ,  **UpperCamelCase				)






     def 	__snake_case    (       self :							Union[str, Any]     ,  UpperCamelCase :							str     ,  UpperCamelCase :							str="np"				)->Any:
      return self.tokenizer.get_prompt_ids(UpperCamelCase     ,  return_tensors=UpperCamelCase				)











import inspect
import os
import unittest

import torch

import accelerate
from accelerate import debug_launcher
from accelerate.test_utils import (
    execute_subprocess_async,
    require_cpu,
    require_huggingface_suite,
    require_multi_gpu,
    require_single_gpu,
)
from accelerate.utils import patch_environment

@require_huggingface_suite
class       _SCREAMING_SNAKE_CASE	(unittest.TestCase ):





     def 	__snake_case    (       self :							Optional[Any]				)->Union[str, Any]:
      __SCREAMING_SNAKE_CASE       :    Optional[Any]       =	inspect.getfile(accelerate.test_utils				)
      __SCREAMING_SNAKE_CASE       :    Union[str, Any]       =	os.path.sep.join(
          mod_file.split(os.path.sep				)[:-1] + ["scripts", "external_deps", "test_metrics.py"]				)

      from accelerate.test_utils.scripts.external_deps import test_metrics  # noqa: F401

      __SCREAMING_SNAKE_CASE       :    Tuple       =	test_metrics





     @require_cpu
     def 	__snake_case    (       self :							Union[str, Any]				)->List[Any]:
      debug_launcher(self.test_metrics.main     ,  num_processes=1				)





     @require_cpu
     def 	__snake_case    (       self :							str				)->List[str]:
      debug_launcher(self.test_metrics.main				)





     @require_single_gpu
     def 	__snake_case    (       self :							Union[str, Any]				)->str:
      self.test_metrics.main()






     @require_multi_gpu
     def 	__snake_case    (       self :							List[str]				)->int:
      print(F"""Found {torch.cuda.device_count()} devices."""				)
      __SCREAMING_SNAKE_CASE       :    Any       =	["torchrun", F"""--nproc_per_node={torch.cuda.device_count()}""", self.test_file_path]
      with patch_environment(omp_num_threads=1				):
       execute_subprocess_async(UpperCamelCase     ,  env=os.environ.copy()				)






