





"""simple docstring"""
import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment



def 					__lowerCamelCase       (							a_					:		Any			,       a_					:		str=()			,       a_					:		Dict=None			,       a_					:		Union[str, Any]="no"			,       a_					:		Optional[int]="29500"							)      ->				Optional[Any]:
      __SCREAMING_SNAKE_CASE       :Dict		    =  False
      __SCREAMING_SNAKE_CASE       :Tuple		    =  False
      if any(key.startswith('''KAGGLE'''							) for key in os.environ.keys()							):
            __SCREAMING_SNAKE_CASE       :int		    =  True
      elif "IPython" in sys.modules:
            __SCREAMING_SNAKE_CASE       :Dict		    =  '''google.colab''' in str(sys.modules['''IPython'''].get_ipython()							)

      try:
            __SCREAMING_SNAKE_CASE       :Dict		    =  PrecisionType(mixed_precision.lower()							)
      except ValueError:
            raise ValueError(
                f'''Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.'''							)

      if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''			,       a_							) is not None):
            # TPU launch
            import torch_xla.distributed.xla_multiprocessing as xmp

            if len(AcceleratorState._shared_state							) > 0:
                  raise ValueError(
                      '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
                      '''your training function. Restart your notebook and make sure no cells initializes an '''
                      '''`Accelerator`.'''							)
            if num_processes is None:
                  __SCREAMING_SNAKE_CASE       :Dict		    =  8

            __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  PrepareForLaunch(a_			,       distributed_type='''TPU'''							)
            print(f'''Launching a training on {num_processes} TPU cores.'''							)
            xmp.spawn(a_			,       args=a_			,       nprocs=a_			,       start_method='''fork'''							)
      elif in_colab:
            # No need for a distributed launch otherwise as it's either CPU or one GPU.
            if torch.cuda.is_available():
                  print('''Launching training on one GPU.'''							)
            else:
                  print('''Launching training on one CPU.'''							)
            function(*a_							)
      else:
            if num_processes is None:
                  raise ValueError(
                      '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.'''							)

            if num_processes > 1:
                  # Multi-GPU launch
                  from torch.multiprocessing import start_processes
                  from torch.multiprocessing.spawn import ProcessRaisedException

                  if len(AcceleratorState._shared_state							) > 0:
                        raise ValueError(
                            '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
                            '''inside your training function. Restart your notebook and make sure no cells initializes an '''
                            '''`Accelerator`.'''							)

                  if torch.cuda.is_initialized():
                        raise ValueError(
                            '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
                            '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
                            '''function.'''							)

                  # torch.distributed will expect a few environment variable to be here. We set the ones common to each
                  # process here (the other ones will be set be the launcher).
                  with patch_environment(
                      world_size=a_			,       master_addr='''127.0.01'''			,       master_port=a_			,       mixed_precision=a_							):
                        __SCREAMING_SNAKE_CASE       :str		    =  PrepareForLaunch(a_			,       distributed_type='''MULTI_GPU'''							)
                        print(f'''Launching training on {num_processes} GPUs.'''							)
                        try:
                              start_processes(a_			,       args=a_			,       nprocs=a_			,       start_method='''fork'''							)
                        except ProcessRaisedException as e:
                              if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
                                    raise RuntimeError(
                                        '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
                                        '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
                                        '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
                                        '''which one is problematic.'''							) from e

            else:
                  # No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
                  if is_mps_available():
                        __SCREAMING_SNAKE_CASE       :Dict		    =  '''1'''
                        print('''Launching training on MPS.'''							)
                  elif torch.cuda.is_available():
                        print('''Launching training on one GPU.'''							)
                  else:
                        print('''Launching training on CPU.'''							)
                  function(*a_							)







def 					__lowerCamelCase       (							a_					:		List[Any]			,       a_					:		Union[str, Any]=()			,       a_					:		List[Any]=2							)      ->				List[Any]:
      from torch.multiprocessing import start_processes

      with tempfile.NamedTemporaryFile() as tmp_file:
            # torch.distributed will expect a few environment variable to be here. We set the ones common to each
            # process here (the other ones will be set be the launcher).
            with patch_environment(
                world_size=a_			,       master_addr='''127.0.01'''			,       master_port='''29500'''			,       accelerate_mixed_precision='''no'''			,       accelerate_debug_rdv_file=tmp_file.name			,       accelerate_use_cpu='''yes'''			,       ):
                  __SCREAMING_SNAKE_CASE       :Any		    =  PrepareForLaunch(a_			,       debug=a_							)
                  start_processes(a_			,       args=a_			,       nprocs=a_			,       start_method='''fork'''							)










"""simple docstring"""
import unittest

from datasets import load_dataset

from transformers.pipelines import pipeline
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_torch, slow






@is_pipeline_test
@require_torch
class  _SCREAMING_SNAKE_CASE(       unittest.TestCase  ):




       @require_torch
       def    _UpperCamelCase					(						self							)      ->				str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE       :Dict		    =  pipeline(
                 task='''zero-shot-audio-classification'''		,model='''hf-internal-testing/tiny-clap-htsat-unfused'''							)
             __SCREAMING_SNAKE_CASE       :Any		    =  load_dataset('''ashraq/esc50'''							)
             __SCREAMING_SNAKE_CASE       :int		    =  dataset['''train''']['''audio'''][-1]['''array''']
             __SCREAMING_SNAKE_CASE       :Dict		    =  audio_classifier(SCREAMING_SNAKE_CASE__		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']							)
             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[{'''score''': 0.5_0_1, '''label''': '''Sound of a dog'''}, {'''score''': 0.4_9_9, '''label''': '''Sound of vaccum cleaner'''}]		,)




       @unittest.skip('''No models are available in TF'''							)
       def    _UpperCamelCase					(						self							)      ->				Optional[int]:





             """simple docstring"""






             pass




       @slow
       @require_torch
       def    _UpperCamelCase					(						self							)      ->				Tuple:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE       :Optional[int]		    =  pipeline(
                 task='''zero-shot-audio-classification'''		,model='''laion/clap-htsat-unfused'''		,)
             # This is an audio of a dog
             __SCREAMING_SNAKE_CASE       :List[Any]		    =  load_dataset('''ashraq/esc50'''							)
             __SCREAMING_SNAKE_CASE       :Tuple		    =  dataset['''train''']['''audio'''][-1]['''array''']
             __SCREAMING_SNAKE_CASE       :str		    =  audio_classifier(SCREAMING_SNAKE_CASE__		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']							)

             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[
                     {'''score''': 0.9_9_9, '''label''': '''Sound of a dog'''},
                     {'''score''': 0.0_0_1, '''label''': '''Sound of vaccum cleaner'''},
                 ]		,)

             __SCREAMING_SNAKE_CASE       :Dict		    =  audio_classifier([audio] * 5		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']							)
             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[
                     [
                         {'''score''': 0.9_9_9, '''label''': '''Sound of a dog'''},
                         {'''score''': 0.0_0_1, '''label''': '''Sound of vaccum cleaner'''},
                     ],
                 ]
                 * 5		,)
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  audio_classifier(
                 [audio] * 5		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']		,batch_size=5							)
             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[
                     [
                         {'''score''': 0.9_9_9, '''label''': '''Sound of a dog'''},
                         {'''score''': 0.0_0_1, '''label''': '''Sound of vaccum cleaner'''},
                     ],
                 ]
                 * 5		,)




       @unittest.skip('''No models are available in TF'''							)
       def    _UpperCamelCase					(						self							)      ->				List[str]:





             """simple docstring"""






             pass