





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					):
					lowerCAmelCase_							=	MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
					lowerCAmelCase_							=	TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING




					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							Any:
										_snake_case									=				AudioClassificationPipeline(model=lowerCAmelCase_				,						feature_extractor=lowerCAmelCase_  )

										# test with a raw waveform
										_snake_case									=				np.zeros((3_4000,)  )
										_snake_case									=				np.zeros((1_4000,)  )
										return audio_classifier, [audioa, audio]




					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							List[str]:
										_snake_case							,     _snake_case									=				examples
										_snake_case									=				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_  )},
										    ]				,						)
										_snake_case									=				audio_classifier(lowerCAmelCase_				,						top_k=1  )
										self.assertEqual(
										    lowerCAmelCase_				,						[
										        {'score': ANY(lowerCAmelCase_  ), 'label': ANY(lowerCAmelCase_  )},
										    ]				,						)

										self.run_torchaudio(lowerCAmelCase_  )




					@require_torchaudio
					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							List[Any]:
										import datasets

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




					@require_torch
					def 	lowerCAmelCase			(		self  )		->							List[Any]:
										_snake_case									=				'anton-l/wav2vec2-random-tiny-classifier'

										_snake_case									=				pipeline('audio-classification'				,						model=lowerCAmelCase_  )

										_snake_case									=				np.ones((8000,)  )
										_snake_case									=				audio_classifier(lowerCAmelCase_				,						top_k=4  )

										_snake_case									=				[
										    {'score': 0.08_42, 'label': 'no'},
										    {'score': 0.08_38, 'label': 'up'},
										    {'score': 0.08_37, 'label': 'go'},
										    {'score': 0.08_34, 'label': 'right'},
										]
										_snake_case									=				[
										    {'score': 0.08_45, 'label': 'stop'},
										    {'score': 0.08_44, 'label': 'on'},
										    {'score': 0.08_41, 'label': 'right'},
										    {'score': 0.08_34, 'label': 'left'},
										]
										self.assertIn(nested_simplify(lowerCAmelCase_				,						decimals=4  )				,						[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]  )

										_snake_case									=				{'array': np.ones((8000,)  ), 'sampling_rate': audio_classifier.feature_extractor.sampling_rate}
										_snake_case									=				audio_classifier(lowerCAmelCase_				,						top_k=4  )
										self.assertIn(nested_simplify(lowerCAmelCase_				,						decimals=4  )				,						[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]  )




					@require_torch
					@slow
					def 	lowerCAmelCase			(		self  )		->							Dict:
										import datasets

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

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

										_snake_case									=				np.array(dataset[3]['speech']				,						dtype=np.floataa  )
										_snake_case									=				audio_classifier(lowerCAmelCase_				,						top_k=4  )
										self.assertEqual(
										    nested_simplify(lowerCAmelCase_				,						decimals=3  )				,						[
										        {'score': 0.9_81, 'label': 'go'},
										        {'score': 0.0_07, 'label': 'up'},
										        {'score': 0.0_06, 'label': '_unknown_'},
										        {'score': 0.0_01, 'label': 'down'},
										    ]				,						)






					@require_tf
					@unittest.skip('Audio classification is not implemented for TF'  )
					def 	lowerCAmelCase			(		self  )		->							List[str]:
										pass










from collections import deque

from .hash_table import HashTable
class 			UpperCamelCase_		(  _lowerCamelCase					):




					def __init__(		self				,						*lowerCAmelCase_				,						**lowerCAmelCase_  )		->							Any:
										super().__init__(*lowerCAmelCase_				,						**lowerCAmelCase_  )




					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							Optional[int]:
										_snake_case									=				deque([]  ) if self.values[key] is None else self.values[key]
										self.values[key].appendleft(lowerCAmelCase_  )
										_snake_case									=				self.values[key]




					def 	lowerCAmelCase			(		self  )		->							Optional[int]:
										return (
										    sum(self.charge_factor - len(lowerCAmelCase_  ) for slot in self.values  )
										    / self.size_table
										    * self.charge_factor
										)






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_=None  )		->							str:
										if not (
										    len(self.values[key]  ) == self.charge_factor and self.values.count(lowerCAmelCase_  ) == 0
										):
															return key
										return super()._collision_resolution(lowerCAmelCase_				,						lowerCAmelCase_  )



