





import argparse
import hashlib
import os
import urllib
import warnings

import torch
from torch import nn
from tqdm import tqdm

from transformers import WhisperConfig, WhisperForConditionalGeneration


A       :   Tuple					= {
    '''tiny.en''': '''https://openaipublic.azureedge.net/main/whisper/models/d3dd57d32accea0b295c96e26691aa14d8822fac7d9d27d5dc00b4ca2826dd03/tiny.en.pt''',
    '''tiny''': '''https://openaipublic.azureedge.net/main/whisper/models/65147644a518d12f04e32d6f3b26facc3f8dd46e5390956a9424a650c0ce22b9/tiny.pt''',
    '''base.en''': '''https://openaipublic.azureedge.net/main/whisper/models/25a8566e1d0c1e2231d1c762132cd20e0f96a85d16145c3a00adf5d1ac670ead/base.en.pt''',
    '''base''': '''https://openaipublic.azureedge.net/main/whisper/models/ed3a0b6b1c0edf879ad9b11b1af5a0e6ab5db9205f891f668f8b0e6c6326e34e/base.pt''',
    '''small.en''': '''https://openaipublic.azureedge.net/main/whisper/models/f953ad0fd29cacd07d5a9eda5624af0f6bcf2258be67c92b79389873d91e0872/small.en.pt''',
    '''small''': '''https://openaipublic.azureedge.net/main/whisper/models/9ecf779972d90ba49c06d968637d720dd632c55bbf19d441fb42bf17a411e794/small.pt''',
    '''medium.en''': '''https://openaipublic.azureedge.net/main/whisper/models/d7440d1dc186f76616474e0ff0b3b6b879abc9d1a4926b7adfa41db2d497ab4f/medium.en.pt''',
    '''medium''': '''https://openaipublic.azureedge.net/main/whisper/models/345ae4da62f9b3d59415adc60127b97c714f32e89e936602e85993674d08dcb1/medium.pt''',
    '''large''': '''https://openaipublic.azureedge.net/main/whisper/models/e4b87e7e0bf463eb8e6956e646f1e277e901512310def2c24bf0e11bd3c28e9a/large.pt''',
    '''large-v2''': '''https://openaipublic.azureedge.net/main/whisper/models/81f7c96c852ee8fc832187b0132e569d6c3065a3252ed18e56effd0b6a73e524/large-v2.pt''',
}







def   __lowerCamelCase   (				__a  :str       )						->							Optional[Any]:






		"""simple docstring"""

		A__     =  ["""layers""", """blocks"""]
		for k in ignore_keys:
				state_dict.pop(__UpperCAmelCase       ,	__UpperCAmelCase       )


A       :   str					= {
    '''blocks''': '''layers''',
    '''mlp.0''': '''fc1''',
    '''mlp.2''': '''fc2''',
    '''mlp_ln''': '''final_layer_norm''',
    '''.attn.query''': '''.self_attn.q_proj''',
    '''.attn.key''': '''.self_attn.k_proj''',
    '''.attn.value''': '''.self_attn.v_proj''',
    '''.attn_ln''': '''.self_attn_layer_norm''',
    '''.attn.out''': '''.self_attn.out_proj''',
    '''.cross_attn.query''': '''.encoder_attn.q_proj''',
    '''.cross_attn.key''': '''.encoder_attn.k_proj''',
    '''.cross_attn.value''': '''.encoder_attn.v_proj''',
    '''.cross_attn_ln''': '''.encoder_attn_layer_norm''',
    '''.cross_attn.out''': '''.encoder_attn.out_proj''',
    '''decoder.ln.''': '''decoder.layer_norm.''',
    '''encoder.ln.''': '''encoder.layer_norm.''',
    '''token_embedding''': '''embed_tokens''',
    '''encoder.positional_embedding''': '''encoder.embed_positions.weight''',
    '''decoder.positional_embedding''': '''decoder.embed_positions.weight''',
    '''ln_post''': '''layer_norm''',
}







def   __lowerCamelCase   (				__a  :Optional[int]       )						->							str:






		"""simple docstring"""

		A__     =  list(s_dict.keys()       )
		for key in keys:
				A__     =  key
				for k, v in WHISPER_MAPPING.items():
						if k in key:
								A__     =  new_key.replace(__UpperCAmelCase       ,	__UpperCAmelCase       )

				print(F'{key} -> {new_key}'       )

				A__     =  s_dict.pop(__UpperCAmelCase       )
		return s_dict







def   __lowerCamelCase   (				__a  :Union[str, Any]       )						->							str:






		"""simple docstring"""

		A__     =  emb.weight.shape
		A__     =  nn.Linear(__UpperCAmelCase       ,	__UpperCAmelCase       ,	bias=__UpperCAmelCase       )
		A__     =  emb.weight.data
		return lin_layer







def   __lowerCamelCase   (				__a  :str       ,	__a  :str       )						->							bytes:






		"""simple docstring"""

		os.makedirs(__UpperCAmelCase       ,	exist_ok=__UpperCAmelCase       )
		A__     =  os.path.basename(__UpperCAmelCase       )

		A__     =  url.split("""/"""       )[-2]
		A__     =  os.path.join(__UpperCAmelCase       ,	__UpperCAmelCase       )

		if os.path.exists(__UpperCAmelCase       ) and not os.path.isfile(__UpperCAmelCase       ):
				raise RuntimeError(F'{download_target} exists and is not a regular file'       )

		if os.path.isfile(__UpperCAmelCase       ):
				A__     =  open(__UpperCAmelCase       ,	"""rb"""       ).read()
				if hashlib.shaaaa(__UpperCAmelCase       ).hexdigest() == expected_shaaaa:
						return model_bytes
				else:
						warnings.warn(F'{download_target} exists, but the SHA256 checksum does not match; re-downloading the file'       )

		with urllib.request.urlopen(__UpperCAmelCase       ) as source, open(__UpperCAmelCase       ,	"""wb"""       ) as output:
				with tqdm(
				    total=int(source.info().get("""Content-Length"""       )       )       ,	ncols=8_0       ,	unit="""iB"""       ,	unit_scale=__UpperCAmelCase       ,	unit_divisor=1_0_2_4       ) as loop:
						while True:
								A__     =  source.read(8_1_9_2       )
								if not buffer:
										break

								output.write(__UpperCAmelCase       )
								loop.update(len(__UpperCAmelCase       )       )

		A__     =  open(__UpperCAmelCase       ,	"""rb"""       ).read()
		if hashlib.shaaaa(__UpperCAmelCase       ).hexdigest() != expected_shaaaa:
				raise RuntimeError(
				    """Model has been downloaded but the SHA256 checksum does not not match. Please retry loading the model."""       )

		return model_bytes







def   __lowerCamelCase   (				__a  :Union[str, Any]       ,	__a  :Optional[int]       )						->							Any:






		"""simple docstring"""

		if ".pt" not in checkpoint_path:
				A__     =  _download(_MODELS[checkpoint_path]       )
		else:
				A__     =  torch.load(__UpperCAmelCase       ,	map_location="""cpu"""       )
		A__     =  original_checkpoint["""dims"""]
		A__     =  original_checkpoint["""model_state_dict"""]
		A__     =  state_dict["""decoder.token_embedding.weight"""]
		remove_ignore_keys_(__UpperCAmelCase       )
		rename_keys(__UpperCAmelCase       )
		A__     =  True
		A__     =  state_dict["""decoder.layers.0.fc1.weight"""].shape[0]

		A__     =  WhisperConfig(
		    vocab_size=dimensions["""n_vocab"""]       ,	encoder_ffn_dim=__UpperCAmelCase       ,	decoder_ffn_dim=__UpperCAmelCase       ,	num_mel_bins=dimensions["""n_mels"""]       ,	d_model=dimensions["""n_audio_state"""]       ,	max_target_positions=dimensions["""n_text_ctx"""]       ,	encoder_layers=dimensions["""n_audio_layer"""]       ,	encoder_attention_heads=dimensions["""n_audio_head"""]       ,	decoder_layers=dimensions["""n_text_layer"""]       ,	decoder_attention_heads=dimensions["""n_text_state"""]       ,	max_source_positions=dimensions["""n_audio_ctx"""]       ,	)

		A__     =  WhisperForConditionalGeneration(__UpperCAmelCase       )
		A__     =  model.model.load_state_dict(__UpperCAmelCase       ,	strict=__UpperCAmelCase       )
		if len(__UpperCAmelCase       ) > 0 and not set(__UpperCAmelCase       ) <= {
		    "encoder.embed_positions.weights",
		    "decoder.embed_positions.weights",
		}:
				raise ValueError(
				    """Only `encoder.embed_positions.weights` and `decoder.embed_positions.weights`  are allowed to be missing,"""
				    F' but all the following weights are missing {missing}'       )

		if tie_embeds:
				A__     =  make_linear_from_emb(model.model.decoder.embed_tokens       )
		else:
				A__     =  proj_out_weights

		model.save_pretrained(__UpperCAmelCase       )


if __name__ == "__main__":
					A       :   int					= argparse.ArgumentParser()
					# # Required parameters
					parser.add_argument('''--checkpoint_path''', type=str, help='''Patht to the downloaded checkpoints''')
					parser.add_argument('''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the output PyTorch model.''')
					A       :   int					= parser.parse_args()

					convert_openai_whisper_to_tfms(args.checkpoint_path, args.pytorch_dump_folder_path)











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

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							Optional[Any]:
											__magic_name__:	List[str]							=       """| <pad> <unk> <s> </s> a b c d e f g h i j k""".split()
											__magic_name__:	List[Any]							=       dict(zip(__snake_case     ,		range(len(__snake_case					)					)					)					)

											__magic_name__:	Union[str, Any]							=       {
											    """unk_token""": """<unk>""",
											    """bos_token""": """<s>""",
											    """eos_token""": """</s>""",
											}
											__magic_name__:	Optional[int]							=       {
											    """feature_size""": 1,
											    """padding_value""": 0.0,
											    """sampling_rate""": 1_6_0_0_0,
											    """return_attention_mask""": False,
											    """do_normalize""": True,
											}

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

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

											# load decoder from hub
											__magic_name__:	Dict							=       """hf-internal-testing/ngram-beam-search-decoder"""

					def   lowerCamelCase__   (			self  :      Any     ,		**__snake_case  :      str					)			->							Optional[int]:
											__magic_name__:	Union[str, Any]							=       self.add_kwargs_tokens_map.copy()
											kwargs.update(__snake_case					)
											return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname     ,		**__snake_case					)

					def   lowerCamelCase__   (			self  :      str     ,		**__snake_case  :      int					)			->							Dict:
											return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname     ,		**__snake_case					)

					def   lowerCamelCase__   (			self  :      int     ,		**__snake_case  :      List[str]					)			->							int:
											return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name     ,		**__snake_case					)

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							Any:
											shutil.rmtree(self.tmpdirname					)

					def   lowerCamelCase__   (			self  :      List[Any]					)			->							Optional[Any]:
											__magic_name__:	Dict							=       self.get_tokenizer()
											__magic_name__:	Any							=       self.get_feature_extractor()
											__magic_name__:	Tuple							=       self.get_decoder()

											__magic_name__:	Tuple							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											processor.save_pretrained(self.tmpdirname					)
											__magic_name__:	Dict							=       WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname					)

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

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

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

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							int:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	Optional[Any]							=       self.get_tokenizer()
											__magic_name__:	List[Any]							=       self.get_decoder()

											__magic_name__:	int							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	Tuple							=       floats_list((3, 1_0_0_0)					)

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

											__magic_name__:	Tuple							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	Optional[int]							=       """This is a test string"""

											__magic_name__:	List[str]							=       processor(text=__snake_case					)

											__magic_name__:	Tuple							=       tokenizer(__snake_case					)

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

					def   lowerCamelCase__   (			self  :      int     ,		__snake_case  :      List[str]=(2, 1_0, 1_6)     ,		__snake_case  :      List[Any]=7_7					)			->							Dict:
											np.random.seed(__snake_case					)
											return np.random.rand(*__snake_case					)

					def   lowerCamelCase__   (			self  :      Any					)			->							Any:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	Tuple							=       self.get_tokenizer()
											__magic_name__:	Any							=       self.get_decoder()

											__magic_name__:	Tuple							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	List[Any]							=       self._get_dummy_logits(shape=(1_0, 1_6)     ,		seed=1_3					)

											__magic_name__:	str							=       processor.decode(__snake_case					)

											__magic_name__:	Optional[int]							=       decoder.decode_beams(__snake_case					)[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  :      int     ,		__snake_case  :      Dict					)			->							Any:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	List[Any]							=       self.get_tokenizer()
											__magic_name__:	int							=       self.get_decoder()

											__magic_name__:	Optional[Any]							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

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

											__magic_name__:	Dict							=       list(__snake_case					)

											with get_context("""fork"""					).Pool() as p:
																	__magic_name__:	List[str]							=       decoder.decode_beams_batch(__snake_case     ,		__snake_case					)

											__magic_name__,			__magic_name__,			__magic_name__:	Optional[int]							=       [], [], []
											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(__snake_case     ,		decoded_processor.text					)
											self.assertListEqual(["""<s> <s> </s>""", """<s> <s> <s>"""]     ,		decoded_processor.text					)
											self.assertListEqual(__snake_case     ,		decoded_processor.logit_score					)
											self.assertListEqual(__snake_case     ,		decoded_processor.lm_score					)

					def   lowerCamelCase__   (			self  :      List[Any]					)			->							List[Any]:
											__magic_name__:	List[str]							=       self.get_feature_extractor()
											__magic_name__:	Optional[Any]							=       self.get_tokenizer()
											__magic_name__:	Optional[int]							=       self.get_decoder()

											__magic_name__:	Dict							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	str							=       self._get_dummy_logits()

											__magic_name__:	Dict							=       1_5
											__magic_name__:	int							=       -20.0
											__magic_name__:	int							=       -4.0

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

											__magic_name__:	Union[str, Any]							=       list(__snake_case					)

											with get_context("""fork"""					).Pool() as pool:
																	__magic_name__:	str							=       decoder.decode_beams_batch(
																	    __snake_case     ,		__snake_case     ,		beam_width=__snake_case     ,		beam_prune_logp=__snake_case     ,		token_min_logp=__snake_case     ,		)

											__magic_name__:	Any							=       [d[0][0] for d in decoded_decoder_out]
											__magic_name__:	Optional[int]							=       [d[0][2] for d in decoded_decoder_out]
											__magic_name__:	Optional[Any]							=       [d[0][3] for d in decoded_decoder_out]

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

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

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							int:
											__magic_name__:	int							=       self.get_feature_extractor()
											__magic_name__:	Any							=       self.get_tokenizer()
											__magic_name__:	Union[str, Any]							=       self.get_decoder()

											__magic_name__:	str							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											__magic_name__:	Any							=       self._get_dummy_logits()

											__magic_name__:	Union[str, Any]							=       2.0
											__magic_name__:	Optional[Any]							=       5.0
											__magic_name__:	Optional[Any]							=       -20.0
											__magic_name__:	List[str]							=       True

											__magic_name__:	List[Any]							=       processor.batch_decode(
											    __snake_case     ,		alpha=__snake_case     ,		beta=__snake_case     ,		unk_score_offset=__snake_case     ,		lm_score_boundary=__snake_case     ,		)
											__magic_name__:	Union[str, Any]							=       decoded_processor_out.text

											__magic_name__:	Union[str, Any]							=       list(__snake_case					)
											decoder.reset_params(
											    alpha=__snake_case     ,		beta=__snake_case     ,		unk_score_offset=__snake_case     ,		lm_score_boundary=__snake_case     ,		)

											with get_context("""fork"""					).Pool() as pool:
																	__magic_name__:	str							=       decoder.decode_beams_batch(
																	    __snake_case     ,		__snake_case     ,		)

											__magic_name__:	List[str]							=       [d[0][0] for d in decoded_decoder_out]

											self.assertListEqual(__snake_case     ,		__snake_case					)
											self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]     ,		__snake_case					)
											__magic_name__:	List[str]							=       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     ,		__snake_case					)

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							List[Any]:
											__magic_name__:	List[Any]							=       WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""					)

											__magic_name__:	Union[str, Any]							=       processor.decoder.model_container[processor.decoder._model_key]
											__magic_name__:	Union[str, Any]							=       Path(language_model._kenlm_model.path.decode("""utf-8"""					)					).parent.parent.absolute()

											__magic_name__:	Optional[int]							=       os.listdir(__snake_case					)
											__magic_name__:	Union[str, 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(__snake_case     ,		__snake_case					)

					def   lowerCamelCase__   (			self  :      Any					)			->							Any:
											__magic_name__:	int							=       snapshot_download("""hf-internal-testing/processor_with_lm"""					)

											__magic_name__:	List[Any]							=       WavaVecaProcessorWithLM.from_pretrained(__snake_case					)

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

											__magic_name__:	str							=       os.listdir(__snake_case					)
											__magic_name__:	Tuple							=       os.listdir(__snake_case					)

											local_decoder_files.sort()
											expected_decoder_files.sort()

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

					def   lowerCamelCase__   (			self  :      Optional[int]					)			->							int:
											__magic_name__:	List[Any]							=       WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""					)
											__magic_name__:	List[str]							=       AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""					)

											__magic_name__:	List[str]							=       floats_list((3, 1_0_0_0)					)

											__magic_name__:	Tuple							=       processor_wavaveca(__snake_case     ,		return_tensors="""np"""					)
											__magic_name__:	Optional[Any]							=       processor_auto(__snake_case     ,		return_tensors="""np"""					)

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

											__magic_name__:	int							=       self._get_dummy_logits()

											__magic_name__:	List[Any]							=       processor_wavaveca.batch_decode(__snake_case					)
											__magic_name__:	Union[str, Any]							=       processor_auto.batch_decode(__snake_case					)

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]					)			->							str:
											__magic_name__:	Optional[int]							=       self.get_feature_extractor()
											__magic_name__:	Any							=       self.get_tokenizer()
											__magic_name__:	Dict							=       self.get_decoder()

											__magic_name__:	List[str]							=       WavaVecaProcessorWithLM(tokenizer=__snake_case     ,		feature_extractor=__snake_case     ,		decoder=__snake_case					)

											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__   (			__snake_case  :      Optional[int]     ,		__snake_case  :      int					)			->							int:
											__magic_name__:	Any							=       [d[key] for d in offsets]
											return retrieved_list

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

											__magic_name__:	List[Any]							=       processor.decode(__snake_case     ,		output_word_offsets=__snake_case					)
											# 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(__snake_case     ,		__snake_case					)					)

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

											__magic_name__:	Any							=       processor.batch_decode(__snake_case     ,		output_word_offsets=__snake_case					)

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

											self.assertListEqual(
											    [""" """.join(self.get_from_offsets(__snake_case     ,		"""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  :      Union[str, Any]					)			->							int:
											import torch

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

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

											with torch.no_grad():
																	__magic_name__:	List[Any]							=       model(__snake_case					).logits.cpu().numpy()

											__magic_name__:	Optional[Any]							=       processor.decode(logits[0]     ,		output_word_offsets=__snake_case					)

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

											__magic_name__:	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(__snake_case     ,		"""word"""					)					)     ,		__snake_case					)
											self.assertEqual(""" """.join(self.get_from_offsets(__snake_case     ,		"""word"""					)					)     ,		output.text					)

											# output times
											__magic_name__:	Dict							=       torch.tensor(self.get_from_offsets(__snake_case     ,		"""start_time"""					)					)
											__magic_name__:	Optional[Any]							=       torch.tensor(self.get_from_offsets(__snake_case     ,		"""end_time"""					)					)

											# fmt: off
											__magic_name__:	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]					)
											__magic_name__:	int							=       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(__snake_case     ,		__snake_case     ,		atol=0.01					)					)
											self.assertTrue(torch.allclose(__snake_case     ,		__snake_case     ,		atol=0.01					)					)



