






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






      '''simple docstring'''

      def _SCREAMING_SNAKE_CASE				(		self  :						List[str]):




          '''simple docstring'''


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

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

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

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

          # load decoder from hub
          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	'''hf-internal-testing/ngram-beam-search-decoder'''
      def _SCREAMING_SNAKE_CASE				(		self  :						List[Any]		,			**lowercase_  :						Dict):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						str							=	self.add_kwargs_tokens_map.copy()
          kwargs.update(lowercase_)
          return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname		,			**lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						Optional[Any]		,			**lowercase_  :						Union[str, Any]):




          '''simple docstring'''


          return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname		,			**lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						List[Any]		,			**lowercase_  :						Any):




          '''simple docstring'''


          return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name		,			**lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						Optional[int]):




          '''simple docstring'''


          shutil.rmtree(self.tmpdirname)
      def _SCREAMING_SNAKE_CASE				(		self  :						Dict):




          '''simple docstring'''


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

          SCREAMING_SNAKE_CASE_   :						Tuple							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

          processor.save_pretrained(self.tmpdirname)
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname)

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

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

          # 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		,			lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						Union[str, Any]):




          '''simple docstring'''


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




          '''simple docstring'''


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




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	self.get_feature_extractor()
          SCREAMING_SNAKE_CASE_   :						Tuple							=	self.get_tokenizer()
          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	self.get_decoder()

          SCREAMING_SNAKE_CASE_   :						Dict							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

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

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




          '''simple docstring'''


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

          SCREAMING_SNAKE_CASE_   :						int							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	'''This is a test string'''

          SCREAMING_SNAKE_CASE_   :						List[str]							=	processor(text=lowercase_)

          SCREAMING_SNAKE_CASE_   :						str							=	tokenizer(lowercase_)

          for key in encoded_tok.keys():
              self.assertListEqual(encoded_tok[key]		,			encoded_processor[key])
      def _SCREAMING_SNAKE_CASE				(		self  :						int		,			lowercase_  :						List[Any]=(2, 10, 16)		,			lowercase_  :						Any=77):




          '''simple docstring'''


          np.random.seed(lowercase_)
          return np.random.rand(*lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						Optional[Any]):




          '''simple docstring'''


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

          SCREAMING_SNAKE_CASE_   :						Dict							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

          SCREAMING_SNAKE_CASE_   :						int							=	self._get_dummy_logits(shape=(10, 16)		,			seed=13)

          SCREAMING_SNAKE_CASE_   :						Dict							=	processor.decode(lowercase_)

          SCREAMING_SNAKE_CASE_   :						str							=	decoder.decode_beams(lowercase_)[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  :						Optional[int]		,			lowercase_  :						List[Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	self.get_feature_extractor()
          SCREAMING_SNAKE_CASE_   :						List[str]							=	self.get_tokenizer()
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	self.get_decoder()

          SCREAMING_SNAKE_CASE_   :						Dict							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

          SCREAMING_SNAKE_CASE_   :						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:
              SCREAMING_SNAKE_CASE_   :						int							=	processor.batch_decode(lowercase_)
          else:
              with get_context(lowercase_).Pool() as pool:
                  SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	processor.batch_decode(lowercase_		,			lowercase_)

          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	list(lowercase_)

          with get_context('''fork''').Pool() as p:
              SCREAMING_SNAKE_CASE_   :						Optional[int]							=	decoder.decode_beams_batch(lowercase_		,			lowercase_)

          SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_   :						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(lowercase_		,			decoded_processor.text)
          self.assertListEqual(['''<s> <s> </s>''', '''<s> <s> <s>''']		,			decoded_processor.text)
          self.assertListEqual(lowercase_		,			decoded_processor.logit_score)
          self.assertListEqual(lowercase_		,			decoded_processor.lm_score)
      def _SCREAMING_SNAKE_CASE				(		self  :						int):




          '''simple docstring'''


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

          SCREAMING_SNAKE_CASE_   :						str							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

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

          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	15
          SCREAMING_SNAKE_CASE_   :						str							=	-20.0
          SCREAMING_SNAKE_CASE_   :						int							=	-4.0

          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	processor.batch_decode(
              lowercase_		,			beam_width=lowercase_		,			beam_prune_logp=lowercase_		,			token_min_logp=lowercase_		,			)
          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	decoded_processor_out.text

          SCREAMING_SNAKE_CASE_   :						List[str]							=	list(lowercase_)

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

          SCREAMING_SNAKE_CASE_   :						List[str]							=	[d[0][0] for d in decoded_decoder_out]
          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	[d[0][2] for d in decoded_decoder_out]
          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	[d[0][3] for d in decoded_decoder_out]

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

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

          self.assertTrue(np.array_equal(lowercase_		,			decoded_processor_out.lm_score))
          self.assertTrue(np.allclose([-15.5_54, -13.94_74]		,			lowercase_		,			atol=1e-3))
      def _SCREAMING_SNAKE_CASE				(		self  :						Dict):




          '''simple docstring'''


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

          SCREAMING_SNAKE_CASE_   :						Any							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

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

          SCREAMING_SNAKE_CASE_   :						List[str]							=	2.0
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	5.0
          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	-20.0
          SCREAMING_SNAKE_CASE_   :						str							=	True

          SCREAMING_SNAKE_CASE_   :						str							=	processor.batch_decode(
              lowercase_		,			alpha=lowercase_		,			beta=lowercase_		,			unk_score_offset=lowercase_		,			lm_score_boundary=lowercase_		,			)
          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	decoded_processor_out.text

          SCREAMING_SNAKE_CASE_   :						int							=	list(lowercase_)
          decoder.reset_params(
              alpha=lowercase_		,			beta=lowercase_		,			unk_score_offset=lowercase_		,			lm_score_boundary=lowercase_		,			)

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

          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	[d[0][0] for d in decoded_decoder_out]

          self.assertListEqual(lowercase_		,			lowercase_)
          self.assertListEqual(['''<s> </s> <s> </s> </s>''', '''</s> </s> <s> </s> </s>''']		,			lowercase_)
          SCREAMING_SNAKE_CASE_   :						int							=	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		,			lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						Tuple):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Any							=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''')

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

          SCREAMING_SNAKE_CASE_   :						Dict							=	os.listdir(lowercase_)
          SCREAMING_SNAKE_CASE_   :						Optional[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(lowercase_		,			lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						int):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	snapshot_download('''hf-internal-testing/processor_with_lm''')

          SCREAMING_SNAKE_CASE_   :						List[str]							=	WavaVecaProcessorWithLM.from_pretrained(lowercase_)

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

          SCREAMING_SNAKE_CASE_   :						Tuple							=	os.listdir(lowercase_)
          SCREAMING_SNAKE_CASE_   :						List[str]							=	os.listdir(lowercase_)

          local_decoder_files.sort()
          expected_decoder_files.sort()

          # test that both decoder form hub and local files in cache are the same
          self.assertListEqual(lowercase_		,			lowercase_)
      def _SCREAMING_SNAKE_CASE				(		self  :						Optional[Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''')
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm''')

          SCREAMING_SNAKE_CASE_   :						List[Any]							=	floats_list((3, 1000))

          SCREAMING_SNAKE_CASE_   :						List[Any]							=	processor_wavaveca(lowercase_		,			return_tensors='''np''')
          SCREAMING_SNAKE_CASE_   :						int							=	processor_auto(lowercase_		,			return_tensors='''np''')

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

          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	self._get_dummy_logits()

          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	processor_wavaveca.batch_decode(lowercase_)
          SCREAMING_SNAKE_CASE_   :						Tuple							=	processor_auto.batch_decode(lowercase_)

          self.assertListEqual(decoded_wavaveca.text		,			decoded_auto.text)
      def _SCREAMING_SNAKE_CASE				(		self  :						int):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						Dict							=	self.get_feature_extractor()
          SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	self.get_tokenizer()
          SCREAMING_SNAKE_CASE_   :						List[str]							=	self.get_decoder()

          SCREAMING_SNAKE_CASE_   :						Dict							=	WavaVecaProcessorWithLM(tokenizer=lowercase_		,			feature_extractor=lowercase_		,			decoder=lowercase_)

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




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						List[Any]							=	[d[key] for d in offsets]
          return retrieved_list
      def _SCREAMING_SNAKE_CASE				(		self  :						Union[str, Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						List[Any]							=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''')
          SCREAMING_SNAKE_CASE_   :						Tuple							=	self._get_dummy_logits()[0]

          SCREAMING_SNAKE_CASE_   :						List[str]							=	processor.decode(lowercase_		,			output_word_offsets=lowercase_)
          # 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(lowercase_		,			lowercase_))

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




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_   :						int							=	WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm''')
          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	self._get_dummy_logits()

          SCREAMING_SNAKE_CASE_   :						List[Any]							=	processor.batch_decode(lowercase_		,			output_word_offsets=lowercase_)

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

          self.assertListEqual(
              [''' '''.join(self.get_from_offsets(lowercase_		,			'''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  :						Optional[Any]):




          '''simple docstring'''


          import torch

          SCREAMING_SNAKE_CASE_   :						List[Any]							=	load_dataset('''common_voice'''		,			'''en'''		,			split='''train'''		,			streaming=lowercase_)
          SCREAMING_SNAKE_CASE_   :						str							=	ds.cast_column('''audio'''		,			datasets.Audio(sampling_rate=16000))
          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	iter(lowercase_)
          SCREAMING_SNAKE_CASE_   :						Union[str, Any]							=	next(lowercase_)

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

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

          SCREAMING_SNAKE_CASE_   :						Optional[int]							=	processor.decode(logits[0]		,			output_word_offsets=lowercase_)

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

          SCREAMING_SNAKE_CASE_   :						List[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(lowercase_		,			'''word'''))		,			lowercase_)
          self.assertEqual(''' '''.join(self.get_from_offsets(lowercase_		,			'''word'''))		,			output.text)

          # output times
          SCREAMING_SNAKE_CASE_   :						List[Any]							=	torch.tensor(self.get_from_offsets(lowercase_		,			'''start_time'''))
          SCREAMING_SNAKE_CASE_   :						List[str]							=	torch.tensor(self.get_from_offsets(lowercase_		,			'''end_time'''))

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

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










"""simple docstring"""






def       _A    (__a = 1_00							)				-> int:
    """simple docstring"""






    SCREAMING_SNAKE_CASE_   :						int							=	(n * (n + 1) // 2) ** 2
    SCREAMING_SNAKE_CASE_   :						Optional[Any]							=	n * (n + 1) * (2 * n + 1) // 6
    return sum_cubes - sum_squares


if __name__ == "__main__":
  print(f'''{solution() = }''')


