





import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
      from huggingface_hub import snapshot_download
      from pyctcdecode import BeamSearchDecoderCTC

      from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
      from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
      from transformers import WavaVecaForCTC

@require_pyctcdecode
class 		UpperCamelCase__ (      unittest.TestCase   ):



     '''simple docstring'''

     def 				SCREAMING_SNAKE_CASE__  (							self						:    str       )     ->		Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						"""| <pad> <unk> <s> </s> a b c d e f g h i j k""".split()
           SCREAMING_SNAKE_CASE		=						dict(zip(lowerCamelCase__						,range(len(lowerCamelCase__       )       )       )       )

           SCREAMING_SNAKE_CASE		=						{
               """unk_token""": """<unk>""",
               """bos_token""": """<s>""",
               """eos_token""": """</s>""",
           }
           SCREAMING_SNAKE_CASE		=						{
               """feature_size""": 1,
               """padding_value""": 0.0,
               """sampling_rate""": 16000,
               """return_attention_mask""": False,
               """do_normalize""": True,
           }

           SCREAMING_SNAKE_CASE		=						tempfile.mkdtemp()
           SCREAMING_SNAKE_CASE		=						os.path.join(self.tmpdirname						,VOCAB_FILES_NAMES["""vocab_file"""]       )
           SCREAMING_SNAKE_CASE		=						os.path.join(self.tmpdirname						,lowerCamelCase__       )
           with open(self.vocab_file						,"""w"""						,encoding="""utf-8"""       ) as fp:
                 fp.write(json.dumps(lowerCamelCase__       ) + """\n"""       )

           with open(self.feature_extraction_file						,"""w"""						,encoding="""utf-8"""       ) as fp:
                 fp.write(json.dumps(lowerCamelCase__       ) + """\n"""       )

           # load decoder from hub
           SCREAMING_SNAKE_CASE		=						"""hf-internal-testing/ngram-beam-search-decoder"""






     def 				SCREAMING_SNAKE_CASE__  (							self						:    int						,**lowerCamelCase__						:    Tuple       )     ->		List[str]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.add_kwargs_tokens_map.copy()
           kwargs.update(lowerCamelCase__       )
           return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname						,**lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Dict						,**lowerCamelCase__						:    int       )     ->		Dict:




           '''simple docstring'''


           return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname						,**lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    List[str]						,**lowerCamelCase__						:    List[Any]       )     ->		Any:




           '''simple docstring'''


           return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name						,**lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		Any:




           '''simple docstring'''


           shutil.rmtree(self.tmpdirname       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           processor.save_pretrained(self.tmpdirname       )
           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname       )

           # tokenizer
           self.assertEqual(processor.tokenizer.get_vocab()						,tokenizer.get_vocab()       )
           self.assertIsInstance(processor.tokenizer						,lowerCamelCase__       )

           # feature extractor
           self.assertEqual(processor.feature_extractor.to_json_string()						,feature_extractor.to_json_string()       )
           self.assertIsInstance(processor.feature_extractor						,lowerCamelCase__       )

           # decoder
           self.assertEqual(processor.decoder._alphabet.labels						,decoder._alphabet.labels       )
           self.assertEqual(
               processor.decoder.model_container[decoder._model_key]._unigram_set						,decoder.model_container[decoder._model_key]._unigram_set						,)
           self.assertIsInstance(processor.decoder						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(
               tokenizer=self.get_tokenizer()						,feature_extractor=self.get_feature_extractor()						,decoder=self.get_decoder()       )
           processor.save_pretrained(self.tmpdirname       )

           # make sure that error is thrown when decoder alphabet doesn't match
           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM.from_pretrained(
               self.tmpdirname						,alpha=5.0						,beta=3.0						,score_boundary=-7.0						,unk_score_offset=3       )

           # decoder
           self.assertEqual(processor.language_model.alpha						,5.0       )
           self.assertEqual(processor.language_model.beta						,3.0       )
           self.assertEqual(processor.language_model.score_boundary						,-7.0       )
           self.assertEqual(processor.language_model.unk_score_offset						,3       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Dict:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           # add token to trigger raise
           tokenizer.add_tokens(["""xx"""]       )
           with self.assertRaisesRegex(lowerCamelCase__						,"""include"""       ):
                 WavaVecaProcessorWithLM(
                     tokenizer=lowerCamelCase__						,feature_extractor=self.get_feature_extractor()						,decoder=self.get_decoder()       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						floats_list((3, 1000)       )

           SCREAMING_SNAKE_CASE		=						feature_extractor(lowerCamelCase__						,return_tensors="""np"""       )
           SCREAMING_SNAKE_CASE		=						processor(lowerCamelCase__						,return_tensors="""np"""       )

           for key in input_feat_extract.keys():
                 self.assertAlmostEqual(input_feat_extract[key].sum()						,input_processor[key].sum()						,delta=1e-2       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Any:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						"""This is a test string"""

           SCREAMING_SNAKE_CASE		=						processor(text=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						tokenizer(lowerCamelCase__       )

           for key in encoded_tok.keys():
                 self.assertListEqual(encoded_tok[key]						,encoded_processor[key]       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    str						,lowerCamelCase__						:    Tuple=(2, 10, 16)						,lowerCamelCase__						:    List[Any]=77       )     ->		Union[str, Any]:




           '''simple docstring'''


           np.random.seed(lowerCamelCase__       )
           return np.random.rand(*lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						self._get_dummy_logits(shape=(10, 16)						,seed=13       )

           SCREAMING_SNAKE_CASE		=						processor.decode(lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						decoder.decode_beams(lowerCamelCase__       )[0]

           self.assertEqual(decoded_decoder[0]						,decoded_processor.text       )
           self.assertEqual("""</s> <s> </s>"""						,decoded_processor.text       )
           self.assertEqual(decoded_decoder[-2]						,decoded_processor.logit_score       )
           self.assertEqual(decoded_decoder[-1]						,decoded_processor.lm_score       )






     @parameterized.expand([[None], ["""fork"""], ["""spawn"""]]       )
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[int]						,lowerCamelCase__						:    Tuple       )     ->		int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						self._get_dummy_logits()

           # note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
           #       otherwise, the LM won't be available to the pool's sub-processes.
           # manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
           if pool_context is None:
                 SCREAMING_SNAKE_CASE		=						processor.batch_decode(lowerCamelCase__       )
           else:
                 with get_context(lowerCamelCase__       ).Pool() as pool:
                       SCREAMING_SNAKE_CASE		=						processor.batch_decode(lowerCamelCase__						,lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						list(lowerCamelCase__       )

           with get_context("""fork"""       ).Pool() as p:
                 SCREAMING_SNAKE_CASE		=						decoder.decode_beams_batch(lowerCamelCase__						,lowerCamelCase__       )

           SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE		=						[], [], []
           for beams in decoded_beams:
                 texts_decoder.append(beams[0][0]       )
                 logit_scores_decoder.append(beams[0][-2]       )
                 lm_scores_decoder.append(beams[0][-1]       )

           self.assertListEqual(lowerCamelCase__						,decoded_processor.text       )
           self.assertListEqual(["""<s> <s> </s>""", """<s> <s> <s>"""]						,decoded_processor.text       )
           self.assertListEqual(lowerCamelCase__						,decoded_processor.logit_score       )
           self.assertListEqual(lowerCamelCase__						,decoded_processor.lm_score       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Tuple       )     ->		Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						self._get_dummy_logits()

           SCREAMING_SNAKE_CASE		=						15
           SCREAMING_SNAKE_CASE		=						-20.0
           SCREAMING_SNAKE_CASE		=						-4.0

           SCREAMING_SNAKE_CASE		=						processor.batch_decode(
               lowerCamelCase__						,beam_width=lowerCamelCase__						,beam_prune_logp=lowerCamelCase__						,token_min_logp=lowerCamelCase__						,)
           SCREAMING_SNAKE_CASE		=						decoded_processor_out.text

           SCREAMING_SNAKE_CASE		=						list(lowerCamelCase__       )

           with get_context("""fork"""       ).Pool() as pool:
                 SCREAMING_SNAKE_CASE		=						decoder.decode_beams_batch(
                     lowerCamelCase__						,lowerCamelCase__						,beam_width=lowerCamelCase__						,beam_prune_logp=lowerCamelCase__						,token_min_logp=lowerCamelCase__						,)

           SCREAMING_SNAKE_CASE		=						[d[0][0] for d in decoded_decoder_out]
           SCREAMING_SNAKE_CASE		=						[d[0][2] for d in decoded_decoder_out]
           SCREAMING_SNAKE_CASE		=						[d[0][3] for d in decoded_decoder_out]

           self.assertListEqual(lowerCamelCase__						,lowerCamelCase__       )
           self.assertListEqual(["""</s> <s> <s>""", """<s> <s> <s>"""]						,lowerCamelCase__       )

           self.assertTrue(np.array_equal(lowerCamelCase__						,decoded_processor_out.logit_score       )       )
           self.assertTrue(np.allclose([-20.054, -18.447]						,lowerCamelCase__						,atol=1e-3       )       )

           self.assertTrue(np.array_equal(lowerCamelCase__						,decoded_processor_out.lm_score       )       )
           self.assertTrue(np.allclose([-15.554, -13.9474]						,lowerCamelCase__						,atol=1e-3       )       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Union[str, Any]       )     ->		Any:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						self._get_dummy_logits()

           SCREAMING_SNAKE_CASE		=						2.0
           SCREAMING_SNAKE_CASE		=						5.0
           SCREAMING_SNAKE_CASE		=						-20.0
           SCREAMING_SNAKE_CASE		=						True

           SCREAMING_SNAKE_CASE		=						processor.batch_decode(
               lowerCamelCase__						,alpha=lowerCamelCase__						,beta=lowerCamelCase__						,unk_score_offset=lowerCamelCase__						,lm_score_boundary=lowerCamelCase__						,)
           SCREAMING_SNAKE_CASE		=						decoded_processor_out.text

           SCREAMING_SNAKE_CASE		=						list(lowerCamelCase__       )
           decoder.reset_params(
               alpha=lowerCamelCase__						,beta=lowerCamelCase__						,unk_score_offset=lowerCamelCase__						,lm_score_boundary=lowerCamelCase__						,)

           with get_context("""fork"""       ).Pool() as pool:
                 SCREAMING_SNAKE_CASE		=						decoder.decode_beams_batch(
                     lowerCamelCase__						,lowerCamelCase__						,)

           SCREAMING_SNAKE_CASE		=						[d[0][0] for d in decoded_decoder_out]

           self.assertListEqual(lowerCamelCase__						,lowerCamelCase__       )
           self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]						,lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						processor.decoder.model_container[processor.decoder._model_key]
           self.assertEqual(lm_model.alpha						,2.0       )
           self.assertEqual(lm_model.beta						,5.0       )
           self.assertEqual(lm_model.unk_score_offset						,-20.0       )
           self.assertEqual(lm_model.score_boundary						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    str       )     ->		int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""       )

           SCREAMING_SNAKE_CASE		=						processor.decoder.model_container[processor.decoder._model_key]
           SCREAMING_SNAKE_CASE		=						Path(language_model._kenlm_model.path.decode("""utf-8"""       )       ).parent.parent.absolute()

           SCREAMING_SNAKE_CASE		=						os.listdir(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						["""alphabet.json""", """language_model"""]

           downloaded_decoder_files.sort()
           expected_decoder_files.sort()

           # test that only decoder relevant files from
           # https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
           # are downloaded and none of the rest (e.g. README.md, ...)
           self.assertListEqual(lowerCamelCase__						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						snapshot_download("""hf-internal-testing/processor_with_lm"""       )

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM.from_pretrained(lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						processor.decoder.model_container[processor.decoder._model_key]
           SCREAMING_SNAKE_CASE		=						Path(language_model._kenlm_model.path.decode("""utf-8"""       )       ).parent.parent.absolute()

           SCREAMING_SNAKE_CASE		=						os.listdir(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						os.listdir(lowerCamelCase__       )

           local_decoder_files.sort()
           expected_decoder_files.sort()

           # test that both decoder form hub and local files in cache are the same
           self.assertListEqual(lowerCamelCase__						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    int       )     ->		List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""       )
           SCREAMING_SNAKE_CASE		=						AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""       )

           SCREAMING_SNAKE_CASE		=						floats_list((3, 1000)       )

           SCREAMING_SNAKE_CASE		=						processor_wavaveca(lowerCamelCase__						,return_tensors="""np"""       )
           SCREAMING_SNAKE_CASE		=						processor_auto(lowerCamelCase__						,return_tensors="""np"""       )

           for key in input_wavaveca.keys():
                 self.assertAlmostEqual(input_wavaveca[key].sum()						,input_auto[key].sum()						,delta=1e-2       )

           SCREAMING_SNAKE_CASE		=						self._get_dummy_logits()

           SCREAMING_SNAKE_CASE		=						processor_wavaveca.batch_decode(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						processor_auto.batch_decode(lowerCamelCase__       )

           self.assertListEqual(decoded_wavaveca.text						,decoded_auto.text       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    List[str]       )     ->		List[str]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						self.get_feature_extractor()
           SCREAMING_SNAKE_CASE		=						self.get_tokenizer()
           SCREAMING_SNAKE_CASE		=						self.get_decoder()

           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM(tokenizer=lowerCamelCase__						,feature_extractor=lowerCamelCase__						,decoder=lowerCamelCase__       )

           self.assertListEqual(
               processor.model_input_names						,feature_extractor.model_input_names						,msg="""`processor` and `feature_extractor` model input names do not match"""						,)






     @staticmethod
     def 				SCREAMING_SNAKE_CASE__  (							lowerCamelCase__						:    Tuple						,lowerCamelCase__						:    Any       )     ->		List[str]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						[d[key] for d in offsets]
           return retrieved_list






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Union[str, Any]       )     ->		Dict:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""       )
           SCREAMING_SNAKE_CASE		=						self._get_dummy_logits()[0]

           SCREAMING_SNAKE_CASE		=						processor.decode(lowerCamelCase__						,output_word_offsets=lowerCamelCase__       )
           # check Wav2Vec2CTCTokenizerOutput keys for word
           self.assertEqual(len(outputs.keys()       )						,4       )
           self.assertTrue("""text""" in outputs       )
           self.assertTrue("""word_offsets""" in outputs       )
           self.assertTrue(isinstance(lowerCamelCase__						,lowerCamelCase__       )       )

           self.assertEqual(""" """.join(self.get_from_offsets(outputs["""word_offsets"""]						,"""word"""       )       )						,outputs.text       )
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]						,"""word"""       )						,["""<s>""", """<s>""", """</s>"""]       )
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]						,"""start_offset"""       )						,[0, 2, 4]       )
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]						,"""end_offset"""       )						,[1, 3, 5]       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Tuple       )     ->		Union[str, Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""       )
           SCREAMING_SNAKE_CASE		=						self._get_dummy_logits()

           SCREAMING_SNAKE_CASE		=						processor.batch_decode(lowerCamelCase__						,output_word_offsets=lowerCamelCase__       )

           # check Wav2Vec2CTCTokenizerOutput keys for word
           self.assertEqual(len(outputs.keys()       )						,4       )
           self.assertTrue("""text""" in outputs       )
           self.assertTrue("""word_offsets""" in outputs       )
           self.assertTrue(isinstance(lowerCamelCase__						,lowerCamelCase__       )       )

           self.assertListEqual(
               [""" """.join(self.get_from_offsets(lowerCamelCase__						,"""word"""       )       ) for o in outputs["""word_offsets"""]]						,outputs.text       )
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]						,"""word"""       )						,["""<s>""", """<s>""", """</s>"""]       )
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]						,"""start_offset"""       )						,[0, 2, 4]       )
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]						,"""end_offset"""       )						,[1, 3, 5]       )






     @slow
     @require_torch
     @require_torchaudio
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Tuple       )     ->		List[Any]:




           '''simple docstring'''


           import torch

           SCREAMING_SNAKE_CASE		=						load_dataset("""common_voice"""						,"""en"""						,split="""train"""						,streaming=lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						ds.cast_column("""audio"""						,datasets.Audio(sampling_rate=16000       )       )
           SCREAMING_SNAKE_CASE		=						iter(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						next(lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						AutoProcessor.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""       )
           SCREAMING_SNAKE_CASE		=						WavaVecaForCTC.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""       )

           # compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
           SCREAMING_SNAKE_CASE		=						processor(sample["""audio"""]["""array"""]						,return_tensors="""pt"""       ).input_values

           with torch.no_grad():
                 SCREAMING_SNAKE_CASE		=						model(lowerCamelCase__       ).logits.cpu().numpy()

           SCREAMING_SNAKE_CASE		=						processor.decode(logits[0]						,output_word_offsets=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
           SCREAMING_SNAKE_CASE		=						[
               {
                   """start_time""": d["""start_offset"""] * time_offset,
                   """end_time""": d["""end_offset"""] * time_offset,
                   """word""": d["""word"""],
               }
               for d in output["""word_offsets"""]
           ]

           SCREAMING_SNAKE_CASE		=						"""WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL"""

           # output words
           self.assertEqual(""" """.join(self.get_from_offsets(lowerCamelCase__						,"""word"""       )       )						,lowerCamelCase__       )
           self.assertEqual(""" """.join(self.get_from_offsets(lowerCamelCase__						,"""word"""       )       )						,output.text       )

           # output times
           SCREAMING_SNAKE_CASE		=						torch.tensor(self.get_from_offsets(lowerCamelCase__						,"""start_time"""       )       )
           SCREAMING_SNAKE_CASE		=						torch.tensor(self.get_from_offsets(lowerCamelCase__						,"""end_time"""       )       )

           # fmt: off
           SCREAMING_SNAKE_CASE		=						torch.tensor([1.4199, 1.6599, 2.2599, 3.0, 3.24, 3.5999, 3.7999, 4.0999, 4.26, 4.94, 5.28, 5.6599, 5.78, 5.94, 6.32, 6.5399, 6.6599]       )
           SCREAMING_SNAKE_CASE		=						torch.tensor([1.5399, 1.8999, 2.9, 3.16, 3.5399, 3.72, 4.0199, 4.1799, 4.76, 5.1599, 5.5599, 5.6999, 5.86, 6.1999, 6.38, 6.6199, 6.94]       )
           # fmt: on

           self.assertTrue(torch.allclose(lowerCamelCase__						,lowerCamelCase__						,atol=0.01       )       )
           self.assertTrue(torch.allclose(lowerCamelCase__						,lowerCamelCase__						,atol=0.01       )       )









import importlib
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
import transformers.models.auto
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.bert.configuration_bert import BertConfig
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / """utils"""))

from test_module.custom_configuration import CustomConfig  # noqa E402


SCREAMING_SNAKE_CASE_						   =       get_tests_dir("""fixtures/dummy-config.json""")

class 		UpperCamelCase__ (      unittest.TestCase   ):



     '''simple docstring'''

     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						0






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		str:




           '''simple docstring'''


           self.assertIsNotNone(transformers.models.auto.__spec__       )
           self.assertIsNotNone(importlib.util.find_spec("""transformers.models.auto"""       )       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Union[str, Any]       )     ->		List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""bert-base-uncased"""       )
           self.assertIsInstance(lowerCamelCase__						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    int       )     ->		Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained(lowerCamelCase__       )
           self.assertIsInstance(lowerCamelCase__						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Union[str, Any]       )     ->		List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained(lowerCamelCase__       )
           self.assertIsInstance(lowerCamelCase__						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Tuple       )     ->		Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						AutoConfig.for_model("""roberta"""       )
           self.assertIsInstance(lowerCamelCase__						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    int       )     ->		int:




           '''simple docstring'''


           with tempfile.TemporaryDirectory() as tmp_dir:
                 # This model name contains bert and roberta, but roberta ends up being picked.
                 SCREAMING_SNAKE_CASE		=						os.path.join(lowerCamelCase__						,"""fake-roberta"""       )
                 os.makedirs(lowerCamelCase__						,exist_ok=lowerCamelCase__       )
                 with open(os.path.join(lowerCamelCase__						,"""config.json"""       )						,"""w"""       ) as f:
                       f.write(json.dumps({}       )       )
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained(lowerCamelCase__       )
                 self.assertEqual(type(lowerCamelCase__       )						,lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Dict       )     ->		str:




           '''simple docstring'''


           try:
                 AutoConfig.register("""custom"""						,lowerCamelCase__       )
                 # Wrong model type will raise an error
                 with self.assertRaises(lowerCamelCase__       ):
                       AutoConfig.register("""model"""						,lowerCamelCase__       )
                 # Trying to register something existing in the Transformers library will raise an error
                 with self.assertRaises(lowerCamelCase__       ):
                       AutoConfig.register("""bert"""						,lowerCamelCase__       )

                 # Now that the config is registered, it can be used as any other config with the auto-API
                 SCREAMING_SNAKE_CASE		=						CustomConfig()
                 with tempfile.TemporaryDirectory() as tmp_dir:
                       config.save_pretrained(lowerCamelCase__       )
                       SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained(lowerCamelCase__       )
                       self.assertIsInstance(lowerCamelCase__						,lowerCamelCase__       )

           finally:
                 if "custom" in CONFIG_MAPPING._extra_content:
                       del CONFIG_MAPPING._extra_content["custom"]






     def 				SCREAMING_SNAKE_CASE__  (							self						:    str       )     ->		Dict:




           '''simple docstring'''


           with self.assertRaisesRegex(
               lowerCamelCase__						,"""bert-base is not a local folder and is not a valid model identifier"""       ):
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""bert-base"""       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Dict       )     ->		str:




           '''simple docstring'''


           with self.assertRaisesRegex(
               lowerCamelCase__						,R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""       ):
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained(lowerCamelCase__						,revision="""aaaaaa"""       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Tuple       )     ->		List[Any]:




           '''simple docstring'''


           with self.assertRaisesRegex(
               lowerCamelCase__						,"""hf-internal-testing/no-config-test-repo does not appear to have a file named config.json."""						,):
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""hf-internal-testing/no-config-test-repo"""       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[int]       )     ->		Union[str, Any]:




           '''simple docstring'''


           with self.assertRaises(lowerCamelCase__       ):
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""       )
           # If remote code is disabled, we can't load this config.
           with self.assertRaises(lowerCamelCase__       ):
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""						,trust_remote_code=lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""						,trust_remote_code=lowerCamelCase__       )
           self.assertEqual(config.__class__.__name__						,"""NewModelConfig"""       )

           # Test config can be reloaded.
           with tempfile.TemporaryDirectory() as tmp_dir:
                 config.save_pretrained(lowerCamelCase__       )
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained(lowerCamelCase__						,trust_remote_code=lowerCamelCase__       )
           self.assertEqual(reloaded_config.__class__.__name__						,"""NewModelConfig"""       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Dict       )     ->		Union[str, Any]:




           '''simple docstring'''


           class 		UpperCamelCase__ (      lowerCAmelCase_   ):



                '''simple docstring'''

                __snake_case    :	Union[str, Any]			=			"new-model"

           try:
                 AutoConfig.register("""new-model"""						,lowerCamelCase__       )
                 # If remote code is not set, the default is to use local
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""       )
                 self.assertEqual(config.__class__.__name__						,"""NewModelConfigLocal"""       )

                 # If remote code is disabled, we load the local one.
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""						,trust_remote_code=lowerCamelCase__       )
                 self.assertEqual(config.__class__.__name__						,"""NewModelConfigLocal"""       )

                 # If remote is enabled, we load from the Hub
                 SCREAMING_SNAKE_CASE		=						AutoConfig.from_pretrained("""hf-internal-testing/test_dynamic_model"""						,trust_remote_code=lowerCamelCase__       )
                 self.assertEqual(config.__class__.__name__						,"""NewModelConfig"""       )

           finally:
                 if "new-model" in CONFIG_MAPPING._extra_content:
                       del CONFIG_MAPPING._extra_content["new-model"]


