





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 			lowercase				(							unittest.TestCase     ):



						"""simple docstring"""

						_UpperCamelCase  :	int								=	MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
						_UpperCamelCase  :	int								=	TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING







						def 		__UpperCAmelCase       (    self			:		List[Any] ,				lowerCamelCase_			:		List[Any] ,				lowerCamelCase_			:		Optional[int] ,				lowerCamelCase_			:		Optional[int]					):





										'''simple docstring'''







										_snake_case :			Optional[Any]  						=   AudioClassificationPipeline(model=lowerCamelCase_ ,				feature_extractor=lowerCamelCase_					)

										# test with a raw waveform
										_snake_case :			List[Any]  						=   np.zeros((3_40_00,)					)
										_snake_case :			Optional[int]  						=   np.zeros((1_40_00,)					)
										return audio_classifier, [audioa, audio]







						def 		__UpperCAmelCase       (    self			:		List[str] ,				lowerCamelCase_			:		int ,				lowerCamelCase_			:		List[Any]					):





										'''simple docstring'''







										_snake_case						,			_snake_case :			Optional[int]  						=   examples
										_snake_case :			Union[str, Any]  						=   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 :			Dict  						=   audio_classifier(lowerCamelCase_ ,				top_k=1					)
										self.assertEqual(
										    lowerCamelCase_ ,				[
										        {'score': ANY(lowerCamelCase_					), 'label': ANY(lowerCamelCase_					)},
										    ] ,				)

										self.run_torchaudio(lowerCamelCase_					)







						@require_torchaudio
						def 		__UpperCAmelCase       (    self			:		List[str] ,				lowerCamelCase_			:		List[Any]					):





										'''simple docstring'''







										import datasets

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







						@require_torch
						def 		__UpperCAmelCase       (    self			:		Optional[Any]					):





										'''simple docstring'''







										_snake_case :			str  						=   'anton-l/wav2vec2-random-tiny-classifier'

										_snake_case :			List[str]  						=   pipeline('audio-classification' ,				model=lowerCamelCase_					)

										_snake_case :			Any  						=   np.ones((80_00,)					)
										_snake_case :			Any  						=   audio_classifier(lowerCamelCase_ ,				top_k=4					)

										_snake_case :			Optional[Any]  						=   [
										    {'score': 0.0842, 'label': 'no'},
										    {'score': 0.0838, 'label': 'up'},
										    {'score': 0.0837, 'label': 'go'},
										    {'score': 0.0834, 'label': 'right'},
										]
										_snake_case :			Any  						=   [
										    {'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]					)

										_snake_case :			Optional[Any]  						=   {'array': np.ones((80_00,)					), 'sampling_rate': audio_classifier.feature_extractor.sampling_rate}
										_snake_case :			str  						=   audio_classifier(lowerCamelCase_ ,				top_k=4					)
										self.assertIn(nested_simplify(lowerCamelCase_ ,				decimals=4					) ,				[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]					)







						@require_torch
						@slow
						def 		__UpperCAmelCase       (    self			:		Optional[int]					):





										'''simple docstring'''







										import datasets

										_snake_case :			List[str]  						=   'superb/wav2vec2-base-superb-ks'

										_snake_case :			Tuple  						=   pipeline('audio-classification' ,				model=lowerCamelCase_					)
										_snake_case :			Optional[int]  						=   datasets.load_dataset('anton-l/superb_dummy' ,				'ks' ,				split='test'					)

										_snake_case :			Union[str, Any]  						=   np.array(dataset[3]['speech'] ,				dtype=np.floataa					)
										_snake_case :			Any  						=   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 		__UpperCAmelCase       (    self			:		Any					):





										'''simple docstring'''







										pass













import re

from flax.core.frozen_dict import freeze
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.experimental import PartitionSpec as P


# Sentinels
lowercase_		:				Optional[int]				=						object()

# For specifying empty leaf dict `{}`
lowercase_		:				List[Any]				=						object()

def       A__(		__lowerCAmelCase	,			__lowerCAmelCase ):
				_snake_case :			Optional[Any]  						=   tuple((re.compile(x + '$' ) for x in qs) )
				for i in range(len(__lowerCAmelCase ) - len(__lowerCAmelCase ) + 1 ):
								_snake_case :			Tuple  						=   [x.match(__lowerCAmelCase ) for x, y in zip(__lowerCAmelCase	,			ks[i:] )]
								if matches and all(__lowerCAmelCase ):
												return True
				return False

def       A__(		__lowerCAmelCase ):
				def replace(__lowerCAmelCase	,			__lowerCAmelCase ):
								for rule, replacement in rules:
												if _match(__lowerCAmelCase	,			__lowerCAmelCase ):
																return replacement
								return val

				return replace

def       A__(		):
				return [
				    # embeddings
				    (("transformer", "wpe", "embedding"), P('mp'	,			__lowerCAmelCase )),
				    (("transformer", "wte", "embedding"), P('mp'	,			__lowerCAmelCase )),
				    # atention
				    (("attention", "(q_proj|k_proj|v_proj)", "kernel"), P(__lowerCAmelCase	,			'mp' )),
				    (("attention", "out_proj", "kernel"), P('mp'	,			__lowerCAmelCase )),
				    (("attention", "out_proj", "bias"), None),
				    # mlp
				    (("mlp", "c_fc", "kernel"), P(__lowerCAmelCase	,			'mp' )),
				    (("mlp", "c_fc", "bias"), P('mp' )),
				    (("mlp", "c_proj", "kernel"), P('mp'	,			__lowerCAmelCase )),
				    (("mlp", "c_proj", "bias"), None),
				    # layer norms
				    ((r"ln_\d+", "bias"), None),
				    ((r"\d+", r"ln_\d+", "scale"), None),
				    (("ln_f", "bias"), None),
				    (("ln_f", "scale"), None),
				]

def       A__(		__lowerCAmelCase ):
				_snake_case :			Optional[Any]  						=   _get_partition_rules()
				_snake_case :			Optional[Any]  						=   _replacement_rules(__lowerCAmelCase )
				_snake_case :			str  						=   {k: _unmatched for k in flatten_dict(__lowerCAmelCase )}
				_snake_case :			str  						=   {k: replace(__lowerCAmelCase	,			__lowerCAmelCase ) for k, v in initd.items()}
				assert _unmatched not in result.values(), "Incomplete partition spec."
				return freeze(unflatten_dict(__lowerCAmelCase ) )






