

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   __lowercase      (unittest.TestCase ):
 def 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





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

       __lowerCAmelCase					:						str					=					{
           '''unk_token''': '''<unk>''',
           '''bos_token''': '''<s>''',
           '''eos_token''': '''</s>''',
       }
       __lowerCAmelCase					:						Dict					=					{
           '''feature_size''': 1,
           '''padding_value''': 0.0,
           '''sampling_rate''': 1_6000,
           '''return_attention_mask''': False,
           '''do_normalize''': True,
       }

       __lowerCAmelCase					:						int					=					tempfile.mkdtemp()
       __lowerCAmelCase					:						List[Any]					=					os.path.join(self.tmpdirname      ,							VOCAB_FILES_NAMES['''vocab_file''']					)
       __lowerCAmelCase					:						List[str]					=					os.path.join(self.tmpdirname      ,							A_					)
       with open(self.vocab_file      ,							'''w'''      ,							encoding='''utf-8'''					) as fp:
             fp.write(json.dumps(A_					) + '''\n'''					)

       with open(self.feature_extraction_file      ,							'''w'''      ,							encoding='''utf-8'''					) as fp:
             fp.write(json.dumps(A_					) + '''\n'''					)

       # load decoder from hub
       __lowerCAmelCase					:						Dict					=					'''hf-internal-testing/ngram-beam-search-decoder'''
 def 	UpperCamelCase__  (							self      ,							**A_					)					->Optional[int]:



       '''simple docstring'''





       __lowerCAmelCase					:						List[str]					=					self.add_kwargs_tokens_map.copy()
       kwargs.update(A_					)
       return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname      ,							**A_					)
 def 	UpperCamelCase__  (							self      ,							**A_					)					->Tuple:



       '''simple docstring'''





       return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname      ,							**A_					)
 def 	UpperCamelCase__  (							self      ,							**A_					)					->str:



       '''simple docstring'''





       return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name      ,							**A_					)
 def 	UpperCamelCase__  (							self					)					->List[str]:



       '''simple docstring'''





       shutil.rmtree(self.tmpdirname					)
 def 	UpperCamelCase__  (							self					)					->List[str]:



       '''simple docstring'''





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

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

       processor.save_pretrained(self.tmpdirname					)
       __lowerCAmelCase					:						Dict					=					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 	UpperCamelCase__  (							self					)					->Any:



       '''simple docstring'''





       __lowerCAmelCase					:						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
       __lowerCAmelCase					:						List[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 	UpperCamelCase__  (							self					)					->List[str]:



       '''simple docstring'''





       __lowerCAmelCase					:						List[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 	UpperCamelCase__  (							self					)					->Tuple:



       '''simple docstring'''





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

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

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

       __lowerCAmelCase					:						Tuple					=					feature_extractor(A_      ,							return_tensors='''np'''					)
       __lowerCAmelCase					:						str					=					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 	UpperCamelCase__  (							self					)					->str:



       '''simple docstring'''





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

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

       __lowerCAmelCase					:						Dict					=					'''This is a test string'''

       __lowerCAmelCase					:						Dict					=					processor(text=A_					)

       __lowerCAmelCase					:						List[Any]					=					tokenizer(A_					)

       for key in encoded_tok.keys():
             self.assertListEqual(encoded_tok[key]      ,							encoded_processor[key]					)
 def 	UpperCamelCase__  (							self      ,							A_=(2, 10, 16)      ,							A_=77					)					->Tuple:



       '''simple docstring'''





       np.random.seed(A_					)
       return np.random.rand(*A_					)
 def 	UpperCamelCase__  (							self					)					->Tuple:



       '''simple docstring'''





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

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

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

       __lowerCAmelCase					:						Any					=					processor.decode(A_					)

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



       '''simple docstring'''





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

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

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

       __lowerCAmelCase					:						List[str]					=					list(A_					)

       with get_context('''fork'''					).Pool() as p:
             __lowerCAmelCase					:						str					=					decoder.decode_beams_batch(A_      ,							A_					)

       __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(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 	UpperCamelCase__  (							self					)					->str:



       '''simple docstring'''





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

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

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

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

       __lowerCAmelCase					:						Optional[int]					=					processor.batch_decode(
           A_      ,							beam_width=A_      ,							beam_prune_logp=A_      ,							token_min_logp=A_      ,							)
       __lowerCAmelCase					:						Union[str, Any]					=					decoded_processor_out.text

       __lowerCAmelCase					:						Union[str, Any]					=					list(A_					)

       with get_context('''fork'''					).Pool() as pool:
             __lowerCAmelCase					:						List[str]					=					decoder.decode_beams_batch(
                 A_      ,							A_      ,							beam_width=A_      ,							beam_prune_logp=A_      ,							token_min_logp=A_      ,							)

       __lowerCAmelCase					:						Any					=					[d[0][0] for d in decoded_decoder_out]
       __lowerCAmelCase					:						Optional[Any]					=					[d[0][2] for d in decoded_decoder_out]
       __lowerCAmelCase					:						List[str]					=					[d[0][3] for d in decoded_decoder_out]

       self.assertListEqual(A_      ,							A_					)
       self.assertListEqual(['''</s> <s> <s>''', '''<s> <s> <s>''']      ,							A_					)

       self.assertTrue(np.array_equal(A_      ,							decoded_processor_out.logit_score					)					)
       self.assertTrue(np.allclose([-20.054, -18.447]      ,							A_      ,							atol=1e-3					)					)

       self.assertTrue(np.array_equal(A_      ,							decoded_processor_out.lm_score					)					)
       self.assertTrue(np.allclose([-15.554, -13.9_474]      ,							A_      ,							atol=1e-3					)					)
 def 	UpperCamelCase__  (							self					)					->Optional[Any]:



       '''simple docstring'''





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

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

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

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

       __lowerCAmelCase					:						Optional[int]					=					processor.batch_decode(
           A_      ,							alpha=A_      ,							beta=A_      ,							unk_score_offset=A_      ,							lm_score_boundary=A_      ,							)
       __lowerCAmelCase					:						Dict					=					decoded_processor_out.text

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

       with get_context('''fork'''					).Pool() as pool:
             __lowerCAmelCase					:						int					=					decoder.decode_beams_batch(
                 A_      ,							A_      ,							)

       __lowerCAmelCase					:						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_					)
       __lowerCAmelCase					:						List[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 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





       __lowerCAmelCase					:						str					=					WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''					)

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

       __lowerCAmelCase					:						Union[str, Any]					=					os.listdir(A_					)
       __lowerCAmelCase					:						Dict					=					['''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 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





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

       __lowerCAmelCase					:						Optional[Any]					=					WavaVecaProcessorWithLM.from_pretrained(A_					)

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

       __lowerCAmelCase					:						List[str]					=					os.listdir(A_					)
       __lowerCAmelCase					:						Union[str, Any]					=					os.listdir(A_					)

       local_decoder_files.sort()
       expected_decoder_files.sort()

       # test that both decoder form hub and local files in cache are the same
       self.assertListEqual(A_      ,							A_					)
 def 	UpperCamelCase__  (							self					)					->str:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[Any]					=					WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''					)
       __lowerCAmelCase					:						Optional[int]					=					AutoProcessor.from_pretrained('''hf-internal-testing/processor_with_lm'''					)

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

       __lowerCAmelCase					:						int					=					processor_wavaveca(A_      ,							return_tensors='''np'''					)
       __lowerCAmelCase					:						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					)

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

       __lowerCAmelCase					:						int					=					processor_wavaveca.batch_decode(A_					)
       __lowerCAmelCase					:						Tuple					=					processor_auto.batch_decode(A_					)

       self.assertListEqual(decoded_wavaveca.text      ,							decoded_auto.text					)
 def 	UpperCamelCase__  (							self					)					->Dict:



       '''simple docstring'''





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

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

       self.assertListEqual(
           processor.model_input_names      ,							feature_extractor.model_input_names      ,							msg='''`processor` and `feature_extractor` model input names do not match'''      ,							)
 @staticmethod
 def 	UpperCamelCase__  (							A_      ,							A_					)					->Union[str, Any]:



       '''simple docstring'''





       __lowerCAmelCase					:						List[str]					=					[d[key] for d in offsets]
       return retrieved_list
 def 	UpperCamelCase__  (							self					)					->List[Any]:



       '''simple docstring'''





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

       __lowerCAmelCase					:						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 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





       __lowerCAmelCase					:						List[str]					=					WavaVecaProcessorWithLM.from_pretrained('''hf-internal-testing/processor_with_lm'''					)
       __lowerCAmelCase					:						Any					=					self._get_dummy_logits()

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



       '''simple docstring'''





       import torch

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

       __lowerCAmelCase					:						str					=					AutoProcessor.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''					)
       __lowerCAmelCase					:						str					=					WavaVecaForCTC.from_pretrained('''patrickvonplaten/wav2vec2-base-100h-with-lm'''					)

       # compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
       __lowerCAmelCase					:						Tuple					=					processor(sample['''audio''']['''array''']      ,							return_tensors='''pt'''					).input_values

       with torch.no_grad():
             __lowerCAmelCase					:						Any					=					model(A_					).logits.cpu().numpy()

       __lowerCAmelCase					:						Dict					=					processor.decode(logits[0]      ,							output_word_offsets=A_					)

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

       __lowerCAmelCase					:						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
       __lowerCAmelCase					:						Tuple					=					torch.tensor(self.get_from_offsets(A_      ,							'''start_time'''					)					)
       __lowerCAmelCase					:						Tuple					=					torch.tensor(self.get_from_offsets(A_      ,							'''end_time'''					)					)

       # fmt: off
       __lowerCAmelCase					:						Tuple					=					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]					)
       __lowerCAmelCase					:						Dict					=					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					)					)





import unittest

import numpy as np
import torch

from diffusers import PNDMPipeline, PNDMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()







class   __lowercase      (unittest.TestCase ):
 @property
 def 	UpperCamelCase__  (							self					)					->Tuple:



       '''simple docstring'''





       torch.manual_seed(0					)
       __lowerCAmelCase					:						List[Any]					=					UNetaDModel(
           block_out_channels=(32, 64)      ,							layers_per_block=2      ,							sample_size=32      ,							in_channels=3      ,							out_channels=3      ,							down_block_types=('''DownBlock2D''', '''AttnDownBlock2D''')      ,							up_block_types=('''AttnUpBlock2D''', '''UpBlock2D''')      ,							)
       return model





 def 	UpperCamelCase__  (							self					)					->int:



       '''simple docstring'''





       __lowerCAmelCase					:						List[str]					=					self.dummy_uncond_unet
       __lowerCAmelCase					:						Any					=					PNDMScheduler()

       __lowerCAmelCase					:						Dict					=					PNDMPipeline(unet=A_      ,							scheduler=A_					)
       pndm.to(A_					)
       pndm.set_progress_bar_config(disable=A_					)

       __lowerCAmelCase					:						Optional[Any]					=					torch.manual_seed(0					)
       __lowerCAmelCase					:						Any					=					pndm(generator=A_      ,							num_inference_steps=20      ,							output_type='''numpy'''					).images

       __lowerCAmelCase					:						Optional[Any]					=					torch.manual_seed(0					)
       __lowerCAmelCase					:						List[Any]					=					pndm(generator=A_      ,							num_inference_steps=20      ,							output_type='''numpy'''      ,							return_dict=A_					)[0]

       __lowerCAmelCase					:						Tuple					=					image[0, -3:, -3:, -1]
       __lowerCAmelCase					:						Optional[Any]					=					image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       __lowerCAmelCase					:						int					=					np.array([1.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0]					)

       assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-2
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice					).max() < 1e-2







@slow
@require_torch
class   __lowercase      (unittest.TestCase ):
 def 	UpperCamelCase__  (							self					)					->Optional[Any]:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[int]					=					'''google/ddpm-cifar10-32'''

       __lowerCAmelCase					:						Union[str, Any]					=					UNetaDModel.from_pretrained(A_					)
       __lowerCAmelCase					:						int					=					PNDMScheduler()

       __lowerCAmelCase					:						Any					=					PNDMPipeline(unet=A_      ,							scheduler=A_					)
       pndm.to(A_					)
       pndm.set_progress_bar_config(disable=A_					)
       __lowerCAmelCase					:						Tuple					=					torch.manual_seed(0					)
       __lowerCAmelCase					:						Any					=					pndm(generator=A_      ,							output_type='''numpy'''					).images

       __lowerCAmelCase					:						Optional[int]					=					image[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       __lowerCAmelCase					:						List[Any]					=					np.array([0.1_564, 0.14_645, 0.1_406, 0.14_715, 0.12_425, 0.14_045, 0.13_115, 0.12_175, 0.125]					)

       assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-2


