



"""simple docstring"""






import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
   from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
   import torch_xla.core.xla_model as xm






def 			snake_case_    ( A_   :						List[Any]		):


       '''simple docstring'''





       if is_torch_version('''<''',							'''2.0.0'''		) or not hasattr(A_,							'''_dynamo'''		):
              return False
       return isinstance(A_,							torch._dynamo.eval_frame.OptimizedModule		)






def 			snake_case_    ( A_   :						List[str],							A_   :						bool = True		):


       '''simple docstring'''





       _lowerCamelCase		:							Tuple            =					(torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

       _lowerCamelCase		:							Any            =					is_compiled_module(A_		)
       if is_compiled:
              _lowerCamelCase		:							Dict            =					model
              _lowerCamelCase		:							Any            =					model._orig_mod

       if is_deepspeed_available():
              options += (DeepSpeedEngine,)

       while isinstance(A_,							A_		):
              _lowerCamelCase		:							List[Any]            =					model.module

       if not keep_fpaa_wrapper:
              _lowerCamelCase		:							int            =					getattr(A_,							'''forward'''		)
              _lowerCamelCase		:							int            =					model.__dict__.pop('''_original_forward''',							A_		)
              if original_forward is not None:
                     while hasattr(A_,							'''__wrapped__'''		):
                            _lowerCamelCase		:							Union[str, Any]            =					forward.__wrapped__
                            if forward == original_forward:
                                   break
                     _lowerCamelCase		:							int            =					forward
              if getattr(A_,							'''_converted_to_transformer_engine''',							A_		):
                     convert_model(A_,							to_transformer_engine=A_		)

       if is_compiled:
              _lowerCamelCase		:							Dict            =					model
              _lowerCamelCase		:							int            =					compiled_model

       return model






def 			snake_case_    ( ):


       '''simple docstring'''





       PartialState().wait_for_everyone()






def 			snake_case_    ( A_   :						Optional[int],							A_   :						str		):


       '''simple docstring'''





       if PartialState().distributed_type == DistributedType.TPU:
              xm.save(A_,							A_		)
       elif PartialState().local_process_index == 0:
              torch.save(A_,							A_		)






@contextmanager
def 			snake_case_    ( **A_   :						Any		):


       '''simple docstring'''





       for key, value in kwargs.items():
              _lowerCamelCase		:							Union[str, Any]            =					str(A_		)

       yield

       for key in kwargs:
              if key.upper() in os.environ:
                     del os.environ[key.upper()]






def 			snake_case_    ( A_   :						Any		):


       '''simple docstring'''





       if not hasattr(A_,							'''__qualname__'''		) and not hasattr(A_,							'''__name__'''		):
              _lowerCamelCase		:							Union[str, Any]            =					getattr(A_,							'''__class__''',							A_		)
       if hasattr(A_,							'''__qualname__'''		):
              return obj.__qualname__
       if hasattr(A_,							'''__name__'''		):
              return obj.__name__
       return str(A_		)






def 			snake_case_    ( A_   :						Any,							A_   :						Dict		):


       '''simple docstring'''





       for key, value in source.items():
              if isinstance(A_,							A_		):
                     _lowerCamelCase		:							Optional[Any]            =					destination.setdefault(A_,							{}		)
                     merge_dicts(A_,							A_		)
              else:
                     _lowerCamelCase		:							Any            =					value

       return destination







def 			snake_case_    ( A_   :						int = None		):


       '''simple docstring'''





       if port is None:
              _lowerCamelCase		:							Optional[int]            =					2_95_00
       with socket.socket(socket.AF_INET,							socket.SOCK_STREAM		) as s:
              return s.connect_ex(('''localhost''', port)		) == 0










import unittest

import numpy as np

from transformers import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING, TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
from transformers.pipelines import AudioClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torchaudio,
    slow,
)

from .test_pipelines_common import ANY
@is_pipeline_test
class  _UpperCAmelCase   (			unittest.TestCase      ):

				"""simple docstring"""




				lowercase__											=			MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
				lowercase__											=			TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING




				def 							lowercase__  (	self		:       Any,    lowerCamelCase		:       Optional[Any],    lowerCamelCase		:       Any,    lowerCamelCase		:       Tuple    ):

								'''simple docstring'''
								lowercase__  					=     AudioClassificationPipeline(model=lowerCamelCase,    feature_extractor=lowerCamelCase    )

								# test with a raw waveform
								lowercase__  					=     np.zeros((34_000,)    )
								lowercase__  					=     np.zeros((14_000,)    )
								return audio_classifier, [audioa, audio]




				def 							lowercase__  (	self		:       str,    lowerCamelCase		:       Dict,    lowerCamelCase		:       Union[str, Any]    ):

								'''simple docstring'''
								lowercase__					,		lowercase__  					=     examples
								lowercase__  					=     audio_classifier(lowerCamelCase    )
								# by default a model is initialized with num_labels=2
								self.assertEqual(
								    lowerCamelCase,    [
								        {'''score''': ANY(lowerCamelCase    ), '''label''': ANY(lowerCamelCase    )},
								        {'''score''': ANY(lowerCamelCase    ), '''label''': ANY(lowerCamelCase    )},
								    ],    )
								lowercase__  					=     audio_classifier(lowerCamelCase,    top_k=1    )
								self.assertEqual(
								    lowerCamelCase,    [
								        {'''score''': ANY(lowerCamelCase    ), '''label''': ANY(lowerCamelCase    )},
								    ],    )

								self.run_torchaudio(lowerCamelCase    )




				@require_torchaudio
				def 							lowercase__  (	self		:       Optional[int],    lowerCamelCase		:       List[Any]    ):

								'''simple docstring'''
								import datasets

								# test with a local file
								lowercase__  					=     datasets.load_dataset('''hf-internal-testing/librispeech_asr_dummy''',    '''clean''',    split='''validation'''    )
								lowercase__  					=     dataset[0]['''audio''']['''array''']
								lowercase__  					=     audio_classifier(lowerCamelCase    )
								self.assertEqual(
								    lowerCamelCase,    [
								        {'''score''': ANY(lowerCamelCase    ), '''label''': ANY(lowerCamelCase    )},
								        {'''score''': ANY(lowerCamelCase    ), '''label''': ANY(lowerCamelCase    )},
								    ],    )




				@require_torch
				def 							lowercase__  (	self		:       int    ):

								'''simple docstring'''
								lowercase__  					=     '''anton-l/wav2vec2-random-tiny-classifier'''

								lowercase__  					=     pipeline('''audio-classification''',    model=lowerCamelCase    )

								lowercase__  					=     np.ones((8_000,)    )
								lowercase__  					=     audio_classifier(lowerCamelCase,    top_k=4    )

								lowercase__  					=     [
								    {'''score''': 0.0842, '''label''': '''no'''},
								    {'''score''': 0.0838, '''label''': '''up'''},
								    {'''score''': 0.0837, '''label''': '''go'''},
								    {'''score''': 0.0834, '''label''': '''right'''},
								]
								lowercase__  					=     [
								    {'''score''': 0.0845, '''label''': '''stop'''},
								    {'''score''': 0.0844, '''label''': '''on'''},
								    {'''score''': 0.0841, '''label''': '''right'''},
								    {'''score''': 0.0834, '''label''': '''left'''},
								]
								self.assertIn(nested_simplify(lowerCamelCase,    decimals=4    ),    [EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]    )

								lowercase__  					=     {'''array''': np.ones((8_000,)    ), '''sampling_rate''': audio_classifier.feature_extractor.sampling_rate}
								lowercase__  					=     audio_classifier(lowerCamelCase,    top_k=4    )
								self.assertIn(nested_simplify(lowerCamelCase,    decimals=4    ),    [EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]    )




				@require_torch
				@slow
				def 							lowercase__  (	self		:       List[str]    ):

								'''simple docstring'''
								import datasets

								lowercase__  					=     '''superb/wav2vec2-base-superb-ks'''

								lowercase__  					=     pipeline('''audio-classification''',    model=lowerCamelCase    )
								lowercase__  					=     datasets.load_dataset('''anton-l/superb_dummy''',    '''ks''',    split='''test'''    )

								lowercase__  					=     np.array(dataset[3]['''speech'''],    dtype=np.floataa    )
								lowercase__  					=     audio_classifier(lowerCamelCase,    top_k=4    )
								self.assertEqual(
								    nested_simplify(lowerCamelCase,    decimals=3    ),    [
								        {'''score''': 0.981, '''label''': '''go'''},
								        {'''score''': 0.007, '''label''': '''up'''},
								        {'''score''': 0.006, '''label''': '''_unknown_'''},
								        {'''score''': 0.001, '''label''': '''down'''},
								    ],    )




				@require_tf
				@unittest.skip('''Audio classification is not implemented for TF'''    )
				def 							lowercase__  (	self		:       Optional[int]    ):

								'''simple docstring'''
								pass




