



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




	'''simple docstring'''





	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[int])				->       List[str]:

				"""simple docstring"""

				__lowerCAmelCase      :  str												=       "| <pad> <unk> <s> </s> a b c d e f g h i j k".split()
				__lowerCAmelCase      :  List[str]												=       dict(zip(_SCREAMING_SNAKE_CASE       ,			range(len(_SCREAMING_SNAKE_CASE))))

				__lowerCAmelCase      :  str												=       {
				    "unk_token": "<unk>",
				    "bos_token": "<s>",
				    "eos_token": "</s>",
				}
				__lowerCAmelCase      :  List[Any]												=       {
				    "feature_size": 1,
				    "padding_value": 0.0,
				    "sampling_rate": 1_6000,
				    "return_attention_mask": False,
				    "do_normalize": True,
				}

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

	def 							_SCREAMING_SNAKE_CASE   (     self:		int       ,			**_SCREAMING_SNAKE_CASE:		int)				->       Any:

				"""simple docstring"""

				__lowerCAmelCase      :  Dict												=       self.add_kwargs_tokens_map.copy()
				kwargs.update(_SCREAMING_SNAKE_CASE)
				return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname       ,			**_SCREAMING_SNAKE_CASE)

	def 							_SCREAMING_SNAKE_CASE   (     self:		str       ,			**_SCREAMING_SNAKE_CASE:		str)				->       List[Any]:

				"""simple docstring"""

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

	def 							_SCREAMING_SNAKE_CASE   (     self:		int       ,			**_SCREAMING_SNAKE_CASE:		Union[str, Any])				->       List[Any]:

				"""simple docstring"""

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

	def 							_SCREAMING_SNAKE_CASE   (     self:		List[Any])				->       Any:

				"""simple docstring"""

				shutil.rmtree(self.tmpdirname)

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

				"""simple docstring"""

				__lowerCAmelCase      :  Any												=       self.get_tokenizer()
				__lowerCAmelCase      :  Tuple												=       self.get_feature_extractor()
				__lowerCAmelCase      :  Any												=       self.get_decoder()

				__lowerCAmelCase      :  int												=       WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE       ,			feature_extractor=_SCREAMING_SNAKE_CASE       ,			decoder=_SCREAMING_SNAKE_CASE)

				processor.save_pretrained(self.tmpdirname)
				__lowerCAmelCase      :  List[str]												=       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 							_SCREAMING_SNAKE_CASE   (     self:		int)				->       List[Any]:

				"""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      :  str												=       WavaVecaProcessorWithLM.from_pretrained(
				    self.tmpdirname       ,			alpha=5.0       ,			beta=3.0       ,			score_boundary=-7.0       ,			unk_score_offset=3)

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

	def 							_SCREAMING_SNAKE_CASE   (     self:		List[str])				->       Tuple:

				"""simple docstring"""

				__lowerCAmelCase      :  List[str]												=       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 							_SCREAMING_SNAKE_CASE   (     self:		List[Any])				->       Optional[int]:

				"""simple docstring"""

				__lowerCAmelCase      :  Any												=       self.get_feature_extractor()
				__lowerCAmelCase      :  Any												=       self.get_tokenizer()
				__lowerCAmelCase      :  Dict												=       self.get_decoder()

				__lowerCAmelCase      :  Tuple												=       WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE       ,			feature_extractor=_SCREAMING_SNAKE_CASE       ,			decoder=_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase      :  int												=       floats_list((3, 1000))

				__lowerCAmelCase      :  List[str]												=       feature_extractor(_SCREAMING_SNAKE_CASE       ,			return_tensors="np")
				__lowerCAmelCase      :  Optional[Any]												=       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 							_SCREAMING_SNAKE_CASE   (     self:		List[str])				->       Dict:

				"""simple docstring"""

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

				__lowerCAmelCase      :  Union[str, Any]												=       WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE       ,			feature_extractor=_SCREAMING_SNAKE_CASE       ,			decoder=_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase      :  Dict												=       "This is a test string"

				__lowerCAmelCase      :  List[str]												=       processor(text=_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase      :  List[str]												=       tokenizer(_SCREAMING_SNAKE_CASE)

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

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

				"""simple docstring"""

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

	def 							_SCREAMING_SNAKE_CASE   (     self:		Tuple)				->       List[Any]:

				"""simple docstring"""

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

				__lowerCAmelCase      :  List[Any]												=       WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE       ,			feature_extractor=_SCREAMING_SNAKE_CASE       ,			decoder=_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase      :  Any												=       self._get_dummy_logits(shape=(10, 16)       ,			seed=13)

				__lowerCAmelCase      :  Union[str, Any]												=       processor.decode(_SCREAMING_SNAKE_CASE)

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

				"""simple docstring"""

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

				__lowerCAmelCase      :  str												=       WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE       ,			feature_extractor=_SCREAMING_SNAKE_CASE       ,			decoder=_SCREAMING_SNAKE_CASE)

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

				__lowerCAmelCase      :  Optional[int]												=       list(_SCREAMING_SNAKE_CASE)

				with get_context("fork").Pool() as p:
							__lowerCAmelCase      :  Tuple												=       decoder.decode_beams_batch(_SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase						,	__lowerCAmelCase						,	__lowerCAmelCase      :  Dict												=       [], [], []
				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 							_SCREAMING_SNAKE_CASE   (     self:		List[str])				->       Optional[int]:

				"""simple docstring"""

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

				__lowerCAmelCase      :  Tuple												=       WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE       ,			feature_extractor=_SCREAMING_SNAKE_CASE       ,			decoder=_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase      :  Any												=       self._get_dummy_logits()

				__lowerCAmelCase      :  Tuple												=       15
				__lowerCAmelCase      :  List[Any]												=       -20.0
				__lowerCAmelCase      :  Optional[int]												=       -4.0

				__lowerCAmelCase      :  Tuple												=       processor.batch_decode(
				    _SCREAMING_SNAKE_CASE       ,			beam_width=_SCREAMING_SNAKE_CASE       ,			beam_prune_logp=_SCREAMING_SNAKE_CASE       ,			token_min_logp=_SCREAMING_SNAKE_CASE       ,			)
				__lowerCAmelCase      :  Dict												=       decoded_processor_out.text

				__lowerCAmelCase      :  Optional[int]												=       list(_SCREAMING_SNAKE_CASE)

				with get_context("fork").Pool() as pool:
							__lowerCAmelCase      :  Optional[int]												=       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       ,			)

				__lowerCAmelCase      :  Tuple												=       [d[0][0] for d in decoded_decoder_out]
				__lowerCAmelCase      :  int												=       [d[0][2] for d in decoded_decoder_out]
				__lowerCAmelCase      :  Tuple												=       [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 							_SCREAMING_SNAKE_CASE   (     self:		List[str])				->       Dict:

				"""simple docstring"""

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

				__lowerCAmelCase      :  str												=       WavaVecaProcessorWithLM(tokenizer=_SCREAMING_SNAKE_CASE       ,			feature_extractor=_SCREAMING_SNAKE_CASE       ,			decoder=_SCREAMING_SNAKE_CASE)

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

				__lowerCAmelCase      :  Union[str, Any]												=       2.0
				__lowerCAmelCase      :  str												=       5.0
				__lowerCAmelCase      :  int												=       -20.0
				__lowerCAmelCase      :  Any												=       True

				__lowerCAmelCase      :  Optional[Any]												=       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       ,			)
				__lowerCAmelCase      :  Tuple												=       decoded_processor_out.text

				__lowerCAmelCase      :  Union[str, Any]												=       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:
							__lowerCAmelCase      :  Any												=       decoder.decode_beams_batch(
							    _SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE       ,			)

				__lowerCAmelCase      :  Optional[Any]												=       [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)
				__lowerCAmelCase      :  Optional[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       ,			_SCREAMING_SNAKE_CASE)

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

				"""simple docstring"""

				__lowerCAmelCase      :  List[Any]												=       WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm")

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

				__lowerCAmelCase      :  List[Any]												=       os.listdir(_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase      :  List[Any]												=       ["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 							_SCREAMING_SNAKE_CASE   (     self:		Optional[Any])				->       Optional[Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  str												=       snapshot_download("hf-internal-testing/processor_with_lm")

				__lowerCAmelCase      :  str												=       WavaVecaProcessorWithLM.from_pretrained(_SCREAMING_SNAKE_CASE)

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

				__lowerCAmelCase      :  Optional[Any]												=       os.listdir(_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase      :  List[Any]												=       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 							_SCREAMING_SNAKE_CASE   (     self:		Optional[int])				->       str:

				"""simple docstring"""

				__lowerCAmelCase      :  Dict												=       WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm")
				__lowerCAmelCase      :  Dict												=       AutoProcessor.from_pretrained("hf-internal-testing/processor_with_lm")

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

				__lowerCAmelCase      :  List[Any]												=       processor_wavaveca(_SCREAMING_SNAKE_CASE       ,			return_tensors="np")
				__lowerCAmelCase      :  Tuple												=       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)

				__lowerCAmelCase      :  Optional[Any]												=       self._get_dummy_logits()

				__lowerCAmelCase      :  List[Any]												=       processor_wavaveca.batch_decode(_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase      :  Tuple												=       processor_auto.batch_decode(_SCREAMING_SNAKE_CASE)

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

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

				"""simple docstring"""

				__lowerCAmelCase      :  Tuple												=       self.get_feature_extractor()
				__lowerCAmelCase      :  int												=       self.get_tokenizer()
				__lowerCAmelCase      :  Tuple												=       self.get_decoder()

				__lowerCAmelCase      :  List[str]												=       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 							_SCREAMING_SNAKE_CASE   (     _SCREAMING_SNAKE_CASE:		Any       ,			_SCREAMING_SNAKE_CASE:		Optional[Any])				->       Any:

				"""simple docstring"""

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

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

				"""simple docstring"""

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

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

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[int]												=       WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm")
				__lowerCAmelCase      :  str												=       self._get_dummy_logits()

				__lowerCAmelCase      :  Tuple												=       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 							_SCREAMING_SNAKE_CASE   (     self:		int)				->       Any:

				"""simple docstring"""

				import torch

				__lowerCAmelCase      :  int												=       load_dataset("common_voice"       ,			"en"       ,			split="train"       ,			streaming=_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase      :  Tuple												=       ds.cast_column("audio"       ,			datasets.Audio(sampling_rate=1_6000))
				__lowerCAmelCase      :  int												=       iter(_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase      :  Any												=       next(_SCREAMING_SNAKE_CASE)

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

				with torch.no_grad():
							__lowerCAmelCase      :  int												=       model(_SCREAMING_SNAKE_CASE).logits.cpu().numpy()

				__lowerCAmelCase      :  Optional[int]												=       processor.decode(logits[0]       ,			output_word_offsets=_SCREAMING_SNAKE_CASE)

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

				__lowerCAmelCase      :  List[str]												=       "WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL"

				# output words
				self.assertEqual(" ".join(self.get_from_offsets(_SCREAMING_SNAKE_CASE       ,			"word"))       ,			_SCREAMING_SNAKE_CASE)
				self.assertEqual(" ".join(self.get_from_offsets(_SCREAMING_SNAKE_CASE       ,			"word"))       ,			output.text)

				# output times
				__lowerCAmelCase      :  Any												=       torch.tensor(self.get_from_offsets(_SCREAMING_SNAKE_CASE       ,			"start_time"))
				__lowerCAmelCase      :  Union[str, Any]												=       torch.tensor(self.get_from_offsets(_SCREAMING_SNAKE_CASE       ,			"end_time"))

				# fmt: off
				__lowerCAmelCase      :  Union[str, Any]												=       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      :  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(_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 numpy as np
from scipy.spatial.distance import cdist
from sklearn.metrics import fa_score

import datasets


__snake_case							:   Optional[int]      =    '\\n    @inproceedings{kakwani2020indicnlpsuite,\n    title={{IndicNLPSuite: Monolingual Corpora, Evaluation Benchmarks and Pre-trained Multilingual Language Models for Indian Languages}},\n    author={Divyanshu Kakwani and Anoop Kunchukuttan and Satish Golla and Gokul N.C. and Avik Bhattacharyya and Mitesh M. Khapra and Pratyush Kumar},\n    year={2020},\n    booktitle={Findings of EMNLP},\n}\n'

__snake_case							:   str      =    '\\n    IndicGLUE is a natural language understanding benchmark for Indian languages. It contains a wide\n    variety of tasks and covers 11 major Indian languages - as, bn, gu, hi, kn, ml, mr, or, pa, ta, te.\n'

__snake_case							:   str      =    '\nCompute IndicGLUE evaluation metric associated to each IndicGLUE dataset.\nArgs:\n    predictions: list of predictions to score (as int64),\n        except for \'cvit-mkb-clsr\' where each prediction is a vector (of float32).\n    references: list of ground truth labels corresponding to the predictions (as int64),\n        except for \'cvit-mkb-clsr\' where each reference is a vector (of float32).\nReturns: depending on the IndicGLUE subset, one or several of:\n    "accuracy": Accuracy\n    "f1": F1 score\n    "precision": Precision@10\nExamples:\n\n    >>> indic_glue_metric = datasets.load_metric(\'indic_glue\', \'wnli\')  # \'wnli\' or any of ["copa", "sna", "csqa", "wstp", "inltkh", "bbca", "iitp-mr", "iitp-pr", "actsa-sc", "md"]\n    >>> references = [0, 1]\n    >>> predictions = [0, 1]\n    >>> results = indic_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'accuracy\': 1.0}\n\n    >>> indic_glue_metric = datasets.load_metric(\'indic_glue\', \'wiki-ner\')\n    >>> references = [0, 1]\n    >>> predictions = [0, 1]\n    >>> results = indic_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'accuracy\': 1.0, \'f1\': 1.0}\n\n    >>> indic_glue_metric = datasets.load_metric(\'indic_glue\', \'cvit-mkb-clsr\')\n    >>> references = [[0.5, 0.5, 0.5], [0.1, 0.2, 0.3]]\n    >>> predictions = [[0.5, 0.5, 0.5], [0.1, 0.2, 0.3]]\n    >>> results = indic_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'precision@10\': 1.0}\n\n'


def 			_lowercase				(  __snake_case					,__snake_case	)						->   Union[str, Any]:
			return float((preds == labels).mean()	)


def 			_lowercase				(  __snake_case					,__snake_case	)						->   str:
			__lowerCAmelCase      :  str												=       simple_accuracy(__snake_case					,__snake_case	)
			__lowerCAmelCase      :  Any												=       float(fa_score(y_true=__snake_case					,y_pred=__snake_case	)	)
			return {
			    "accuracy": acc,
			    "f1": fa,
			}


def 			_lowercase				(  __snake_case					,__snake_case	)						->   int:
			__lowerCAmelCase      :  Union[str, Any]												=       np.array(__snake_case	)
			__lowerCAmelCase      :  Tuple												=       np.array(__snake_case	)
			__lowerCAmelCase      :  List[Any]												=       en_sentvecs.shape[0]

			# mean centering
			__lowerCAmelCase      :  Union[str, Any]												=       en_sentvecs - np.mean(__snake_case					,axis=0	)
			__lowerCAmelCase      :  int												=       in_sentvecs - np.mean(__snake_case					,axis=0	)

			__lowerCAmelCase      :  Optional[Any]												=       cdist(__snake_case					,__snake_case					,"cosine"	)
			__lowerCAmelCase      :  int												=       np.array(range(__snake_case	)	)
			__lowerCAmelCase      :  int												=       sim.argsort(axis=1	)[:, :10]
			__lowerCAmelCase      :  Optional[Any]												=       np.any(preds == actual[:, None]					,axis=1	)
			return float(matches.mean()	)





@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION  ,		_KWARGS_DESCRIPTION							)
class 			A__   (  datasets.Metric							):




	'''simple docstring'''





	def 							_SCREAMING_SNAKE_CASE   (     self:		int)				->       str:

				"""simple docstring"""

				if self.config_name not in [
				    "wnli",
				    "copa",
				    "sna",
				    "csqa",
				    "wstp",
				    "inltkh",
				    "bbca",
				    "cvit-mkb-clsr",
				    "iitp-mr",
				    "iitp-pr",
				    "actsa-sc",
				    "md",
				    "wiki-ner",
				]:
							raise KeyError(
							    "You should supply a configuration name selected in "
							    "[\"wnli\", \"copa\", \"sna\", \"csqa\", \"wstp\", \"inltkh\", \"bbca\", "
							    "\"cvit-mkb-clsr\", \"iitp-mr\", \"iitp-pr\", \"actsa-sc\", \"md\", "
							    "\"wiki-ner\"]")
				return datasets.MetricInfo(
				    description=_DESCRIPTION       ,			citation=_CITATION       ,			inputs_description=_KWARGS_DESCRIPTION       ,			features=datasets.Features(
				        {
				            "predictions": datasets.Value("int64")
				            if self.config_name != "cvit-mkb-clsr"
				            else datasets.Sequence(datasets.Value("float32")),
				            "references": datasets.Value("int64")
				            if self.config_name != "cvit-mkb-clsr"
				            else datasets.Sequence(datasets.Value("float32")),
				        })       ,			codebase_urls=[]       ,			reference_urls=[]       ,			format="numpy" if self.config_name != "cvit-mkb-clsr" else None       ,			)







	def 							_SCREAMING_SNAKE_CASE   (     self:		List[str]       ,			_SCREAMING_SNAKE_CASE:		int       ,			_SCREAMING_SNAKE_CASE:		Optional[Any])				->       int:

				"""simple docstring"""

				if self.config_name == "cvit-mkb-clsr":
							return {"precision@10": precision_at_aa(_SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE)}
				elif self.config_name in ["wiki-ner"]:
							return acc_and_fa(_SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE)
				elif self.config_name in [
				    "wnli",
				    "copa",
				    "sna",
				    "csqa",
				    "wstp",
				    "inltkh",
				    "bbca",
				    "iitp-mr",
				    "iitp-pr",
				    "actsa-sc",
				    "md",
				]:
							return {"accuracy": simple_accuracy(_SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE)}
				else:
							raise KeyError(
							    "You should supply a configuration name selected in "
							    "[\"wnli\", \"copa\", \"sna\", \"csqa\", \"wstp\", \"inltkh\", \"bbca\", "
							    "\"cvit-mkb-clsr\", \"iitp-mr\", \"iitp-pr\", \"actsa-sc\", \"md\", "
							    "\"wiki-ner\"]")