


"""simple docstring"""




import gc
import unittest

import numpy as np
import torch

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, DPMSolverMultistepScheduler, TransformeraDModel
from diffusers.utils import is_xformers_available, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS,
    CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
class 					__lowerCAmelCase					(   __SCREAMING_SNAKE_CASE     ,			unittest.TestCase       ):


		'''simple docstring'''



		__UpperCAmelCase		:			Union[str, Any]										=				DiTPipeline
		__UpperCAmelCase		:			Any										=				CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
		__UpperCAmelCase		:			Optional[Any]										=				PipelineTesterMixin.required_optional_params - {
		    'latents',
		    'num_images_per_prompt',
		    'callback',
		    'callback_steps',
		}
		__UpperCAmelCase		:			Optional[int]										=				CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
		__UpperCAmelCase		:			List[Any]										=				False

		def 			__UpperCAmelCase       (  self      ):
				torch.manual_seed(0      )
				__a           =			TransformeraDModel(
				    sample_size=16							, num_layers=2							, patch_size=4							, attention_head_dim=8							, num_attention_heads=2							, in_channels=4							, out_channels=8							, attention_bias=_SCREAMING_SNAKE_CASE							, activation_fn='''gelu-approximate'''							, num_embeds_ada_norm=1_000							, norm_type='''ada_norm_zero'''							, norm_elementwise_affine=_SCREAMING_SNAKE_CASE							, )
				__a           =			AutoencoderKL()
				__a           =			DDIMScheduler()
				__a           =			{'''transformer''': transformer.eval(), '''vae''': vae.eval(), '''scheduler''': scheduler}
				return components

		def 			__UpperCAmelCase       (  self							, _a							, _a=0      ):





				if str(_SCREAMING_SNAKE_CASE      ).startswith('''mps'''      ):
						__a           =			torch.manual_seed(_SCREAMING_SNAKE_CASE      )
				else:
						__a           =			torch.Generator(device=_SCREAMING_SNAKE_CASE      ).manual_seed(_SCREAMING_SNAKE_CASE      )
				__a           =			{
				    '''class_labels''': [1],
				    '''generator''': generator,
				    '''num_inference_steps''': 2,
				    '''output_type''': '''numpy''',
				}
				return inputs

		def 			__UpperCAmelCase       (  self      ):
				__a           =			'''cpu'''

				__a           =			self.get_dummy_components()
				__a           =			self.pipeline_class(**_SCREAMING_SNAKE_CASE      )
				pipe.to(_SCREAMING_SNAKE_CASE      )
				pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE      )

				__a           =			self.get_dummy_inputs(_SCREAMING_SNAKE_CASE      )
				__a           =			pipe(**_SCREAMING_SNAKE_CASE      ).images
				__a           =			image[0, -3:, -3:, -1]

				self.assertEqual(image.shape							, (1, 16, 16, 3)      )
				__a           =			np.array([0.2946, 0.6601, 0.4329, 0.3296, 0.4144, 0.5319, 0.7273, 0.5013, 0.4457]      )
				__a           =			np.abs(image_slice.flatten() - expected_slice      ).max()
				self.assertLessEqual(_SCREAMING_SNAKE_CASE							, 1E-3      )

		def 			__UpperCAmelCase       (  self      ):
				self._test_inference_batch_single_identical(relax_max_difference=_SCREAMING_SNAKE_CASE							, expected_max_diff=1E-3      )





		@unittest.skipIf(
		    torch_device != '''cuda''' or not is_xformers_available()							, reason='''XFormers attention is only available with CUDA and `xformers` installed'''							, )
		def 			__UpperCAmelCase       (  self      ):
				self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3      )







@require_torch_gpu
@slow
class 					__lowerCAmelCase					(   unittest.TestCase       ):


		'''simple docstring'''



		def 			__UpperCAmelCase       (  self      ):
				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()

		def 			__UpperCAmelCase       (  self      ):
				__a           =			torch.manual_seed(0      )

				__a           =			DiTPipeline.from_pretrained('''facebook/DiT-XL-2-256'''      )
				pipe.to('''cuda'''      )

				__a           =			['''vase''', '''umbrella''', '''white shark''', '''white wolf''']
				__a           =			pipe.get_label_ids(_SCREAMING_SNAKE_CASE      )

				__a           =			pipe(_SCREAMING_SNAKE_CASE							, generator=_SCREAMING_SNAKE_CASE							, num_inference_steps=40							, output_type='''np'''      ).images

				for word, image in zip(_SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE      ):
						__a           =			load_numpy(
						    f'''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/dit/{word}.npy'''      )
						assert np.abs((expected_image - image).max()      ) < 1E-2





		def 			__UpperCAmelCase       (  self      ):
				__a           =			DiTPipeline.from_pretrained('''facebook/DiT-XL-2-512'''      )
				__a           =			DPMSolverMultistepScheduler.from_config(pipe.scheduler.config      )
				pipe.to('''cuda'''      )

				__a           =			['''vase''', '''umbrella''']
				__a           =			pipe.get_label_ids(_SCREAMING_SNAKE_CASE      )

				__a           =			torch.manual_seed(0      )
				__a           =			pipe(_SCREAMING_SNAKE_CASE							, generator=_SCREAMING_SNAKE_CASE							, num_inference_steps=25							, output_type='''np'''      ).images

				for word, image in zip(_SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE      ):
						__a           =			load_numpy(
						    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
						    f'''/dit/{word}_512.npy'''      )

						assert np.abs((expected_image - image).max()      ) < 1E-1










import copy
import tempfile
import unittest

from huggingface_hub import HfFolder, delete_repo
from parameterized import parameterized
from requests.exceptions import HTTPError

from transformers import AutoConfig, GenerationConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test






class   _lowerCamelCase (					unittest.TestCase			):




       """simple docstring"""



       @parameterized.expand([(None,), ('''foo.json''',)]		)
       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE		)->Tuple:




         '''simple docstring'''





         A_      :				Dict		     =					GenerationConfig(
             do_sample=_SCREAMING_SNAKE_CASE				,       temperature=0.7				,       length_penalty=1.0				,       bad_words_ids=[[1, 2, 3], [4, 5]]				,       )
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(_SCREAMING_SNAKE_CASE				,       config_name=_SCREAMING_SNAKE_CASE		)
           A_      :				Tuple		     =					GenerationConfig.from_pretrained(_SCREAMING_SNAKE_CASE				,       config_name=_SCREAMING_SNAKE_CASE		)

         # Checks parameters that were specified
         self.assertEqual(loaded_config.do_sample				,       _SCREAMING_SNAKE_CASE		)
         self.assertEqual(loaded_config.temperature				,       0.7		)
         self.assertEqual(loaded_config.length_penalty				,       1.0		)
         self.assertEqual(loaded_config.bad_words_ids				,       [[1, 2, 3], [4, 5]]		)

         # Checks parameters that were not specified (defaults)
         self.assertEqual(loaded_config.top_k				,       50		)
         self.assertEqual(loaded_config.max_length				,       20		)
         self.assertEqual(loaded_config.max_time				,       _SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self		)->Dict:




         '''simple docstring'''





         A_      :				str		     =					AutoConfig.from_pretrained('''gpt2'''		)
         A_      :				int		     =					GenerationConfig.from_model_config(_SCREAMING_SNAKE_CASE		)
         A_      :				Dict		     =					GenerationConfig()

         # The generation config has loaded a few non-default parameters from the model config
         self.assertNotEqual(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)

         # One of those parameters is eos_token_id -- check if it matches
         self.assertNotEqual(generation_config_from_model.eos_token_id				,       default_generation_config.eos_token_id		)
         self.assertEqual(generation_config_from_model.eos_token_id				,       model_config.eos_token_id		)



       def        _snake_case       (      self		)->Optional[int]:




         '''simple docstring'''





         A_      :				List[Any]		     =					GenerationConfig()
         A_      :				Optional[Any]		     =					{
             '''max_new_tokens''': 1024,
             '''foo''': '''bar''',
         }
         A_      :				Dict		     =					copy.deepcopy(_SCREAMING_SNAKE_CASE		)
         A_      :				Optional[int]		     =					generation_config.update(**_SCREAMING_SNAKE_CASE		)

         # update_kwargs was not modified (no side effects)
         self.assertEqual(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)

         # update_kwargs was used to update the config on valid attributes
         self.assertEqual(generation_config.max_new_tokens				,       1024		)

         # `.update()` returns a dictionary of unused kwargs
         self.assertEqual(_SCREAMING_SNAKE_CASE				,       {'''foo''': '''bar'''}		)



       def        _snake_case       (      self		)->str:




         '''simple docstring'''





         A_      :				List[str]		     =					GenerationConfig()
         A_      :				int		     =					'''bar'''

         with tempfile.TemporaryDirectory('''test-generation-config'''		) as tmp_dir:
           generation_config.save_pretrained(_SCREAMING_SNAKE_CASE		)

           A_      :				Optional[Any]		     =					GenerationConfig.from_pretrained(_SCREAMING_SNAKE_CASE		)
         # update_kwargs was used to update the config on valid attributes
         self.assertEqual(new_config.foo				,       '''bar'''		)

         A_      :				Optional[int]		     =					GenerationConfig.from_model_config(_SCREAMING_SNAKE_CASE		)
         assert not hasattr(_SCREAMING_SNAKE_CASE				,       '''foo'''		)  # no new kwargs should be initialized if from config







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




         '''simple docstring'''





         A_      :				str		     =					GenerationConfig()
         self.assertEqual(default_config.temperature				,       1.0		)
         self.assertEqual(default_config.do_sample				,       _SCREAMING_SNAKE_CASE		)
         self.assertEqual(default_config.num_beams				,       1		)

         A_      :				str		     =					GenerationConfig(
             do_sample=_SCREAMING_SNAKE_CASE				,       temperature=0.7				,       length_penalty=1.0				,       bad_words_ids=[[1, 2, 3], [4, 5]]				,       )
         self.assertEqual(config.temperature				,       0.7		)
         self.assertEqual(config.do_sample				,       _SCREAMING_SNAKE_CASE		)
         self.assertEqual(config.num_beams				,       1		)

         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(_SCREAMING_SNAKE_CASE		)
           A_      :				Dict		     =					GenerationConfig.from_pretrained(_SCREAMING_SNAKE_CASE				,       temperature=1.0		)

         self.assertEqual(loaded_config.temperature				,       1.0		)
         self.assertEqual(loaded_config.do_sample				,       _SCREAMING_SNAKE_CASE		)
         self.assertEqual(loaded_config.num_beams				,       1		)  # default value






@is_staging_test
class   _lowerCamelCase (					unittest.TestCase			):




       """simple docstring"""



       @classmethod
       def        _snake_case       (      cls		)->Optional[int]:




         '''simple docstring'''





         A_      :				str		     =					TOKEN
         HfFolder.save_token(_SCREAMING_SNAKE_CASE		)



       @classmethod
       def        _snake_case       (      cls		)->List[Any]:




         '''simple docstring'''





         try:
           delete_repo(token=cls._token				,       repo_id='''test-generation-config'''		)
         except HTTPError:
           pass

         try:
           delete_repo(token=cls._token				,       repo_id='''valid_org/test-generation-config-org'''		)
         except HTTPError:
           pass



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




         '''simple docstring'''





         A_      :				List[Any]		     =					GenerationConfig(
             do_sample=_SCREAMING_SNAKE_CASE				,       temperature=0.7				,       length_penalty=1.0				,       )
         config.push_to_hub('''test-generation-config'''				,       use_auth_token=self._token		)

         A_      :				Optional[int]		     =					GenerationConfig.from_pretrained(F'''{USER}/test-generation-config'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(_SCREAMING_SNAKE_CASE				,       getattr(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)		)

        # Reset repo
         delete_repo(token=self._token				,       repo_id='''test-generation-config'''		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(
               _SCREAMING_SNAKE_CASE				,       repo_id='''test-generation-config'''				,       push_to_hub=_SCREAMING_SNAKE_CASE				,       use_auth_token=self._token		)

         A_      :				Optional[Any]		     =					GenerationConfig.from_pretrained(F'''{USER}/test-generation-config'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(_SCREAMING_SNAKE_CASE				,       getattr(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)		)







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




         '''simple docstring'''





         A_      :				Tuple		     =					GenerationConfig(
             do_sample=_SCREAMING_SNAKE_CASE				,       temperature=0.7				,       length_penalty=1.0				,       )
         config.push_to_hub('''valid_org/test-generation-config-org'''				,       use_auth_token=self._token		)

         A_      :				Union[str, Any]		     =					GenerationConfig.from_pretrained('''valid_org/test-generation-config-org'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(_SCREAMING_SNAKE_CASE				,       getattr(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)		)

        # Reset repo
         delete_repo(token=self._token				,       repo_id='''valid_org/test-generation-config-org'''		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(
               _SCREAMING_SNAKE_CASE				,       repo_id='''valid_org/test-generation-config-org'''				,       push_to_hub=_SCREAMING_SNAKE_CASE				,       use_auth_token=self._token		)

         A_      :				Dict		     =					GenerationConfig.from_pretrained('''valid_org/test-generation-config-org'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(_SCREAMING_SNAKE_CASE				,       getattr(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)		)




