




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       _A				(		unittest.TestCase     ):
		__a					    =       MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
		__a					    =       TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			AudioClassificationPipeline(model=_SCREAMING_SNAKE_CASE						,		feature_extractor=_SCREAMING_SNAKE_CASE			)

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






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase    ,					_UpperCAmelCase      =			examples
								_UpperCAmelCase      =			audio_classifier(_SCREAMING_SNAKE_CASE			)
								# by default a model is initialized with num_labels=2
								self.assertEqual(
								    _SCREAMING_SNAKE_CASE						,		[
								        {"""score""": ANY(_SCREAMING_SNAKE_CASE			), """label""": ANY(_SCREAMING_SNAKE_CASE			)},
								        {"""score""": ANY(_SCREAMING_SNAKE_CASE			), """label""": ANY(_SCREAMING_SNAKE_CASE			)},
								    ]						,		)
								_UpperCAmelCase      =			audio_classifier(_SCREAMING_SNAKE_CASE						,		top_k=1			)
								self.assertEqual(
								    _SCREAMING_SNAKE_CASE						,		[
								        {"""score""": ANY(_SCREAMING_SNAKE_CASE			), """label""": ANY(_SCREAMING_SNAKE_CASE			)},
								    ]						,		)

								self.run_torchaudio(_SCREAMING_SNAKE_CASE			)






		@require_torchaudio
		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								import datasets

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






		@require_torch
		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			"""anton-l/wav2vec2-random-tiny-classifier"""

								_UpperCAmelCase      =			pipeline("""audio-classification"""						,		model=_SCREAMING_SNAKE_CASE			)

								_UpperCAmelCase      =			np.ones((8000,)			)
								_UpperCAmelCase      =			audio_classifier(_SCREAMING_SNAKE_CASE						,		top_k=4			)

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

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






		@require_torch
		@slow
		def 				UpperCAmelCase (    self			):
								import datasets

								_UpperCAmelCase      =			"""superb/wav2vec2-base-superb-ks"""

								_UpperCAmelCase      =			pipeline("""audio-classification"""						,		model=_SCREAMING_SNAKE_CASE			)
								_UpperCAmelCase      =			datasets.load_dataset("""anton-l/superb_dummy"""						,		"""ks"""						,		split="""test"""			)

								_UpperCAmelCase      =			np.array(dataset[3]["""speech"""]						,		dtype=np.floataa			)
								_UpperCAmelCase      =			audio_classifier(_SCREAMING_SNAKE_CASE						,		top_k=4			)
								self.assertEqual(
								    nested_simplify(_SCREAMING_SNAKE_CASE						,		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			):
								pass





from typing import Dict

from .base import GenericTensor, Pipeline




class       _A				(		__lowercase     ):






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		_SCREAMING_SNAKE_CASE=None						,		**_SCREAMING_SNAKE_CASE			):
								if tokenize_kwargs is None:
														_UpperCAmelCase      =			{}

								if truncation is not None:
														if "truncation" in tokenize_kwargs:
																				raise ValueError(
																				    """truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)"""			)
														_UpperCAmelCase      =			truncation

								_UpperCAmelCase      =			tokenize_kwargs

								_UpperCAmelCase      =			{}
								if return_tensors is not None:
														_UpperCAmelCase      =			return_tensors

								return preprocess_params, {}, postprocess_params






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			self.framework
								_UpperCAmelCase      =			self.tokenizer(_SCREAMING_SNAKE_CASE						,		return_tensors=_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			)
								return model_inputs






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE			):
								_UpperCAmelCase      =			self.model(**_SCREAMING_SNAKE_CASE			)
								return model_outputs






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=False			):
								# [0] is the first available tensor, logits or last_hidden_state.
								if return_tensors:
														return model_outputs[0]
								if self.framework == "pt":
														return model_outputs[0].tolist()
								elif self.framework == "tf":
														return model_outputs[0].numpy().tolist()






		def __call__(    self						,		*_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			):
								return super().__call__(*_SCREAMING_SNAKE_CASE						,		**_SCREAMING_SNAKE_CASE			)