






'''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 		A      (  self  :		int      )    ->							int:
								UpperCAmelCase_      :       str          =       '''| <pad> <unk> <s> </s> a b c d e f g h i j k'''.split()
								UpperCAmelCase_      :       Dict          =       dict(zip(_A				, range(len(_A      )      )      )      )

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

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




					def 		A      (  self  :		Any				, **_A  :		Tuple      )    ->							List[str]:
								UpperCAmelCase_      :       Any          =       self.add_kwargs_tokens_map.copy()
								kwargs.update(_A      )
								return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname				, **_A      )




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




					def 		A      (  self  :		Tuple				, **_A  :		str      )    ->							str:
								return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name				, **_A      )




					def 		A      (  self  :		int      )    ->							int:
								shutil.rmtree(self.tmpdirname      )




					def 		A      (  self  :		str      )    ->							Optional[Any]:
								UpperCAmelCase_      :       Optional[int]          =       self.get_tokenizer()
								UpperCAmelCase_      :       List[str]          =       self.get_feature_extractor()
								UpperCAmelCase_      :       List[Any]          =       self.get_decoder()

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

								processor.save_pretrained(self.tmpdirname      )
								UpperCAmelCase_      :       Any          =       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 		A      (  self  :		List[str]      )    ->							Any:
								UpperCAmelCase_      :       Union[str, Any]          =       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
								UpperCAmelCase_      :       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 		A      (  self  :		List[Any]      )    ->							Optional[Any]:
								UpperCAmelCase_      :       str          =       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 		A      (  self  :		List[str]      )    ->							List[Any]:
								UpperCAmelCase_      :       str          =       self.get_feature_extractor()
								UpperCAmelCase_      :       List[str]          =       self.get_tokenizer()
								UpperCAmelCase_      :       Tuple          =       self.get_decoder()

								UpperCAmelCase_      :       Union[str, Any]          =       WavaVecaProcessorWithLM(tokenizer=_A				, feature_extractor=_A				, decoder=_A      )

								UpperCAmelCase_      :       Dict          =       floats_list((3, 10_00)      )

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

								UpperCAmelCase_      :       List[str]          =       WavaVecaProcessorWithLM(tokenizer=_A				, feature_extractor=_A				, decoder=_A      )

								UpperCAmelCase_      :       str          =       '''This is a test string'''

								UpperCAmelCase_      :       List[str]          =       processor(text=_A      )

								UpperCAmelCase_      :       str          =       tokenizer(_A      )

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




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




					def 		A      (  self  :		Optional[int]      )    ->							List[str]:
								UpperCAmelCase_      :       List[Any]          =       self.get_feature_extractor()
								UpperCAmelCase_      :       str          =       self.get_tokenizer()
								UpperCAmelCase_      :       Optional[int]          =       self.get_decoder()

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

								UpperCAmelCase_      :       List[str]          =       self._get_dummy_logits(shape=(10, 16)				, seed=13      )

								UpperCAmelCase_      :       Union[str, Any]          =       processor.decode(_A      )

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

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

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

								UpperCAmelCase_      :       Tuple          =       list(_A      )

								with get_context('''fork'''      ).Pool() as p:
											UpperCAmelCase_      :       Optional[int]          =       decoder.decode_beams_batch(_A				, _A      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       Tuple          =       [], [], []
								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 		A      (  self  :		Dict      )    ->							str:
								UpperCAmelCase_      :       Optional[int]          =       self.get_feature_extractor()
								UpperCAmelCase_      :       str          =       self.get_tokenizer()
								UpperCAmelCase_      :       Tuple          =       self.get_decoder()

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

								UpperCAmelCase_      :       Dict          =       self._get_dummy_logits()

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

								UpperCAmelCase_      :       Tuple          =       processor.batch_decode(
								    _A				, beam_width=_A				, beam_prune_logp=_A				, token_min_logp=_A				, )
								UpperCAmelCase_      :       Optional[int]          =       decoded_processor_out.text

								UpperCAmelCase_      :       int          =       list(_A      )

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

								UpperCAmelCase_      :       Optional[int]          =       [d[0][0] for d in decoded_decoder_out]
								UpperCAmelCase_      :       str          =       [d[0][2] for d in decoded_decoder_out]
								UpperCAmelCase_      :       int          =       [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 		A      (  self  :		Optional[int]      )    ->							Tuple:
								UpperCAmelCase_      :       Optional[Any]          =       self.get_feature_extractor()
								UpperCAmelCase_      :       Union[str, Any]          =       self.get_tokenizer()
								UpperCAmelCase_      :       Optional[Any]          =       self.get_decoder()

								UpperCAmelCase_      :       List[Any]          =       WavaVecaProcessorWithLM(tokenizer=_A				, feature_extractor=_A				, decoder=_A      )

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

								UpperCAmelCase_      :       List[Any]          =       2.0
								UpperCAmelCase_      :       Optional[Any]          =       5.0
								UpperCAmelCase_      :       List[str]          =       -20.0
								UpperCAmelCase_      :       Union[str, Any]          =       True

								UpperCAmelCase_      :       Optional[Any]          =       processor.batch_decode(
								    _A				, alpha=_A				, beta=_A				, unk_score_offset=_A				, lm_score_boundary=_A				, )
								UpperCAmelCase_      :       List[Any]          =       decoded_processor_out.text

								UpperCAmelCase_      :       Dict          =       list(_A      )
								decoder.reset_params(
								    alpha=_A				, beta=_A				, unk_score_offset=_A				, lm_score_boundary=_A				, )

								with get_context('''fork'''      ).Pool() as pool:
											UpperCAmelCase_      :       str          =       decoder.decode_beams_batch(
											    _A				, _A				, )

								UpperCAmelCase_      :       Optional[Any]          =       [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      )
								UpperCAmelCase_      :       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 		A      (  self  :		Tuple      )    ->							Union[str, Any]:
								UpperCAmelCase_      :       Optional[int]          =       WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )

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

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

								UpperCAmelCase_      :       Union[str, Any]          =       WavaVecaProcessorWithLM.from_pretrained(_A      )

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

								UpperCAmelCase_      :       Optional[int]          =       os.listdir(_A      )
								UpperCAmelCase_      :       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 		A      (  self  :		int      )    ->							List[str]:
								UpperCAmelCase_      :       Optional[Any]          =       WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )
								UpperCAmelCase_      :       Union[str, Any]          =       AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''      )

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

								UpperCAmelCase_      :       str          =       processor_wavaveca(_A				, return_tensors='''np'''      )
								UpperCAmelCase_      :       int          =       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      )

								UpperCAmelCase_      :       int          =       self._get_dummy_logits()

								UpperCAmelCase_      :       Any          =       processor_wavaveca.batch_decode(_A      )
								UpperCAmelCase_      :       int          =       processor_auto.batch_decode(_A      )

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




					def 		A      (  self  :		Dict      )    ->							Union[str, Any]:
								UpperCAmelCase_      :       Any          =       self.get_feature_extractor()
								UpperCAmelCase_      :       Optional[int]          =       self.get_tokenizer()
								UpperCAmelCase_      :       Any          =       self.get_decoder()

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




					def 		A      (  self  :		int      )    ->							int:
								UpperCAmelCase_      :       Any          =       WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''      )
								UpperCAmelCase_      :       Any          =       self._get_dummy_logits()[0]

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

								UpperCAmelCase_      :       Optional[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 		A      (  self  :		str      )    ->							int:
								import torch

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

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

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

								UpperCAmelCase_      :       int          =       processor.decode(logits[0]				, output_word_offsets=_A      )

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

								UpperCAmelCase_      :       Union[str, 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
								UpperCAmelCase_      :       List[str]          =       torch.tensor(self.get_from_offsets(_A				, '''start_time'''      )      )
								UpperCAmelCase_      :       Any          =       torch.tensor(self.get_from_offsets(_A				, '''end_time'''      )      )

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





import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
						import torch

						from transformers.generation import DisjunctiveConstraint


@require_torch
class  snake_case__      (					unittest.TestCase):




					def 		A      (  self  :		Union[str, Any]      )    ->							List[str]:
								# For consistency across different places the DisjunctiveConstraint is called,
								# dc.token_ids is a list of integers. It is also initialized only by integers.

								UpperCAmelCase_      :       Union[str, Any]          =       [[1, 2, 4], [1, 2, 3, 4]]
								UpperCAmelCase_      :       Dict          =       DisjunctiveConstraint(_A      )
								self.assertTrue(isinstance(dc.token_ids				, _A      )      )

								with self.assertRaises(_A      ):
											DisjunctiveConstraint(torch.LongTensor([[1, 2, 4], [1, 2, 3]]      )      )

								with self.assertRaises(_A      ):
											DisjunctiveConstraint([torch.LongTensor([1, 2, 4]      ), torch.LongTensor([1, 2, 3, 4, 5]      )]      )




					def 		A      (  self  :		Optional[int]      )    ->							List[Any]:
								# We can't have constraints that are complete subsets of another. This leads to a preverse
								# interpretation of "constraint fulfillment": does generating [1,2,3] fulfill the constraint?
								# It would mean that it generated [1,2] which fulfills it, but it's in the middle of potentially
								# fulfilling [1,2,3,4]. If we believe that [1,2,3] does fulfill the constraint, then the algorithm
								# will necessarily never reach [1,2,3,4], giving users a false sense of control (better to just not allow it).
								UpperCAmelCase_      :       Optional[Any]          =       [[1, 2], [1, 2, 3, 4]]

								with self.assertRaises(_A      ):
											DisjunctiveConstraint(_A      )  # fails here




					def 		A      (  self  :		List[str]      )    ->							int:
								UpperCAmelCase_      :       Dict          =       [[1, 2, 3], [1, 2, 4]]

								UpperCAmelCase_      :       List[Any]          =       DisjunctiveConstraint(_A      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       List[str]          =       dc.update(1      )
								UpperCAmelCase_      :       int          =       stepped is True and completed is False and reset is False
								self.assertTrue(_A      )
								self.assertTrue(not dc.completed      )
								self.assertTrue(dc.current_seq == [1]      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       Tuple          =       dc.update(2      )
								UpperCAmelCase_      :       Optional[int]          =       stepped is True and completed is False and reset is False
								self.assertTrue(_A      )
								self.assertTrue(not dc.completed      )
								self.assertTrue(dc.current_seq == [1, 2]      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       Any          =       dc.update(3      )
								UpperCAmelCase_      :       List[Any]          =       stepped is True and completed is True and reset is False
								self.assertTrue(_A      )
								self.assertTrue(dc.completed      )  # Completed!
								self.assertTrue(dc.current_seq == [1, 2, 3]      )





					def 		A      (  self  :		List[Any]      )    ->							Any:
								UpperCAmelCase_      :       int          =       [[1, 2, 3], [1, 2, 4, 5], [1, 2, 5]]

								UpperCAmelCase_      :       str          =       DisjunctiveConstraint(_A      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       Tuple          =       dc.update(1      )
								self.assertTrue(not dc.completed      )
								self.assertTrue(dc.current_seq == [1]      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       str          =       dc.update(2      )
								self.assertTrue(not dc.completed      )
								self.assertTrue(dc.current_seq == [1, 2]      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       str          =       dc.update(4      )
								self.assertTrue(not dc.completed      )
								self.assertTrue(dc.current_seq == [1, 2, 4]      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       Any          =       dc.update(5      )
								self.assertTrue(dc.completed      )  # Completed!
								self.assertTrue(dc.current_seq == [1, 2, 4, 5]      )

								dc.reset()

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       List[str]          =       dc.update(1      )
								self.assertTrue(not dc.completed      )
								self.assertTrue(dc.remaining() == 3      )
								self.assertTrue(dc.current_seq == [1]      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       Optional[int]          =       dc.update(2      )
								self.assertTrue(not dc.completed      )
								self.assertTrue(dc.remaining() == 2      )
								self.assertTrue(dc.current_seq == [1, 2]      )

								UpperCAmelCase_    ,      UpperCAmelCase_    ,      UpperCAmelCase_      :       int          =       dc.update(5      )
								self.assertTrue(dc.completed      )  # Completed!
								self.assertTrue(dc.remaining() == 0      )
								self.assertTrue(dc.current_seq == [1, 2, 5]      )


