

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



 """simple docstring"""







 def    __magic_name__       (		self :	Union[str, Any]       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					int        =    {
          "unk_token": "<unk>",
          "bos_token": "<s>",
          "eos_token": "</s>",
      }
      _lowerCAmelCase   :					Any        =    {
          "feature_size": 1,
          "padding_value": 0.0,
          "sampling_rate": 1_6_0_0_0,
          "return_attention_mask": False,
          "do_normalize": True,
      }

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

 def    __magic_name__       (		self :	Tuple     ,						**A_ :	Optional[int]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					str        =    self.add_kwargs_tokens_map.copy()
      kwargs.update(A_       )
      return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname     ,						**A_       )

 def    __magic_name__       (		self :	Tuple     ,						**A_ :	Tuple       ):






      '''simple docstring'''




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

 def    __magic_name__       (		self :	str     ,						**A_ :	Tuple       ):






      '''simple docstring'''




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

 def    __magic_name__       (		self :	Any       ):






      '''simple docstring'''




      shutil.rmtree(self.tmpdirname       )

 def    __magic_name__       (		self :	Dict       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					List[str]        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

      processor.save_pretrained(self.tmpdirname       )
      _lowerCAmelCase   :					int        =    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    __magic_name__       (		self :	Tuple       ):






      '''simple docstring'''




      _lowerCAmelCase   :					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
      _lowerCAmelCase   :					Union[str, Any]        =    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    __magic_name__       (		self :	str       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Union[str, Any]        =    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    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					Optional[Any]        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

      _lowerCAmelCase   :					int        =    floats_list((3, 1_0_0_0)       )

      _lowerCAmelCase   :					Tuple        =    feature_extractor(A_     ,						return_tensors="np"       )
      _lowerCAmelCase   :					Any        =    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    __magic_name__       (		self :	List[Any]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Tuple        =    self.get_feature_extractor()
      _lowerCAmelCase   :					Optional[int]        =    self.get_tokenizer()
      _lowerCAmelCase   :					List[str]        =    self.get_decoder()

      _lowerCAmelCase   :					List[str]        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

      _lowerCAmelCase   :					int        =    "This is a test string"

      _lowerCAmelCase   :					List[str]        =    processor(text=A_       )

      _lowerCAmelCase   :					int        =    tokenizer(A_       )

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

 def    __magic_name__       (		self :	Optional[int]     ,						A_ :	Tuple=(2, 1_0, 1_6)     ,						A_ :	str=7_7       ):






      '''simple docstring'''




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

 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					List[str]        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

      _lowerCAmelCase   :					Tuple        =    self._get_dummy_logits(shape=(1_0, 1_6)     ,						seed=1_3       )

      _lowerCAmelCase   :					List[Any]        =    processor.decode(A_       )

      _lowerCAmelCase   :					Any        =    decoder.decode_beams(A_       )[0]

      self.assertEqual(decoded_decoder[0]     ,						decoded_processor.text       )
      self.assertEqual("</s> <s> </s>"     ,						decoded_processor.text       )
      self.assertEqual(decoded_decoder[-2]     ,						decoded_processor.logit_score       )
      self.assertEqual(decoded_decoder[-1]     ,						decoded_processor.lm_score       )

 @parameterized.expand([[None], ["fork"], ["spawn"]]       )
 def    __magic_name__       (		self :	Optional[int]     ,						A_ :	Union[str, Any]       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					Tuple        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

      _lowerCAmelCase   :					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:
           _lowerCAmelCase   :					int        =    processor.batch_decode(A_       )
      else:
           with get_context(A_       ).Pool() as pool:
                _lowerCAmelCase   :					List[Any]        =    processor.batch_decode(A_     ,						A_       )

      _lowerCAmelCase   :					List[Any]        =    list(A_       )

      with get_context("fork"       ).Pool() as p:
           _lowerCAmelCase   :					Any        =    decoder.decode_beams_batch(A_     ,						A_       )

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

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

 def    __magic_name__       (		self :	Union[str, Any]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					List[str]        =    self.get_feature_extractor()
      _lowerCAmelCase   :					str        =    self.get_tokenizer()
      _lowerCAmelCase   :					Tuple        =    self.get_decoder()

      _lowerCAmelCase   :					int        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

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

      _lowerCAmelCase   :					Optional[Any]        =    1_5
      _lowerCAmelCase   :					str        =    -20.0
      _lowerCAmelCase   :					Any        =    -4.0

      _lowerCAmelCase   :					int        =    processor.batch_decode(
          A_     ,						beam_width=A_     ,						beam_prune_logp=A_     ,						token_min_logp=A_     ,						)
      _lowerCAmelCase   :					Tuple        =    decoded_processor_out.text

      _lowerCAmelCase   :					Optional[Any]        =    list(A_       )

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

      _lowerCAmelCase   :					List[Any]        =    [d[0][0] for d in decoded_decoder_out]
      _lowerCAmelCase   :					Tuple        =    [d[0][2] for d in decoded_decoder_out]
      _lowerCAmelCase   :					str        =    [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    __magic_name__       (		self :	str       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Any        =    self.get_feature_extractor()
      _lowerCAmelCase   :					str        =    self.get_tokenizer()
      _lowerCAmelCase   :					int        =    self.get_decoder()

      _lowerCAmelCase   :					Union[str, Any]        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

      _lowerCAmelCase   :					Any        =    self._get_dummy_logits()

      _lowerCAmelCase   :					int        =    2.0
      _lowerCAmelCase   :					int        =    5.0
      _lowerCAmelCase   :					Tuple        =    -20.0
      _lowerCAmelCase   :					List[str]        =    True

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

      _lowerCAmelCase   :					Optional[int]        =    list(A_       )
      decoder.reset_params(
          alpha=A_     ,						beta=A_     ,						unk_score_offset=A_     ,						lm_score_boundary=A_     ,						)

      with get_context("fork"       ).Pool() as pool:
           _lowerCAmelCase   :					List[Any]        =    decoder.decode_beams_batch(
               A_     ,						A_     ,						)

      _lowerCAmelCase   :					str        =    [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_       )
      _lowerCAmelCase   :					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     ,						-20.0       )
      self.assertEqual(lm_model.score_boundary     ,						A_       )

 def    __magic_name__       (		self :	str       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[int]        =    WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"       )

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

      _lowerCAmelCase   :					Any        =    os.listdir(A_       )
      _lowerCAmelCase   :					List[str]        =    ["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    __magic_name__       (		self :	str       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					str        =    WavaVecaProcessorWithLM.from_pretrained(A_       )

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

      _lowerCAmelCase   :					Optional[Any]        =    os.listdir(A_       )
      _lowerCAmelCase   :					Optional[Any]        =    os.listdir(A_       )

      local_decoder_files.sort()
      expected_decoder_files.sort()

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

 def    __magic_name__       (		self :	Any       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[int]        =    WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"       )
      _lowerCAmelCase   :					Optional[int]        =    AutoProcessor.from_pretrained("hf-internal-testing/processor_with_lm"       )

      _lowerCAmelCase   :					Any        =    floats_list((3, 1_0_0_0)       )

      _lowerCAmelCase   :					Optional[int]        =    processor_wavaveca(A_     ,						return_tensors="np"       )
      _lowerCAmelCase   :					Optional[Any]        =    processor_auto(A_     ,						return_tensors="np"       )

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

      _lowerCAmelCase   :					Any        =    self._get_dummy_logits()

      _lowerCAmelCase   :					Any        =    processor_wavaveca.batch_decode(A_       )
      _lowerCAmelCase   :					Union[str, Any]        =    processor_auto.batch_decode(A_       )

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

 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					Any        =    WavaVecaProcessorWithLM(tokenizer=A_     ,						feature_extractor=A_     ,						decoder=A_       )

      self.assertListEqual(
          processor.model_input_names     ,						feature_extractor.model_input_names     ,						msg="`processor` and `feature_extractor` model input names do not match"     ,						)

 @staticmethod
 def    __magic_name__       (		A_ :	Union[str, Any]     ,						A_ :	Optional[int]       ):






      '''simple docstring'''




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

 def    __magic_name__       (		self :	Optional[int]       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					Union[str, Any]        =    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    __magic_name__       (		self :	str       ):






      '''simple docstring'''




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

      _lowerCAmelCase   :					Union[str, Any]        =    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    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      import torch

      _lowerCAmelCase   :					Any        =    load_dataset("common_voice"     ,						"en"     ,						split="train"     ,						streaming=A_       )
      _lowerCAmelCase   :					Optional[int]        =    ds.cast_column("audio"     ,						datasets.Audio(sampling_rate=1_6_0_0_0       )       )
      _lowerCAmelCase   :					Union[str, Any]        =    iter(A_       )
      _lowerCAmelCase   :					Optional[int]        =    next(A_       )

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

      with torch.no_grad():
           _lowerCAmelCase   :					List[str]        =    model(A_       ).logits.cpu().numpy()

      _lowerCAmelCase   :					List[Any]        =    processor.decode(logits[0]     ,						output_word_offsets=A_       )

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

      _lowerCAmelCase   :					Optional[int]        =    "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
      _lowerCAmelCase   :					Dict        =    torch.tensor(self.get_from_offsets(A_     ,						"start_time"       )       )
      _lowerCAmelCase   :					Tuple        =    torch.tensor(self.get_from_offsets(A_     ,						"end_time"       )       )

      # fmt: off
      _lowerCAmelCase   :					List[str]        =    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]       )
      _lowerCAmelCase   :					Optional[Any]        =    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       )       )









from __future__ import annotations

import inspect
import unittest
from math import floor

import numpy as np

from transformers import CvtConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
  import tensorflow as tf

  from transformers import TFCvtForImageClassification, TFCvtModel
  from transformers.models.cvt.modeling_tf_cvt import TF_CVT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
  from PIL import Image

  from transformers import AutoImageProcessor





class A__    ( A ):



 """simple docstring"""







 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					int        =    self.config_class(**self.inputs_dict       )
      self.parent.assertTrue(hasattr(A_     ,						"embed_dim"       )       )
      self.parent.assertTrue(hasattr(A_     ,						"num_heads"       )       )





class A__    :



 """simple docstring"""







 def __init__(		self :	Tuple     ,						A_ :	int     ,						A_ :	Dict=1_3     ,						A_ :	int=6_4     ,						A_ :	str=3     ,						A_ :	Optional[int]=[1_6, 4_8, 9_6]     ,						A_ :	int=[1, 3, 6]     ,						A_ :	Optional[int]=[1, 2, 1_0]     ,						A_ :	Any=[7, 3, 3]     ,						A_ :	Tuple=[4, 2, 2]     ,						A_ :	str=[2, 1, 1]     ,						A_ :	Optional[Any]=[2, 2, 2]     ,						A_ :	Union[str, Any]=[False, False, True]     ,						A_ :	Union[str, Any]=[0.0, 0.0, 0.0]     ,						A_ :	Any=0.02     ,						A_ :	Optional[int]=1E-12     ,						A_ :	str=True     ,						A_ :	List[Any]=True     ,						A_ :	Union[str, Any]=2     ,						):






      '''simple docstring'''




      _lowerCAmelCase   :					Dict        =    parent
      _lowerCAmelCase   :					int        =    batch_size
      _lowerCAmelCase   :					Union[str, Any]        =    image_size
      _lowerCAmelCase   :					Optional[Any]        =    patch_sizes
      _lowerCAmelCase   :					Optional[int]        =    patch_stride
      _lowerCAmelCase   :					List[str]        =    patch_padding
      _lowerCAmelCase   :					str        =    is_training
      _lowerCAmelCase   :					Any        =    use_labels
      _lowerCAmelCase   :					Dict        =    num_labels
      _lowerCAmelCase   :					Tuple        =    num_channels
      _lowerCAmelCase   :					Tuple        =    embed_dim
      _lowerCAmelCase   :					List[Any]        =    num_heads
      _lowerCAmelCase   :					Union[str, Any]        =    stride_kv
      _lowerCAmelCase   :					Tuple        =    depth
      _lowerCAmelCase   :					List[str]        =    cls_token
      _lowerCAmelCase   :					Tuple        =    attention_drop_rate
      _lowerCAmelCase   :					Optional[Any]        =    initializer_range
      _lowerCAmelCase   :					Optional[int]        =    layer_norm_eps

 def    __magic_name__       (		self :	Dict       ):






      '''simple docstring'''




      _lowerCAmelCase   :					str        =    floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]       )

      _lowerCAmelCase   :					int        =    None
      if self.use_labels:
           # create a random int32 tensor of given shape
           _lowerCAmelCase   :					List[Any]        =    ids_tensor([self.batch_size]     ,						self.num_labels       )

      _lowerCAmelCase   :					Tuple        =    self.get_config()
      return config, pixel_values, labels

 def    __magic_name__       (		self :	Union[str, Any]       ):






      '''simple docstring'''




      return CvtConfig(
          image_size=self.image_size     ,						num_labels=self.num_labels     ,						num_channels=self.num_channels     ,						embed_dim=self.embed_dim     ,						num_heads=self.num_heads     ,						patch_sizes=self.patch_sizes     ,						patch_padding=self.patch_padding     ,						patch_stride=self.patch_stride     ,						stride_kv=self.stride_kv     ,						depth=self.depth     ,						cls_token=self.cls_token     ,						attention_drop_rate=self.attention_drop_rate     ,						initializer_range=self.initializer_range     ,						)

 def    __magic_name__       (		self :	Union[str, Any]     ,						A_ :	str     ,						A_ :	Optional[int]     ,						A_ :	str       ):






      '''simple docstring'''




      _lowerCAmelCase   :					str        =    TFCvtModel(config=A_       )
      _lowerCAmelCase   :					str        =    model(A_     ,						training=A_       )
      _lowerCAmelCase   :					List[Any]        =    (self.image_size, self.image_size)
      _lowerCAmelCase      ,			_lowerCAmelCase   :					Any        =    image_size[0], image_size[1]
      for i in range(len(self.depth       )       ):
           _lowerCAmelCase   :					Union[str, Any]        =    floor(((height + 2 * self.patch_padding[i] - self.patch_sizes[i]) / self.patch_stride[i]) + 1       )
           _lowerCAmelCase   :					List[Any]        =    floor(((width + 2 * self.patch_padding[i] - self.patch_sizes[i]) / self.patch_stride[i]) + 1       )
      self.parent.assertEqual(result.last_hidden_state.shape     ,						(self.batch_size, self.embed_dim[-1], height, width)       )

 def    __magic_name__       (		self :	Any     ,						A_ :	Tuple     ,						A_ :	int     ,						A_ :	Any       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[int]        =    self.num_labels
      _lowerCAmelCase   :					Optional[Any]        =    TFCvtForImageClassification(A_       )
      _lowerCAmelCase   :					Optional[Any]        =    model(A_     ,						labels=A_     ,						training=A_       )
      self.parent.assertEqual(result.logits.shape     ,						(self.batch_size, self.num_labels)       )


 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[Any]        =    self.prepare_config_and_inputs()
      _lowerCAmelCase      ,			_lowerCAmelCase      ,			_lowerCAmelCase   :					List[str]        =    config_and_inputs
      _lowerCAmelCase   :					Optional[int]        =    {"pixel_values": pixel_values}
      return config, inputs_dict





@require_tf
class A__    ( A  ,		A  ,		unittest.TestCase ):



 """simple docstring"""







 _lowercase     :					Any      =						(TFCvtModel, TFCvtForImageClassification) if is_tf_available() else ()
 _lowercase     :					Tuple      =						(
     {'''feature-extraction''': TFCvtModel, '''image-classification''': TFCvtForImageClassification}
     if is_tf_available()
     else {}
 )
 _lowercase     :					Optional[int]      =						False
 _lowercase     :					Any      =						False
 _lowercase     :					int      =						False
 _lowercase     :					int      =						False
 _lowercase     :					Tuple      =						False

 def    __magic_name__       (		self :	Optional[Any]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					List[Any]        =    TFCvtModelTester(self       )
      _lowerCAmelCase   :					Optional[Any]        =    TFCvtConfigTester(self     ,						config_class=A_     ,						has_text_modality=A_     ,						hidden_size=3_7       )

 def    __magic_name__       (		self :	List[Any]       ):






      '''simple docstring'''




      self.config_tester.create_and_test_config_common_properties()
      self.config_tester.create_and_test_config_to_json_string()
      self.config_tester.create_and_test_config_to_json_file()
      self.config_tester.create_and_test_config_from_and_save_pretrained()
      self.config_tester.create_and_test_config_with_num_labels()
      self.config_tester.check_config_can_be_init_without_params()
      self.config_tester.check_config_arguments_init()

 @unittest.skip(reason="Cvt does not output attentions"       )
 def    __magic_name__       (		self :	str       ):






      '''simple docstring'''




      pass

 @unittest.skip(reason="Cvt does not use inputs_embeds"       )
 def    __magic_name__       (		self :	Tuple       ):






      '''simple docstring'''




      pass

 @unittest.skip(reason="Cvt does not support input and output embeddings"       )
 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      pass

 @unittest.skipIf(
     not is_tf_available() or len(tf.config.list_physical_devices("GPU"       )       ) == 0     ,						reason="TF does not support backprop for grouped convolutions on CPU."     ,						)
 def    __magic_name__       (		self :	Dict       ):






      '''simple docstring'''




      super().test_dataset_conversion()

 @unittest.skipIf(
     not is_tf_available() or len(tf.config.list_physical_devices("GPU"       )       ) == 0     ,						reason="TF does not support backprop for grouped convolutions on CPU."     ,						)
 @slow
 def    __magic_name__       (		self :	Dict       ):






      '''simple docstring'''




      super().test_keras_fit()

 @unittest.skip(reason="Get `Failed to determine best cudnn convolution algo.` error after using TF 2.12+cuda 11.8"       )
 def    __magic_name__       (		self :	Dict       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[int]        =    tf.keras.mixed_precision.Policy("mixed_float16"       )
      tf.keras.mixed_precision.set_global_policy(A_       )
      super().test_keras_fit()
      tf.keras.mixed_precision.set_global_policy("float32"       )

 def    __magic_name__       (		self :	str       ):






      '''simple docstring'''




      _lowerCAmelCase      ,			_lowerCAmelCase   :					List[Any]        =    self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
           _lowerCAmelCase   :					Dict        =    model_class(A_       )
           _lowerCAmelCase   :					Optional[int]        =    inspect.signature(model.call       )
           # signature.parameters is an OrderedDict => so arg_names order is deterministic
           _lowerCAmelCase   :					List[Any]        =    [*signature.parameters.keys()]

           _lowerCAmelCase   :					Union[str, Any]        =    ["pixel_values"]
           self.assertListEqual(arg_names[:1]     ,						A_       )

 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      def check_hidden_states_output(A_ :	Any     ,						A_ :	Any     ,						A_ :	Dict       ):
           _lowerCAmelCase   :					Tuple        =    model_class(A_       )

           _lowerCAmelCase   :					List[Any]        =    model(**self._prepare_for_class(A_     ,						A_       )       )
           _lowerCAmelCase   :					Tuple        =    outputs.hidden_states

           _lowerCAmelCase   :					Optional[Any]        =    len(self.model_tester.depth       )
           self.assertEqual(len(A_       )     ,						A_       )

           # verify the first hidden states (first block)
           self.assertListEqual(
               list(hidden_states[0].shape[-3:]       )     ,						[
                   self.model_tester.embed_dim[0],
                   self.model_tester.image_size // 4,
                   self.model_tester.image_size // 4,
               ]     ,						)

      _lowerCAmelCase      ,			_lowerCAmelCase   :					Tuple        =    self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
           _lowerCAmelCase   :					Union[str, Any]        =    True
           check_hidden_states_output(A_     ,						A_     ,						A_       )

           # check that output_hidden_states also work using config
           del inputs_dict["output_hidden_states"]
           _lowerCAmelCase   :					List[str]        =    True

           check_hidden_states_output(A_     ,						A_     ,						A_       )

 def    __magic_name__       (		self :	Tuple       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Tuple        =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*A_       )

 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					List[Any]        =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_image_classification(*A_       )


 @slow
 def    __magic_name__       (		self :	Dict       ):






      '''simple docstring'''




      for model_name in TF_CVT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           _lowerCAmelCase   :					Optional[Any]        =    TFCvtModel.from_pretrained(A_       )
           self.assertIsNotNone(A_       )







def 	_snake_case		(    )       ->      Dict:

     """simple docstring"""







     _lowerCAmelCase   :					int        =    Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"				)
     return image





@require_tf
@require_vision
class A__    ( unittest.TestCase ):



 """simple docstring"""







 @cached_property
 def    __magic_name__       (		self :	str       ):






      '''simple docstring'''




      return AutoImageProcessor.from_pretrained(TF_CVT_PRETRAINED_MODEL_ARCHIVE_LIST[0]       )


 @slow
 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[int]        =    TFCvtForImageClassification.from_pretrained(TF_CVT_PRETRAINED_MODEL_ARCHIVE_LIST[0]       )

      _lowerCAmelCase   :					int        =    self.default_image_processor
      _lowerCAmelCase   :					Optional[Any]        =    prepare_img()
      _lowerCAmelCase   :					Optional[Any]        =    image_processor(images=A_     ,						return_tensors="tf"       )

      # forward pass
      _lowerCAmelCase   :					str        =    model(**A_       )

      # verify the logits
      _lowerCAmelCase   :					List[str]        =    tf.TensorShape((1, 1_0_0_0)       )
      self.assertEqual(outputs.logits.shape     ,						A_       )

      _lowerCAmelCase   :					Any        =    tf.constant([0.9285, 0.9015, -0.3150]       )
      self.assertTrue(np.allclose(outputs.logits[0, :3].numpy()     ,						A_     ,						atol=1E-4       )       )






