
"""simple docstring"""






import argparse
import json
import os

import fairseq
import torch
from torch import nn

from transformers import (
    SpeechaTextaConfig,
    SpeechaTextaForCausalLM,
    SpeechaTextaTokenizer,
    SpeechEncoderDecoderConfig,
    SpeechEncoderDecoderModel,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaModel,
    logging,
)


logging.set_verbosity_info()
UpperCAmelCase        =				logging.get_logger(__name__)

UpperCAmelCase        =				{
    """post_extract_proj""": """feature_projection.projection""",
    """encoder.pos_conv.0""": """encoder.pos_conv_embed.conv""",
    """self_attn.k_proj""": """encoder.layers.*.attention.k_proj""",
    """self_attn.v_proj""": """encoder.layers.*.attention.v_proj""",
    """self_attn.q_proj""": """encoder.layers.*.attention.q_proj""",
    """self_attn.out_proj""": """encoder.layers.*.attention.out_proj""",
    """self_attn_layer_norm""": """encoder.layers.*.layer_norm""",
    """fc1""": """encoder.layers.*.feed_forward.intermediate_dense""",
    """fc2""": """encoder.layers.*.feed_forward.output_dense""",
    """final_layer_norm""": """encoder.layers.*.final_layer_norm""",
    """encoder.layer_norm""": """encoder.layer_norm""",
    """w2v_model.layer_norm""": """feature_projection.layer_norm""",
    """quantizer.weight_proj""": """quantizer.weight_proj""",
    """quantizer.vars""": """quantizer.codevectors""",
    """project_q""": """project_q""",
    """final_proj""": """project_hid""",
    """w2v_encoder.proj""": """lm_head""",
    """mask_emb""": """masked_spec_embed""",
}
UpperCAmelCase        =				[
    """lm_head""",
    """quantizer.weight_proj""",
    """quantizer.codevectors""",
    """project_q""",
    """project_hid""",
]




def __magic_name__	(				_lowerCamelCase:		List[Any],					_lowerCamelCase:		List[Any],					_lowerCamelCase:		Optional[int],					_lowerCamelCase:		int,					_lowerCamelCase:		List[str]			)					->						Any:

	'''simple docstring'''




	for attribute in key.split('''.'''			):
		lowerCAmelCase												=  getattr(_lowerCamelCase,					_lowerCamelCase			)

	if weight_type is not None:
		lowerCAmelCase												=  getattr(_lowerCamelCase,					_lowerCamelCase			).shape
	else:
		lowerCAmelCase												=  hf_pointer.shape

	assert hf_shape == value.shape, (
	    F"""Shape of hf {key + "." + weight_type if weight_type is not None else ""} is {hf_shape}, but should be"""
	    F""" {value.shape} for {full_name}"""
	)

	if weight_type == "weight":
		lowerCAmelCase												=  value
	elif weight_type == "weight_g":
		lowerCAmelCase												=  value
	elif weight_type == "weight_v":
		lowerCAmelCase												=  value
	elif weight_type == "bias":
		lowerCAmelCase												=  value
	else:
		lowerCAmelCase												=  value

	logger.info(F"""{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}."""			)




def __magic_name__	(				_lowerCamelCase:		str,					_lowerCamelCase:		int			)					->						Tuple:

	'''simple docstring'''




	lowerCAmelCase												=  []
	lowerCAmelCase												=  fairseq_model.state_dict()

	lowerCAmelCase												=  hf_model.feature_extractor

	# if encoder has different dim to decoder -> use proj_weight
	lowerCAmelCase												=  None

	for name, value in fairseq_dict.items():
		lowerCAmelCase												=  False
		if "conv_layers" in name:
			load_conv_layer(
			    _lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					hf_model.config.feat_extract_norm == '''group''',					)
			lowerCAmelCase												=  True
		elif name.split('''.'''			)[0] == "proj":
			lowerCAmelCase												=  fairseq_model.proj
			lowerCAmelCase												=  True
		else:
			for key, mapped_key in MAPPING.items():
				if key in name or key.split('''w2v_model.'''			)[-1] == name.split('''.'''			)[0]:
					lowerCAmelCase												=  True
					if "*" in mapped_key:
						lowerCAmelCase												=  name.split(_lowerCamelCase			)[0].split('''.'''			)[-2]
						lowerCAmelCase												=  mapped_key.replace('''*''',					_lowerCamelCase			)
					if "weight_g" in name:
						lowerCAmelCase												=  '''weight_g'''
					elif "weight_v" in name:
						lowerCAmelCase												=  '''weight_v'''
					elif "bias" in name:
						lowerCAmelCase												=  '''bias'''
					elif "weight" in name:
						lowerCAmelCase												=  '''weight'''
					else:
						lowerCAmelCase												=  None
					set_recursively(_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase			)
				continue
		if not is_used:
			unused_weights.append(_lowerCamelCase			)

	logger.warning(F"""Unused weights: {unused_weights}"""			)

	return proj_weight




def __magic_name__	(				_lowerCamelCase:		Optional[int],					_lowerCamelCase:		Optional[int],					_lowerCamelCase:		Optional[int],					_lowerCamelCase:		Union[str, Any],					_lowerCamelCase:		Optional[Any]			)					->						Dict:

	'''simple docstring'''




	lowerCAmelCase												=  full_name.split('''conv_layers.'''			)[-1]
	lowerCAmelCase												=  name.split('''.'''			)
	lowerCAmelCase												=  int(items[0]			)
	lowerCAmelCase												=  int(items[1]			)

	if type_id == 0:
		if "bias" in name:
			assert value.shape == feature_extractor.conv_layers[layer_id].conv.bias.data.shape, (
			    F"""{full_name} has size {value.shape}, but"""
			    F""" {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found."""
			)
			lowerCAmelCase												=  value
			logger.info(F"""Feat extract conv layer {layer_id} was initialized from {full_name}."""			)
		elif "weight" in name:
			assert value.shape == feature_extractor.conv_layers[layer_id].conv.weight.data.shape, (
			    F"""{full_name} has size {value.shape}, but"""
			    F""" {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found."""
			)
			lowerCAmelCase												=  value
			logger.info(F"""Feat extract conv layer {layer_id} was initialized from {full_name}."""			)
	elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
		if "bias" in name:
			assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape, (
			    F"""{full_name} has size {value.shape}, but {feature_extractor[layer_id].layer_norm.bias.data.shape} was"""
			    " found."
			)
			lowerCAmelCase												=  value
			logger.info(F"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""			)
		elif "weight" in name:
			assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape, (
			    F"""{full_name} has size {value.shape}, but"""
			    F""" {feature_extractor[layer_id].layer_norm.weight.data.shape} was found."""
			)
			lowerCAmelCase												=  value
			logger.info(F"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""			)
	else:
		unused_weights.append(_lowerCamelCase			)




def __magic_name__	(				_lowerCamelCase:		Any			)					->						int:

	'''simple docstring'''




	lowerCAmelCase							,       lowerCAmelCase												=  emb.weight.shape
	lowerCAmelCase												=  nn.Linear(_lowerCamelCase,					_lowerCamelCase,					bias=_lowerCamelCase			)
	lowerCAmelCase												=  emb.weight.data
	return lin_layer




def __magic_name__	(				_lowerCamelCase:		List[str]			)					->						Any:

	'''simple docstring'''




	with open(_lowerCamelCase,					'''r''',					encoding='''utf-8'''			) as f:
		lowerCAmelCase												=  f.readlines()
		lowerCAmelCase												=  [line.split(''' '''			)[0] for line in lines]

	lowerCAmelCase												=  len(_lowerCamelCase			)

	lowerCAmelCase												=  {
	    '''<s>''': 0,
	    '''<pad>''': 1,
	    '''</s>''': 2,
	    '''<unk>''': 3,
	}

	vocab_dict.update(dict(zip(_lowerCamelCase,					range(4,					num_words + 4			)			)			)			)
	return vocab_dict




@torch.no_grad()
def __magic_name__	(				_lowerCamelCase:		List[Any],					_lowerCamelCase:		int,					_lowerCamelCase:		Optional[Any],					_lowerCamelCase:		Union[str, Any],					_lowerCamelCase:		List[str],					_lowerCamelCase:		Any,					_lowerCamelCase:		Dict,					)					->						List[str]:

	'''simple docstring'''




	lowerCAmelCase												=  WavaVecaConfig.from_pretrained(_lowerCamelCase			)
	lowerCAmelCase												=  SpeechaTextaConfig.from_pretrained(
	    _lowerCamelCase,					vocab_size=_lowerCamelCase,					decoder_layers=_lowerCamelCase,					do_stable_layer_norm=_lowerCamelCase			)

	lowerCAmelCase												=  WavaVecaFeatureExtractor(
	    feature_size=1,					sampling_rate=16_000,					padding_value=0,					do_normalize=_lowerCamelCase,					return_attention_mask=_lowerCamelCase,					)

	lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase												=  fairseq.checkpoint_utils.load_model_ensemble_and_task(
	    [checkpoint_path],					arg_overrides={'''data''': '''/'''.join(dict_path.split('''/'''			)[:-1]			)}			)
	lowerCAmelCase												=  model[0].eval()

	# set weights for wav2vec2 encoder
	lowerCAmelCase												=  WavaVecaModel(_lowerCamelCase			)
	lowerCAmelCase												=  recursively_load_weights_wavaveca(model.encoder,					_lowerCamelCase			)

	lowerCAmelCase												=  SpeechaTextaForCausalLM(_lowerCamelCase			)
	lowerCAmelCase							,       lowerCAmelCase												=  hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict(),					strict=_lowerCamelCase			)

	# set output linear layer
	unexpected_keys.remove('''embed_out'''			)
	lowerCAmelCase												=  nn.Parameter(model.decoder.embed_out.detach()			)

	# layer norm is init to identity matrix so leaving it is fine
	logger.warning(F"""The following keys are missing when loading the decoder weights: {missing_keys}"""			)
	logger.warning(F"""The following keys are unexpected when loading the decoder weights: {unexpected_keys}"""			)

	lowerCAmelCase												=  SpeechEncoderDecoderModel(encoder=_lowerCamelCase,					decoder=_lowerCamelCase			)
	lowerCAmelCase												=  False

	# add projection layer
	lowerCAmelCase												=  nn.Parameter(projection_layer.weight			)
	lowerCAmelCase												=  nn.Parameter(projection_layer.bias			)

	lowerCAmelCase												=  create_vocab_dict(_lowerCamelCase			)

	with open(os.path.join(_lowerCamelCase,					'''vocab.json'''			),					'''w'''			) as fp:
		json.dump(_lowerCamelCase,					_lowerCamelCase			)

	lowerCAmelCase												=  SpeechaTextaTokenizer(os.path.join(_lowerCamelCase,					'''vocab.json'''			)			)
	tokenizer.save_pretrained(_lowerCamelCase			)

	lowerCAmelCase												=  hf_wavavec.config.to_dict()
	lowerCAmelCase												=  tokenizer.pad_token_id
	lowerCAmelCase												=  tokenizer.bos_token_id
	lowerCAmelCase												=  tokenizer.eos_token_id
	lowerCAmelCase												=  '''speech_to_text_2'''
	lowerCAmelCase												=  '''wav2vec2'''

	lowerCAmelCase												=  SpeechEncoderDecoderConfig.from_dict(_lowerCamelCase			)

	hf_wavavec.save_pretrained(_lowerCamelCase			)
	feature_extractor.save_pretrained(_lowerCamelCase			)


if __name__ == "__main__":
					UpperCAmelCase        =				argparse.ArgumentParser()
					parser.add_argument("""--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the output PyTorch model.""")
					parser.add_argument("""--checkpoint_path""", default=None, type=str, help="""Path to fairseq checkpoint""")
					parser.add_argument("""--dict_path""", default=None, type=str, help="""Path to dict of fine-tuned model""")
					parser.add_argument(
					    """--encoder_config_path""",
					    default="""facebook/wav2vec2-large-lv60""",
					    type=str,
					    help="""Path to hf encoder wav2vec2 checkpoint config""",
					)
					parser.add_argument(
					    """--decoder_config_path""",
					    default="""facebook/s2t-small-mustc-en-fr-st""",
					    type=str,
					    help="""Path to hf decoder s2t checkpoint config""",
					)
					parser.add_argument("""--vocab_size""", default=1_0_2_2_4, type=int, help="""Vocab size of decoder""")
					parser.add_argument("""--num_decoder_layers""", default=7, type=int, help="""Number of decoder layers""")

					UpperCAmelCase        =				parser.parse_args()
					convert_wavaveca_checkpoint(
					    args.checkpoint_path,
					    args.pytorch_dump_folder_path,
					    args.dict_path,
					    encoder_config_path=args.encoder_config_path,
					    decoder_config_path=args.decoder_config_path,
					    vocab_size=args.vocab_size,
					    num_decoder_layers=args.num_decoder_layers,
					)




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






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





				"""simple docstring"""


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

				lowerCAmelCase												=  {
				    '''unk_token''': '''<unk>''',
				    '''bos_token''': '''<s>''',
				    '''eos_token''': '''</s>''',
				}
				lowerCAmelCase												=  {
				    '''feature_size''': 1,
				    '''padding_value''': 0.0,
				    '''sampling_rate''': 16_000,
				    '''return_attention_mask''': False,
				    '''do_normalize''': True,
				}

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

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

				# load decoder from hub
				lowerCAmelCase												=  '''hf-internal-testing/ngram-beam-search-decoder'''






			def 					UpperCAmelCase							(self  :	List[str]      ,**SCREAMING_SNAKE_CASE_  :	List[Any]			)       ->	Any:





				"""simple docstring"""


				lowerCAmelCase												=  self.add_kwargs_tokens_map.copy()
				kwargs.update(SCREAMING_SNAKE_CASE_			)
				return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname      ,**SCREAMING_SNAKE_CASE_			)






			def 					UpperCAmelCase							(self  :	str      ,**SCREAMING_SNAKE_CASE_  :	Union[str, Any]			)       ->	Union[str, Any]:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,**SCREAMING_SNAKE_CASE_  :	Dict			)       ->	Optional[int]:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	Dict			)       ->	List[str]:





				"""simple docstring"""


				shutil.rmtree(self.tmpdirname			)






			def 					UpperCAmelCase							(self  :	Optional[int]			)       ->	int:





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_SNAKE_CASE_			)

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

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

				# feature extractor
				self.assertEqual(processor.feature_extractor.to_json_string()      ,feature_extractor.to_json_string()			)
				self.assertIsInstance(processor.feature_extractor      ,SCREAMING_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      ,SCREAMING_SNAKE_CASE_			)






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





				"""simple docstring"""


				lowerCAmelCase												=  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
				lowerCAmelCase												=  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 					UpperCAmelCase							(self  :	List[Any]			)       ->	int:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	Dict			)       ->	str:





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  floats_list((3, 1_000)			)

				lowerCAmelCase												=  feature_extractor(SCREAMING_SNAKE_CASE_      ,return_tensors='''np'''			)
				lowerCAmelCase												=  processor(SCREAMING_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 					UpperCAmelCase							(self  :	str			)       ->	Optional[Any]:





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  '''This is a test string'''

				lowerCAmelCase												=  processor(text=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  tokenizer(SCREAMING_SNAKE_CASE_			)

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






			def 					UpperCAmelCase							(self  :	Tuple      ,SCREAMING_SNAKE_CASE_  :	Union[str, Any]=(2, 10, 16)      ,SCREAMING_SNAKE_CASE_  :	List[Any]=77			)       ->	Optional[int]:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	int			)       ->	List[str]:





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_SNAKE_CASE_			)

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

				lowerCAmelCase												=  processor.decode(SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  decoder.decode_beams(SCREAMING_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 					UpperCAmelCase							(self  :	int      ,SCREAMING_SNAKE_CASE_  :	Dict			)       ->	List[str]:





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  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:
					lowerCAmelCase												=  processor.batch_decode(SCREAMING_SNAKE_CASE_			)
				else:
					with get_context(SCREAMING_SNAKE_CASE_			).Pool() as pool:
						lowerCAmelCase												=  processor.batch_decode(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  list(SCREAMING_SNAKE_CASE_			)

				with get_context('''fork'''			).Pool() as p:
					lowerCAmelCase												=  decoder.decode_beams_batch(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase							,       lowerCAmelCase							,       lowerCAmelCase												=  [], [], []
				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(SCREAMING_SNAKE_CASE_      ,decoded_processor.text			)
				self.assertListEqual(['''<s> <s> </s>''', '''<s> <s> <s>''']      ,decoded_processor.text			)
				self.assertListEqual(SCREAMING_SNAKE_CASE_      ,decoded_processor.logit_score			)
				self.assertListEqual(SCREAMING_SNAKE_CASE_      ,decoded_processor.lm_score			)






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





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  self._get_dummy_logits()

				lowerCAmelCase												=  15
				lowerCAmelCase												=  -20.0
				lowerCAmelCase												=  -4.0

				lowerCAmelCase												=  processor.batch_decode(
				    SCREAMING_SNAKE_CASE_      ,beam_width=SCREAMING_SNAKE_CASE_      ,beam_prune_logp=SCREAMING_SNAKE_CASE_      ,token_min_logp=SCREAMING_SNAKE_CASE_      ,)
				lowerCAmelCase												=  decoded_processor_out.text

				lowerCAmelCase												=  list(SCREAMING_SNAKE_CASE_			)

				with get_context('''fork'''			).Pool() as pool:
					lowerCAmelCase												=  decoder.decode_beams_batch(
					    SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_      ,beam_width=SCREAMING_SNAKE_CASE_      ,beam_prune_logp=SCREAMING_SNAKE_CASE_      ,token_min_logp=SCREAMING_SNAKE_CASE_      ,)

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

				self.assertListEqual(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)
				self.assertListEqual(['''</s> <s> <s>''', '''<s> <s> <s>''']      ,SCREAMING_SNAKE_CASE_			)

				self.assertTrue(np.array_equal(SCREAMING_SNAKE_CASE_      ,decoded_processor_out.logit_score			)			)
				self.assertTrue(np.allclose([-20.0_54, -18.4_47]      ,SCREAMING_SNAKE_CASE_      ,atol=1e-3			)			)

				self.assertTrue(np.array_equal(SCREAMING_SNAKE_CASE_      ,decoded_processor_out.lm_score			)			)
				self.assertTrue(np.allclose([-15.5_54, -13.94_74]      ,SCREAMING_SNAKE_CASE_      ,atol=1e-3			)			)






			def 					UpperCAmelCase							(self  :	List[str]			)       ->	Any:





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  self._get_dummy_logits()

				lowerCAmelCase												=  2.0
				lowerCAmelCase												=  5.0
				lowerCAmelCase												=  -20.0
				lowerCAmelCase												=  True

				lowerCAmelCase												=  processor.batch_decode(
				    SCREAMING_SNAKE_CASE_      ,alpha=SCREAMING_SNAKE_CASE_      ,beta=SCREAMING_SNAKE_CASE_      ,unk_score_offset=SCREAMING_SNAKE_CASE_      ,lm_score_boundary=SCREAMING_SNAKE_CASE_      ,)
				lowerCAmelCase												=  decoded_processor_out.text

				lowerCAmelCase												=  list(SCREAMING_SNAKE_CASE_			)
				decoder.reset_params(
				    alpha=SCREAMING_SNAKE_CASE_      ,beta=SCREAMING_SNAKE_CASE_      ,unk_score_offset=SCREAMING_SNAKE_CASE_      ,lm_score_boundary=SCREAMING_SNAKE_CASE_      ,)

				with get_context('''fork'''			).Pool() as pool:
					lowerCAmelCase												=  decoder.decode_beams_batch(
					    SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_      ,)

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

				self.assertListEqual(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)
				self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>''']      ,SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  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      ,SCREAMING_SNAKE_CASE_			)






			def 					UpperCAmelCase							(self  :	Dict			)       ->	str:





				"""simple docstring"""


				lowerCAmelCase												=  WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''			)

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

				lowerCAmelCase												=  os.listdir(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  ['''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(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)






			def 					UpperCAmelCase							(self  :	Any			)       ->	Dict:





				"""simple docstring"""


				lowerCAmelCase												=  snapshot_download('''hf-internal-testing/processor_with_lm'''			)

				lowerCAmelCase												=  WavaVecaProcessorWithLM.from_pretrained(SCREAMING_SNAKE_CASE_			)

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

				lowerCAmelCase												=  os.listdir(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  os.listdir(SCREAMING_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(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)






			def 					UpperCAmelCase							(self  :	Any			)       ->	int:





				"""simple docstring"""


				lowerCAmelCase												=  WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''			)
				lowerCAmelCase												=  AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''			)

				lowerCAmelCase												=  floats_list((3, 1_000)			)

				lowerCAmelCase												=  processor_wavaveca(SCREAMING_SNAKE_CASE_      ,return_tensors='''np'''			)
				lowerCAmelCase												=  processor_auto(SCREAMING_SNAKE_CASE_      ,return_tensors='''np'''			)

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

				lowerCAmelCase												=  self._get_dummy_logits()

				lowerCAmelCase												=  processor_wavaveca.batch_decode(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  processor_auto.batch_decode(SCREAMING_SNAKE_CASE_			)

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






			def 					UpperCAmelCase							(self  :	List[str]			)       ->	Any:





				"""simple docstring"""


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

				lowerCAmelCase												=  WavaVecaProcessorWithLM(tokenizer=SCREAMING_SNAKE_CASE_      ,feature_extractor=SCREAMING_SNAKE_CASE_      ,decoder=SCREAMING_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 					UpperCAmelCase							(SCREAMING_SNAKE_CASE_  :	Tuple      ,SCREAMING_SNAKE_CASE_  :	str			)       ->	List[str]:





				"""simple docstring"""


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






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





				"""simple docstring"""


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

				lowerCAmelCase												=  processor.decode(SCREAMING_SNAKE_CASE_      ,output_word_offsets=SCREAMING_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(SCREAMING_SNAKE_CASE_      ,SCREAMING_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 					UpperCAmelCase							(self  :	List[Any]			)       ->	List[str]:





				"""simple docstring"""


				lowerCAmelCase												=  WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''			)
				lowerCAmelCase												=  self._get_dummy_logits()

				lowerCAmelCase												=  processor.batch_decode(SCREAMING_SNAKE_CASE_      ,output_word_offsets=SCREAMING_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(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)			)

				self.assertListEqual(
				    [''' '''.join(self.get_from_offsets(SCREAMING_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 					UpperCAmelCase							(self  :	Tuple			)       ->	Any:





				"""simple docstring"""


				import torch

				lowerCAmelCase												=  load_dataset('''common_voice'''      ,'''en'''      ,split='''train'''      ,streaming=SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  ds.cast_column('''audio'''      ,datasets.Audio(sampling_rate=16_000			)			)
				lowerCAmelCase												=  iter(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  next(SCREAMING_SNAKE_CASE_			)

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

				with torch.no_grad():
					lowerCAmelCase												=  model(SCREAMING_SNAKE_CASE_			).logits.cpu().numpy()

				lowerCAmelCase												=  processor.decode(logits[0]      ,output_word_offsets=SCREAMING_SNAKE_CASE_			)

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

				lowerCAmelCase												=  '''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(SCREAMING_SNAKE_CASE_      ,'''word'''			)			)      ,SCREAMING_SNAKE_CASE_			)
				self.assertEqual(''' '''.join(self.get_from_offsets(SCREAMING_SNAKE_CASE_      ,'''word'''			)			)      ,output.text			)

				# output times
				lowerCAmelCase												=  torch.tensor(self.get_from_offsets(SCREAMING_SNAKE_CASE_      ,'''start_time'''			)			)
				lowerCAmelCase												=  torch.tensor(self.get_from_offsets(SCREAMING_SNAKE_CASE_      ,'''end_time'''			)			)

				# fmt: off
				lowerCAmelCase												=  torch.tensor([1.41_99, 1.65_99, 2.25_99, 3.0, 3.24, 3.59_99, 3.79_99, 4.09_99, 4.26, 4.94, 5.28, 5.65_99, 5.78, 5.94, 6.32, 6.53_99, 6.65_99]			)
				lowerCAmelCase												=  torch.tensor([1.53_99, 1.89_99, 2.9, 3.16, 3.53_99, 3.72, 4.01_99, 4.17_99, 4.76, 5.15_99, 5.55_99, 5.69_99, 5.86, 6.19_99, 6.38, 6.61_99, 6.94]			)
				# fmt: on

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


