



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 				_snake_case			(					lowercase__    ,	unittest.TestCase):
				UpperCamelCase__				:					Dict     		=DiTPipeline
				UpperCamelCase__				:					Tuple     		=CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
				UpperCamelCase__				:					Tuple     		=PipelineTesterMixin.required_optional_params - {
				    """latents""",
				    """num_images_per_prompt""",
				    """callback""",
				    """callback_steps""",
				}
				UpperCamelCase__				:					List[str]     		=CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
				UpperCamelCase__				:					int     		=False




				def 							A__      (						self					:						int						):
							torch.manual_seed(0						)
							lowercase__					  =				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=__lowercase,					activation_fn="gelu-approximate",					num_embeds_ada_norm=1000,					norm_type="ada_norm_zero",					norm_elementwise_affine=__lowercase,					)
							lowercase__					  =				AutoencoderKL()
							lowercase__					  =				DDIMScheduler()
							lowercase__					  =				{"transformer": transformer.eval(), "vae": vae.eval(), "scheduler": scheduler}
							return components




				def 							A__      (						self					:						Optional[int],					__lowercase					:						Dict,					__lowercase					:						Optional[int]=0						):
							if str(__lowercase						).startswith("mps"						):
										lowercase__					  =				torch.manual_seed(__lowercase						)
							else:
										lowercase__					  =				torch.Generator(device=__lowercase						).manual_seed(__lowercase						)
							lowercase__					  =				{
							    "class_labels": [1],
							    "generator": generator,
							    "num_inference_steps": 2,
							    "output_type": "numpy",
							}
							return inputs




				def 							A__      (						self					:						List[Any]						):
							lowercase__					  =				"cpu"

							lowercase__					  =				self.get_dummy_components()
							lowercase__					  =				self.pipeline_class(**__lowercase						)
							pipe.to(__lowercase						)
							pipe.set_progress_bar_config(disable=__lowercase						)

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

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




				def 							A__      (						self					:						Optional[int]						):
							self._test_inference_batch_single_identical(relax_max_difference=__lowercase,					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 							A__      (						self					:						List[Any]						):
							self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3						)
@require_torch_gpu
@slow
class 				_snake_case			(					unittest.TestCase):




				def 							A__      (						self					:						List[Any]						):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()




				def 							A__      (						self					:						Any						):
							lowercase__					  =				torch.manual_seed(0						)

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

							lowercase__					  =				["vase", "umbrella", "white shark", "white wolf"]
							lowercase__					  =				pipe.get_label_ids(__lowercase						)

							lowercase__					  =				pipe(__lowercase,					generator=__lowercase,					num_inference_steps=40,					output_type="np"						).images

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

							lowercase__					  =				["vase", "umbrella"]
							lowercase__					  =				pipe.get_label_ids(__lowercase						)

							lowercase__					  =				torch.manual_seed(0						)
							lowercase__					  =				pipe(__lowercase,					generator=__lowercase,					num_inference_steps=25,					output_type="np"						).images

							for word, image in zip(__lowercase,					__lowercase						):
										lowercase__					  =				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 argparse
import os

import numpy as np
import tensorflow as tf
import torch

from transformers import BertModel



def __lowerCAmelCase (     SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  ):
			lowercase__					  =				("dense.weight", "attention.self.query", "attention.self.key", "attention.self.value")

			lowercase__					  =				(
			    ("layer.", "layer_"),
			    ("word_embeddings.weight", "word_embeddings"),
			    ("position_embeddings.weight", "position_embeddings"),
			    ("token_type_embeddings.weight", "token_type_embeddings"),
			    (".", "/"),
			    ("LayerNorm/weight", "LayerNorm/gamma"),
			    ("LayerNorm/bias", "LayerNorm/beta"),
			    ("weight", "kernel"),
			)

			if not os.path.isdir(SCREAMING_SNAKE_CASE_  ):
						os.makedirs(SCREAMING_SNAKE_CASE_  )

			lowercase__					  =				model.state_dict()

			def to_tf_var_name(SCREAMING_SNAKE_CASE_  ):
						for patt, repl in iter(SCREAMING_SNAKE_CASE_  ):
									lowercase__					  =				name.replace(SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  )
						return f'''bert/{name}'''

			def create_tf_var(SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  ):
						lowercase__					  =				tf.dtypes.as_dtype(tensor.dtype  )
						lowercase__					  =				tf.get_variable(dtype=SCREAMING_SNAKE_CASE_						,  shape=tensor.shape						,  name=SCREAMING_SNAKE_CASE_						,  initializer=tf.zeros_initializer()  )
						session.run(tf.variables_initializer([tf_var]  )  )
						session.run(SCREAMING_SNAKE_CASE_  )
						return tf_var

			tf.reset_default_graph()
			with tf.Session() as session:
						for var_name in state_dict:
									lowercase__					  =				to_tf_var_name(SCREAMING_SNAKE_CASE_  )
									lowercase__					  =				state_dict[var_name].numpy()
									if any(x in var_name for x in tensors_to_transpose  ):
												lowercase__					  =				torch_tensor.T
									lowercase__					  =				create_tf_var(tensor=SCREAMING_SNAKE_CASE_						,  name=SCREAMING_SNAKE_CASE_						,  session=SCREAMING_SNAKE_CASE_  )
									tf.keras.backend.set_value(SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  )
									lowercase__					  =				session.run(SCREAMING_SNAKE_CASE_  )
									print(f'''Successfully created {tf_name}: {np.allclose(SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  )}'''  )

						lowercase__					  =				tf.train.Saver(tf.trainable_variables()  )
						saver.save(SCREAMING_SNAKE_CASE_						,  os.path.join(SCREAMING_SNAKE_CASE_						,  model_name.replace("-"						,  "_"  ) + ".ckpt"  )  )



def __lowerCAmelCase (     SCREAMING_SNAKE_CASE_=None  ):
			lowercase__					  =				argparse.ArgumentParser()
			parser.add_argument("--model_name"						,  type=SCREAMING_SNAKE_CASE_						,  required=SCREAMING_SNAKE_CASE_						,  help="model name e.g. bert-base-uncased"  )
			parser.add_argument(
			    "--cache_dir"						,  type=SCREAMING_SNAKE_CASE_						,  default=SCREAMING_SNAKE_CASE_						,  required=SCREAMING_SNAKE_CASE_						,  help="Directory containing pytorch model"  )
			parser.add_argument("--pytorch_model_path"						,  type=SCREAMING_SNAKE_CASE_						,  required=SCREAMING_SNAKE_CASE_						,  help="/path/to/<pytorch-model-name>.bin"  )
			parser.add_argument("--tf_cache_dir"						,  type=SCREAMING_SNAKE_CASE_						,  required=SCREAMING_SNAKE_CASE_						,  help="Directory in which to save tensorflow model"  )
			lowercase__					  =				parser.parse_args(SCREAMING_SNAKE_CASE_  )

			lowercase__					  =				BertModel.from_pretrained(
			    pretrained_model_name_or_path=args.model_name						,  state_dict=torch.load(args.pytorch_model_path  )						,  cache_dir=args.cache_dir						,  )

			convert_pytorch_checkpoint_to_tf(model=SCREAMING_SNAKE_CASE_						,  ckpt_dir=args.tf_cache_dir						,  model_name=args.model_name  )


if __name__ == "__main__":
							main()




