






"""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 							SCREAMING_SNAKE_CASE__						(    a__			,       unittest.TestCase						):





					"""simple docstring"""






					a_     =		DiTPipeline
					a_     =		CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
					a_     =		PipelineTesterMixin.required_optional_params - {
					    "latents",
					    "num_images_per_prompt",
					    "callback",
					    "callback_steps",
					}
					a_     =		CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
					a_     =		False



					def 	_lowercase					( self					:  Tuple	):
						torch.manual_seed(0	)
						snake_case__			:				Optional[Any]       =  TransformeraDModel(
						    sample_size=1_6							,		num_layers=2							,		patch_size=4							,		attention_head_dim=8							,		num_attention_heads=2							,		in_channels=4							,		out_channels=8							,		attention_bias=__A							,		activation_fn="gelu-approximate"							,		num_embeds_ada_norm=1_0_0_0							,		norm_type="ada_norm_zero"							,		norm_elementwise_affine=__A							,		)
						snake_case__			:				List[str]       =  AutoencoderKL()
						snake_case__			:				List[Any]       =  DDIMScheduler()
						snake_case__			:				Optional[Any]       =  {"transformer": transformer.eval(), "vae": vae.eval(), "scheduler": scheduler}
						return components



					def 	_lowercase					( self					:  List[str]							,		__A					:  Any							,		__A					:  Union[str, Any]=0	):
						if str(__A	).startswith("mps"	):
							snake_case__			:				List[str]       =  torch.manual_seed(__A	)
						else:
							snake_case__			:				List[str]       =  torch.Generator(device=__A	).manual_seed(__A	)
						snake_case__			:				str       =  {
						    "class_labels": [1],
						    "generator": generator,
						    "num_inference_steps": 2,
						    "output_type": "numpy",
						}
						return inputs



					def 	_lowercase					( self					:  Optional[Any]	):
						snake_case__			:				Dict       =  "cpu"

						snake_case__			:				int       =  self.get_dummy_components()
						snake_case__			:				Optional[Any]       =  self.pipeline_class(**__A	)
						pipe.to(__A	)
						pipe.set_progress_bar_config(disable=__A	)

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

						self.assertEqual(image.shape							,		(1, 1_6, 1_6, 3)	)
						snake_case__			:				Optional[int]       =  np.array([0.2_9_4_6, 0.6_6_0_1, 0.4_3_2_9, 0.3_2_9_6, 0.4_1_4_4, 0.5_3_1_9, 0.7_2_7_3, 0.5_0_1_3, 0.4_4_5_7]	)
						snake_case__			:				Optional[int]       =  np.abs(image_slice.flatten() - expected_slice	).max()
						self.assertLessEqual(__A							,		1e-3	)



					def 	_lowercase					( self					:  Any	):
						self._test_inference_batch_single_identical(relax_max_difference=__A							,		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 	_lowercase					( self					:  List[str]	):
						self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3	)



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





					"""simple docstring"""






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



					def 	_lowercase					( self					:  List[str]	):
						snake_case__			:				Optional[int]       =  torch.manual_seed(0	)

						snake_case__			:				str       =  DiTPipeline.from_pretrained("facebook/DiT-XL-2-256"	)
						pipe.to("cuda"	)

						snake_case__			:				Tuple       =  ["vase", "umbrella", "white shark", "white wolf"]
						snake_case__			:				Optional[int]       =  pipe.get_label_ids(__A	)

						snake_case__			:				Optional[Any]       =  pipe(__A							,		generator=__A							,		num_inference_steps=4_0							,		output_type="np"	).images

						for word, image in zip(__A							,		__A	):
							snake_case__			:				Dict       =  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 	_lowercase					( self					:  str	):
						snake_case__			:				Tuple       =  DiTPipeline.from_pretrained("facebook/DiT-XL-2-512"	)
						snake_case__			:				Any       =  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config	)
						pipe.to("cuda"	)

						snake_case__			:				Union[str, Any]       =  ["vase", "umbrella"]
						snake_case__			:				int       =  pipe.get_label_ids(__A	)

						snake_case__			:				Dict       =  torch.manual_seed(0	)
						snake_case__			:				Dict       =  pipe(__A							,		generator=__A							,		num_inference_steps=2_5							,		output_type="np"	).images

						for word, image in zip(__A							,		__A	):
							snake_case__			:				Union[str, Any]       =  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











from tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch



@require_torch
@require_tf
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[Any]	):
						snake_case__			:				Optional[Any]       =  SMALL_MODEL_IDENTIFIER
						snake_case__			:				Any       =  "pt"
						snake_case__			:				Any       =  "tf"



					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  List[Any]	):
						snake_case__			:				int       =  AutoModel.from_pretrained(self.test_model	)
						model_pt.save_pretrained(__A	)



					def 	_lowercase					( self					:  Optional[int]							,		__A					:  Tuple	):
						snake_case__			:				List[Any]       =  TFAutoModel.from_pretrained(self.test_model							,		from_pt=__A	)
						model_tf.save_pretrained(__A	)



					def 	_lowercase					( self					:  str	):
						snake_case__			:				Optional[Any]       =  "mock_framework"

						# Framework provided - return whatever the user provides
						snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model							,		__A	)
						self.assertEqual(__A							,		__A	)

						# Local checkpoint and framework provided - return provided framework
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				Optional[int]       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				int       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)



					def 	_lowercase					( self					:  Dict	):
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				List[str]       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_pt	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				Tuple       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Invalid local checkpoint
						with TemporaryDirectory() as local_invalid_ckpt:
							with self.assertRaises(__A	):
								snake_case__			:				int       =  FeaturesManager.determine_framework(__A	)






					def 	_lowercase					( self					:  Dict	):
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	):
							snake_case__			:				List[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# PyTorch not in environment -> use TensorFlow
						snake_case__			:				Tuple       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				int       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Both in environment -> use PyTorch
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[int]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# Both not in environment -> raise error
						snake_case__			:				List[str]       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[Any]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							with self.assertRaises(__A	):
								snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)



