






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





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

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

				UpperCAmelCase_  :				Optional[Any]      =      tempfile.mkdtemp()
				UpperCAmelCase_  :				int      =      os.path.join(self.tmpdirname			,			VOCAB_FILES_NAMES["vocab_file"]		)
				UpperCAmelCase_  :				List[str]      =      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
				UpperCAmelCase_  :				Any      =      "hf-internal-testing/ngram-beam-search-decoder"





	def   _SCREAMING_SNAKE_CASE   (   self      :				Tuple			,			**lowerCAmelCase_      :				Union[str, Any]		)		->    Union[str, Any]:
				UpperCAmelCase_  :				str      =      self.add_kwargs_tokens_map.copy()
				kwargs.update(lowerCAmelCase_		)
				return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname			,			**lowerCAmelCase_		)





	def   _SCREAMING_SNAKE_CASE   (   self      :				Optional[Any]			,			**lowerCAmelCase_      :				Union[str, Any]		)		->    Union[str, Any]:
				return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname			,			**lowerCAmelCase_		)





	def   _SCREAMING_SNAKE_CASE   (   self      :				Optional[Any]			,			**lowerCAmelCase_      :				Union[str, Any]		)		->    List[Any]:
				return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name			,			**lowerCAmelCase_		)





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





	def   _SCREAMING_SNAKE_CASE   (   self      :				Tuple		)		->    List[str]:
				UpperCAmelCase_  :				Dict      =      self.get_tokenizer()
				UpperCAmelCase_  :				str      =      self.get_feature_extractor()
				UpperCAmelCase_  :				List[str]      =      self.get_decoder()

				UpperCAmelCase_  :				Any      =      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,			feature_extractor=lowerCAmelCase_			,			decoder=lowerCAmelCase_		)

				processor.save_pretrained(self.tmpdirname		)
				UpperCAmelCase_  :				List[Any]      =      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   _SCREAMING_SNAKE_CASE   (   self      :				List[Any]		)		->    Union[str, Any]:
				UpperCAmelCase_  :				Optional[int]      =      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
				UpperCAmelCase_  :				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   _SCREAMING_SNAKE_CASE   (   self      :				str		)		->    Union[str, Any]:
				UpperCAmelCase_  :				Optional[int]      =      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   _SCREAMING_SNAKE_CASE   (   self      :				Union[str, Any]		)		->    Dict:
				UpperCAmelCase_  :				Union[str, Any]      =      self.get_feature_extractor()
				UpperCAmelCase_  :				str      =      self.get_tokenizer()
				UpperCAmelCase_  :				Optional[int]      =      self.get_decoder()

				UpperCAmelCase_  :				int      =      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,			feature_extractor=lowerCAmelCase_			,			decoder=lowerCAmelCase_		)

				UpperCAmelCase_  :				Optional[int]      =      floats_list((3, 1_000)		)

				UpperCAmelCase_  :				int      =      feature_extractor(lowerCAmelCase_			,			return_tensors="np"		)
				UpperCAmelCase_  :				List[str]      =      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   _SCREAMING_SNAKE_CASE   (   self      :				List[str]		)		->    Optional[Any]:
				UpperCAmelCase_  :				Union[str, Any]      =      self.get_feature_extractor()
				UpperCAmelCase_  :				Tuple      =      self.get_tokenizer()
				UpperCAmelCase_  :				List[str]      =      self.get_decoder()

				UpperCAmelCase_  :				Union[str, Any]      =      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,			feature_extractor=lowerCAmelCase_			,			decoder=lowerCAmelCase_		)

				UpperCAmelCase_  :				Any      =      "This is a test string"

				UpperCAmelCase_  :				Union[str, Any]      =      processor(text=lowerCAmelCase_		)

				UpperCAmelCase_  :				int      =      tokenizer(lowerCAmelCase_		)

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





	def   _SCREAMING_SNAKE_CASE   (   self      :				List[str]			,			lowerCAmelCase_      :				str=(2, 10, 16)			,			lowerCAmelCase_      :				Dict=77		)		->    Optional[int]:
				np.random.seed(lowerCAmelCase_		)
				return np.random.rand(*lowerCAmelCase_		)





	def   _SCREAMING_SNAKE_CASE   (   self      :				Union[str, Any]		)		->    Tuple:
				UpperCAmelCase_  :				Any      =      self.get_feature_extractor()
				UpperCAmelCase_  :				List[Any]      =      self.get_tokenizer()
				UpperCAmelCase_  :				List[Any]      =      self.get_decoder()

				UpperCAmelCase_  :				Union[str, Any]      =      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,			feature_extractor=lowerCAmelCase_			,			decoder=lowerCAmelCase_		)

				UpperCAmelCase_  :				Optional[Any]      =      self._get_dummy_logits(shape=(10, 16)			,			seed=13		)

				UpperCAmelCase_  :				Union[str, Any]      =      processor.decode(lowerCAmelCase_		)

				UpperCAmelCase_  :				Dict      =      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   _SCREAMING_SNAKE_CASE   (   self      :				Tuple			,			lowerCAmelCase_      :				Dict		)		->    List[str]:
				UpperCAmelCase_  :				int      =      self.get_feature_extractor()
				UpperCAmelCase_  :				Optional[Any]      =      self.get_tokenizer()
				UpperCAmelCase_  :				List[Any]      =      self.get_decoder()

				UpperCAmelCase_  :				Optional[Any]      =      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,			feature_extractor=lowerCAmelCase_			,			decoder=lowerCAmelCase_		)

				UpperCAmelCase_  :				Union[str, 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:
							UpperCAmelCase_  :				Tuple      =      processor.batch_decode(lowerCAmelCase_		)
				else:
							with get_context(lowerCAmelCase_		).Pool() as pool:
										UpperCAmelCase_  :				Tuple      =      processor.batch_decode(lowerCAmelCase_			,			lowerCAmelCase_		)

				UpperCAmelCase_  :				Tuple      =      list(lowerCAmelCase_		)

				with get_context("fork"		).Pool() as p:
							UpperCAmelCase_  :				str      =      decoder.decode_beams_batch(lowerCAmelCase_			,			lowerCAmelCase_		)

				UpperCAmelCase_  ,			UpperCAmelCase_  ,			UpperCAmelCase_  :				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(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   _SCREAMING_SNAKE_CASE   (   self      :				Dict		)		->    Tuple:
				UpperCAmelCase_  :				List[Any]      =      self.get_feature_extractor()
				UpperCAmelCase_  :				Union[str, Any]      =      self.get_tokenizer()
				UpperCAmelCase_  :				int      =      self.get_decoder()

				UpperCAmelCase_  :				Optional[Any]      =      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,			feature_extractor=lowerCAmelCase_			,			decoder=lowerCAmelCase_		)

				UpperCAmelCase_  :				str      =      self._get_dummy_logits()

				UpperCAmelCase_  :				Dict      =      15
				UpperCAmelCase_  :				int      =      -2_0.0
				UpperCAmelCase_  :				Any      =      -4.0

				UpperCAmelCase_  :				Dict      =      processor.batch_decode(
				    lowerCAmelCase_			,			beam_width=lowerCAmelCase_			,			beam_prune_logp=lowerCAmelCase_			,			token_min_logp=lowerCAmelCase_			,			)
				UpperCAmelCase_  :				Dict      =      decoded_processor_out.text

				UpperCAmelCase_  :				List[str]      =      list(lowerCAmelCase_		)

				with get_context("fork"		).Pool() as pool:
							UpperCAmelCase_  :				Dict      =      decoder.decode_beams_batch(
							    lowerCAmelCase_			,			lowerCAmelCase_			,			beam_width=lowerCAmelCase_			,			beam_prune_logp=lowerCAmelCase_			,			token_min_logp=lowerCAmelCase_			,			)

				UpperCAmelCase_  :				Union[str, Any]      =      [d[0][0] for d in decoded_decoder_out]
				UpperCAmelCase_  :				List[str]      =      [d[0][2] for d in decoded_decoder_out]
				UpperCAmelCase_  :				Optional[int]      =      [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   _SCREAMING_SNAKE_CASE   (   self      :				Tuple		)		->    int:
				UpperCAmelCase_  :				int      =      self.get_feature_extractor()
				UpperCAmelCase_  :				Optional[Any]      =      self.get_tokenizer()
				UpperCAmelCase_  :				Tuple      =      self.get_decoder()

				UpperCAmelCase_  :				List[str]      =      WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase_			,			feature_extractor=lowerCAmelCase_			,			decoder=lowerCAmelCase_		)

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

				UpperCAmelCase_  :				Dict      =      2.0
				UpperCAmelCase_  :				Union[str, Any]      =      5.0
				UpperCAmelCase_  :				str      =      -2_0.0
				UpperCAmelCase_  :				List[str]      =      True

				UpperCAmelCase_  :				str      =      processor.batch_decode(
				    lowerCAmelCase_			,			alpha=lowerCAmelCase_			,			beta=lowerCAmelCase_			,			unk_score_offset=lowerCAmelCase_			,			lm_score_boundary=lowerCAmelCase_			,			)
				UpperCAmelCase_  :				List[Any]      =      decoded_processor_out.text

				UpperCAmelCase_  :				str      =      list(lowerCAmelCase_		)
				decoder.reset_params(
				    alpha=lowerCAmelCase_			,			beta=lowerCAmelCase_			,			unk_score_offset=lowerCAmelCase_			,			lm_score_boundary=lowerCAmelCase_			,			)

				with get_context("fork"		).Pool() as pool:
							UpperCAmelCase_  :				Dict      =      decoder.decode_beams_batch(
							    lowerCAmelCase_			,			lowerCAmelCase_			,			)

				UpperCAmelCase_  :				Any      =      [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_		)
				UpperCAmelCase_  :				Union[str, Any]      =      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   _SCREAMING_SNAKE_CASE   (   self      :				Optional[Any]		)		->    List[Any]:
				UpperCAmelCase_  :				Tuple      =      WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)

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

				UpperCAmelCase_  :				Dict      =      os.listdir(lowerCAmelCase_		)
				UpperCAmelCase_  :				int      =      ["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   _SCREAMING_SNAKE_CASE   (   self      :				Dict		)		->    Optional[Any]:
				UpperCAmelCase_  :				Tuple      =      snapshot_download("hf-internal-testing/processor_with_lm"		)

				UpperCAmelCase_  :				int      =      WavaVecaProcessorWithLM.from_pretrained(lowerCAmelCase_		)

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

				UpperCAmelCase_  :				Any      =      os.listdir(lowerCAmelCase_		)
				UpperCAmelCase_  :				List[Any]      =      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   _SCREAMING_SNAKE_CASE   (   self      :				Union[str, Any]		)		->    Optional[Any]:
				UpperCAmelCase_  :				int      =      WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)
				UpperCAmelCase_  :				Optional[Any]      =      AutoProcessor.from_pretrained("hf-internal-testing/processor_with_lm"		)

				UpperCAmelCase_  :				List[Any]      =      floats_list((3, 1_000)		)

				UpperCAmelCase_  :				Union[str, Any]      =      processor_wavaveca(lowerCAmelCase_			,			return_tensors="np"		)
				UpperCAmelCase_  :				List[str]      =      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		)

				UpperCAmelCase_  :				Optional[int]      =      self._get_dummy_logits()

				UpperCAmelCase_  :				str      =      processor_wavaveca.batch_decode(lowerCAmelCase_		)
				UpperCAmelCase_  :				int      =      processor_auto.batch_decode(lowerCAmelCase_		)

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





	def   _SCREAMING_SNAKE_CASE   (   self      :				Any		)		->    List[str]:
				UpperCAmelCase_  :				Dict      =      self.get_feature_extractor()
				UpperCAmelCase_  :				List[Any]      =      self.get_tokenizer()
				UpperCAmelCase_  :				str      =      self.get_decoder()

				UpperCAmelCase_  :				Dict      =      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   _SCREAMING_SNAKE_CASE   (   lowerCAmelCase_      :				Optional[int]			,			lowerCAmelCase_      :				str		)		->    Optional[int]:
				UpperCAmelCase_  :				str      =      [d[key] for d in offsets]
				return retrieved_list





	def   _SCREAMING_SNAKE_CASE   (   self      :				Optional[Any]		)		->    Optional[Any]:
				UpperCAmelCase_  :				List[str]      =      WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)
				UpperCAmelCase_  :				Optional[int]      =      self._get_dummy_logits()[0]

				UpperCAmelCase_  :				List[Any]      =      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   _SCREAMING_SNAKE_CASE   (   self      :				Union[str, Any]		)		->    Optional[Any]:
				UpperCAmelCase_  :				int      =      WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"		)
				UpperCAmelCase_  :				List[str]      =      self._get_dummy_logits()

				UpperCAmelCase_  :				Optional[int]      =      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   _SCREAMING_SNAKE_CASE   (   self      :				List[str]		)		->    Optional[int]:
				import torch

				UpperCAmelCase_  :				Optional[Any]      =      load_dataset("common_voice"			,			"en"			,			split="train"			,			streaming=lowerCAmelCase_		)
				UpperCAmelCase_  :				int      =      ds.cast_column("audio"			,			datasets.Audio(sampling_rate=16_000		)		)
				UpperCAmelCase_  :				Optional[Any]      =      iter(lowerCAmelCase_		)
				UpperCAmelCase_  :				int      =      next(lowerCAmelCase_		)

				UpperCAmelCase_  :				Any      =      AutoProcessor.from_pretrained("patrickvonplaten/wav2vec2-base-100h-with-lm"		)
				UpperCAmelCase_  :				Dict      =      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
				UpperCAmelCase_  :				Union[str, Any]      =      processor(sample["audio"]["array"]			,			return_tensors="pt"		).input_values

				with torch.no_grad():
							UpperCAmelCase_  :				Union[str, Any]      =      model(lowerCAmelCase_		).logits.cpu().numpy()

				UpperCAmelCase_  :				List[str]      =      processor.decode(logits[0]			,			output_word_offsets=lowerCAmelCase_		)

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

				UpperCAmelCase_  :				List[Any]      =      "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
				UpperCAmelCase_  :				Tuple      =      torch.tensor(self.get_from_offsets(lowerCAmelCase_			,			"start_time"		)		)
				UpperCAmelCase_  :				List[str]      =      torch.tensor(self.get_from_offsets(lowerCAmelCase_			,			"end_time"		)		)

				# fmt: off
				UpperCAmelCase_  :				Optional[int]      =      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]		)
				UpperCAmelCase_  :				int      =      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		)		)











"""simple docstring"""
lowerCamelCase_       			=   {
    '''a''': '''AAAAA''',
    '''b''': '''AAAAB''',
    '''c''': '''AAABA''',
    '''d''': '''AAABB''',
    '''e''': '''AABAA''',
    '''f''': '''AABAB''',
    '''g''': '''AABBA''',
    '''h''': '''AABBB''',
    '''i''': '''ABAAA''',
    '''j''': '''BBBAA''',
    '''k''': '''ABAAB''',
    '''l''': '''ABABA''',
    '''m''': '''ABABB''',
    '''n''': '''ABBAA''',
    '''o''': '''ABBAB''',
    '''p''': '''ABBBA''',
    '''q''': '''ABBBB''',
    '''r''': '''BAAAA''',
    '''s''': '''BAAAB''',
    '''t''': '''BAABA''',
    '''u''': '''BAABB''',
    '''v''': '''BBBAB''',
    '''w''': '''BABAA''',
    '''x''': '''BABAB''',
    '''y''': '''BABBA''',
    '''z''': '''BABBB''',
    ''' ''': ''' ''',
}


lowerCamelCase_       			=   {value: key for key, value in encode_dict.items()}

def        snake_case   (     A__		):
			UpperCAmelCase_  :				Union[str, Any]      =      ""
			for letter in word.lower():
						if letter.isalpha() or letter == " ":
									encoded += encode_dict[letter]
						else:
									raise Exception("encode() accepts only letters of the alphabet and spaces"		)
			return encoded

def        snake_case   (     A__		):
			if set(A__		) - {"A", "B", " "} != set():
						raise Exception("decode() accepts only 'A', 'B' and spaces"		)
			UpperCAmelCase_  :				Dict      =      ""
			for word in coded.split():
						while len(A__		) != 0:
									decoded += decode_dict[word[:5]]
									UpperCAmelCase_  :				str      =      word[5:]
						decoded += " "
			return decoded.strip()


if __name__ == "__main__":
			from doctest import testmod

			testmod()



