



import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFGPTaLMHeadModel, is_keras_nlp_available, is_tf_available
from transformers.models.gpta.tokenization_gpta import GPTaTokenizer
from transformers.testing_utils import require_keras_nlp, require_tf, slow


if is_tf_available():
		import tensorflow as tf

if is_keras_nlp_available():
		from transformers.models.gpta import TFGPTaTokenizer


_lowercase						: Tuple     						=["gpt2"]
_lowercase						: Optional[Any]     						="gpt2"

if is_tf_available():

		class   snake_case__							(tf.Module						):







								"""simple docstring"""




								def __init__(  self   ,  __lowercase    )   -> Optional[int]:
											"""simple docstring"""




											super().__init__()
											a__  :		List[str]					=    tokenizer
											a__  :		Optional[int]					=    AutoConfig.from_pretrained(__UpperCAmelCase    )
											a__  :		Optional[Any]					=    TFGPTaLMHeadModel.from_config(__UpperCAmelCase    )

								@tf.function(input_signature=(tf.TensorSpec((None,)   ,  tf.string   ,  name="""text"""    ),)    )
								def 				SCREAMING_SNAKE_CASE__(  self   ,  __lowercase    )   -> Optional[Any]:
											"""simple docstring"""




											a__  :		str					=    self.tokenizer(__UpperCAmelCase    )
											a__  :		int					=    tokenized['input_ids'].to_tensor()

											a__  :		Optional[Any]					=    tf.cast(input_ids_dense > 0   ,  tf.intaa    )
											# input_mask = tf.reshape(input_mask, [-1, MAX_SEQ_LEN])

											a__  :		int					=    self.model(input_ids=__UpperCAmelCase   ,  attention_mask=__UpperCAmelCase    )['logits']

											return outputs

@require_tf
@require_keras_nlp
class   snake_case__							(unittest.TestCase						):







						"""simple docstring"""




						def 				SCREAMING_SNAKE_CASE__(  self    )   -> List[Any]:
									"""simple docstring"""




									super().setUp()

									a__  :		List[str]					=    [GPTaTokenizer.from_pretrained(__UpperCAmelCase    ) for checkpoint in (TOKENIZER_CHECKPOINTS)]
									a__  :		Union[str, Any]					=    [TFGPTaTokenizer.from_pretrained(__UpperCAmelCase    ) for checkpoint in TOKENIZER_CHECKPOINTS]
									assert len(self.tokenizers    ) == len(self.tf_tokenizers    )

									a__  :		int					=    [
									    'This is a straightforward English test sentence.',
									    'This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.',
									    'Now we\'re going to add some Chinese: 一 二 三 一二三',
									    'And some much more rare Chinese: 齉 堃 齉堃',
									    'Je vais aussi écrire en français pour tester les accents',
									    'Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ',
									]
									a__  :		str					=    list(zip(self.test_sentences   ,  self.test_sentences[::-1]    )    )

						def 				SCREAMING_SNAKE_CASE__(  self    )   -> Optional[int]:
									"""simple docstring"""




									for tokenizer, tf_tokenizer in zip(self.tokenizers   ,  self.tf_tokenizers    ):
												for test_inputs in self.test_sentences:
															a__  :		int					=    tokenizer([test_inputs]   ,  return_tensors="""tf"""    )
															a__  :		Optional[int]					=    tf_tokenizer([test_inputs]    )

															for key in python_outputs.keys():
																		# convert them to numpy to avoid messing with ragged tensors
																		a__  :		Optional[int]					=    python_outputs[key].numpy()
																		a__  :		List[str]					=    tf_outputs[key].numpy()

																		self.assertTrue(tf.reduce_all(python_outputs_values.shape == tf_outputs_values.shape    )    )
																		self.assertTrue(tf.reduce_all(tf.cast(__UpperCAmelCase   ,  tf.intaa    ) == tf_outputs_values    )    )

						@slow
						def 				SCREAMING_SNAKE_CASE__(  self    )   -> Optional[Any]:
									"""simple docstring"""




									for tf_tokenizer in self.tf_tokenizers:
												a__  :		Optional[int]					=    tf.function(__UpperCAmelCase    )
												for test_inputs in self.test_sentences:
															a__  :		Any					=    tf.constant(__UpperCAmelCase    )
															a__  :		int					=    compiled_tokenizer(__UpperCAmelCase    )
															a__  :		Any					=    tf_tokenizer(__UpperCAmelCase    )

															for key in eager_outputs.keys():
																		self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]    )    )

						@slow
						def 				SCREAMING_SNAKE_CASE__(  self    )   -> Optional[int]:
									"""simple docstring"""




									for tf_tokenizer in self.tf_tokenizers:
												a__  :		List[str]					=    ModelToSave(tokenizer=__UpperCAmelCase    )
												a__  :		Union[str, Any]					=    tf.convert_to_tensor([self.test_sentences[0]]    )
												a__  :		List[str]					=    model.serving(__UpperCAmelCase    )  # Build model with some sample inputs
												with TemporaryDirectory() as tempdir:
															a__  :		Union[str, Any]					=    Path(__UpperCAmelCase    ) / 'saved.model'
															tf.saved_model.save(__UpperCAmelCase   ,  __UpperCAmelCase   ,  signatures={"""serving_default""": model.serving}    )
															a__  :		str					=    tf.saved_model.load(__UpperCAmelCase    )
												a__  :		Dict					=    loaded_model.signatures['serving_default'](__UpperCAmelCase    )['output_0']
												# We may see small differences because the loaded model is compiled, so we need an epsilon for the test
												self.assertTrue(tf.reduce_all(out == loaded_output    )    )

						@slow
						def 				SCREAMING_SNAKE_CASE__(  self    )   -> Optional[Any]:
									"""simple docstring"""




									for tf_tokenizer in self.tf_tokenizers:
												a__  :		List[str]					=    tf.convert_to_tensor([self.test_sentences[0]]    )
												a__  :		List[str]					=    tf_tokenizer(__UpperCAmelCase    )  # Build model with some sample inputs

												a__  :		Union[str, Any]					=    tf_tokenizer.get_config()
												a__  :		Tuple					=    TFGPTaTokenizer.from_config(__UpperCAmelCase    )
												a__  :		Tuple					=    model_from_config(__UpperCAmelCase    )

												for key in from_config_output.keys():
															self.assertTrue(tf.reduce_all(from_config_output[key] == out[key]    )    )

						@slow
						def 				SCREAMING_SNAKE_CASE__(  self    )   -> int:
									"""simple docstring"""




									for tf_tokenizer in self.tf_tokenizers:
												# for the test to run
												a__  :		int					=    1_2_3_1_2_3

												for max_length in [3, 5, 1_0_2_4]:
															a__  :		Tuple					=    tf.convert_to_tensor([self.test_sentences[0]]    )
															a__  :		List[str]					=    tf_tokenizer(__UpperCAmelCase   ,  max_length=__UpperCAmelCase    )

															a__  :		Union[str, Any]					=    out['input_ids'].numpy().shape[1]

															assert out_length == max_length











"""simple docstring"""





import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
  import onnxruntime as ort






class    _lowerCAmelCase  (	a    ,  unittest.TestCase			):
       """simple docstring"""




       __magic_name__	:int						    =   """hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"""






       def 	snake_case		(       self       ,					__UpperCAmelCase=0 ):



         '''simple docstring'''

         lowerCAmelCase__     :List[str]											=				floats_tensor((1, 3, 1_2_8, 1_2_8)       ,					rng=random.Random(__UpperCAmelCase ) )
         lowerCAmelCase__     :List[str]											=				np.random.RandomState(__UpperCAmelCase )
         lowerCAmelCase__     :List[str]											=				{
             'prompt': 'A painting of a squirrel eating a burger',
             'image': image,
             'generator': generator,
             'num_inference_steps': 3,
             'strength': 0.75,
             'guidance_scale': 7.5,
             'output_type': 'numpy',
         }
         return inputs






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[Any]											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       ,					provider='CPUExecutionProvider' )
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :Union[str, Any]											=				self.get_dummy_inputs()
         lowerCAmelCase__     :Optional[int]											=				pipe(**__UpperCAmelCase ).images
         lowerCAmelCase__     :Union[str, Any]											=				image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         lowerCAmelCase__     :Union[str, Any]											=				np.array([0.6_96_43, 0.5_84_84, 0.5_03_14, 0.5_87_60, 0.5_53_68, 0.5_96_43, 0.5_15_29, 0.4_12_17, 0.4_90_87] )
         assert np.abs(image_slice - expected_slice ).max() < 1E-1






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[int]											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       ,					provider='CPUExecutionProvider' )
         lowerCAmelCase__     :str											=				PNDMScheduler.from_config(pipe.scheduler.config       ,					skip_prk_steps=__UpperCAmelCase )
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :Tuple											=				self.get_dummy_inputs()
         lowerCAmelCase__     :Optional[Any]											=				pipe(**__UpperCAmelCase ).images
         lowerCAmelCase__     :Any											=				image[0, -3:, -3:, -1]

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         lowerCAmelCase__     :int											=				np.array([0.6_17_37, 0.5_46_42, 0.5_31_83, 0.5_44_65, 0.5_27_42, 0.6_05_25, 0.4_99_69, 0.4_06_55, 0.4_81_54] )

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






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[int]											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       ,					provider='CPUExecutionProvider' )
         lowerCAmelCase__     :List[str]											=				LMSDiscreteScheduler.from_config(pipe.scheduler.config )
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         # warmup pass to apply optimizations
         lowerCAmelCase__     :List[Any]											=				pipe(**self.get_dummy_inputs() )

         lowerCAmelCase__     :Tuple											=				self.get_dummy_inputs()
         lowerCAmelCase__     :int											=				pipe(**__UpperCAmelCase ).images
         lowerCAmelCase__     :Any											=				image[0, -3:, -3:, -1]

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         lowerCAmelCase__     :Union[str, Any]											=				np.array([0.5_27_61, 0.5_99_77, 0.4_90_33, 0.4_96_19, 0.5_42_82, 0.5_03_11, 0.4_76_00, 0.4_09_18, 0.4_52_03] )

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






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[Any]											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       ,					provider='CPUExecutionProvider' )
         lowerCAmelCase__     :Dict											=				EulerDiscreteScheduler.from_config(pipe.scheduler.config )
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :Any											=				self.get_dummy_inputs()
         lowerCAmelCase__     :List[str]											=				pipe(**__UpperCAmelCase ).images
         lowerCAmelCase__     :List[str]											=				image[0, -3:, -3:, -1]

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         lowerCAmelCase__     :Optional[int]											=				np.array([0.5_29_11, 0.6_00_04, 0.4_92_29, 0.4_98_05, 0.5_45_02, 0.5_06_80, 0.4_77_77, 0.4_10_28, 0.4_53_04] )

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






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :str											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       ,					provider='CPUExecutionProvider' )
         lowerCAmelCase__     :str											=				EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config )
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :Tuple											=				self.get_dummy_inputs()
         lowerCAmelCase__     :Any											=				pipe(**__UpperCAmelCase ).images
         lowerCAmelCase__     :Optional[int]											=				image[0, -3:, -3:, -1]

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         lowerCAmelCase__     :int											=				np.array([0.5_29_11, 0.6_00_04, 0.4_92_29, 0.4_98_05, 0.5_45_02, 0.5_06_80, 0.4_77_77, 0.4_10_28, 0.4_53_04] )

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






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :List[Any]											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       ,					provider='CPUExecutionProvider' )
         lowerCAmelCase__     :List[Any]											=				DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :Any											=				self.get_dummy_inputs()
         lowerCAmelCase__     :List[Any]											=				pipe(**__UpperCAmelCase ).images
         lowerCAmelCase__     :int											=				image[0, -3:, -3:, -1]

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         lowerCAmelCase__     :Optional[Any]											=				np.array([0.6_53_31, 0.5_82_77, 0.4_82_04, 0.5_60_59, 0.5_36_65, 0.5_62_35, 0.5_09_69, 0.4_00_09, 0.4_65_52] )

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






@nightly
@require_onnxruntime
@require_torch_gpu
class    _lowerCAmelCase  (	unittest.TestCase			):
       """simple docstring"""




       @property
       def 	snake_case		(       self ):



         '''simple docstring'''

         return (
             "CUDAExecutionProvider",
             {
                 "gpu_mem_limit": "15000000000",  # 15GB
                 "arena_extend_strategy": "kSameAsRequested",
             },
         )






       @property
       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Tuple											=				ort.SessionOptions()
         lowerCAmelCase__     :Optional[int]											=				False
         return options






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Any											=				load_image(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
             '/img2img/sketch-mountains-input.jpg' )
         lowerCAmelCase__     :Any											=				init_image.resize((7_6_8, 5_1_2) )
         # using the PNDM scheduler by default
         lowerCAmelCase__     :Optional[int]											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(
             'CompVis/stable-diffusion-v1-4'       ,					revision='onnx'       ,					safety_checker=__UpperCAmelCase       ,					feature_extractor=__UpperCAmelCase       ,					provider=self.gpu_provider       ,					sess_options=self.gpu_options       ,					)
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :List[Any]											=				'A fantasy landscape, trending on artstation'

         lowerCAmelCase__     :Optional[Any]											=				np.random.RandomState(0 )
         lowerCAmelCase__     :List[str]											=				pipe(
             prompt=__UpperCAmelCase       ,					image=__UpperCAmelCase       ,					strength=0.75       ,					guidance_scale=7.5       ,					num_inference_steps=1_0       ,					generator=__UpperCAmelCase       ,					output_type='np'       ,					)
         lowerCAmelCase__     :Any											=				output.images
         lowerCAmelCase__     :List[str]											=				images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

         assert images.shape == (1, 5_1_2, 7_6_8, 3)
         lowerCAmelCase__     :List[Any]											=				np.array([0.49_09, 0.50_59, 0.53_72, 0.46_23, 0.48_76, 0.50_49, 0.48_20, 0.49_56, 0.50_19] )
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__     :Union[str, Any]											=				load_image(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
             '/img2img/sketch-mountains-input.jpg' )
         lowerCAmelCase__     :Optional[Any]											=				init_image.resize((7_6_8, 5_1_2) )
         lowerCAmelCase__     :List[Any]											=				LMSDiscreteScheduler.from_pretrained(
             'runwayml/stable-diffusion-v1-5'       ,					subfolder='scheduler'       ,					revision='onnx' )
         lowerCAmelCase__     :Optional[Any]											=				OnnxStableDiffusionImgaImgPipeline.from_pretrained(
             'runwayml/stable-diffusion-v1-5'       ,					revision='onnx'       ,					scheduler=__UpperCAmelCase       ,					safety_checker=__UpperCAmelCase       ,					feature_extractor=__UpperCAmelCase       ,					provider=self.gpu_provider       ,					sess_options=self.gpu_options       ,					)
         pipe.set_progress_bar_config(disable=__UpperCAmelCase )

         lowerCAmelCase__     :List[Any]											=				'A fantasy landscape, trending on artstation'

         lowerCAmelCase__     :List[Any]											=				np.random.RandomState(0 )
         lowerCAmelCase__     :List[Any]											=				pipe(
             prompt=__UpperCAmelCase       ,					image=__UpperCAmelCase       ,					strength=0.75       ,					guidance_scale=7.5       ,					num_inference_steps=2_0       ,					generator=__UpperCAmelCase       ,					output_type='np'       ,					)
         lowerCAmelCase__     :Optional[Any]											=				output.images
         lowerCAmelCase__     :int											=				images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

         assert images.shape == (1, 5_1_2, 7_6_8, 3)
         lowerCAmelCase__     :List[Any]											=				np.array([0.80_43, 0.9_26, 0.95_81, 0.81_19, 0.89_54, 0.9_13, 0.72_09, 0.74_63, 0.74_31] )
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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