






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'''




      A     :   Any						=					MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
      A     :   List[Any]						=					TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING






      def       _lowerCAmelCase       (	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				) ->				Union[str, Any]:
        snake_case_				:							Dict						 =       AudioClassificationPipeline(model=lowercase_		,				feature_extractor=lowercase_				)

        # test with a raw waveform
        snake_case_				:							int						 =       np.zeros((3_4000,)				)
        snake_case_				:							Union[str, Any]						 =       np.zeros((1_4000,)				)
        return audio_classifier, [audioa, audio]






      def       _lowerCAmelCase       (	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				) ->				int:
        snake_case_				:							Tuple						 =       examples
        snake_case_				:							List[str]						 =       audio_classifier(lowercase_				)
        # by default a model is initialized with num_labels=2
        self.assertEqual(
            lowercase_		,				[
                {"score": ANY(lowercase_				), "label": ANY(lowercase_				)},
                {"score": ANY(lowercase_				), "label": ANY(lowercase_				)},
            ]		,				)
        snake_case_				:							Tuple						 =       audio_classifier(lowercase_		,				top_k=1				)
        self.assertEqual(
            lowercase_		,				[
                {"score": ANY(lowercase_				), "label": ANY(lowercase_				)},
            ]		,				)

        self.run_torchaudio(lowercase_				)






      @require_torchaudio
      def       _lowerCAmelCase       (	self		,				_SCREAMING_SNAKE_CASE				) ->				List[str]:
        import datasets

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






      @require_torch
      def       _lowerCAmelCase       (	self				) ->				Optional[Any]:
        snake_case_				:							Tuple						 =       "anton-l/wav2vec2-random-tiny-classifier"

        snake_case_				:							str						 =       pipeline("audio-classification"		,				model=lowercase_				)

        snake_case_				:							Any						 =       np.ones((8000,)				)
        snake_case_				:							Dict						 =       audio_classifier(lowercase_		,				top_k=4				)

        snake_case_				:							str						 =       [
            {"score": 0.0842, "label": "no"},
            {"score": 0.0838, "label": "up"},
            {"score": 0.0837, "label": "go"},
            {"score": 0.0834, "label": "right"},
        ]
        snake_case_				:							Optional[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(lowercase_		,				decimals=4				)		,				[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]				)

        snake_case_				:							List[Any]						 =       {"array": np.ones((8000,)				), "sampling_rate": audio_classifier.feature_extractor.sampling_rate}
        snake_case_				:							Optional[Any]						 =       audio_classifier(lowercase_		,				top_k=4				)
        self.assertIn(nested_simplify(lowercase_		,				decimals=4				)		,				[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]				)






      @require_torch
      @slow
      def       _lowerCAmelCase       (	self				) ->				List[Any]:
        import datasets

        snake_case_				:							Optional[Any]						 =       "superb/wav2vec2-base-superb-ks"

        snake_case_				:							Tuple						 =       pipeline("audio-classification"		,				model=lowercase_				)
        snake_case_				:							List[str]						 =       datasets.load_dataset("anton-l/superb_dummy"		,				"ks"		,				split="test"				)

        snake_case_				:							List[Any]						 =       np.array(dataset[3]["speech"]		,				dtype=np.floataa				)
        snake_case_				:							str						 =       audio_classifier(lowercase_		,				top_k=4				)
        self.assertEqual(
            nested_simplify(lowercase_		,				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       _lowerCAmelCase       (	self				) ->				Optional[int]:
        pass








import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType




class 		UpperCAmelCase_						(	SCREAMING_SNAKE_CASE__					):


						'''simple docstring'''




						A     :   Tuple						=					['image_processor', 'tokenizer']
						A     :   List[Any]						=					'ViltImageProcessor'
						A     :   Optional[Any]						=					('BertTokenizer', 'BertTokenizerFast')






						def __init__(	self		,				_SCREAMING_SNAKE_CASE=None		,				_SCREAMING_SNAKE_CASE=None		,				**_SCREAMING_SNAKE_CASE				) ->				int:
								snake_case_				:							str						 =       None
								if "feature_extractor" in kwargs:
										warnings.warn(
										    "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
										    " instead."		,				_SCREAMING_SNAKE_CASE		,				)
										snake_case_				:							str						 =       kwargs.pop("feature_extractor"				)

								snake_case_				:							Tuple						 =       image_processor if image_processor is not None else feature_extractor
								if image_processor is None:
										raise ValueError("You need to specify an `image_processor`."				)
								if tokenizer is None:
										raise ValueError("You need to specify a `tokenizer`."				)

								super().__init__(_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				)
								snake_case_				:							str						 =       self.image_processor






						def __call__(	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = True		,				_SCREAMING_SNAKE_CASE = False		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = 0		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = None		,				_SCREAMING_SNAKE_CASE = False		,				_SCREAMING_SNAKE_CASE = False		,				_SCREAMING_SNAKE_CASE = False		,				_SCREAMING_SNAKE_CASE = False		,				_SCREAMING_SNAKE_CASE = True		,				_SCREAMING_SNAKE_CASE = None		,				**_SCREAMING_SNAKE_CASE		,				) ->				BatchEncoding:
								snake_case_				:							List[Any]						 =       self.tokenizer(
								    text=_SCREAMING_SNAKE_CASE		,				add_special_tokens=_SCREAMING_SNAKE_CASE		,				padding=_SCREAMING_SNAKE_CASE		,				truncation=_SCREAMING_SNAKE_CASE		,				max_length=_SCREAMING_SNAKE_CASE		,				stride=_SCREAMING_SNAKE_CASE		,				pad_to_multiple_of=_SCREAMING_SNAKE_CASE		,				return_token_type_ids=_SCREAMING_SNAKE_CASE		,				return_attention_mask=_SCREAMING_SNAKE_CASE		,				return_overflowing_tokens=_SCREAMING_SNAKE_CASE		,				return_special_tokens_mask=_SCREAMING_SNAKE_CASE		,				return_offsets_mapping=_SCREAMING_SNAKE_CASE		,				return_length=_SCREAMING_SNAKE_CASE		,				verbose=_SCREAMING_SNAKE_CASE		,				return_tensors=_SCREAMING_SNAKE_CASE		,				**_SCREAMING_SNAKE_CASE		,				)
								# add pixel_values + pixel_mask
								snake_case_				:							str						 =       self.image_processor(_SCREAMING_SNAKE_CASE		,				return_tensors=_SCREAMING_SNAKE_CASE				)
								encoding.update(_SCREAMING_SNAKE_CASE				)

								return encoding






						def       _lowerCAmelCase       (	self		,				*_SCREAMING_SNAKE_CASE		,				**_SCREAMING_SNAKE_CASE				) ->				List[str]:
								return self.tokenizer.batch_decode(*_SCREAMING_SNAKE_CASE		,				**_SCREAMING_SNAKE_CASE				)






						def       _lowerCAmelCase       (	self		,				*_SCREAMING_SNAKE_CASE		,				**_SCREAMING_SNAKE_CASE				) ->				Dict:
								return self.tokenizer.decode(*_SCREAMING_SNAKE_CASE		,				**_SCREAMING_SNAKE_CASE				)






						@property
						def       _lowerCAmelCase       (	self				) ->				Tuple:
								snake_case_				:							List[Any]						 =       self.tokenizer.model_input_names
								snake_case_				:							Tuple						 =       self.image_processor.model_input_names
								return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names				)				)






						@property
						def       _lowerCAmelCase       (	self				) ->				Optional[Any]:
								warnings.warn(
								    "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."		,				_SCREAMING_SNAKE_CASE		,				)
								return self.image_processor_class






						@property
						def       _lowerCAmelCase       (	self				) ->				Optional[int]:
								warnings.warn(
								    "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."		,				_SCREAMING_SNAKE_CASE		,				)
								return self.image_processor
