






import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import CLIPTokenizer, CLIPTokenizerFast
from transformers.models.clip.tokenization_clip import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
	from PIL import Image

	from transformers import CLIPImageProcessor, CLIPProcessor

@require_vision
class 		lowerCAmelCase     ( unittest.TestCase			):

					'''simple docstring'''

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


												"""simple docstring"""

												__lowercase      :							str				=     tempfile.mkdtemp()

												# fmt: off
												__lowercase      :							str				=     ["""l""", """o""", """w""", """e""", """r""", """s""", """t""", """i""", """d""", """n""", """lo""", """l</w>""", """w</w>""", """r</w>""", """t</w>""", """low</w>""", """er</w>""", """lowest</w>""", """newer</w>""", """wider""", """<unk>""", """<|startoftext|>""", """<|endoftext|>"""]
												# fmt: on
												__lowercase      :							Optional[Any]				=     dict(zip(__a		,     range(len(__a							)							)							)							)
												__lowercase      :							List[str]				=     ["""#version: 0.2""", """l o""", """lo w</w>""", """e r</w>""", """"""]
												__lowercase      :							Optional[Any]				=     {"""unk_token""": """<unk>"""}

												__lowercase      :							List[Any]				=     os.path.join(self.tmpdirname		,     VOCAB_FILES_NAMES["""vocab_file"""]							)
												__lowercase      :							Tuple				=     os.path.join(self.tmpdirname		,     VOCAB_FILES_NAMES["""merges_file"""]							)
												with open(self.vocab_file		,     """w"""		,     encoding="""utf-8"""							) as fp:
																			fp.write(json.dumps(__a							) + """\n"""							)
												with open(self.merges_file		,     """w"""		,     encoding="""utf-8"""							) as fp:
																			fp.write("""\n""".join(__a							)							)

												__lowercase      :							List[str]				=     {
												    """do_resize""": True,
												    """size""": 20,
												    """do_center_crop""": True,
												    """crop_size""": 18,
												    """do_normalize""": True,
												    """image_mean""": [0.48145466, 0.4578275, 0.40821073],
												    """image_std""": [0.26862954, 0.26130258, 0.27577711],
												}
												__lowercase      :							Optional[Any]				=     os.path.join(self.tmpdirname		,     __a							)
												with open(self.image_processor_file		,     """w"""		,     encoding="""utf-8"""							) as fp:
																			json.dump(__a		,     __a							)



					def       lowerCAmelCase				(  self				:	Any		,     **__a				:	List[str]							)			->  int:


												"""simple docstring"""

												return CLIPTokenizer.from_pretrained(self.tmpdirname		,     **__a							)



					def       lowerCAmelCase				(  self				:	Tuple		,     **__a				:	Dict							)			->  int:


												"""simple docstring"""

												return CLIPTokenizerFast.from_pretrained(self.tmpdirname		,     **__a							)



					def       lowerCAmelCase				(  self				:	Union[str, Any]		,     **__a				:	List[str]							)			->  Union[str, Any]:


												"""simple docstring"""

												return CLIPImageProcessor.from_pretrained(self.tmpdirname		,     **__a							)



					def       lowerCAmelCase				(  self				:	Tuple							)			->  Dict:


												"""simple docstring"""

												shutil.rmtree(self.tmpdirname							)



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


												"""simple docstring"""

												__lowercase      :							Optional[int]				=     [np.random.randint(255		,     size=(3, 30, 400)		,     dtype=np.uinta							)]

												__lowercase      :							Dict				=     [Image.fromarray(np.moveaxis(__a		,     0		,     -1							)							) for x in image_inputs]

												return image_inputs



					def       lowerCAmelCase				(  self				:	Dict							)			->  List[Any]:


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     self.get_tokenizer()
												__lowercase      :							str				=     self.get_rust_tokenizer()
												__lowercase      :							Tuple				=     self.get_image_processor()

												__lowercase      :							Dict				=     CLIPProcessor(tokenizer=__a		,     image_processor=__a							)
												processor_slow.save_pretrained(self.tmpdirname							)
												__lowercase      :							Dict				=     CLIPProcessor.from_pretrained(self.tmpdirname		,     use_fast=__a							)

												__lowercase      :							Optional[int]				=     CLIPProcessor(tokenizer=__a		,     image_processor=__a							)
												processor_fast.save_pretrained(self.tmpdirname							)
												__lowercase      :							Dict				=     CLIPProcessor.from_pretrained(self.tmpdirname							)

												self.assertEqual(processor_slow.tokenizer.get_vocab()		,     tokenizer_slow.get_vocab()							)
												self.assertEqual(processor_fast.tokenizer.get_vocab()		,     tokenizer_fast.get_vocab()							)
												self.assertEqual(tokenizer_slow.get_vocab()		,     tokenizer_fast.get_vocab()							)
												self.assertIsInstance(processor_slow.tokenizer		,     __a							)
												self.assertIsInstance(processor_fast.tokenizer		,     __a							)

												self.assertEqual(processor_slow.image_processor.to_json_string()		,     image_processor.to_json_string()							)
												self.assertEqual(processor_fast.image_processor.to_json_string()		,     image_processor.to_json_string()							)
												self.assertIsInstance(processor_slow.image_processor		,     __a							)
												self.assertIsInstance(processor_fast.image_processor		,     __a							)



					def       lowerCAmelCase				(  self				:	Dict							)			->  str:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     CLIPProcessor(tokenizer=self.get_tokenizer()		,     image_processor=self.get_image_processor()							)
												processor.save_pretrained(self.tmpdirname							)

												__lowercase      :							Tuple				=     self.get_tokenizer(bos_token="""(BOS)"""		,     eos_token="""(EOS)"""							)
												__lowercase      :							Optional[int]				=     self.get_image_processor(do_normalize=__a		,     padding_value=1.0							)

												__lowercase      :							Tuple				=     CLIPProcessor.from_pretrained(
												    self.tmpdirname		,     bos_token="""(BOS)"""		,     eos_token="""(EOS)"""		,     do_normalize=__a		,     padding_value=1.0							)

												self.assertEqual(processor.tokenizer.get_vocab()		,     tokenizer_add_kwargs.get_vocab()							)
												self.assertIsInstance(processor.tokenizer		,     __a							)

												self.assertEqual(processor.image_processor.to_json_string()		,     image_processor_add_kwargs.to_json_string()							)
												self.assertIsInstance(processor.image_processor		,     __a							)



					def       lowerCAmelCase				(  self				:	List[Any]							)			->  Tuple:


												"""simple docstring"""

												__lowercase      :							Any				=     self.get_image_processor()
												__lowercase      :							int				=     self.get_tokenizer()

												__lowercase      :							Dict				=     CLIPProcessor(tokenizer=__a		,     image_processor=__a							)

												__lowercase      :							List[Any]				=     self.prepare_image_inputs()

												__lowercase      :							Dict				=     image_processor(__a		,     return_tensors="""np"""							)
												__lowercase      :							Optional[int]				=     processor(images=__a		,     return_tensors="""np"""							)

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



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


												"""simple docstring"""

												__lowercase      :							List[str]				=     self.get_image_processor()
												__lowercase      :							Optional[int]				=     self.get_tokenizer()

												__lowercase      :							List[str]				=     CLIPProcessor(tokenizer=__a		,     image_processor=__a							)

												__lowercase      :							str				=     """lower newer"""

												__lowercase      :							Union[str, Any]				=     processor(text=__a							)

												__lowercase      :							List[str]				=     tokenizer(__a							)

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



					def       lowerCAmelCase				(  self				:	Dict							)			->  Optional[int]:


												"""simple docstring"""

												__lowercase      :							Dict				=     self.get_image_processor()
												__lowercase      :							Dict				=     self.get_tokenizer()

												__lowercase      :							Dict				=     CLIPProcessor(tokenizer=__a		,     image_processor=__a							)

												__lowercase      :							int				=     """lower newer"""
												__lowercase      :							Union[str, Any]				=     self.prepare_image_inputs()

												__lowercase      :							int				=     processor(text=__a		,     images=__a							)

												self.assertListEqual(list(inputs.keys()							)		,     ["""input_ids""", """attention_mask""", """pixel_values"""]							)

												# test if it raises when no input is passed
												with pytest.raises(__a							):
																			processor()



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


												"""simple docstring"""

												__lowercase      :							str				=     self.get_image_processor()
												__lowercase      :							Optional[Any]				=     self.get_tokenizer()

												__lowercase      :							List[str]				=     CLIPProcessor(tokenizer=__a		,     image_processor=__a							)

												__lowercase      :							Dict				=     [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

												__lowercase      :							Any				=     processor.batch_decode(__a							)
												__lowercase      :							str				=     tokenizer.batch_decode(__a							)

												self.assertListEqual(__a		,     __a							)



					def       lowerCAmelCase				(  self				:	Optional[Any]							)			->  int:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     self.get_image_processor()
												__lowercase      :							Any				=     self.get_tokenizer()

												__lowercase      :							Tuple				=     CLIPProcessor(tokenizer=__a		,     image_processor=__a							)

												__lowercase      :							Union[str, Any]				=     """lower newer"""
												__lowercase      :							Dict				=     self.prepare_image_inputs()

												__lowercase      :							Any				=     processor(text=__a		,     images=__a							)

												self.assertListEqual(list(inputs.keys()							)		,     processor.model_input_names							)







import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
	from huggingface_hub import snapshot_download
	from pyctcdecode import BeamSearchDecoderCTC

	from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
	from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
	from transformers import WavaVecaForCTC

@require_pyctcdecode
class 		lowerCAmelCase     ( unittest.TestCase			):

					'''simple docstring'''

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


												"""simple docstring"""

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

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

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

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

												# load decoder from hub
												__lowercase      :							Optional[int]				=     """hf-internal-testing/ngram-beam-search-decoder"""



					def       lowerCAmelCase				(  self				:	Optional[Any]		,     **__a				:	Dict							)			->  Tuple:


												"""simple docstring"""

												__lowercase      :							Union[str, Any]				=     self.add_kwargs_tokens_map.copy()
												kwargs.update(__a							)
												return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname		,     **__a							)



					def       lowerCAmelCase				(  self				:	str		,     **__a				:	int							)			->  Tuple:


												"""simple docstring"""

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



					def       lowerCAmelCase				(  self				:	Union[str, Any]		,     **__a				:	Union[str, Any]							)			->  List[Any]:


												"""simple docstring"""

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



					def       lowerCAmelCase				(  self				:	int							)			->  Tuple:


												"""simple docstring"""

												shutil.rmtree(self.tmpdirname							)



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


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     self.get_tokenizer()
												__lowercase      :							Any				=     self.get_feature_extractor()
												__lowercase      :							str				=     self.get_decoder()

												__lowercase      :							Tuple				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

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

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

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

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



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


												"""simple docstring"""

												__lowercase      :							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
												__lowercase      :							str				=     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				:	Any							)			->  List[str]:


												"""simple docstring"""

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



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


												"""simple docstring"""

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

												__lowercase      :							int				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

												__lowercase      :							Optional[int]				=     floats_list((3, 1000)							)

												__lowercase      :							List[Any]				=     feature_extractor(__a		,     return_tensors="""np"""							)
												__lowercase      :							List[str]				=     processor(__a		,     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]							)			->  Optional[Any]:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     self.get_feature_extractor()
												__lowercase      :							int				=     self.get_tokenizer()
												__lowercase      :							Dict				=     self.get_decoder()

												__lowercase      :							Union[str, Any]				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

												__lowercase      :							Dict				=     """This is a test string"""

												__lowercase      :							Any				=     processor(text=__a							)

												__lowercase      :							Dict				=     tokenizer(__a							)

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



					def       lowerCAmelCase				(  self				:	str		,     __a				:	Tuple=(2, 10, 16)		,     __a				:	int=77							)			->  Optional[Any]:


												"""simple docstring"""

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



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


												"""simple docstring"""

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

												__lowercase      :							int				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

												__lowercase      :							List[str]				=     self._get_dummy_logits(shape=(10, 16)		,     seed=13							)

												__lowercase      :							Optional[Any]				=     processor.decode(__a							)

												__lowercase      :							Any				=     decoder.decode_beams(__a							)[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				:	List[str]		,     __a				:	Dict							)			->  List[Any]:


												"""simple docstring"""

												__lowercase      :							str				=     self.get_feature_extractor()
												__lowercase      :							Dict				=     self.get_tokenizer()
												__lowercase      :							Optional[int]				=     self.get_decoder()

												__lowercase      :							Any				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

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

												# note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
												#       otherwise, the LM won't be available to the pool's sub-processes.
												# manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
												if pool_context is None:
																			__lowercase      :							Union[str, Any]				=     processor.batch_decode(__a							)
												else:
																			with get_context(__a							).Pool() as pool:
																										__lowercase      :							Optional[Any]				=     processor.batch_decode(__a		,     __a							)

												__lowercase      :							Union[str, Any]				=     list(__a							)

												with get_context("""fork"""							).Pool() as p:
																			__lowercase      :							Optional[Any]				=     decoder.decode_beams_batch(__a		,     __a							)

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

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



					def       lowerCAmelCase				(  self				:	Any							)			->  Optional[int]:


												"""simple docstring"""

												__lowercase      :							int				=     self.get_feature_extractor()
												__lowercase      :							Dict				=     self.get_tokenizer()
												__lowercase      :							List[str]				=     self.get_decoder()

												__lowercase      :							int				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

												__lowercase      :							Dict				=     self._get_dummy_logits()

												__lowercase      :							Tuple				=     15
												__lowercase      :							Tuple				=     -20.0
												__lowercase      :							Dict				=     -4.0

												__lowercase      :							Dict				=     processor.batch_decode(
												    __a		,     beam_width=__a		,     beam_prune_logp=__a		,     token_min_logp=__a		,     )
												__lowercase      :							Tuple				=     decoded_processor_out.text

												__lowercase      :							List[Any]				=     list(__a							)

												with get_context("""fork"""							).Pool() as pool:
																			__lowercase      :							Any				=     decoder.decode_beams_batch(
																			    __a		,     __a		,     beam_width=__a		,     beam_prune_logp=__a		,     token_min_logp=__a		,     )

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

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

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

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



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


												"""simple docstring"""

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

												__lowercase      :							Dict				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

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

												__lowercase      :							Optional[int]				=     2.0
												__lowercase      :							Tuple				=     5.0
												__lowercase      :							Optional[Any]				=     -20.0
												__lowercase      :							Tuple				=     True

												__lowercase      :							Union[str, Any]				=     processor.batch_decode(
												    __a		,     alpha=__a		,     beta=__a		,     unk_score_offset=__a		,     lm_score_boundary=__a		,     )
												__lowercase      :							Any				=     decoded_processor_out.text

												__lowercase      :							List[Any]				=     list(__a							)
												decoder.reset_params(
												    alpha=__a		,     beta=__a		,     unk_score_offset=__a		,     lm_score_boundary=__a		,     )

												with get_context("""fork"""							).Pool() as pool:
																			__lowercase      :							Tuple				=     decoder.decode_beams_batch(
																			    __a		,     __a		,     )

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

												self.assertListEqual(__a		,     __a							)
												self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]		,     __a							)
												__lowercase      :							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		,     __a							)



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


												"""simple docstring"""

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

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

												__lowercase      :							int				=     os.listdir(__a							)
												__lowercase      :							Optional[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(__a		,     __a							)



					def       lowerCAmelCase				(  self				:	str							)			->  Optional[int]:


												"""simple docstring"""

												__lowercase      :							List[str]				=     snapshot_download("""hf-internal-testing/processor_with_lm"""							)

												__lowercase      :							Union[str, Any]				=     WavaVecaProcessorWithLM.from_pretrained(__a							)

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

												__lowercase      :							Dict				=     os.listdir(__a							)
												__lowercase      :							List[Any]				=     os.listdir(__a							)

												local_decoder_files.sort()
												expected_decoder_files.sort()

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



					def       lowerCAmelCase				(  self				:	Tuple							)			->  int:


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""							)
												__lowercase      :							Dict				=     AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""							)

												__lowercase      :							Any				=     floats_list((3, 1000)							)

												__lowercase      :							List[str]				=     processor_wavaveca(__a		,     return_tensors="""np"""							)
												__lowercase      :							List[Any]				=     processor_auto(__a		,     return_tensors="""np"""							)

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

												__lowercase      :							List[str]				=     self._get_dummy_logits()

												__lowercase      :							List[str]				=     processor_wavaveca.batch_decode(__a							)
												__lowercase      :							Optional[int]				=     processor_auto.batch_decode(__a							)

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



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


												"""simple docstring"""

												__lowercase      :							Any				=     self.get_feature_extractor()
												__lowercase      :							Union[str, Any]				=     self.get_tokenizer()
												__lowercase      :							Dict				=     self.get_decoder()

												__lowercase      :							Optional[Any]				=     WavaVecaProcessorWithLM(tokenizer=__a		,     feature_extractor=__a		,     decoder=__a							)

												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				(  __a				:	Union[str, Any]		,     __a				:	List[Any]							)			->  Dict:


												"""simple docstring"""

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



					def       lowerCAmelCase				(  self				:	Optional[int]							)			->  str:


												"""simple docstring"""

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

												__lowercase      :							Dict				=     processor.decode(__a		,     output_word_offsets=__a							)
												# 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(__a		,     __a							)							)

												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				:	List[str]							)			->  Tuple:


												"""simple docstring"""

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

												__lowercase      :							Dict				=     processor.batch_decode(__a		,     output_word_offsets=__a							)

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

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



					@slow
					@require_torch
					@require_torchaudio
					def       lowerCAmelCase				(  self				:	Dict							)			->  int:


												"""simple docstring"""

												import torch

												__lowercase      :							Any				=     load_dataset("""common_voice"""		,     """en"""		,     split="""train"""		,     streaming=__a							)
												__lowercase      :							str				=     ds.cast_column("""audio"""		,     datasets.Audio(sampling_rate=16000							)							)
												__lowercase      :							Tuple				=     iter(__a							)
												__lowercase      :							Union[str, Any]				=     next(__a							)

												__lowercase      :							int				=     AutoProcessor.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""							)
												__lowercase      :							int				=     WavaVecaForCTC.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""							)

												# compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
												__lowercase      :							Union[str, Any]				=     processor(sample["""audio"""]["""array"""]		,     return_tensors="""pt"""							).input_values

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

												__lowercase      :							Tuple				=     processor.decode(logits[0]		,     output_word_offsets=__a							)

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

												__lowercase      :							str				=     """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(__a		,     """word"""							)							)		,     __a							)
												self.assertEqual(""" """.join(self.get_from_offsets(__a		,     """word"""							)							)		,     output.text							)

												# output times
												__lowercase      :							Tuple				=     torch.tensor(self.get_from_offsets(__a		,     """start_time"""							)							)
												__lowercase      :							Dict				=     torch.tensor(self.get_from_offsets(__a		,     """end_time"""							)							)

												# fmt: off
												__lowercase      :							List[Any]				=     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]							)
												__lowercase      :							Optional[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(__a		,     __a		,     atol=0.01							)							)
												self.assertTrue(torch.allclose(__a		,     __a		,     atol=0.01							)							)