


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


     def        __snake_case(  self							,					A_							,					A_							,					A_    ):
       _UpperCAmelCase		:							Optional[Any]							=     AudioClassificationPipeline(model=__UpperCAmelCase							,					feature_extractor=__UpperCAmelCase    )

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


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

       self.run_torchaudio(__UpperCAmelCase    )


     @require_torchaudio
     def        __snake_case(  self							,					A_    ):
       import datasets

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


     @require_torch
     def        __snake_case(  self    ):
       _UpperCAmelCase		:							Union[str, Any]							=     "anton-l/wav2vec2-random-tiny-classifier"

       _UpperCAmelCase		:							Dict							=     pipeline("""audio-classification"""							,					model=__UpperCAmelCase    )

       _UpperCAmelCase		:							Any							=     np.ones((80_00,)    )
       _UpperCAmelCase		:							str							=     audio_classifier(__UpperCAmelCase							,					top_k=4    )

       _UpperCAmelCase		:							Union[str, Any]							=     [
           {"score": 0.0_8_4_2, "label": "no"},
           {"score": 0.0_8_3_8, "label": "up"},
           {"score": 0.0_8_3_7, "label": "go"},
           {"score": 0.0_8_3_4, "label": "right"},
       ]
       _UpperCAmelCase		:							List[str]							=     [
           {"score": 0.0_8_4_5, "label": "stop"},
           {"score": 0.0_8_4_4, "label": "on"},
           {"score": 0.0_8_4_1, "label": "right"},
           {"score": 0.0_8_3_4, "label": "left"},
       ]
       self.assertIn(nested_simplify(__UpperCAmelCase							,					decimals=4    )							,					[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]    )

       _UpperCAmelCase		:							Union[str, Any]							=     {"array": np.ones((80_00,)    ), "sampling_rate": audio_classifier.feature_extractor.sampling_rate}
       _UpperCAmelCase		:							Any							=     audio_classifier(__UpperCAmelCase							,					top_k=4    )
       self.assertIn(nested_simplify(__UpperCAmelCase							,					decimals=4    )							,					[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]    )


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

       _UpperCAmelCase		:							Union[str, Any]							=     "superb/wav2vec2-base-superb-ks"

       _UpperCAmelCase		:							Union[str, Any]							=     pipeline("""audio-classification"""							,					model=__UpperCAmelCase    )
       _UpperCAmelCase		:							str							=     datasets.load_dataset("""anton-l/superb_dummy"""							,					"""ks"""							,					split="""test"""    )

       _UpperCAmelCase		:							Optional[int]							=     np.array(dataset[3]["""speech"""]							,					dtype=np.floataa    )
       _UpperCAmelCase		:							Tuple							=     audio_classifier(__UpperCAmelCase							,					top_k=4    )
       self.assertEqual(
           nested_simplify(__UpperCAmelCase							,					decimals=3    )							,					[
               {"""score""": 0.9_8_1, """label""": """go"""},
               {"""score""": 0.0_0_7, """label""": """up"""},
               {"""score""": 0.0_0_6, """label""": """_unknown_"""},
               {"""score""": 0.0_0_1, """label""": """down"""},
           ]							,					)




     @require_tf
     @unittest.skip("""Audio classification is not implemented for TF"""    )
     def        __snake_case(  self    ):
       pass










# This script creates a super tiny model that is useful inside tests, when we just want to test that
# the machinery works, without needing to the check the quality of the outcomes.
#
# This version creates a tiny model through reduction of a normal pre-trained model, but keeping the
# full vocab, merges file, and thus also resulting in a larger model due to a large vocab size.
# This gives ~3MB in total for all files.
#
# If you want a 50 times smaller than this see `fsmt-make-super-tiny-model.py`, which is slightly more complicated
#
#
# It will be used then as "stas/tiny-wmt19-en-de"

# Build
from transformers import FSMTTokenizer, FSMTConfig, FSMTForConditionalGeneration

lowerCamelCase		       =      """facebook/wmt19-en-de"""
lowerCamelCase		       =      FSMTTokenizer.from_pretrained(mname)
# get the correct vocab sizes, etc. from the master model
lowerCamelCase		       =      FSMTConfig.from_pretrained(mname)
config.update(
    dict(
        d_model=4,
        encoder_layers=1,
        decoder_layers=1,
        encoder_ffn_dim=4,
        decoder_ffn_dim=4,
        encoder_attention_heads=1,
        decoder_attention_heads=1,
    )
)

lowerCamelCase		       =      FSMTForConditionalGeneration(config)
print(F'num of params {tiny_model.num_parameters()}')

# Test
lowerCamelCase		       =      tokenizer(["""Making tiny model"""], return_tensors="""pt""")
lowerCamelCase		       =      tiny_model(**batch)

print("""test output:""", len(outputs.logits[0]))

# Save
lowerCamelCase		       =      """tiny-wmt19-en-de"""
tiny_model.half()  # makes it smaller
tiny_model.save_pretrained(mname_tiny)
tokenizer.save_pretrained(mname_tiny)

print(F'Generated {mname_tiny}')

# Upload
# transformers-cli upload tiny-wmt19-en-de


