





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


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





     """simple docstring"""







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

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

     lowercase__           =  tempfile.mkdtemp()
     lowercase__           =  os.path.join(self.tmpdirname      ,					VOCAB_FILES_NAMES['vocab_file']	)
     lowercase__           =  os.path.join(self.tmpdirname      ,					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__           =  'hf-internal-testing/ngram-beam-search-decoder'


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Tuple      ,					**a      :    Union[str, Any]	)->       str:





     """simple docstring"""







     lowercase__           =  self.add_kwargs_tokens_map.copy()
     kwargs.update(a	)
     return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname      ,					**a	)


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Any      ,					**a      :    Tuple	)->       Tuple:





     """simple docstring"""







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


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Dict      ,					**a      :    Dict	)->       Tuple:





     """simple docstring"""







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


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





     """simple docstring"""







     shutil.rmtree(self.tmpdirname	)


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Optional[Any]	)->       Dict:





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM(tokenizer=a      ,					feature_extractor=a      ,					decoder=a	)

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

     # tokenizer
     self.assertEqual(processor.tokenizer.get_vocab()      ,					tokenizer.get_vocab()	)
     self.assertIsInstance(processor.tokenizer      ,					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 	SCREAMING_SNAKE_CASE_							(   self      :    Union[str, Any]	)->       Tuple:





     """simple docstring"""







     lowercase__           =  WavaVecaProcessorWithLM(
         tokenizer=self.get_tokenizer()      ,					feature_extractor=self.get_feature_extractor()      ,					decoder=self.get_decoder()	)
     processor.save_pretrained(self.tmpdirname	)

     # make sure that error is thrown when decoder alphabet doesn't match
     lowercase__           =  WavaVecaProcessorWithLM.from_pretrained(
         self.tmpdirname      ,					alpha=5.0      ,					beta=3.0      ,					score_boundary=-7.0      ,					unk_score_offset=3	)

     # decoder
     self.assertEqual(processor.language_model.alpha      ,					5.0	)
     self.assertEqual(processor.language_model.beta      ,					3.0	)
     self.assertEqual(processor.language_model.score_boundary      ,					-7.0	)
     self.assertEqual(processor.language_model.unk_score_offset      ,					3	)


   def 	SCREAMING_SNAKE_CASE_							(   self      :    str	)->       str:





     """simple docstring"""







     lowercase__           =  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 	SCREAMING_SNAKE_CASE_							(   self      :    Tuple	)->       Optional[int]:





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM(tokenizer=a      ,					feature_extractor=a      ,					decoder=a	)

     lowercase__           =  floats_list((3, 1_000)	)

     lowercase__           =  feature_extractor(a      ,					return_tensors='np'	)
     lowercase__           =  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 	SCREAMING_SNAKE_CASE_							(   self      :    Dict	)->       Optional[Any]:





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM(tokenizer=a      ,					feature_extractor=a      ,					decoder=a	)

     lowercase__           =  'This is a test string'

     lowercase__           =  processor(text=a	)

     lowercase__           =  tokenizer(a	)

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


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





     """simple docstring"""







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


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Optional[Any]	)->       str:





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM(tokenizer=a      ,					feature_extractor=a      ,					decoder=a	)

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

     lowercase__           =  processor.decode(a	)

     lowercase__           =  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 	SCREAMING_SNAKE_CASE_							(   self      :    List[Any]      ,					a      :    Optional[Any]	)->       Any:





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM(tokenizer=a      ,					feature_extractor=a      ,					decoder=a	)

     lowercase__           =  self._get_dummy_logits()

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

     lowercase__           =  list(a	)

     with get_context('fork'	).Pool() as p:
       lowercase__           =  decoder.decode_beams_batch(a      ,					a	)

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

     self.assertListEqual(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 	SCREAMING_SNAKE_CASE_							(   self      :    Optional[Any]	)->       List[Any]:





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM(tokenizer=a      ,					feature_extractor=a      ,					decoder=a	)

     lowercase__           =  self._get_dummy_logits()

     lowercase__           =  15
     lowercase__           =  -20.0
     lowercase__           =  -4.0

     lowercase__           =  processor.batch_decode(
         a      ,					beam_width=a      ,					beam_prune_logp=a      ,					token_min_logp=a      ,					)
     lowercase__           =  decoded_processor_out.text

     lowercase__           =  list(a	)

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

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

     self.assertListEqual(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 	SCREAMING_SNAKE_CASE_							(   self      :    List[str]	)->       List[str]:





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM(tokenizer=a      ,					feature_extractor=a      ,					decoder=a	)

     lowercase__           =  self._get_dummy_logits()

     lowercase__           =  2.0
     lowercase__           =  5.0
     lowercase__           =  -20.0
     lowercase__           =  True

     lowercase__           =  processor.batch_decode(
         a      ,					alpha=a      ,					beta=a      ,					unk_score_offset=a      ,					lm_score_boundary=a      ,					)
     lowercase__           =  decoded_processor_out.text

     lowercase__           =  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__           =  decoder.decode_beams_batch(
           a      ,					a      ,					)

     lowercase__           =  [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__           =  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 	SCREAMING_SNAKE_CASE_							(   self      :    Optional[int]	)->       Dict:





     """simple docstring"""







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

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

     lowercase__           =  os.listdir(a	)
     lowercase__           =  ['alphabet.json', 'language_model']

     downloaded_decoder_files.sort()
     expected_decoder_files.sort()

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


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





     """simple docstring"""







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

     lowercase__           =  WavaVecaProcessorWithLM.from_pretrained(a	)

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

     lowercase__           =  os.listdir(a	)
     lowercase__           =  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 	SCREAMING_SNAKE_CASE_							(   self      :    Tuple	)->       Union[str, Any]:





     """simple docstring"""







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

     lowercase__           =  floats_list((3, 1_000)	)

     lowercase__           =  processor_wavaveca(a      ,					return_tensors='np'	)
     lowercase__           =  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__           =  self._get_dummy_logits()

     lowercase__           =  processor_wavaveca.batch_decode(a	)
     lowercase__           =  processor_auto.batch_decode(a	)

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


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Dict	)->       Any:





     """simple docstring"""







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

     lowercase__           =  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 	SCREAMING_SNAKE_CASE_							(   a      :    List[str]      ,					a      :    Union[str, Any]	)->       Tuple:





     """simple docstring"""







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


   def 	SCREAMING_SNAKE_CASE_							(   self      :    Optional[Any]	)->       Dict:





     """simple docstring"""







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

     lowercase__           =  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 	SCREAMING_SNAKE_CASE_							(   self      :    Optional[int]	)->       Optional[int]:





     """simple docstring"""







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

     lowercase__           =  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 	SCREAMING_SNAKE_CASE_							(   self      :    Tuple	)->       int:





     """simple docstring"""







     import torch

     lowercase__           =  load_dataset('common_voice'      ,					'en'      ,					split='train'      ,					streaming=a	)
     lowercase__           =  ds.cast_column('audio'      ,					datasets.Audio(sampling_rate=16_000	)	)
     lowercase__           =  iter(a	)
     lowercase__           =  next(a	)

     lowercase__           =  AutoProcessor.from_pretrained('patrickvonplaten/wav2vec2-base-100h-with-lm'	)
     lowercase__           =  WavaVecaForCTC.from_pretrained('patrickvonplaten/wav2vec2-base-100h-with-lm'	)

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

     with torch.no_grad():
       lowercase__           =  model(a	).logits.cpu().numpy()

     lowercase__           =  processor.decode(logits[0]      ,					output_word_offsets=a	)

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

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

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

     # output times
     lowercase__           =  torch.tensor(self.get_from_offsets(a      ,					'start_time'	)	)
     lowercase__           =  torch.tensor(self.get_from_offsets(a      ,					'end_time'	)	)

     # fmt: off
     lowercase__           =  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__           =  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	)	)









import sys

lowercase_          =		(
    """73167176531330624919225119674426574742355349194934"""
    """96983520312774506326239578318016984801869478851843"""
    """85861560789112949495459501737958331952853208805511"""
    """12540698747158523863050715693290963295227443043557"""
    """66896648950445244523161731856403098711121722383113"""
    """62229893423380308135336276614282806444486645238749"""
    """30358907296290491560440772390713810515859307960866"""
    """70172427121883998797908792274921901699720888093776"""
    """65727333001053367881220235421809751254540594752243"""
    """52584907711670556013604839586446706324415722155397"""
    """53697817977846174064955149290862569321978468622482"""
    """83972241375657056057490261407972968652414535100474"""
    """82166370484403199890008895243450658541227588666881"""
    """16427171479924442928230863465674813919123162824586"""
    """17866458359124566529476545682848912883142607690042"""
    """24219022671055626321111109370544217506941658960408"""
    """07198403850962455444362981230987879927244284909188"""
    """84580156166097919133875499200524063689912560717606"""
    """05886116467109405077541002256983155200055935729725"""
    """71636269561882670428252483600823257530420752963450"""
)




def       __UpperCamelCase  (_SCREAMING_SNAKE_CASE = N		)    ->       int:
  lowercase__           =  -sys.maxsize - 1
  for i in range(len(_SCREAMING_SNAKE_CASE		) - 12		):
    lowercase__           =  1
    for j in range(13		):
      product *= int(n[i + j]		)
    if product > largest_product:
      lowercase__           =  product
  return largest_product


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


