

'''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       UpperCAmelCase__		(		self					:       Any			)  -> List[str]:


      """simple docstring"""




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

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

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

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

      # load decoder from hub
      __SCREAMING_SNAKE_CASE     =      """hf-internal-testing/ngram-beam-search-decoder"""
 def       UpperCAmelCase__		(		self					:       Dict					,							**__SCREAMING_SNAKE_CASE					:       str			)  -> Optional[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self.add_kwargs_tokens_map.copy()
      kwargs.update(__SCREAMING_SNAKE_CASE			)
      return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname					,							**__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       str					,							**__SCREAMING_SNAKE_CASE					:       Dict			)  -> Optional[Any]:


      """simple docstring"""




      return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname					,							**__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Optional[int]					,							**__SCREAMING_SNAKE_CASE					:       Optional[Any]			)  -> Any:


      """simple docstring"""




      return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name					,							**__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> Optional[int]:


      """simple docstring"""




      shutil.rmtree(self.tmpdirname			)
 def       UpperCAmelCase__		(		self					:       Optional[int]			)  -> int:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

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

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

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

      # decoder
      self.assertEqual(processor.decoder._alphabet.labels					,							decoder._alphabet.labels			)
      self.assertEqual(
          processor.decoder.model_container[decoder._model_key]._unigram_set					,							decoder.model_container[decoder._model_key]._unigram_set					,							)
      self.assertIsInstance(processor.decoder					,							__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       int			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      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     =      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       UpperCAmelCase__		(		self					:       List[Any]			)  -> Dict:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self.get_tokenizer()
      # add token to trigger raise
      tokenizer.add_tokens(["""xx"""]			)
      with self.assertRaisesRegex(__SCREAMING_SNAKE_CASE					,							"""include"""			):
           WavaVecaProcessorWithLM(
               tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=self.get_feature_extractor()					,							decoder=self.get_decoder()			)
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> Any:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      floats_list((3, 1_000)			)

      __SCREAMING_SNAKE_CASE     =      feature_extractor(__SCREAMING_SNAKE_CASE					,							return_tensors="""np"""			)
      __SCREAMING_SNAKE_CASE     =      processor(__SCREAMING_SNAKE_CASE					,							return_tensors="""np"""			)

      for key in input_feat_extract.keys():
           self.assertAlmostEqual(input_feat_extract[key].sum()					,							input_processor[key].sum()					,							delta=1E-2			)
 def       UpperCAmelCase__		(		self					:       Any			)  -> List[Any]:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      """This is a test string"""

      __SCREAMING_SNAKE_CASE     =      processor(text=__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      tokenizer(__SCREAMING_SNAKE_CASE			)

      for key in encoded_tok.keys():
           self.assertListEqual(encoded_tok[key]					,							encoded_processor[key]			)
 def       UpperCAmelCase__		(		self					:       Dict					,							__SCREAMING_SNAKE_CASE					:       Dict=(2, 10, 16)					,							__SCREAMING_SNAKE_CASE					:       List[str]=77			)  -> str:


      """simple docstring"""




      np.random.seed(__SCREAMING_SNAKE_CASE			)
      return np.random.rand(*__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Any			)  -> List[str]:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

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

      __SCREAMING_SNAKE_CASE     =      processor.decode(__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      decoder.decode_beams(__SCREAMING_SNAKE_CASE			)[0]

      self.assertEqual(decoded_decoder[0]					,							decoded_processor.text			)
      self.assertEqual("""</s> <s> </s>"""					,							decoded_processor.text			)
      self.assertEqual(decoded_decoder[-2]					,							decoded_processor.logit_score			)
      self.assertEqual(decoded_decoder[-1]					,							decoded_processor.lm_score			)
 @parameterized.expand([[None], ["""fork"""], ["""spawn"""]]			)
 def       UpperCAmelCase__		(		self					:       Union[str, Any]					,							__SCREAMING_SNAKE_CASE					:       Dict			)  -> Tuple:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      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     =      processor.batch_decode(__SCREAMING_SNAKE_CASE			)
      else:
           with get_context(__SCREAMING_SNAKE_CASE			).Pool() as pool:
                __SCREAMING_SNAKE_CASE     =      processor.batch_decode(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      list(__SCREAMING_SNAKE_CASE			)

      with get_context("""fork"""			).Pool() as p:
           __SCREAMING_SNAKE_CASE     =      decoder.decode_beams_batch(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      [], [], []
      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(__SCREAMING_SNAKE_CASE					,							decoded_processor.text			)
      self.assertListEqual(["""<s> <s> </s>""", """<s> <s> <s>"""]					,							decoded_processor.text			)
      self.assertListEqual(__SCREAMING_SNAKE_CASE					,							decoded_processor.logit_score			)
      self.assertListEqual(__SCREAMING_SNAKE_CASE					,							decoded_processor.lm_score			)
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> List[Any]:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      self._get_dummy_logits()

      __SCREAMING_SNAKE_CASE     =      15
      __SCREAMING_SNAKE_CASE     =      -20.0
      __SCREAMING_SNAKE_CASE     =      -4.0

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

      __SCREAMING_SNAKE_CASE     =      list(__SCREAMING_SNAKE_CASE			)

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

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

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

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

      self.assertTrue(np.array_equal(__SCREAMING_SNAKE_CASE					,							decoded_processor_out.lm_score			)			)
      self.assertTrue(np.allclose([-15.554, -13.9474]					,							__SCREAMING_SNAKE_CASE					,							atol=1E-3			)			)
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> Optional[int]:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      self._get_dummy_logits()

      __SCREAMING_SNAKE_CASE     =      2.0
      __SCREAMING_SNAKE_CASE     =      5.0
      __SCREAMING_SNAKE_CASE     =      -20.0
      __SCREAMING_SNAKE_CASE     =      True

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

      __SCREAMING_SNAKE_CASE     =      list(__SCREAMING_SNAKE_CASE			)
      decoder.reset_params(
          alpha=__SCREAMING_SNAKE_CASE					,							beta=__SCREAMING_SNAKE_CASE					,							unk_score_offset=__SCREAMING_SNAKE_CASE					,							lm_score_boundary=__SCREAMING_SNAKE_CASE					,							)

      with get_context("""fork"""			).Pool() as pool:
           __SCREAMING_SNAKE_CASE     =      decoder.decode_beams_batch(
               __SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE					,							)

      __SCREAMING_SNAKE_CASE     =      [d[0][0] for d in decoded_decoder_out]

      self.assertListEqual(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)
      self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]					,							__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      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					,							__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Tuple			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""			)

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

      __SCREAMING_SNAKE_CASE     =      os.listdir(__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      ["""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(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Tuple			)  -> List[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      snapshot_download("""hf-internal-testing/processor_with_lm"""			)

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM.from_pretrained(__SCREAMING_SNAKE_CASE			)

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

      __SCREAMING_SNAKE_CASE     =      os.listdir(__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      os.listdir(__SCREAMING_SNAKE_CASE			)

      local_decoder_files.sort()
      expected_decoder_files.sort()

      # test that both decoder form hub and local files in cache are the same
      self.assertListEqual(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Tuple			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""			)
      __SCREAMING_SNAKE_CASE     =      AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""			)

      __SCREAMING_SNAKE_CASE     =      floats_list((3, 1_000)			)

      __SCREAMING_SNAKE_CASE     =      processor_wavaveca(__SCREAMING_SNAKE_CASE					,							return_tensors="""np"""			)
      __SCREAMING_SNAKE_CASE     =      processor_auto(__SCREAMING_SNAKE_CASE					,							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     =      self._get_dummy_logits()

      __SCREAMING_SNAKE_CASE     =      processor_wavaveca.batch_decode(__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      processor_auto.batch_decode(__SCREAMING_SNAKE_CASE			)

      self.assertListEqual(decoded_wavaveca.text					,							decoded_auto.text			)
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> int:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM(tokenizer=__SCREAMING_SNAKE_CASE					,							feature_extractor=__SCREAMING_SNAKE_CASE					,							decoder=__SCREAMING_SNAKE_CASE			)

      self.assertListEqual(
          processor.model_input_names					,							feature_extractor.model_input_names					,							msg="""`processor` and `feature_extractor` model input names do not match"""					,							)
 @staticmethod
 def       UpperCAmelCase__		(		__SCREAMING_SNAKE_CASE					:       Union[str, Any]					,							__SCREAMING_SNAKE_CASE					:       Tuple			)  -> Dict:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      [d[key] for d in offsets]
      return retrieved_list
 def       UpperCAmelCase__		(		self					:       int			)  -> Optional[Any]:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      processor.decode(__SCREAMING_SNAKE_CASE					,							output_word_offsets=__SCREAMING_SNAKE_CASE			)
      # check Wav2Vec2CTCTokenizerOutput keys for word
      self.assertEqual(len(outputs.keys()			)					,							4			)
      self.assertTrue("""text""" in outputs			)
      self.assertTrue("""word_offsets""" in outputs			)
      self.assertTrue(isinstance(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)			)

      self.assertEqual(""" """.join(self.get_from_offsets(outputs["""word_offsets"""]					,							"""word"""			)			)					,							outputs.text			)
      self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]					,							"""word"""			)					,							["""<s>""", """<s>""", """</s>"""]			)
      self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]					,							"""start_offset"""			)					,							[0, 2, 4]			)
      self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]					,							"""end_offset"""			)					,							[1, 3, 5]			)
 def       UpperCAmelCase__		(		self					:       List[str]			)  -> Union[str, Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""			)
      __SCREAMING_SNAKE_CASE     =      self._get_dummy_logits()

      __SCREAMING_SNAKE_CASE     =      processor.batch_decode(__SCREAMING_SNAKE_CASE					,							output_word_offsets=__SCREAMING_SNAKE_CASE			)

      # check Wav2Vec2CTCTokenizerOutput keys for word
      self.assertEqual(len(outputs.keys()			)					,							4			)
      self.assertTrue("""text""" in outputs			)
      self.assertTrue("""word_offsets""" in outputs			)
      self.assertTrue(isinstance(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)			)

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


 @slow
 @require_torch
 @require_torchaudio
 def       UpperCAmelCase__		(		self					:       Tuple			)  -> Dict:


      """simple docstring"""




      import torch

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

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

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

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

      __SCREAMING_SNAKE_CASE     =      model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
      __SCREAMING_SNAKE_CASE     =      [
          {
              """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     =      """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(__SCREAMING_SNAKE_CASE					,							"""word"""			)			)					,							__SCREAMING_SNAKE_CASE			)
      self.assertEqual(""" """.join(self.get_from_offsets(__SCREAMING_SNAKE_CASE					,							"""word"""			)			)					,							output.text			)

      # output times
      __SCREAMING_SNAKE_CASE     =      torch.tensor(self.get_from_offsets(__SCREAMING_SNAKE_CASE					,							"""start_time"""			)			)
      __SCREAMING_SNAKE_CASE     =      torch.tensor(self.get_from_offsets(__SCREAMING_SNAKE_CASE					,							"""end_time"""			)			)

      # fmt: off
      __SCREAMING_SNAKE_CASE     =      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]			)
      __SCREAMING_SNAKE_CASE     =      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(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE					,							atol=0.01			)			)
      self.assertTrue(torch.allclose(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE					,							atol=0.01			)			)









'''simple docstring'''



import unittest

from diffusers import FlaxAutoencoderKL
from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import require_flax

from .test_modeling_common_flax import FlaxModelTesterMixin


if is_flax_available():
  import jax







@require_flax
class 						lowerCAmelCase__   (      a ,					unittest.TestCase					):



 """simple docstring"""



 lowerCAmelCase__  					=      FlaxAutoencoderKL
 @property
 def       UpperCAmelCase__		(		self					:       Tuple			)  -> List[str]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      4
      __SCREAMING_SNAKE_CASE     =      3
      __SCREAMING_SNAKE_CASE     =      (32, 32)

      __SCREAMING_SNAKE_CASE     =      jax.random.PRNGKey(0			)
      __SCREAMING_SNAKE_CASE     =      jax.random.uniform(__SCREAMING_SNAKE_CASE					,							((batch_size, num_channels) + sizes)			)

      return {"sample": image, "prng_key": prng_key}


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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      {
          """block_out_channels""": [32, 64],
          """in_channels""": 3,
          """out_channels""": 3,
          """down_block_types""": ["""DownEncoderBlock2D""", """DownEncoderBlock2D"""],
          """up_block_types""": ["""UpDecoderBlock2D""", """UpDecoderBlock2D"""],
          """latent_channels""": 4,
      }
      __SCREAMING_SNAKE_CASE     =      self.dummy_input
      return init_dict, inputs_dict






