


"""simple docstring"""



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 _lowerCAmelCase       (   unittest.TestCase			):







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

       '''simple docstring'''




       snake_case					:   List[str]         =					"| <pad> <unk> <s> </s> a b c d e f g h i j k".split()
       snake_case					:   List[str]         =					dict(zip(UpperCamelCase__		,			range(len(UpperCamelCase__		)		)		)		)

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

       snake_case					:   Dict         =					tempfile.mkdtemp()
       snake_case					:   Tuple         =					os.path.join(self.tmpdirname		,			VOCAB_FILES_NAMES["vocab_file"]		)
       snake_case					:   Dict         =					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
       snake_case					:   List[Any]         =					"hf-internal-testing/ngram-beam-search-decoder"







   def 				lowerCamelCase	(			self		,			**UpperCamelCase__		)			->	Tuple:

       '''simple docstring'''




       snake_case					:   List[str]         =					self.add_kwargs_tokens_map.copy()
       kwargs.update(UpperCamelCase__		)
       return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname		,			**UpperCamelCase__		)







   def 				lowerCamelCase	(			self		,			**UpperCamelCase__		)			->	Optional[Any]:

       '''simple docstring'''




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







   def 				lowerCamelCase	(			self		,			**UpperCamelCase__		)			->	str:

       '''simple docstring'''




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







   def 				lowerCamelCase	(			self		)			->	Tuple:

       '''simple docstring'''




       shutil.rmtree(self.tmpdirname		)







   def 				lowerCamelCase	(			self		)			->	str:

       '''simple docstring'''




       snake_case					:   str         =					self.get_tokenizer()
       snake_case					:   str         =					self.get_feature_extractor()
       snake_case					:   Any         =					self.get_decoder()

       snake_case					:   Union[str, Any]         =					WavaVecaProcessorWithLM(tokenizer=UpperCamelCase__		,			feature_extractor=UpperCamelCase__		,			decoder=UpperCamelCase__		)

       processor.save_pretrained(self.tmpdirname		)
       snake_case					:   Optional[int]         =					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 				lowerCamelCase	(			self		)			->	List[Any]:

       '''simple docstring'''




       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
       snake_case					:   Dict         =					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 				lowerCamelCase	(			self		)			->	Tuple:

       '''simple docstring'''




       snake_case					:   Dict         =					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 				lowerCamelCase	(			self		)			->	List[str]:

       '''simple docstring'''




       snake_case					:   Tuple         =					self.get_feature_extractor()
       snake_case					:   int         =					self.get_tokenizer()
       snake_case					:   Optional[int]         =					self.get_decoder()

       snake_case					:   str         =					WavaVecaProcessorWithLM(tokenizer=UpperCamelCase__		,			feature_extractor=UpperCamelCase__		,			decoder=UpperCamelCase__		)

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

       snake_case					:   int         =					feature_extractor(UpperCamelCase__		,			return_tensors="np"		)
       snake_case					:   Optional[int]         =					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 				lowerCamelCase	(			self		)			->	List[str]:

       '''simple docstring'''




       snake_case					:   int         =					self.get_feature_extractor()
       snake_case					:   Tuple         =					self.get_tokenizer()
       snake_case					:   Optional[Any]         =					self.get_decoder()

       snake_case					:   List[Any]         =					WavaVecaProcessorWithLM(tokenizer=UpperCamelCase__		,			feature_extractor=UpperCamelCase__		,			decoder=UpperCamelCase__		)

       snake_case					:   List[str]         =					"This is a test string"

       snake_case					:   Dict         =					processor(text=UpperCamelCase__		)

       snake_case					:   str         =					tokenizer(UpperCamelCase__		)

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







   def 				lowerCamelCase	(			self		,			UpperCamelCase__=(2, 10, 16)		,			UpperCamelCase__=77		)			->	Dict:

       '''simple docstring'''




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







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

       '''simple docstring'''




       snake_case					:   Union[str, Any]         =					self.get_feature_extractor()
       snake_case					:   str         =					self.get_tokenizer()
       snake_case					:   List[Any]         =					self.get_decoder()

       snake_case					:   Tuple         =					WavaVecaProcessorWithLM(tokenizer=UpperCamelCase__		,			feature_extractor=UpperCamelCase__		,			decoder=UpperCamelCase__		)

       snake_case					:   str         =					self._get_dummy_logits(shape=(10, 16)		,			seed=13		)

       snake_case					:   Optional[int]         =					processor.decode(UpperCamelCase__		)

       snake_case					:   Optional[Any]         =					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 				lowerCamelCase	(			self		,			UpperCamelCase__		)			->	Union[str, Any]:

       '''simple docstring'''




       snake_case					:   List[str]         =					self.get_feature_extractor()
       snake_case					:   Dict         =					self.get_tokenizer()
       snake_case					:   Any         =					self.get_decoder()

       snake_case					:   Any         =					WavaVecaProcessorWithLM(tokenizer=UpperCamelCase__		,			feature_extractor=UpperCamelCase__		,			decoder=UpperCamelCase__		)

       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:
           snake_case					:   Tuple         =					processor.batch_decode(UpperCamelCase__		)
       else:
           with get_context(UpperCamelCase__		).Pool() as pool:
               snake_case					:   List[str]         =					processor.batch_decode(UpperCamelCase__		,			UpperCamelCase__		)

       snake_case					:   Union[str, Any]         =					list(UpperCamelCase__		)

       with get_context("fork"		).Pool() as p:
           snake_case					:   Any         =					decoder.decode_beams_batch(UpperCamelCase__		,			UpperCamelCase__		)

       snake_case      ,snake_case      ,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 				lowerCamelCase	(			self		)			->	Optional[Any]:

       '''simple docstring'''




       snake_case					:   str         =					self.get_feature_extractor()
       snake_case					:   Any         =					self.get_tokenizer()
       snake_case					:   Dict         =					self.get_decoder()

       snake_case					:   int         =					WavaVecaProcessorWithLM(tokenizer=UpperCamelCase__		,			feature_extractor=UpperCamelCase__		,			decoder=UpperCamelCase__		)

       snake_case					:   Optional[Any]         =					self._get_dummy_logits()

       snake_case					:   int         =					15
       snake_case					:   List[str]         =					-20.0
       snake_case					:   Tuple         =					-4.0

       snake_case					:   Optional[int]         =					processor.batch_decode(
           UpperCamelCase__		,			beam_width=UpperCamelCase__		,			beam_prune_logp=UpperCamelCase__		,			token_min_logp=UpperCamelCase__		,			)
       snake_case					:   Dict         =					decoded_processor_out.text

       snake_case					:   Dict         =					list(UpperCamelCase__		)

       with get_context("fork"		).Pool() as pool:
           snake_case					:   Tuple         =					decoder.decode_beams_batch(
               UpperCamelCase__		,			UpperCamelCase__		,			beam_width=UpperCamelCase__		,			beam_prune_logp=UpperCamelCase__		,			token_min_logp=UpperCamelCase__		,			)

       snake_case					:   int         =					[d[0][0] for d in decoded_decoder_out]
       snake_case					:   Union[str, Any]         =					[d[0][2] for d in decoded_decoder_out]
       snake_case					:   Dict         =					[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.9474]		,			UpperCamelCase__		,			atol=1e-3		)		)







   def 				lowerCamelCase	(			self		)			->	int:

       '''simple docstring'''




       snake_case					:   Dict         =					self.get_feature_extractor()
       snake_case					:   Optional[Any]         =					self.get_tokenizer()
       snake_case					:   Optional[int]         =					self.get_decoder()

       snake_case					:   Any         =					WavaVecaProcessorWithLM(tokenizer=UpperCamelCase__		,			feature_extractor=UpperCamelCase__		,			decoder=UpperCamelCase__		)

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

       snake_case					:   List[str]         =					2.0
       snake_case					:   Union[str, Any]         =					5.0
       snake_case					:   int         =					-20.0
       snake_case					:   List[Any]         =					True

       snake_case					:   Optional[Any]         =					processor.batch_decode(
           UpperCamelCase__		,			alpha=UpperCamelCase__		,			beta=UpperCamelCase__		,			unk_score_offset=UpperCamelCase__		,			lm_score_boundary=UpperCamelCase__		,			)
       snake_case					:   Optional[int]         =					decoded_processor_out.text

       snake_case					:   str         =					list(UpperCamelCase__		)
       decoder.reset_params(
           alpha=UpperCamelCase__		,			beta=UpperCamelCase__		,			unk_score_offset=UpperCamelCase__		,			lm_score_boundary=UpperCamelCase__		,			)

       with get_context("fork"		).Pool() as pool:
           snake_case					:   int         =					decoder.decode_beams_batch(
               UpperCamelCase__		,			UpperCamelCase__		,			)

       snake_case					:   Union[str, 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__		)
       snake_case					:   Any         =					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 				lowerCamelCase	(			self		)			->	Any:

       '''simple docstring'''




       snake_case					:   Union[str, Any]         =					WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)

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

       snake_case					:   str         =					os.listdir(UpperCamelCase__		)
       snake_case					:   List[str]         =					["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 				lowerCamelCase	(			self		)			->	str:

       '''simple docstring'''




       snake_case					:   Tuple         =					snapshot_download("hf-internal-testing/processor_with_lm"		)

       snake_case					:   Tuple         =					WavaVecaProcessorWithLM.from_pretrained(UpperCamelCase__		)

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

       snake_case					:   List[str]         =					os.listdir(UpperCamelCase__		)
       snake_case					:   str         =					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 				lowerCamelCase	(			self		)			->	Dict:

       '''simple docstring'''




       snake_case					:   List[str]         =					WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)
       snake_case					:   List[str]         =					AutoProcessor.from_pretrained("hf-internal-testing/processor_with_lm"		)

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

       snake_case					:   Dict         =					processor_wavaveca(UpperCamelCase__		,			return_tensors="np"		)
       snake_case					:   Tuple         =					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		)

       snake_case					:   Optional[Any]         =					self._get_dummy_logits()

       snake_case					:   List[str]         =					processor_wavaveca.batch_decode(UpperCamelCase__		)
       snake_case					:   int         =					processor_auto.batch_decode(UpperCamelCase__		)

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







   def 				lowerCamelCase	(			self		)			->	List[Any]:

       '''simple docstring'''




       snake_case					:   List[str]         =					self.get_feature_extractor()
       snake_case					:   Optional[int]         =					self.get_tokenizer()
       snake_case					:   Optional[Any]         =					self.get_decoder()

       snake_case					:   Optional[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 				lowerCamelCase	(			UpperCamelCase__		,			UpperCamelCase__		)			->	Union[str, Any]:

       '''simple docstring'''




       snake_case					:   int         =					[d[key] for d in offsets]
       return retrieved_list







   def 				lowerCamelCase	(			self		)			->	Tuple:

       '''simple docstring'''




       snake_case					:   str         =					WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)
       snake_case					:   Any         =					self._get_dummy_logits()[0]

       snake_case					:   Union[str, Any]         =					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 				lowerCamelCase	(			self		)			->	Optional[Any]:

       '''simple docstring'''




       snake_case					:   Optional[int]         =					WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)
       snake_case					:   List[Any]         =					self._get_dummy_logits()

       snake_case					:   List[str]         =					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 				lowerCamelCase	(			self		)			->	Dict:

       '''simple docstring'''




       import torch

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

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

       with torch.no_grad():
           snake_case					:   Dict         =					model(UpperCamelCase__		).logits.cpu().numpy()

       snake_case					:   List[str]         =					processor.decode(logits[0]		,			output_word_offsets=UpperCamelCase__		)

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

       snake_case					:   Tuple         =					"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
       snake_case					:   List[Any]         =					torch.tensor(self.get_from_offsets(UpperCamelCase__		,			"start_time"		)		)
       snake_case					:   Dict         =					torch.tensor(self.get_from_offsets(UpperCamelCase__		,			"end_time"		)		)

       # fmt: off
       snake_case					:   Tuple         =					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]		)
       snake_case					:   Dict         =					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(UpperCamelCase__		,			UpperCamelCase__		,			atol=0.01		)		)
       self.assertTrue(torch.allclose(UpperCamelCase__		,			UpperCamelCase__		,			atol=0.01		)		)









"""simple docstring"""



import argparse
import os
import re


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_dummies.py
__snake_case         =					"""src/diffusers"""

# Matches is_xxx_available()
__snake_case         =					re.compile(R"""is\_([a-z_]*)_available\(\)""")
# Matches from xxx import bla
__snake_case         =					re.compile(R"""\s+from\s+\S*\s+import\s+([^\(\s].*)\n""")


__snake_case         =					"""
{0} = None
"""

__snake_case         =					"""
class {0}(metaclass=DummyObject):
    _backends = {1}

    def __init__(self, *args, **kwargs):
        requires_backends(self, {1})

    @classmethod
    def from_config(cls, *args, **kwargs):
        requires_backends(cls, {1})

    @classmethod
    def from_pretrained(cls, *args, **kwargs):
        requires_backends(cls, {1})
"""


__snake_case         =					"""
def {0}(*args, **kwargs):
    requires_backends({0}, {1})
"""







def 						__lowerCAmelCase		(		lowercase						:						Tuple				)			->		str:


    """simple docstring"""





    snake_case					:   Optional[Any]         =					_re_backend.findall(lowercase				)
    if len(lowercase				) == 0:
        return None

    return "_and_".join(lowercase				)







def 						__lowerCAmelCase		(		)			->		Tuple:


    """simple docstring"""





    with open(os.path.join(lowercase		,							"__init__.py"				)		,							"r"		,							encoding="utf-8"		,							newline="\n"				) as f:
        snake_case					:   Tuple         =					f.readlines()

    # Get to the point we do the actual imports for type checking
    snake_case					:   Tuple         =					0
    snake_case					:   Optional[Any]         =					{}
    # Go through the end of the file
    while line_index < len(lowercase				):
        # If the line contains is_backend_available, we grab all objects associated with the `else` block
        snake_case					:   List[Any]         =					find_backend(lines[line_index]				)
        if backend is not None:
            while not lines[line_index].startswith("else:"				):
                line_index += 1
            line_index += 1
            snake_case					:   str         =					[]
            # Until we unindent, add backend objects to the list
            while line_index < len(lowercase				) and len(lines[line_index]				) > 1:
                snake_case					:   List[str]         =					lines[line_index]
                snake_case					:   Any         =					_re_single_line_import.search(lowercase				)
                if single_line_import_search is not None:
                    objects.extend(single_line_import_search.groups()[0].split(", "				)				)
                elif line.startswith(" " * 8				):
                    objects.append(line[8:-2]				)
                line_index += 1

            if len(lowercase				) > 0:
                snake_case					:   Any         =					objects
        else:
            line_index += 1

    return backend_specific_objects







def 						__lowerCAmelCase		(		lowercase						:						Union[str, Any]		,							lowercase						:						Tuple				)			->		int:


    """simple docstring"""





    if name.isupper():
        return DUMMY_CONSTANT.format(lowercase				)
    elif name.islower():
        return DUMMY_FUNCTION.format(lowercase		,							lowercase				)
    else:
        return DUMMY_CLASS.format(lowercase		,							lowercase				)







def 						__lowerCAmelCase		(		lowercase						:						Union[str, Any]=None				)			->		Optional[int]:


    """simple docstring"""





    if backend_specific_objects is None:
        snake_case					:   List[str]         =					read_init()
    # For special correspondence backend to module name as used in the function requires_modulename
    snake_case					:   Dict         =					{}

    for backend, objects in backend_specific_objects.items():
        snake_case					:   Union[str, Any]         =					"[" + ", ".join(F'"{b}"' for b in backend.split("_and_"				)				) + "]"
        snake_case					:   Union[str, Any]         =					"# This file is autogenerated by the command `make fix-copies`, do not edit.\n"
        dummy_file += "from ..utils import DummyObject, requires_backends\n\n"
        dummy_file += "\n".join([create_dummy_object(lowercase		,							lowercase				) for o in objects]				)
        snake_case					:   Optional[Any]         =					dummy_file

    return dummy_files







def 						__lowerCAmelCase		(		lowercase						:						Optional[Any]=False				)			->		Any:


    """simple docstring"""





    snake_case					:   str         =					create_dummy_files()
    # For special correspondence backend to shortcut as used in utils/dummy_xxx_objects.py
    snake_case					:   Dict         =					{"torch": "pt"}

    # Locate actual dummy modules and read their content.
    snake_case					:   Optional[Any]         =					os.path.join(lowercase		,							"utils"				)
    snake_case					:   Union[str, Any]         =					{
        backend: os.path.join(lowercase		,							F'dummy_{short_names.get(lowercase		,							lowercase				)}_objects.py'				)
        for backend in dummy_files.keys()
    }

    snake_case					:   Optional[Any]         =					{}
    for backend, file_path in dummy_file_paths.items():
        if os.path.isfile(lowercase				):
            with open(lowercase		,							"r"		,							encoding="utf-8"		,							newline="\n"				) as f:
                snake_case					:   int         =					f.read()
        else:
            snake_case					:   Optional[Any]         =					""

    for backend in dummy_files.keys():
        if dummy_files[backend] != actual_dummies[backend]:
            if overwrite:
                print(
                    F'Updating diffusers.utils.dummy_{short_names.get(lowercase		,							lowercase				)}_objects.py as the main '
                    "__init__ has new objects."				)
                with open(dummy_file_paths[backend]		,							"w"		,							encoding="utf-8"		,							newline="\n"				) as f:
                    f.write(dummy_files[backend]				)
            else:
                raise ValueError(
                    "The main __init__ has objects that are not present in "
                    F'diffusers.utils.dummy_{short_names.get(lowercase		,							lowercase				)}_objects.py. Run `make fix-copies` '
                    "to fix this."				)


if __name__ == "__main__":
  __snake_case         =					argparse.ArgumentParser()
  parser.add_argument("""--fix_and_overwrite""", action="""store_true""", help="""Whether to fix inconsistencies.""")
  __snake_case         =					parser.parse_args()

  check_dummies(args.fix_and_overwrite)





