





'''simple docstring'''



import json
import logging
import math
import os
import sys
from dataclasses import dataclass, field
from typing import Optional

from datasets import Dataset, load_dataset

import transformers
from transformers import (
    CONFIG_MAPPING,
    MODEL_FOR_MASKED_LM_MAPPING,
    AutoConfig,
    AutoModelForMaskedLM,
    AutoTokenizer,
    DataCollatorForWholeWordMask,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import get_last_checkpoint, is_main_process


_SCREAMING_SNAKE_CASE					=    logging.getLogger(__name__)
_SCREAMING_SNAKE_CASE					=    list(MODEL_FOR_MASKED_LM_MAPPING.keys())
_SCREAMING_SNAKE_CASE					=    tuple(conf.model_type for conf in MODEL_CONFIG_CLASSES)






@dataclass
class 							__lowercase       :







	'''simple docstring'''

	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={
	        "help": (
	            "The model checkpoint for weights initialization.Don't set if you want to train a model from scratch."
	        )
	    } ,      )
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "If training from scratch, pass a model type from the list: " + ", ".join(lowerCAmelCase__						)} ,      )
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={
	        "help": (
	            "Override some existing default config settings when a model is trained from scratch. Example: "
	            "n_embd=10,resid_pdrop=0.2,scale_attn_weights=false,summary_type=cls_index"
	        )
	    } ,      )
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "Pretrained config name or path if not the same as model_name"}						)
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "Pretrained tokenizer name or path if not the same as model_name"}						)
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"} ,      )
	a    :				bool											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "Whether to use one of the fast tokenizer (backed by the tokenizers library) or not."} ,      )
	a    :				str											=       field(
	    default="main" ,      metadata={"help": "The specific model version to use (can be a branch name, tag name or commit id)."} ,      )
	a    :				bool											=       field(
	    default=lowerCAmelCase__ ,      metadata={
	        "help": (
	            "Will use the token generated when running `huggingface-cli login` (necessary to use this script "
	            "with private models)."
	        )
	    } ,      )


	def 						_UpperCAmelCase     (self )							->							str:







				'''simple docstring'''





				if self.config_overrides is not None and (self.config_name is not None or self.model_name_or_path is not None):
							raise ValueError(
							    '''--config_overrides can\'t be used in combination with --config_name or --model_name_or_path''' )






@dataclass
class 							__lowercase       :







	'''simple docstring'''

	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "The name of the dataset to use (via the datasets library)."}						)
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "The configuration name of the dataset to use (via the datasets library)."}						)
	a    :				Optional[str]											=       field(default=lowerCAmelCase__ ,      metadata={"help": "The input training data file (a text file)."}						)
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "An optional input evaluation data file to evaluate the perplexity on (a text file)."} ,      )
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "An optional input train ref data file for whole word masking in Chinese."} ,      )
	a    :				Optional[str]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "An optional input validation ref data file for whole word masking in Chinese."} ,      )
	a    :				bool											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "Overwrite the cached training and evaluation sets"}						)
	a    :				Optional[int]											=       field(
	    default=5 ,      metadata={
	        "help": "The percentage of the train set used as validation set in case there's no validation split"
	    } ,      )
	a    :				Optional[int]											=       field(
	    default=lowerCAmelCase__ ,      metadata={
	        "help": (
	            "The maximum total input sequence length after tokenization. Sequences longer "
	            "than this will be truncated. Default to the max input length of the model."
	        )
	    } ,      )
	a    :				Optional[int]											=       field(
	    default=lowerCAmelCase__ ,      metadata={"help": "The number of processes to use for the preprocessing."} ,      )
	a    :				float											=       field(
	    default=0.15 ,      metadata={"help": "Ratio of tokens to mask for masked language modeling loss"}						)
	a    :				bool											=       field(
	    default=lowerCAmelCase__ ,      metadata={
	        "help": (
	            "Whether to pad all samples to `max_seq_length`. "
	            "If False, will pad the samples dynamically when batching to the maximum length in the batch."
	        )
	    } ,      )


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







				'''simple docstring'''





				if self.train_file is not None:
							__lowercase										=							self.train_file.split('''.''' )[-1]
							assert extension in ["csv", "json", "txt"], "`train_file` should be a csv, a json or a txt file."
				if self.validation_file is not None:
							__lowercase										=							self.validation_file.split('''.''' )[-1]
							assert extension in ["csv", "json", "txt"], "`validation_file` should be a csv, a json or a txt file."





def    _lowerCAmelCase						(				lowerCamelCase_			:  Union[str, Any]		,       lowerCamelCase_			:  Any				):
			with open(lowerCamelCase_		,       '''r'''		,       encoding='''utf-8'''				) as f:
						__lowercase										=							[json.loads(lowerCamelCase_				) for line in f.read().splitlines() if (len(lowerCamelCase_				) > 0 and not line.isspace())]
			assert len(lowerCamelCase_				) == len(lowerCamelCase_				)

			__lowercase										=							{c: dataset[c] for c in dataset.column_names}
			__lowercase										=							refs
			return Dataset.from_dict(lowerCamelCase_				)




def    _lowerCAmelCase						(				):
			# See all possible arguments in src/transformers/training_args.py
			# or by passing the --help flag to this script.
			# We now keep distinct sets of args, for a cleaner separation of concerns.

			__lowercase										=							HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)				)
			if len(sys.argv				) == 2 and sys.argv[1].endswith('''.json'''				):
						# If we pass only one argument to the script and it's the path to a json file,
						# let's parse it to get our arguments.
						__lowercase						,		__lowercase						,		__lowercase										=							parser.parse_json_file(json_file=os.path.abspath(sys.argv[1]				)				)
			else:
						__lowercase						,		__lowercase						,		__lowercase										=							parser.parse_args_into_dataclasses()

			# Detecting last checkpoint.
			__lowercase										=							None
			if os.path.isdir(training_args.output_dir				) and training_args.do_train and not training_args.overwrite_output_dir:
						__lowercase										=							get_last_checkpoint(training_args.output_dir				)
						if last_checkpoint is None and len(os.listdir(training_args.output_dir				)				) > 0:
									raise ValueError(
									    f"Output directory ({training_args.output_dir}) already exists and is not empty. "
									    '''Use --overwrite_output_dir to overcome.'''				)
						elif last_checkpoint is not None:
									logger.info(
									    f"Checkpoint detected, resuming training at {last_checkpoint}. To avoid this behavior, change "
									    '''the `--output_dir` or add `--overwrite_output_dir` to train from scratch.'''				)

    # Setup logging
			logging.basicConfig(
			    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s'''		,       datefmt='''%m/%d/%Y %H:%M:%S'''		,       handlers=[logging.StreamHandler(sys.stdout				)]		,       )
			logger.setLevel(logging.INFO if is_main_process(training_args.local_rank				) else logging.WARN				)

			# Log on each process the small summary:
			logger.warning(
			    f"Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}"
			    + f"distributed training: {bool(training_args.local_rank != -1				)}, 16-bits training: {training_args.fpaa}"				)
			# Set the verbosity to info of the Transformers logger (on main process only):
			if is_main_process(training_args.local_rank				):
						transformers.utils.logging.set_verbosity_info()
						transformers.utils.logging.enable_default_handler()
						transformers.utils.logging.enable_explicit_format()
			logger.info('''Training/evaluation parameters %s'''		,       lowerCamelCase_				)

			# Set seed before initializing model.
			set_seed(training_args.seed				)

			# Get the datasets: you can either provide your own CSV/JSON/TXT training and evaluation files (see below)
			# or just provide the name of one of the public datasets available on the hub at https://huggingface.co/datasets/
			# (the dataset will be downloaded automatically from the datasets Hub).
			#
			# For CSV/JSON files, this script will use the column called 'text' or the first column if no column called
			# 'text' is found. You can easily tweak this behavior (see below).
			#
			# In distributed training, the load_dataset function guarantee that only one local process can concurrently
			# download the dataset.
			if data_args.dataset_name is not None:
						# Downloading and loading a dataset from the hub.
						__lowercase										=							load_dataset(data_args.dataset_name		,       data_args.dataset_config_name				)
						if "validation" not in datasets.keys():
									__lowercase										=							load_dataset(
									    data_args.dataset_name		,       data_args.dataset_config_name		,       split=f"train[:{data_args.validation_split_percentage}%]"		,       )
									__lowercase										=							load_dataset(
									    data_args.dataset_name		,       data_args.dataset_config_name		,       split=f"train[{data_args.validation_split_percentage}%:]"		,       )
			else:
						__lowercase										=							{}
						if data_args.train_file is not None:
									__lowercase										=							data_args.train_file
						if data_args.validation_file is not None:
									__lowercase										=							data_args.validation_file
						__lowercase										=							data_args.train_file.split('''.'''				)[-1]
						if extension == "txt":
									__lowercase										=							'''text'''
						__lowercase										=							load_dataset(lowerCamelCase_		,       data_files=lowerCamelCase_				)
			# See more about loading any type of standard or custom dataset (from files, python dict, pandas DataFrame, etc) at
			# https://huggingface.co/docs/datasets/loading_datasets.html.

			# Load pretrained model and tokenizer
			#
			# Distributed training:
			# The .from_pretrained methods guarantee that only one local process can concurrently
			# download model & vocab.
			__lowercase										=							{
			    '''cache_dir''': model_args.cache_dir,
			    '''revision''': model_args.model_revision,
			    '''use_auth_token''': True if model_args.use_auth_token else None,
			}
			if model_args.config_name:
						__lowercase										=							AutoConfig.from_pretrained(model_args.config_name		,       **lowerCamelCase_				)
			elif model_args.model_name_or_path:
						__lowercase										=							AutoConfig.from_pretrained(model_args.model_name_or_path		,       **lowerCamelCase_				)
			else:
						__lowercase										=							CONFIG_MAPPING[model_args.model_type]()
						logger.warning('''You are instantiating a new config instance from scratch.'''				)
						if model_args.config_overrides is not None:
									logger.info(f"Overriding config: {model_args.config_overrides}"				)
									config.update_from_string(model_args.config_overrides				)
									logger.info(f"New config: {config}"				)

			__lowercase										=							{
			    '''cache_dir''': model_args.cache_dir,
			    '''use_fast''': model_args.use_fast_tokenizer,
			    '''revision''': model_args.model_revision,
			    '''use_auth_token''': True if model_args.use_auth_token else None,
			}
			if model_args.tokenizer_name:
						__lowercase										=							AutoTokenizer.from_pretrained(model_args.tokenizer_name		,       **lowerCamelCase_				)
			elif model_args.model_name_or_path:
						__lowercase										=							AutoTokenizer.from_pretrained(model_args.model_name_or_path		,       **lowerCamelCase_				)
			else:
						raise ValueError(
						    '''You are instantiating a new tokenizer from scratch. This is not supported by this script.'''
						    '''You can do it from another script, save it, and load it from here, using --tokenizer_name.'''				)

			if model_args.model_name_or_path:
						__lowercase										=							AutoModelForMaskedLM.from_pretrained(
						    model_args.model_name_or_path		,       from_tf=bool('''.ckpt''' in model_args.model_name_or_path				)		,       config=lowerCamelCase_		,       cache_dir=model_args.cache_dir		,       revision=model_args.model_revision		,       use_auth_token=True if model_args.use_auth_token else None		,       )
			else:
						logger.info('''Training new model from scratch'''				)
						__lowercase										=							AutoModelForMaskedLM.from_config(lowerCamelCase_				)

			model.resize_token_embeddings(len(lowerCamelCase_				)				)

			# Preprocessing the datasets.
			# First we tokenize all the texts.
			if training_args.do_train:
						__lowercase										=							datasets['''train'''].column_names
			else:
						__lowercase										=							datasets['''validation'''].column_names
			__lowercase										=							'''text''' if '''text''' in column_names else column_names[0]

			__lowercase										=							'''max_length''' if data_args.pad_to_max_length else False

			def tokenize_function(lowerCamelCase_			:  int				):
						# Remove empty lines
						__lowercase										=							[line for line in examples['''text'''] if len(lowerCamelCase_				) > 0 and not line.isspace()]
						return tokenizer(examples['''text''']		,       padding=lowerCamelCase_		,       truncation=lowerCamelCase_		,       max_length=data_args.max_seq_length				)

			__lowercase										=							datasets.map(
			    lowerCamelCase_		,       batched=lowerCamelCase_		,       num_proc=data_args.preprocessing_num_workers		,       remove_columns=[text_column_name]		,       load_from_cache_file=not data_args.overwrite_cache		,       )

			# Add the chinese references if provided
			if data_args.train_ref_file is not None:
						__lowercase										=							add_chinese_references(tokenized_datasets['''train''']		,       data_args.train_ref_file				)
			if data_args.validation_ref_file is not None:
						__lowercase										=							add_chinese_references(
						    tokenized_datasets['''validation''']		,       data_args.validation_ref_file				)
			# If we have ref files, need to avoid it removed by trainer
			__lowercase										=							data_args.train_ref_file or data_args.validation_ref_file
			if has_ref:
						__lowercase										=							False

			# Data collator
			# This one will take care of randomly masking the tokens.
			__lowercase										=							DataCollatorForWholeWordMask(tokenizer=lowerCamelCase_		,       mlm_probability=data_args.mlm_probability				)

			# Initialize our Trainer
			__lowercase										=							Trainer(
			    model=lowerCamelCase_		,       args=lowerCamelCase_		,       train_dataset=tokenized_datasets['''train'''] if training_args.do_train else None		,       eval_dataset=tokenized_datasets['''validation'''] if training_args.do_eval else None		,       tokenizer=lowerCamelCase_		,       data_collator=lowerCamelCase_		,       )

			# Training
			if training_args.do_train:
						if last_checkpoint is not None:
									__lowercase										=							last_checkpoint
						elif model_args.model_name_or_path is not None and os.path.isdir(model_args.model_name_or_path				):
									__lowercase										=							model_args.model_name_or_path
						else:
									__lowercase										=							None
						__lowercase										=							trainer.train(resume_from_checkpoint=lowerCamelCase_				)
						trainer.save_model()  # Saves the tokenizer too for easy upload

						__lowercase										=							os.path.join(training_args.output_dir		,       '''train_results.txt'''				)
						if trainer.is_world_process_zero():
									with open(lowerCamelCase_		,       '''w'''				) as writer:
												logger.info('''***** Train results *****'''				)
												for key, value in sorted(train_result.metrics.items()				):
															logger.info(f"  {key} = {value}"				)
															writer.write(f"{key} = {value}\n"				)

            # Need to save the state, since Trainer.save_model saves only the tokenizer with the model
									trainer.state.save_to_json(os.path.join(training_args.output_dir		,       '''trainer_state.json'''				)				)

    # Evaluation
			__lowercase										=							{}
			if training_args.do_eval:
						logger.info('''*** Evaluate ***'''				)

						__lowercase										=							trainer.evaluate()

						__lowercase										=							math.exp(eval_output['''eval_loss''']				)
						__lowercase										=							perplexity

						__lowercase										=							os.path.join(training_args.output_dir		,       '''eval_results_mlm_wwm.txt'''				)
						if trainer.is_world_process_zero():
									with open(lowerCamelCase_		,       '''w'''				) as writer:
												logger.info('''***** Eval results *****'''				)
												for key, value in sorted(results.items()				):
															logger.info(f"  {key} = {value}"				)
															writer.write(f"{key} = {value}\n"				)

			return results




def    _lowerCAmelCase						(				lowerCamelCase_			:  Any				):
			# For xla_spawn (TPUs)
			main()


if __name__ == "__main__":
	main()












'''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						):







	'''simple docstring'''

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







				'''simple docstring'''





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

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

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

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

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


	def 						_UpperCAmelCase     (self     ,**_lowerCamelCase )							->							List[str]:







				'''simple docstring'''





				__lowercase										=							self.add_kwargs_tokens_map.copy()
				kwargs.update(_lowerCamelCase )
				return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname     ,**_lowerCamelCase )


	def 						_UpperCAmelCase     (self     ,**_lowerCamelCase )							->							List[Any]:







				'''simple docstring'''





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


	def 						_UpperCAmelCase     (self     ,**_lowerCamelCase )							->							Dict:







				'''simple docstring'''





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


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







				'''simple docstring'''





				shutil.rmtree(self.tmpdirname )


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







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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

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

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

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


	def 						_UpperCAmelCase     (self )							->							Any:







				'''simple docstring'''





				__lowercase										=							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
				__lowercase										=							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 )							->							Dict:







				'''simple docstring'''





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


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







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							floats_list((3, 1000) )

				__lowercase										=							feature_extractor(_lowerCamelCase     ,return_tensors='''np''' )
				__lowercase										=							processor(_lowerCamelCase     ,return_tensors='''np''' )

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


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







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							'''This is a test string'''

				__lowercase										=							processor(text=_lowerCamelCase )

				__lowercase										=							tokenizer(_lowerCamelCase )

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


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase=(2, 10, 16)     ,_lowerCamelCase=77 )							->							Optional[int]:







				'''simple docstring'''





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


	def 						_UpperCAmelCase     (self )							->							str:







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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

				__lowercase										=							processor.decode(_lowerCamelCase )

				__lowercase										=							decoder.decode_beams(_lowerCamelCase )[0]

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


	@parameterized.expand([[None], ['''fork'''], ['''spawn''']] )
	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							Optional[Any]:







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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

				__lowercase										=							list(_lowerCamelCase )

				with get_context('''fork''' ).Pool() as p:
							__lowercase										=							decoder.decode_beams_batch(_lowerCamelCase     ,_lowerCamelCase )

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

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


	def 						_UpperCAmelCase     (self )							->							Any:







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							self._get_dummy_logits()

				__lowercase										=							15
				__lowercase										=							-2_0.0
				__lowercase										=							-4.0

				__lowercase										=							processor.batch_decode(
				    _lowerCamelCase     ,beam_width=_lowerCamelCase     ,beam_prune_logp=_lowerCamelCase     ,token_min_logp=_lowerCamelCase     ,)
				__lowercase										=							decoded_processor_out.text

				__lowercase										=							list(_lowerCamelCase )

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

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

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

				self.assertTrue(np.array_equal(_lowerCamelCase     ,decoded_processor_out.logit_score ) )
				self.assertTrue(np.allclose([-2_0.0_5_4, -1_8.4_4_7]     ,_lowerCamelCase     ,atol=1E-3 ) )

				self.assertTrue(np.array_equal(_lowerCamelCase     ,decoded_processor_out.lm_score ) )
				self.assertTrue(np.allclose([-1_5.5_5_4, -1_3.9_4_7_4]     ,_lowerCamelCase     ,atol=1E-3 ) )


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







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

				__lowercase										=							self._get_dummy_logits()

				__lowercase										=							2.0
				__lowercase										=							5.0
				__lowercase										=							-2_0.0
				__lowercase										=							True

				__lowercase										=							processor.batch_decode(
				    _lowerCamelCase     ,alpha=_lowerCamelCase     ,beta=_lowerCamelCase     ,unk_score_offset=_lowerCamelCase     ,lm_score_boundary=_lowerCamelCase     ,)
				__lowercase										=							decoded_processor_out.text

				__lowercase										=							list(_lowerCamelCase )
				decoder.reset_params(
				    alpha=_lowerCamelCase     ,beta=_lowerCamelCase     ,unk_score_offset=_lowerCamelCase     ,lm_score_boundary=_lowerCamelCase     ,)

				with get_context('''fork''' ).Pool() as pool:
							__lowercase										=							decoder.decode_beams_batch(
							    _lowerCamelCase     ,_lowerCamelCase     ,)

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

				self.assertListEqual(_lowerCamelCase     ,_lowerCamelCase )
				self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>''']     ,_lowerCamelCase )
				__lowercase										=							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     ,-2_0.0 )
				self.assertEqual(lm_model.score_boundary     ,_lowerCamelCase )


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







				'''simple docstring'''





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

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

				__lowercase										=							os.listdir(_lowerCamelCase )
				__lowercase										=							['''alphabet.json''', '''language_model''']

				downloaded_decoder_files.sort()
				expected_decoder_files.sort()

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


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







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM.from_pretrained(_lowerCamelCase )

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

				__lowercase										=							os.listdir(_lowerCamelCase )
				__lowercase										=							os.listdir(_lowerCamelCase )

				local_decoder_files.sort()
				expected_decoder_files.sort()

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


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





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

				__lowercase										=							floats_list((3, 1000) )

				__lowercase										=							processor_wavaveca(_lowerCamelCase     ,return_tensors='''np''' )
				__lowercase										=							processor_auto(_lowerCamelCase     ,return_tensors='''np''' )

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

				__lowercase										=							self._get_dummy_logits()

				__lowercase										=							processor_wavaveca.batch_decode(_lowerCamelCase )
				__lowercase										=							processor_auto.batch_decode(_lowerCamelCase )

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


	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





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

				__lowercase										=							WavaVecaProcessorWithLM(tokenizer=_lowerCamelCase     ,feature_extractor=_lowerCamelCase     ,decoder=_lowerCamelCase )

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


	@staticmethod
	def 						_UpperCAmelCase     (_lowerCamelCase     ,_lowerCamelCase )							->							Optional[Any]:







				'''simple docstring'''





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


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







				'''simple docstring'''





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

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

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


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







				'''simple docstring'''





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

				__lowercase										=							processor.batch_decode(_lowerCamelCase     ,output_word_offsets=_lowerCamelCase )

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

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


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







				'''simple docstring'''





				import torch

				__lowercase										=							load_dataset('''common_voice'''     ,'''en'''     ,split='''train'''     ,streaming=_lowerCamelCase )
				__lowercase										=							ds.cast_column('''audio'''     ,datasets.Audio(sampling_rate=16000 ) )
				__lowercase										=							iter(_lowerCamelCase )
				__lowercase										=							next(_lowerCamelCase )

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

				with torch.no_grad():
							__lowercase										=							model(_lowerCamelCase ).logits.cpu().numpy()

				__lowercase										=							processor.decode(logits[0]     ,output_word_offsets=_lowerCamelCase )

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

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

				# output words
				self.assertEqual(''' '''.join(self.get_from_offsets(_lowerCamelCase     ,'''word''' ) )     ,_lowerCamelCase )
				self.assertEqual(''' '''.join(self.get_from_offsets(_lowerCamelCase     ,'''word''' ) )     ,output.text )

				# output times
				__lowercase										=							torch.tensor(self.get_from_offsets(_lowerCamelCase     ,'''start_time''' ) )
				__lowercase										=							torch.tensor(self.get_from_offsets(_lowerCamelCase     ,'''end_time''' ) )

				# fmt: off
				__lowercase										=							torch.tensor([1.4_1_9_9, 1.6_5_9_9, 2.2_5_9_9, 3.0, 3.2_4, 3.5_9_9_9, 3.7_9_9_9, 4.0_9_9_9, 4.2_6, 4.9_4, 5.2_8, 5.6_5_9_9, 5.7_8, 5.9_4, 6.3_2, 6.5_3_9_9, 6.6_5_9_9] )
				__lowercase										=							torch.tensor([1.5_3_9_9, 1.8_9_9_9, 2.9, 3.1_6, 3.5_3_9_9, 3.7_2, 4.0_1_9_9, 4.1_7_9_9, 4.7_6, 5.1_5_9_9, 5.5_5_9_9, 5.6_9_9_9, 5.8_6, 6.1_9_9_9, 6.3_8, 6.6_1_9_9, 6.9_4] )
				# fmt: on

				self.assertTrue(torch.allclose(_lowerCamelCase     ,_lowerCamelCase     ,atol=0.0_1 ) )
				self.assertTrue(torch.allclose(_lowerCamelCase     ,_lowerCamelCase     ,atol=0.0_1 ) )





