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



				def  _lowercase					(self    :				List[str])   ->     List[str]:
					__snake_case      :							int						=			'| <pad> <unk> <s> </s> a b c d e f g h i j k'.split()
					__snake_case      :							str						=			dict(zip(_A     ,  range(len(_A))))

					__snake_case      :							List[Any]						=			{
					    'unk_token': '<unk>',
					    'bos_token': '<s>',
					    'eos_token': '</s>',
					}
					__snake_case      :							Dict						=			{
					    'feature_size': 1,
					    'padding_value': 0.0,
					    'sampling_rate': 1_60_00,
					    'return_attention_mask': False,
					    'do_normalize': True,
					}

					__snake_case      :							List[str]						=			tempfile.mkdtemp()
					__snake_case      :							Dict						=			os.path.join(self.tmpdirname     ,  VOCAB_FILES_NAMES['vocab_file'])
					__snake_case      :							Optional[int]						=			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      :							str						=			'hf-internal-testing/ngram-beam-search-decoder'



				def  _lowercase					(self    :				str     ,  **_A    :				Any)   ->     Optional[int]:
					__snake_case      :							Tuple						=			self.add_kwargs_tokens_map.copy()
					kwargs.update(_A)
					return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname     ,  **_A)



				def  _lowercase					(self    :				List[str]     ,  **_A    :				Any)   ->     Optional[int]:
					return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname     ,  **_A)



				def  _lowercase					(self    :				Optional[Any]     ,  **_A    :				Union[str, Any])   ->     Union[str, Any]:
					return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name     ,  **_A)



				def  _lowercase					(self    :				Any)   ->     str:
					shutil.rmtree(self.tmpdirname)



				def  _lowercase					(self    :				str)   ->     Optional[Any]:
					__snake_case      :							Dict						=			self.get_tokenizer()
					__snake_case      :							Tuple						=			self.get_feature_extractor()
					__snake_case      :							Union[str, Any]						=			self.get_decoder()

					__snake_case      :							int						=			WavaVecaProcessorWithLM(tokenizer=_A     ,  feature_extractor=_A     ,  decoder=_A)

					processor.save_pretrained(self.tmpdirname)
					__snake_case      :							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  _lowercase					(self    :				Dict)   ->     List[str]:
					__snake_case      :							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
					__snake_case      :							Optional[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  _lowercase					(self    :				int)   ->     str:
					__snake_case      :							Tuple						=			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  _lowercase					(self    :				Optional[Any])   ->     List[Any]:
					__snake_case      :							Dict						=			self.get_feature_extractor()
					__snake_case      :							Optional[Any]						=			self.get_tokenizer()
					__snake_case      :							List[Any]						=			self.get_decoder()

					__snake_case      :							Optional[Any]						=			WavaVecaProcessorWithLM(tokenizer=_A     ,  feature_extractor=_A     ,  decoder=_A)

					__snake_case      :							int						=			floats_list((3, 10_00))

					__snake_case      :							Tuple						=			feature_extractor(_A     ,  return_tensors='np')
					__snake_case      :							Optional[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  _lowercase					(self    :				Union[str, Any])   ->     Dict:
					__snake_case      :							Optional[int]						=			self.get_feature_extractor()
					__snake_case      :							List[str]						=			self.get_tokenizer()
					__snake_case      :							str						=			self.get_decoder()

					__snake_case      :							Optional[int]						=			WavaVecaProcessorWithLM(tokenizer=_A     ,  feature_extractor=_A     ,  decoder=_A)

					__snake_case      :							Any						=			'This is a test string'

					__snake_case      :							Any						=			processor(text=_A)

					__snake_case      :							str						=			tokenizer(_A)

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



				def  _lowercase					(self    :				List[str]     ,  _A    :				Any=(2, 10, 16)     ,  _A    :				int=77)   ->     List[str]:
					np.random.seed(_A)
					return np.random.rand(*_A)



				def  _lowercase					(self    :				str)   ->     Optional[Any]:
					__snake_case      :							Dict						=			self.get_feature_extractor()
					__snake_case      :							Any						=			self.get_tokenizer()
					__snake_case      :							Tuple						=			self.get_decoder()

					__snake_case      :							str						=			WavaVecaProcessorWithLM(tokenizer=_A     ,  feature_extractor=_A     ,  decoder=_A)

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

					__snake_case      :							Optional[int]						=			processor.decode(_A)

					__snake_case      :							Dict						=			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  _lowercase					(self    :				str     ,  _A    :				str)   ->     Any:
					__snake_case      :							Tuple						=			self.get_feature_extractor()
					__snake_case      :							int						=			self.get_tokenizer()
					__snake_case      :							List[Any]						=			self.get_decoder()

					__snake_case      :							Any						=			WavaVecaProcessorWithLM(tokenizer=_A     ,  feature_extractor=_A     ,  decoder=_A)

					__snake_case      :							Optional[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:
						__snake_case      :							Tuple						=			processor.batch_decode(_A)
					else:
						with get_context(_A).Pool() as pool:
							__snake_case      :							Any						=			processor.batch_decode(_A     ,  _A)

					__snake_case      :							Any						=			list(_A)

					with get_context('fork').Pool() as p:
						__snake_case      :							Dict						=			decoder.decode_beams_batch(_A     ,  _A)

					__snake_case		,    __snake_case		,    __snake_case      :							List[str]						=			[], [], []
					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  _lowercase					(self    :				Optional[int])   ->     List[str]:
					__snake_case      :							Dict						=			self.get_feature_extractor()
					__snake_case      :							Any						=			self.get_tokenizer()
					__snake_case      :							Union[str, Any]						=			self.get_decoder()

					__snake_case      :							Any						=			WavaVecaProcessorWithLM(tokenizer=_A     ,  feature_extractor=_A     ,  decoder=_A)

					__snake_case      :							List[str]						=			self._get_dummy_logits()

					__snake_case      :							int						=			15
					__snake_case      :							str						=			-20.0
					__snake_case      :							Dict						=			-4.0

					__snake_case      :							Tuple						=			processor.batch_decode(
					    _A     ,  beam_width=_A     ,  beam_prune_logp=_A     ,  token_min_logp=_A     ,  )
					__snake_case      :							str						=			decoded_processor_out.text

					__snake_case      :							int						=			list(_A)

					with get_context('fork').Pool() as pool:
						__snake_case      :							Dict						=			decoder.decode_beams_batch(
						    _A     ,  _A     ,  beam_width=_A     ,  beam_prune_logp=_A     ,  token_min_logp=_A     ,  )

					__snake_case      :							Union[str, Any]						=			[d[0][0] for d in decoded_decoder_out]
					__snake_case      :							str						=			[d[0][2] for d in decoded_decoder_out]
					__snake_case      :							Dict						=			[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.9_474]     ,  _A     ,  atol=1E-3))



				def  _lowercase					(self    :				List[str])   ->     Tuple:
					__snake_case      :							int						=			self.get_feature_extractor()
					__snake_case      :							Union[str, Any]						=			self.get_tokenizer()
					__snake_case      :							Optional[Any]						=			self.get_decoder()

					__snake_case      :							str						=			WavaVecaProcessorWithLM(tokenizer=_A     ,  feature_extractor=_A     ,  decoder=_A)

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

					__snake_case      :							List[Any]						=			2.0
					__snake_case      :							List[Any]						=			5.0
					__snake_case      :							List[Any]						=			-20.0
					__snake_case      :							List[str]						=			True

					__snake_case      :							Tuple						=			processor.batch_decode(
					    _A     ,  alpha=_A     ,  beta=_A     ,  unk_score_offset=_A     ,  lm_score_boundary=_A     ,  )
					__snake_case      :							Dict						=			decoded_processor_out.text

					__snake_case      :							str						=			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      :							List[str]						=			decoder.decode_beams_batch(
						    _A     ,  _A     ,  )

					__snake_case      :							int						=			[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      :							int						=			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  _lowercase					(self    :				List[Any])   ->     Optional[int]:
					__snake_case      :							List[Any]						=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm')

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

					__snake_case      :							Optional[int]						=			os.listdir(_A)
					__snake_case      :							Tuple						=			['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  _lowercase					(self    :				str)   ->     Tuple:
					__snake_case      :							str						=			snapshot_download('hf-internal-testing/processor_with_lm')

					__snake_case      :							int						=			WavaVecaProcessorWithLM.from_pretrained(_A)

					__snake_case      :							Optional[int]						=			processor.decoder.model_container[processor.decoder._model_key]
					__snake_case      :							Tuple						=			Path(language_model._kenlm_model.path.decode('utf-8')).parent.parent.absolute()

					__snake_case      :							Any						=			os.listdir(_A)
					__snake_case      :							Dict						=			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  _lowercase					(self    :				List[Any])   ->     List[Any]:
					__snake_case      :							Tuple						=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm')
					__snake_case      :							Dict						=			AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm')

					__snake_case      :							Optional[Any]						=			floats_list((3, 10_00))

					__snake_case      :							Dict						=			processor_wavaveca(_A     ,  return_tensors='np')
					__snake_case      :							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)

					__snake_case      :							List[str]						=			self._get_dummy_logits()

					__snake_case      :							List[Any]						=			processor_wavaveca.batch_decode(_A)
					__snake_case      :							List[Any]						=			processor_auto.batch_decode(_A)

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



				def  _lowercase					(self    :				Optional[Any])   ->     Optional[Any]:
					__snake_case      :							List[Any]						=			self.get_feature_extractor()
					__snake_case      :							int						=			self.get_tokenizer()
					__snake_case      :							Union[str, Any]						=			self.get_decoder()

					__snake_case      :							Tuple						=			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  _lowercase					(_A    :				Any     ,  _A    :				Tuple)   ->     str:
					__snake_case      :							Tuple						=			[d[key] for d in offsets]
					return retrieved_list



				def  _lowercase					(self    :				int)   ->     int:
					__snake_case      :							str						=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm')
					__snake_case      :							List[Any]						=			self._get_dummy_logits()[0]

					__snake_case      :							List[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  _lowercase					(self    :				List[str])   ->     List[Any]:
					__snake_case      :							List[Any]						=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm')
					__snake_case      :							Any						=			self._get_dummy_logits()

					__snake_case      :							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  _lowercase					(self    :				str)   ->     Union[str, Any]:
					import torch

					__snake_case      :							Optional[Any]						=			load_dataset('common_voice'     ,  'en'     ,  split='train'     ,  streaming=_A)
					__snake_case      :							Tuple						=			ds.cast_column('audio'     ,  datasets.Audio(sampling_rate=1_60_00))
					__snake_case      :							int						=			iter(_A)
					__snake_case      :							Optional[int]						=			next(_A)

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

					with torch.no_grad():
						__snake_case      :							Union[str, Any]						=			model(_A).logits.cpu().numpy()

					__snake_case      :							Union[str, Any]						=			processor.decode(logits[0]     ,  output_word_offsets=_A)

					__snake_case      :							List[str]						=			model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
					__snake_case      :							Dict						=			[
					    {
					        '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      :							Optional[Any]						=			'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      :							Tuple						=			torch.tensor(self.get_from_offsets(_A     ,  'start_time'))
					__snake_case      :							List[Any]						=			torch.tensor(self.get_from_offsets(_A     ,  'end_time'))

					# fmt: off
					__snake_case      :							List[Any]						=			torch.tensor([1.4_199, 1.6_599, 2.2_599, 3.0, 3.24, 3.5_999, 3.7_999, 4.0_999, 4.26, 4.94, 5.28, 5.6_599, 5.78, 5.94, 6.32, 6.5_399, 6.6_599])
					__snake_case      :							str						=			torch.tensor([1.5_399, 1.8_999, 2.9, 3.16, 3.5_399, 3.72, 4.0_199, 4.1_799, 4.76, 5.1_599, 5.5_599, 5.6_999, 5.86, 6.1_999, 6.38, 6.6_199, 6.94])
					# fmt: on

					self.assertTrue(torch.allclose(_A     ,  _A     ,  atol=0.01))
					self.assertTrue(torch.allclose(_A     ,  _A     ,  atol=0.01))






"""simple docstring"""







from __future__ import annotations

import os
from typing import Any

import requests

_a					:					int=			"https://api.github.com"

# https://docs.github.com/en/free-pro-team@latest/rest/reference/users#get-the-authenticated-user
_a					:					Dict=			BASE_URL + "/user"

# https://github.com/settings/tokens
_a					:					Union[str, Any]=			os.environ.get("USER_TOKEN", "")
def   __UpperCAmelCase    (   UpperCAmelCase_   :    str  )							->      dict[Any, Any]:

	'''simple docstring'''





	__snake_case      :							Tuple						=			{
	    'Authorization': F"token {auth_token}",
	    'Accept': 'application/vnd.github.v3+json',
	}
	return requests.get(UpperCAmelCase_      ,							headers=UpperCAmelCase_  ).json()


if __name__ == "__main__":  # pragma: no cover
	if USER_TOKEN:
		for key, value in fetch_github_info(USER_TOKEN).items():
			print(f'''{key}: {value}''')
	else:
		raise ValueError("'USER_TOKEN' field cannot be empty.")





