






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 			SCREAMING_SNAKE_CASE      (      unittest.TestCase       ):
     '''simple docstring'''


     def 					_A							(					self							:						str   ):
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     "| <pad> <unk> <s> </s> a b c d e f g h i j k".split()
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     dict(zip(UpperCAmelCase_       ,			range(len(UpperCAmelCase_   )   )   )   )

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

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

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

       # load decoder from hub
       SCREAMING_SNAKE_CASE							:   List[str]						    =     "hf-internal-testing/ngram-beam-search-decoder"


     def 					_A							(					self							:						int       ,			**UpperCAmelCase_							:						List[str]   ):
       SCREAMING_SNAKE_CASE							:   int						    =     self.add_kwargs_tokens_map.copy()
       kwargs.update(UpperCAmelCase_   )
       return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname       ,			**UpperCAmelCase_   )


     def 					_A							(					self							:						str       ,			**UpperCAmelCase_							:						List[str]   ):
       return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname       ,			**UpperCAmelCase_   )


     def 					_A							(					self							:						Optional[Any]       ,			**UpperCAmelCase_							:						Optional[int]   ):
       return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name       ,			**UpperCAmelCase_   )


     def 					_A							(					self							:						List[Any]   ):
       shutil.rmtree(self.tmpdirname   )


     def 					_A							(					self							:						List[str]   ):
       SCREAMING_SNAKE_CASE							:   int						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   Any						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   List[str]						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   List[Any]						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

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

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

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

       # 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       ,			UpperCAmelCase_   )


     def 					_A							(					self							:						Optional[Any]   ):
       SCREAMING_SNAKE_CASE							:   List[str]						    =     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							:   List[str]						    =     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 					_A							(					self							:						Dict   ):
       SCREAMING_SNAKE_CASE							:   int						    =     self.get_tokenizer()
       # add token to trigger raise
       tokenizer.add_tokens(["xx"]   )
       with self.assertRaisesRegex(UpperCAmelCase_       ,			"include"   ):
         WavaVecaProcessorWithLM(
             tokenizer=UpperCAmelCase_       ,			feature_extractor=self.get_feature_extractor()       ,			decoder=self.get_decoder()   )


     def 					_A							(					self							:						Optional[Any]   ):
       SCREAMING_SNAKE_CASE							:   Tuple						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   List[str]						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   List[Any]						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   List[Any]						    =     floats_list((3, 1000)   )

       SCREAMING_SNAKE_CASE							:   Tuple						    =     feature_extractor(UpperCAmelCase_       ,			return_tensors="np"   )
       SCREAMING_SNAKE_CASE							:   Any						    =     processor(UpperCAmelCase_       ,			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 					_A							(					self							:						List[str]   ):
       SCREAMING_SNAKE_CASE							:   List[Any]						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   str						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   List[Any]						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   int						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Any						    =     "This is a test string"

       SCREAMING_SNAKE_CASE							:   List[str]						    =     processor(text=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     tokenizer(UpperCAmelCase_   )

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


     def 					_A							(					self							:						str       ,			UpperCAmelCase_							:						List[Any]=(2, 10, 16)       ,			UpperCAmelCase_							:						Optional[Any]=77   ):
       np.random.seed(UpperCAmelCase_   )
       return np.random.rand(*UpperCAmelCase_   )


     def 					_A							(					self							:						Optional[int]   ):
       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   List[Any]						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   Tuple						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     self._get_dummy_logits(shape=(10, 16)       ,			seed=13   )

       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     processor.decode(UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Dict						    =     decoder.decode_beams(UpperCAmelCase_   )[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 					_A							(					self							:						str       ,			UpperCAmelCase_							:						Tuple   ):
       SCREAMING_SNAKE_CASE							:   List[str]						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   List[str]						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   Tuple						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   List[str]						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Any						    =     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							:   Optional[int]						    =     processor.batch_decode(UpperCAmelCase_   )
       else:
         with get_context(UpperCAmelCase_   ).Pool() as pool:
           SCREAMING_SNAKE_CASE							:   List[Any]						    =     processor.batch_decode(UpperCAmelCase_       ,			UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Dict						    =     list(UpperCAmelCase_   )

       with get_context("fork"   ).Pool() as p:
         SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     decoder.decode_beams_batch(UpperCAmelCase_       ,			UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE    ,	SCREAMING_SNAKE_CASE    ,	SCREAMING_SNAKE_CASE							:   Any						    =     [], [], []
       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(UpperCAmelCase_       ,			decoded_processor.text   )
       self.assertListEqual(["<s> <s> </s>", "<s> <s> <s>"]       ,			decoded_processor.text   )
       self.assertListEqual(UpperCAmelCase_       ,			decoded_processor.logit_score   )
       self.assertListEqual(UpperCAmelCase_       ,			decoded_processor.lm_score   )


     def 					_A							(					self							:						Optional[int]   ):
       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   Tuple						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   Any						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   List[Any]						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   int						    =     self._get_dummy_logits()

       SCREAMING_SNAKE_CASE							:   Union[str, Any]						    =     15
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     -20.0
       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     -4.0

       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     processor.batch_decode(
           UpperCAmelCase_       ,			beam_width=UpperCAmelCase_       ,			beam_prune_logp=UpperCAmelCase_       ,			token_min_logp=UpperCAmelCase_       ,			)
       SCREAMING_SNAKE_CASE							:   Union[str, Any]						    =     decoded_processor_out.text

       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     list(UpperCAmelCase_   )

       with get_context("fork"   ).Pool() as pool:
         SCREAMING_SNAKE_CASE							:   List[Any]						    =     decoder.decode_beams_batch(
             UpperCAmelCase_       ,			UpperCAmelCase_       ,			beam_width=UpperCAmelCase_       ,			beam_prune_logp=UpperCAmelCase_       ,			token_min_logp=UpperCAmelCase_       ,			)

       SCREAMING_SNAKE_CASE							:   Tuple						    =     [d[0][0] for d in decoded_decoder_out]
       SCREAMING_SNAKE_CASE							:   Union[str, Any]						    =     [d[0][2] for d in decoded_decoder_out]
       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     [d[0][3] for d in decoded_decoder_out]

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

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

       self.assertTrue(np.array_equal(UpperCAmelCase_       ,			decoded_processor_out.lm_score   )   )
       self.assertTrue(np.allclose([-15.554, -13.9_474]       ,			UpperCAmelCase_       ,			atol=1E-3   )   )


     def 					_A							(					self							:						Dict   ):
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   Any						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   Union[str, Any]						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   int						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   List[Any]						    =     self._get_dummy_logits()

       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     2.0
       SCREAMING_SNAKE_CASE							:   Tuple						    =     5.0
       SCREAMING_SNAKE_CASE							:   Dict						    =     -20.0
       SCREAMING_SNAKE_CASE							:   Tuple						    =     True

       SCREAMING_SNAKE_CASE							:   str						    =     processor.batch_decode(
           UpperCAmelCase_       ,			alpha=UpperCAmelCase_       ,			beta=UpperCAmelCase_       ,			unk_score_offset=UpperCAmelCase_       ,			lm_score_boundary=UpperCAmelCase_       ,			)
       SCREAMING_SNAKE_CASE							:   List[Any]						    =     decoded_processor_out.text

       SCREAMING_SNAKE_CASE							:   Any						    =     list(UpperCAmelCase_   )
       decoder.reset_params(
           alpha=UpperCAmelCase_       ,			beta=UpperCAmelCase_       ,			unk_score_offset=UpperCAmelCase_       ,			lm_score_boundary=UpperCAmelCase_       ,			)

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

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

       self.assertListEqual(UpperCAmelCase_       ,			UpperCAmelCase_   )
       self.assertListEqual(["<s> </s> <s> </s> </s>", "</s> </s> <s> </s> </s>"]       ,			UpperCAmelCase_   )
       SCREAMING_SNAKE_CASE							:   Dict						    =     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       ,			UpperCAmelCase_   )


     def 					_A							(					self							:						Tuple   ):
       SCREAMING_SNAKE_CASE							:   int						    =     WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )

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

       SCREAMING_SNAKE_CASE							:   Tuple						    =     os.listdir(UpperCAmelCase_   )
       SCREAMING_SNAKE_CASE							:   Any						    =     ["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(UpperCAmelCase_       ,			UpperCAmelCase_   )


     def 					_A							(					self							:						List[str]   ):
       SCREAMING_SNAKE_CASE							:   List[str]						    =     snapshot_download("hf-internal-testing/processor_with_lm"   )

       SCREAMING_SNAKE_CASE							:   str						    =     WavaVecaProcessorWithLM.from_pretrained(UpperCAmelCase_   )

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

       SCREAMING_SNAKE_CASE							:   List[Any]						    =     os.listdir(UpperCAmelCase_   )
       SCREAMING_SNAKE_CASE							:   Any						    =     os.listdir(UpperCAmelCase_   )

       local_decoder_files.sort()
       expected_decoder_files.sort()

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


     def 					_A							(					self							:						Tuple   ):
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )
       SCREAMING_SNAKE_CASE							:   str						    =     AutoProcessor.from_pretrained("hf-internal-testing/processor_with_lm"   )

       SCREAMING_SNAKE_CASE							:   Tuple						    =     floats_list((3, 1000)   )

       SCREAMING_SNAKE_CASE							:   str						    =     processor_wavaveca(UpperCAmelCase_       ,			return_tensors="np"   )
       SCREAMING_SNAKE_CASE							:   Union[str, Any]						    =     processor_auto(UpperCAmelCase_       ,			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							:   Dict						    =     self._get_dummy_logits()

       SCREAMING_SNAKE_CASE							:   List[Any]						    =     processor_wavaveca.batch_decode(UpperCAmelCase_   )
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     processor_auto.batch_decode(UpperCAmelCase_   )

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


     def 					_A							(					self							:						Tuple   ):
       SCREAMING_SNAKE_CASE							:   List[Any]						    =     self.get_feature_extractor()
       SCREAMING_SNAKE_CASE							:   Any						    =     self.get_tokenizer()
       SCREAMING_SNAKE_CASE							:   Dict						    =     self.get_decoder()

       SCREAMING_SNAKE_CASE							:   int						    =     WavaVecaProcessorWithLM(tokenizer=UpperCAmelCase_       ,			feature_extractor=UpperCAmelCase_       ,			decoder=UpperCAmelCase_   )

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


     @staticmethod
     def 					_A							(					UpperCAmelCase_							:						int       ,			UpperCAmelCase_							:						Optional[Any]   ):
       SCREAMING_SNAKE_CASE							:   str						    =     [d[key] for d in offsets]
       return retrieved_list


     def 					_A							(					self							:						Union[str, Any]   ):
       SCREAMING_SNAKE_CASE							:   Tuple						    =     WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )
       SCREAMING_SNAKE_CASE							:   Dict						    =     self._get_dummy_logits()[0]

       SCREAMING_SNAKE_CASE							:   Dict						    =     processor.decode(UpperCAmelCase_       ,			output_word_offsets=UpperCAmelCase_   )
       # 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(UpperCAmelCase_       ,			UpperCAmelCase_   )   )

       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 					_A							(					self							:						str   ):
       SCREAMING_SNAKE_CASE							:   int						    =     WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     self._get_dummy_logits()

       SCREAMING_SNAKE_CASE							:   List[Any]						    =     processor.batch_decode(UpperCAmelCase_       ,			output_word_offsets=UpperCAmelCase_   )

       # 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(UpperCAmelCase_       ,			UpperCAmelCase_   )   )

       self.assertListEqual(
           [" ".join(self.get_from_offsets(UpperCAmelCase_       ,			"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 					_A							(					self							:						List[str]   ):
       import torch

       SCREAMING_SNAKE_CASE							:   Any						    =     load_dataset("common_voice"       ,			"en"       ,			split="train"       ,			streaming=UpperCAmelCase_   )
       SCREAMING_SNAKE_CASE							:   Union[str, Any]						    =     ds.cast_column("audio"       ,			datasets.Audio(sampling_rate=1_6000   )   )
       SCREAMING_SNAKE_CASE							:   Dict						    =     iter(UpperCAmelCase_   )
       SCREAMING_SNAKE_CASE							:   Optional[int]						    =     next(UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     AutoProcessor.from_pretrained("patrickvonplaten/wav2vec2-base-100h-with-lm"   )
       SCREAMING_SNAKE_CASE							:   Union[str, Any]						    =     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							:   Union[str, Any]						    =     processor(sample["audio"]["array"]       ,			return_tensors="pt"   ).input_values

       with torch.no_grad():
         SCREAMING_SNAKE_CASE							:   List[str]						    =     model(UpperCAmelCase_   ).logits.cpu().numpy()

       SCREAMING_SNAKE_CASE							:   List[str]						    =     processor.decode(logits[0]       ,			output_word_offsets=UpperCAmelCase_   )

       SCREAMING_SNAKE_CASE							:   Optional[Any]						    =     model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
       SCREAMING_SNAKE_CASE							:   List[Any]						    =     [
           {
               "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							:   List[Any]						    =     "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(UpperCAmelCase_       ,			"word"   )   )       ,			UpperCAmelCase_   )
       self.assertEqual(" ".join(self.get_from_offsets(UpperCAmelCase_       ,			"word"   )   )       ,			output.text   )

       # output times
       SCREAMING_SNAKE_CASE							:   Tuple						    =     torch.tensor(self.get_from_offsets(UpperCAmelCase_       ,			"start_time"   )   )
       SCREAMING_SNAKE_CASE							:   Dict						    =     torch.tensor(self.get_from_offsets(UpperCAmelCase_       ,			"end_time"   )   )

       # fmt: off
       SCREAMING_SNAKE_CASE							:   List[str]						    =     torch.tensor([1.4_199, 1.6_599, 2.2_599, 3.0, 3.24, 3.5_999, 3.7_999, 4.0_999, 4.26, 4.94, 5.28, 5.6_599, 5.78, 5.94, 6.32, 6.5_399, 6.6_599]   )
       SCREAMING_SNAKE_CASE							:   List[str]						    =     torch.tensor([1.5_399, 1.8_999, 2.9, 3.16, 3.5_399, 3.72, 4.0_199, 4.1_799, 4.76, 5.1_599, 5.5_599, 5.6_999, 5.86, 6.1_999, 6.38, 6.6_199, 6.94]   )
       # fmt: on

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










import tempfile

import torch

from diffusers import IPNDMScheduler

from .test_schedulers import SchedulerCommonTest






class    _UpperCamelCase     (       __snake_case					):

				"""simple docstring"""







				lowerCAmelCase					=			(IPNDMScheduler,)
				lowerCAmelCase					=			(('num_inference_steps', 5_0),)

				def   _UpperCAmelCase       (    self					, **a__ )						->					List[str]:
					A             =					{"""num_train_timesteps""": 1000}
					config.update(**a__ )
					return config

				def   _UpperCAmelCase       (    self					, a__=0					, **a__ )						->					Tuple:
					A             =					dict(self.forward_default_kwargs )
					A             =					kwargs.pop("""num_inference_steps"""					, a__ )
					A             =					self.dummy_sample
					A             =					0.1 * sample
					A             =					[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

					for scheduler_class in self.scheduler_classes:
						A             =					self.get_scheduler_config(**a__ )
						A             =					scheduler_class(**a__ )
						scheduler.set_timesteps(a__ )
						# copy over dummy past residuals
						A             =					dummy_past_residuals[:]

						if time_step is None:
							A             =					scheduler.timesteps[len(scheduler.timesteps ) // 2]

						with tempfile.TemporaryDirectory() as tmpdirname:
							scheduler.save_config(a__ )
							A             =					scheduler_class.from_pretrained(a__ )
							new_scheduler.set_timesteps(a__ )
							# copy over dummy past residuals
							A             =					dummy_past_residuals[:]

						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample
						A             =					new_scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample

						assert torch.sum(torch.abs(output - new_output ) ) < 1e-5, "Scheduler outputs are not identical"

						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample
						A             =					new_scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample

						assert torch.sum(torch.abs(output - new_output ) ) < 1e-5, "Scheduler outputs are not identical"

				def   _UpperCAmelCase       (    self )						->					Any:
					pass

				def   _UpperCAmelCase       (    self					, a__=0					, **a__ )						->					int:
					A             =					dict(self.forward_default_kwargs )
					A             =					kwargs.pop("""num_inference_steps"""					, a__ )
					A             =					self.dummy_sample
					A             =					0.1 * sample
					A             =					[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

					for scheduler_class in self.scheduler_classes:
						A             =					self.get_scheduler_config()
						A             =					scheduler_class(**a__ )
						scheduler.set_timesteps(a__ )

						# copy over dummy past residuals (must be after setting timesteps)
						A             =					dummy_past_residuals[:]

						if time_step is None:
							A             =					scheduler.timesteps[len(scheduler.timesteps ) // 2]

						with tempfile.TemporaryDirectory() as tmpdirname:
							scheduler.save_config(a__ )
							A             =					scheduler_class.from_pretrained(a__ )
							# copy over dummy past residuals
							new_scheduler.set_timesteps(a__ )

							# copy over dummy past residual (must be after setting timesteps)
							A             =					dummy_past_residuals[:]

						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample
						A             =					new_scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample

						assert torch.sum(torch.abs(output - new_output ) ) < 1e-5, "Scheduler outputs are not identical"

						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample
						A             =					new_scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample

						assert torch.sum(torch.abs(output - new_output ) ) < 1e-5, "Scheduler outputs are not identical"

				def   _UpperCAmelCase       (    self					, **a__ )						->					List[Any]:
					A             =					self.scheduler_classes[0]
					A             =					self.get_scheduler_config(**a__ )
					A             =					scheduler_class(**a__ )

					A             =					10
					A             =					self.dummy_model()
					A             =					self.dummy_sample_deter
					scheduler.set_timesteps(a__ )

					for i, t in enumerate(scheduler.timesteps ):
						A             =					model(a__					, a__ )
						A             =					scheduler.step(a__					, a__					, a__ ).prev_sample

					for i, t in enumerate(scheduler.timesteps ):
						A             =					model(a__					, a__ )
						A             =					scheduler.step(a__					, a__					, a__ ).prev_sample

					return sample

				def   _UpperCAmelCase       (    self )						->					Any:
					A             =					dict(self.forward_default_kwargs )

					A             =					kwargs.pop("""num_inference_steps"""					, a__ )

					for scheduler_class in self.scheduler_classes:
						A             =					self.get_scheduler_config()
						A             =					scheduler_class(**a__ )

						A             =					self.dummy_sample
						A             =					0.1 * sample

						if num_inference_steps is not None and hasattr(a__					, """set_timesteps""" ):
							scheduler.set_timesteps(a__ )
						elif num_inference_steps is not None and not hasattr(a__					, """set_timesteps""" ):
							A             =					num_inference_steps

						# copy over dummy past residuals (must be done after set_timesteps)
						A             =					[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
						A             =					dummy_past_residuals[:]

						A             =					scheduler.timesteps[5]
						A             =					scheduler.timesteps[6]

						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample
						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample

						self.assertEqual(output_a.shape					, sample.shape )
						self.assertEqual(output_a.shape					, output_a.shape )

						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample
						A             =					scheduler.step(a__					, a__					, a__					, **a__ ).prev_sample

						self.assertEqual(output_a.shape					, sample.shape )
						self.assertEqual(output_a.shape					, output_a.shape )

				def   _UpperCAmelCase       (    self )						->					Tuple:
					for timesteps in [100, 1000]:
						self.check_over_configs(num_train_timesteps=a__					, time_step=a__ )

				def   _UpperCAmelCase       (    self )						->					Any:
					for t, num_inference_steps in zip([1, 5, 10]					, [10, 50, 100] ):
						self.check_over_forward(num_inference_steps=a__					, time_step=a__ )

				def   _UpperCAmelCase       (    self )						->					int:
					A             =					self.full_loop()
					A             =					torch.mean(torch.abs(a__ ) )

					assert abs(result_mean.item() - 254_0529 ) < 10



