




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




			def    lowerCAmelCase__				(					self							)				->				List[str]:

										'''simple docstring'''




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

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

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




			def    lowerCAmelCase__				(					self				,							**a__							)				->				Tuple:

										'''simple docstring'''




										snake_case_						=      self.add_kwargs_tokens_map.copy()
										kwargs.update(a__							)
										return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname				,							**a__							)




			def    lowerCAmelCase__				(					self				,							**a__							)				->				int:

										'''simple docstring'''




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




			def    lowerCAmelCase__				(					self				,							**a__							)				->				Optional[int]:

										'''simple docstring'''




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




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

										'''simple docstring'''




										shutil.rmtree(self.tmpdirname							)




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

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM(tokenizer=a__				,							feature_extractor=a__				,							decoder=a__							)

										processor.save_pretrained(self.tmpdirname							)
										snake_case_						=      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    lowerCAmelCase__				(					self							)				->				int:

										'''simple docstring'''




										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
										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    lowerCAmelCase__				(					self							)				->				Dict:

										'''simple docstring'''




										snake_case_						=      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    lowerCAmelCase__				(					self							)				->				List[Any]:

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM(tokenizer=a__				,							feature_extractor=a__				,							decoder=a__							)

										snake_case_						=      floats_list((3, 1_000)							)

										snake_case_						=      feature_extractor(a__				,							return_tensors="np"							)
										snake_case_						=      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    lowerCAmelCase__				(					self							)				->				Tuple:

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM(tokenizer=a__				,							feature_extractor=a__				,							decoder=a__							)

										snake_case_						=      "This is a test string"

										snake_case_						=      processor(text=a__							)

										snake_case_						=      tokenizer(a__							)

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




			def    lowerCAmelCase__				(					self				,							a__=(2, 10, 16)				,							a__=77							)				->				Union[str, Any]:

										'''simple docstring'''




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




			def    lowerCAmelCase__				(					self							)				->				Tuple:

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM(tokenizer=a__				,							feature_extractor=a__				,							decoder=a__							)

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

										snake_case_						=      processor.decode(a__							)

										snake_case_						=      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    lowerCAmelCase__				(					self				,							a__							)				->				List[Any]:

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM(tokenizer=a__				,							feature_extractor=a__				,							decoder=a__							)

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

										snake_case_						=      list(a__							)

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

										snake_case_					,      snake_case_					,      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(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    lowerCAmelCase__				(					self							)				->				Optional[int]:

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM(tokenizer=a__				,							feature_extractor=a__				,							decoder=a__							)

										snake_case_						=      self._get_dummy_logits()

										snake_case_						=      15
										snake_case_						=      -2_0.0
										snake_case_						=      -4.0

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

										snake_case_						=      list(a__							)

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

										snake_case_						=      [d[0][0] for d in decoded_decoder_out]
										snake_case_						=      [d[0][2] for d in decoded_decoder_out]
										snake_case_						=      [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([-2_0.0_5_4, -1_8.4_4_7]				,							a__				,							atol=1e-3							)							)

										self.assertTrue(np.array_equal(a__				,							decoded_processor_out.lm_score							)							)
										self.assertTrue(np.allclose([-1_5.5_5_4, -1_3.9_4_7_4]				,							a__				,							atol=1e-3							)							)




			def    lowerCAmelCase__				(					self							)				->				Tuple:

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM(tokenizer=a__				,							feature_extractor=a__				,							decoder=a__							)

										snake_case_						=      self._get_dummy_logits()

										snake_case_						=      2.0
										snake_case_						=      5.0
										snake_case_						=      -2_0.0
										snake_case_						=      True

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

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

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

										snake_case_						=      [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__							)
										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				,							-2_0.0							)
										self.assertEqual(lm_model.score_boundary				,							a__							)




			def    lowerCAmelCase__				(					self							)				->				Dict:

										'''simple docstring'''




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

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

										snake_case_						=      os.listdir(a__							)
										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(a__				,							a__							)




			def    lowerCAmelCase__				(					self							)				->				Optional[int]:

										'''simple docstring'''




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

										snake_case_						=      WavaVecaProcessorWithLM.from_pretrained(a__							)

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

										snake_case_						=      os.listdir(a__							)
										snake_case_						=      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    lowerCAmelCase__				(					self							)				->				Optional[Any]:

										'''simple docstring'''




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

										snake_case_						=      floats_list((3, 1_000)							)

										snake_case_						=      processor_wavaveca(a__				,							return_tensors="np"							)
										snake_case_						=      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							)

										snake_case_						=      self._get_dummy_logits()

										snake_case_						=      processor_wavaveca.batch_decode(a__							)
										snake_case_						=      processor_auto.batch_decode(a__							)

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




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

										'''simple docstring'''




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

										snake_case_						=      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    lowerCAmelCase__				(					a__				,							a__							)				->				int:

										'''simple docstring'''




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




			def    lowerCAmelCase__				(					self							)				->				Optional[Any]:

										'''simple docstring'''




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

										snake_case_						=      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    lowerCAmelCase__				(					self							)				->				str:

										'''simple docstring'''




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

										snake_case_						=      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    lowerCAmelCase__				(					self							)				->				Tuple:

										'''simple docstring'''




										import torch

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

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

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

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

										snake_case_						=      model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
										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"]
										]

										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(a__				,							"word"							)							)				,							a__							)
										self.assertEqual(" ".join(self.get_from_offsets(a__				,							"word"							)							)				,							output.text							)

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

										# fmt: off
										snake_case_						=      torch.tensor([1.4_1_9_9, 1.6_5_9_9, 2.2_5_9_9, 3.0, 3.2_4, 3.5_9_9_9, 3.7_9_9_9, 4.0_9_9_9, 4.2_6, 4.9_4, 5.2_8, 5.6_5_9_9, 5.7_8, 5.9_4, 6.3_2, 6.5_3_9_9, 6.6_5_9_9]							)
										snake_case_						=      torch.tensor([1.5_3_9_9, 1.8_9_9_9, 2.9, 3.1_6, 3.5_3_9_9, 3.7_2, 4.0_1_9_9, 4.1_7_9_9, 4.7_6, 5.1_5_9_9, 5.5_5_9_9, 5.6_9_9_9, 5.8_6, 6.1_9_9_9, 6.3_8, 6.6_1_9_9, 6.9_4]							)
										# fmt: on

										self.assertTrue(torch.allclose(a__				,							a__				,							atol=0.0_1							)							)
										self.assertTrue(torch.allclose(a__				,							a__				,							atol=0.0_1							)							)









"""simple docstring"""

from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowercase__    :  str							    =	{
    '''configuration_x_clip''': [
        '''XCLIP_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''XCLIPConfig''',
        '''XCLIPTextConfig''',
        '''XCLIPVisionConfig''',
    ],
    '''processing_x_clip''': ['''XCLIPProcessor'''],
}

try:
     if not is_torch_available():
          raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
     pass
else:
     lowercase__    :  Tuple							    =	[
         '''XCLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
         '''XCLIPModel''',
         '''XCLIPPreTrainedModel''',
         '''XCLIPTextModel''',
         '''XCLIPVisionModel''',
     ]

if TYPE_CHECKING:
     from .configuration_x_clip import (
         XCLIP_PRETRAINED_CONFIG_ARCHIVE_MAP,
         XCLIPConfig,
         XCLIPTextConfig,
         XCLIPVisionConfig,
     )
     from .processing_x_clip import XCLIPProcessor

     try:
          if not is_torch_available():
               raise OptionalDependencyNotAvailable()
     except OptionalDependencyNotAvailable:
          pass
     else:
          from .modeling_x_clip import (
              XCLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
              XCLIPModel,
              XCLIPPreTrainedModel,
              XCLIPTextModel,
              XCLIPVisionModel,
          )

else:
     import sys

     lowercase__    :  Tuple							    =	_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)





