






"""simple docstring"""





import logging
import os
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

from filelock import FileLock

from transformers import PreTrainedTokenizer, is_tf_available, is_torch_available


__lowerCamelCase							=  logging.getLogger(__name__)



@dataclass
class       __A		:
					UpperCAmelCase__								=				42
					UpperCAmelCase__								=				42
					UpperCAmelCase__								=				42



@dataclass
class       __A		:
					UpperCAmelCase__								=				42
					UpperCAmelCase__								=				42
					UpperCAmelCase__								=				None
					UpperCAmelCase__								=				None



class       __A		(   SCREAMING_SNAKE_CASE_			):
					UpperCAmelCase__								=				"train"
					UpperCAmelCase__								=				"dev"
					UpperCAmelCase__								=				"test"



class       __A		:

					@staticmethod
					def   lowerCamelCase__   (			__snake_case  :      Union[str, Any]     ,		__snake_case  :      Union[Split, str]					)			->							List[InputExample]:
											raise NotImplementedError

					@staticmethod
					def   lowerCamelCase__   (			__snake_case  :      str					)			->							List[str]:
											raise NotImplementedError






					@staticmethod
					def   lowerCamelCase__   (			__snake_case  :      List[InputExample]     ,		__snake_case  :      List[str]     ,		__snake_case  :      int     ,		__snake_case  :      PreTrainedTokenizer     ,		__snake_case  :      int=False     ,		__snake_case  :      Union[str, Any]="[CLS]"     ,		__snake_case  :      str=1     ,		__snake_case  :      Dict="[SEP]"     ,		__snake_case  :      List[Any]=False     ,		__snake_case  :      Dict=False     ,		__snake_case  :      Dict=0     ,		__snake_case  :      Dict=0     ,		__snake_case  :      Optional[int]=-1_0_0     ,		__snake_case  :      Optional[int]=0     ,		__snake_case  :      str=True     ,		)			->							List[InputFeatures]:
											__magic_name__:	str							=       {label: i for i, label in enumerate(__snake_case					)}

											__magic_name__:	Optional[Any]							=       []
											for ex_index, example in enumerate(__snake_case					):
																	if ex_index % 1_0_0_0_0 == 0:
																							logger.info("""Writing example %d of %d"""     ,		__snake_case     ,		len(__snake_case					)					)

																	__magic_name__:	int							=       []
																	__magic_name__:	List[Any]							=       []
																	for word, label in zip(example.words     ,		example.labels					):
																							__magic_name__:	Optional[Any]							=       tokenizer.tokenize(__snake_case					)

																							# bert-base-multilingual-cased sometimes output "nothing ([]) when calling tokenize with just a space.
																							if len(__snake_case					) > 0:
																													tokens.extend(__snake_case					)
																													# Use the real label id for the first token of the word, and padding ids for the remaining tokens
																													label_ids.extend([label_map[label]] + [pad_token_label_id] * (len(__snake_case					) - 1)					)

            # Account for [CLS] and [SEP] with "- 2" and with "- 3" for RoBERTa.
																	__magic_name__:	int							=       tokenizer.num_special_tokens_to_add()
																	if len(__snake_case					) > max_seq_length - special_tokens_count:
																							__magic_name__:	List[str]							=       tokens[: (max_seq_length - special_tokens_count)]
																							__magic_name__:	Optional[int]							=       label_ids[: (max_seq_length - special_tokens_count)]

																	# The convention in BERT is:
																	# (a) For sequence pairs:
																	#  tokens:   [CLS] is this jack ##son ##ville ? [SEP] no it is not . [SEP]
																	#  type_ids:   0   0  0    0    0     0       0   0   1  1  1  1   1   1
																	# (b) For single sequences:
																	#  tokens:   [CLS] the dog is hairy . [SEP]
																	#  type_ids:   0   0   0   0  0     0   0
																	#
																	# Where "type_ids" are used to indicate whether this is the first
																	# sequence or the second sequence. The embedding vectors for `type=0` and
																	# `type=1` were learned during pre-training and are added to the wordpiece
																	# embedding vector (and position vector). This is not *strictly* necessary
																	# since the [SEP] token unambiguously separates the sequences, but it makes
																	# it easier for the model to learn the concept of sequences.
																	#
																	# For classification tasks, the first vector (corresponding to [CLS]) is
																	# used as the "sentence vector". Note that this only makes sense because
																	# the entire model is fine-tuned.
																	tokens += [sep_token]
																	label_ids += [pad_token_label_id]
																	if sep_token_extra:
																							# roberta uses an extra separator b/w pairs of sentences
																							tokens += [sep_token]
																							label_ids += [pad_token_label_id]
																	__magic_name__:	Union[str, Any]							=       [sequence_a_segment_id] * len(__snake_case					)

																	if cls_token_at_end:
																							tokens += [cls_token]
																							label_ids += [pad_token_label_id]
																							segment_ids += [cls_token_segment_id]
																	else:
																							__magic_name__:	Union[str, Any]							=       [cls_token] + tokens
																							__magic_name__:	List[str]							=       [pad_token_label_id] + label_ids
																							__magic_name__:	Dict							=       [cls_token_segment_id] + segment_ids

																	__magic_name__:	str							=       tokenizer.convert_tokens_to_ids(__snake_case					)

																	# The mask has 1 for real tokens and 0 for padding tokens. Only real
																	# tokens are attended to.
																	__magic_name__:	int							=       [1 if mask_padding_with_zero else 0] * len(__snake_case					)

																	# Zero-pad up to the sequence length.
																	__magic_name__:	Union[str, Any]							=       max_seq_length - len(__snake_case					)
																	if pad_on_left:
																							__magic_name__:	Union[str, Any]							=       ([pad_token] * padding_length) + input_ids
																							__magic_name__:	Dict							=       ([0 if mask_padding_with_zero else 1] * padding_length) + input_mask
																							__magic_name__:	Dict							=       ([pad_token_segment_id] * padding_length) + segment_ids
																							__magic_name__:	Optional[int]							=       ([pad_token_label_id] * padding_length) + label_ids
																	else:
																							input_ids += [pad_token] * padding_length
																							input_mask += [0 if mask_padding_with_zero else 1] * padding_length
																							segment_ids += [pad_token_segment_id] * padding_length
																							label_ids += [pad_token_label_id] * padding_length

																	assert len(__snake_case					) == max_seq_length
																	assert len(__snake_case					) == max_seq_length
																	assert len(__snake_case					) == max_seq_length
																	assert len(__snake_case					) == max_seq_length

																	if ex_index < 5:
																							logger.info("""*** Example ***"""					)
																							logger.info("""guid: %s"""     ,		example.guid					)
																							logger.info("""tokens: %s"""     ,		""" """.join([str(__snake_case					) for x in tokens]					)					)
																							logger.info("""input_ids: %s"""     ,		""" """.join([str(__snake_case					) for x in input_ids]					)					)
																							logger.info("""input_mask: %s"""     ,		""" """.join([str(__snake_case					) for x in input_mask]					)					)
																							logger.info("""segment_ids: %s"""     ,		""" """.join([str(__snake_case					) for x in segment_ids]					)					)
																							logger.info("""label_ids: %s"""     ,		""" """.join([str(__snake_case					) for x in label_ids]					)					)

																	if "token_type_ids" not in tokenizer.model_input_names:
																							__magic_name__:	Any							=       None

																	features.append(
																	    InputFeatures(
																	        input_ids=__snake_case     ,		attention_mask=__snake_case     ,		token_type_ids=__snake_case     ,		label_ids=__snake_case					)					)
											return features


if is_torch_available():
							import torch
							from torch import nn
							from torch.utils.data import Dataset



							class       __A		(   SCREAMING_SNAKE_CASE_			):
												UpperCAmelCase__								=				42
												UpperCAmelCase__								=				nn.CrossEntropyLoss().ignore_index

												def __init__(			self  :      int     ,		__snake_case  :      TokenClassificationTask     ,		__snake_case  :      str     ,		__snake_case  :      PreTrainedTokenizer     ,		__snake_case  :      List[str]     ,		__snake_case  :      str     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      Any=False     ,		__snake_case  :      Split = Split.train     ,		)			->							Optional[int]:
																		# Load data features from cache or dataset file
																		__magic_name__:	Union[str, Any]							=       os.path.join(
																		    __snake_case     ,		"""cached_{}_{}_{}""".format(mode.value     ,		tokenizer.__class__.__name__     ,		str(__snake_case					)					)     ,		)

																		# Make sure only the first process in distributed training processes the dataset,
																		# and the others will use the cache.
																		__magic_name__:	Union[str, Any]							=       cached_features_file + """.lock"""
																		with FileLock(__snake_case					):
																								if os.path.exists(__snake_case					) and not overwrite_cache:
																														logger.info(F'Loading features from cached file {cached_features_file}'					)
																														__magic_name__:	List[Any]							=       torch.load(__snake_case					)
																								else:
																														logger.info(F'Creating features from dataset file at {data_dir}'					)
																														__magic_name__:	int							=       token_classification_task.read_examples_from_file(__snake_case     ,		__snake_case					)
																														# TODO clean up all this to leverage built-in features of tokenizers
																														__magic_name__:	List[str]							=       token_classification_task.convert_examples_to_features(
																														    __snake_case     ,		__snake_case     ,		__snake_case     ,		__snake_case     ,		cls_token_at_end=bool(model_type in ["""xlnet"""]					)     ,		cls_token=tokenizer.cls_token     ,		cls_token_segment_id=2 if model_type in ["""xlnet"""] else 0     ,		sep_token=tokenizer.sep_token     ,		sep_token_extra=__snake_case     ,		pad_on_left=bool(tokenizer.padding_side == """left"""					)     ,		pad_token=tokenizer.pad_token_id     ,		pad_token_segment_id=tokenizer.pad_token_type_id     ,		pad_token_label_id=self.pad_token_label_id     ,		)
																														logger.info(F'Saving features into cached file {cached_features_file}'					)
																														torch.save(self.features     ,		__snake_case					)

												def __len__(			self  :      str					)			->							str:
																		return len(self.features					)






												def __getitem__(			self  :      Any     ,		__snake_case  :      Optional[int]					)			->							InputFeatures:
																		return self.features[i]


if is_tf_available():
							import tensorflow as tf



							class       __A		:
												UpperCAmelCase__								=				42
												UpperCAmelCase__								=				-1_0_0

												def __init__(			self  :      Tuple     ,		__snake_case  :      TokenClassificationTask     ,		__snake_case  :      str     ,		__snake_case  :      PreTrainedTokenizer     ,		__snake_case  :      List[str]     ,		__snake_case  :      str     ,		__snake_case  :      Optional[int] = None     ,		__snake_case  :      str=False     ,		__snake_case  :      Split = Split.train     ,		)			->							List[Any]:
																		__magic_name__:	Optional[int]							=       token_classification_task.read_examples_from_file(__snake_case     ,		__snake_case					)
																		# TODO clean up all this to leverage built-in features of tokenizers
																		__magic_name__:	Optional[int]							=       token_classification_task.convert_examples_to_features(
																		    __snake_case     ,		__snake_case     ,		__snake_case     ,		__snake_case     ,		cls_token_at_end=bool(model_type in ["""xlnet"""]					)     ,		cls_token=tokenizer.cls_token     ,		cls_token_segment_id=2 if model_type in ["""xlnet"""] else 0     ,		sep_token=tokenizer.sep_token     ,		sep_token_extra=__snake_case     ,		pad_on_left=bool(tokenizer.padding_side == """left"""					)     ,		pad_token=tokenizer.pad_token_id     ,		pad_token_segment_id=tokenizer.pad_token_type_id     ,		pad_token_label_id=self.pad_token_label_id     ,		)

																		def gen():
																								for ex in self.features:
																														if ex.token_type_ids is None:
																																				yield (
																																				    {"input_ids": ex.input_ids, "attention_mask": ex.attention_mask},
																																				    ex.label_ids,
																																				)
																														else:
																																				yield (
																																				    {
																																				        "input_ids": ex.input_ids,
																																				        "attention_mask": ex.attention_mask,
																																				        "token_type_ids": ex.token_type_ids,
																																				    },
																																				    ex.label_ids,
																																				)

																		if "token_type_ids" not in tokenizer.model_input_names:
																								__magic_name__:	Union[str, Any]							=       tf.data.Dataset.from_generator(
																								    __snake_case     ,		({"""input_ids""": tf.intaa, """attention_mask""": tf.intaa}, tf.intaa)     ,		(
																								        {"""input_ids""": tf.TensorShape([None]					), """attention_mask""": tf.TensorShape([None]					)},
																								        tf.TensorShape([None]					),
																								    )     ,		)
																		else:
																								__magic_name__:	str							=       tf.data.Dataset.from_generator(
																								    __snake_case     ,		({"""input_ids""": tf.intaa, """attention_mask""": tf.intaa, """token_type_ids""": tf.intaa}, tf.intaa)     ,		(
																								        {
																								            """input_ids""": tf.TensorShape([None]					),
																								            """attention_mask""": tf.TensorShape([None]					),
																								            """token_type_ids""": tf.TensorShape([None]					),
																								        },
																								        tf.TensorShape([None]					),
																								    )     ,		)

												def   lowerCamelCase__   (			self  :      Tuple					)			->							Dict:
																		__magic_name__:	Optional[Any]							=       self.dataset.apply(tf.data.experimental.assert_cardinality(len(self.features					)					)					)

																		return self.dataset

												def __len__(			self  :      Union[str, Any]					)			->							int:
																		return len(self.features					)






												def __getitem__(			self  :      Dict     ,		__snake_case  :      int					)			->							InputFeatures:
																		return self.features[i]











"""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					)					)



