

'''simple docstring'''





import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, LDMTextToImagePipeline, UNetaDConditionModel
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    load_numpy,
    nightly,
    require_torch_gpu,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





class 		__SCREAMING_SNAKE_CASE       ( lowercase__	,       unittest.TestCase			):
				lowerCamelCase_							=						LDMTextToImagePipeline
				lowerCamelCase_							=						TEXT_TO_IMAGE_PARAMS - {
				    'negative_prompt',
				    'negative_prompt_embeds',
				    'cross_attention_kwargs',
				    'prompt_embeds',
				}
				lowerCamelCase_							=						PipelineTesterMixin.required_optional_params - {
				    'num_images_per_prompt',
				    'callback',
				    'callback_steps',
				}
				lowerCamelCase_							=						TEXT_TO_IMAGE_BATCH_PARAMS
				lowerCamelCase_							=						False







				def    lowerCamelCase_  (					self    :		Optional[Any]      ):
										'''simple docstring'''
										torch.manual_seed(0      )
										lowercase   :       List[Any]								=UNetaDConditionModel(
										    block_out_channels=(32, 64)		,				layers_per_block=2		,				sample_size=32		,				in_channels=4		,				out_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,				cross_attention_dim=32		,				)
										lowercase   :       Any								=DDIMScheduler(
										    beta_start=0.0_00_85		,				beta_end=0.0_12		,				beta_schedule='''scaled_linear'''		,				clip_sample=UpperCAmelCase__		,				set_alpha_to_one=UpperCAmelCase__		,				)
										torch.manual_seed(0      )
										lowercase   :       Dict								=AutoencoderKL(
										    block_out_channels=(32, 64)		,				in_channels=3		,				out_channels=3		,				down_block_types=('''DownEncoderBlock2D''', '''DownEncoderBlock2D''')		,				up_block_types=('''UpDecoderBlock2D''', '''UpDecoderBlock2D''')		,				latent_channels=4		,				)
										torch.manual_seed(0      )
										lowercase   :       List[Any]								=CLIPTextConfig(
										    bos_token_id=0		,				eos_token_id=2		,				hidden_size=32		,				intermediate_size=37		,				layer_norm_eps=1E-05		,				num_attention_heads=4		,				num_hidden_layers=5		,				pad_token_id=1		,				vocab_size=1000		,				)
										lowercase   :       Tuple								=CLIPTextModel(UpperCAmelCase__      )
										lowercase   :       Tuple								=CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

										lowercase   :       str								={
										    '''unet''': unet,
										    '''scheduler''': scheduler,
										    '''vqvae''': vae,
										    '''bert''': text_encoder,
										    '''tokenizer''': tokenizer,
										}
										return components







				def    lowerCamelCase_  (					self    :		Optional[int]		,				UpperCAmelCase__    :		Optional[int]		,				UpperCAmelCase__    :		List[Any]=0      ):
										'''simple docstring'''
										if str(UpperCAmelCase__      ).startswith('''mps'''      ):
																lowercase   :       Optional[Any]								=torch.manual_seed(UpperCAmelCase__      )
										else:
																lowercase   :       Tuple								=torch.Generator(device=UpperCAmelCase__      ).manual_seed(UpperCAmelCase__      )
										lowercase   :       Tuple								={
										    '''prompt''': '''A painting of a squirrel eating a burger''',
										    '''generator''': generator,
										    '''num_inference_steps''': 2,
										    '''guidance_scale''': 6.0,
										    '''output_type''': '''numpy''',
										}
										return inputs







				def    lowerCamelCase_  (					self    :		str      ):
										'''simple docstring'''
										lowercase   :       str								='''cpu'''  # ensure determinism for the device-dependent torch.Generator

										lowercase   :       Dict								=self.get_dummy_components()
										lowercase   :       Optional[Any]								=LDMTextToImagePipeline(**UpperCAmelCase__      )
										pipe.to(UpperCAmelCase__      )
										pipe.set_progress_bar_config(disable=UpperCAmelCase__      )

										lowercase   :       Optional[Any]								=self.get_dummy_inputs(UpperCAmelCase__      )
										lowercase   :       str								=pipe(**UpperCAmelCase__      ).images
										lowercase   :       Union[str, Any]								=image[0, -3:, -3:, -1]

										assert image.shape == (1, 16, 16, 3)
										lowercase   :       Tuple								=np.array([0.61_01, 0.61_56, 0.56_22, 0.48_95, 0.66_61, 0.38_04, 0.57_48, 0.61_36, 0.50_14]      )

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





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







				def    lowerCamelCase_  (					self    :		Any      ):
										'''simple docstring'''
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()







				def    lowerCamelCase_  (					self    :		int		,				UpperCAmelCase__    :		Dict		,				UpperCAmelCase__    :		Dict=torch.floataa		,				UpperCAmelCase__    :		str=0      ):
										'''simple docstring'''
										lowercase   :       str								=torch.manual_seed(UpperCAmelCase__      )
										lowercase   :       Union[str, Any]								=np.random.RandomState(UpperCAmelCase__      ).standard_normal((1, 4, 32, 32)      )
										lowercase   :       str								=torch.from_numpy(UpperCAmelCase__      ).to(device=UpperCAmelCase__		,				dtype=UpperCAmelCase__      )
										lowercase   :       Tuple								={
										    '''prompt''': '''A painting of a squirrel eating a burger''',
										    '''latents''': latents,
										    '''generator''': generator,
										    '''num_inference_steps''': 3,
										    '''guidance_scale''': 6.0,
										    '''output_type''': '''numpy''',
										}
										return inputs







				def    lowerCamelCase_  (					self    :		Union[str, Any]      ):
										'''simple docstring'''
										lowercase   :       Optional[int]								=LDMTextToImagePipeline.from_pretrained('''CompVis/ldm-text2im-large-256'''      ).to(UpperCAmelCase__      )
										pipe.set_progress_bar_config(disable=UpperCAmelCase__      )

										lowercase   :       int								=self.get_inputs(UpperCAmelCase__      )
										lowercase   :       Tuple								=pipe(**UpperCAmelCase__      ).images
										lowercase   :       Tuple								=image[0, -3:, -3:, -1].flatten()

										assert image.shape == (1, 256, 256, 3)
										lowercase   :       List[Any]								=np.array([0.5_18_25, 0.5_28_50, 0.5_25_43, 0.5_42_58, 0.5_23_04, 0.5_25_69, 0.5_43_63, 0.5_52_76, 0.5_68_78]      )
										lowercase   :       str								=np.abs(expected_slice - image_slice      ).max()
										assert max_diff < 1E-3





@nightly
@require_torch_gpu
class 		__SCREAMING_SNAKE_CASE       ( unittest.TestCase			):







				def    lowerCamelCase_  (					self    :		int      ):
										'''simple docstring'''
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()







				def    lowerCamelCase_  (					self    :		str		,				UpperCAmelCase__    :		Tuple		,				UpperCAmelCase__    :		Optional[int]=torch.floataa		,				UpperCAmelCase__    :		Any=0      ):
										'''simple docstring'''
										lowercase   :       Optional[Any]								=torch.manual_seed(UpperCAmelCase__      )
										lowercase   :       str								=np.random.RandomState(UpperCAmelCase__      ).standard_normal((1, 4, 32, 32)      )
										lowercase   :       Union[str, Any]								=torch.from_numpy(UpperCAmelCase__      ).to(device=UpperCAmelCase__		,				dtype=UpperCAmelCase__      )
										lowercase   :       Union[str, Any]								={
										    '''prompt''': '''A painting of a squirrel eating a burger''',
										    '''latents''': latents,
										    '''generator''': generator,
										    '''num_inference_steps''': 50,
										    '''guidance_scale''': 6.0,
										    '''output_type''': '''numpy''',
										}
										return inputs







				def    lowerCamelCase_  (					self    :		Any      ):
										'''simple docstring'''
										lowercase   :       Tuple								=LDMTextToImagePipeline.from_pretrained('''CompVis/ldm-text2im-large-256'''      ).to(UpperCAmelCase__      )
										pipe.set_progress_bar_config(disable=UpperCAmelCase__      )

										lowercase   :       Union[str, Any]								=self.get_inputs(UpperCAmelCase__      )
										lowercase   :       List[Any]								=pipe(**UpperCAmelCase__      ).images[0]

										lowercase   :       Tuple								=load_numpy(
										    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/ldm_text2img/ldm_large_256_ddim.npy'''      )
										lowercase   :       str								=np.abs(expected_image - image      ).max()
										assert max_diff < 1E-3








'''simple docstring'''





import argparse
import json
import os
from collections import OrderedDict

import numpy as np
import tensorflow as tf
import torch






def 							_lowerCAmelCase (      __magic_name__	:    str	)							->     Union[str, Any]:
						lowercase   :       Union[str, Any]								=os.path.join(args.tf_model_dir      ,   '''parameters.json'''	)
						lowercase   :       List[str]								=json.loads(open(__magic_name__	).read()	)
						if not params:
												raise ValueError(
												    f'''It seems that the json file at {parameter_file} is empty. Make sure you have a correct json file.'''	)
						if not args.output.endswith('''.pt'''	):
												lowercase   :       Tuple								=args.output + '''.pt'''
						lowercase   :       int								=OrderedDict()
						with tf.device('''/CPU:0'''	):
												lowercase   :       List[Any]								=tf.train.load_checkpoint(args.tf_model_dir	)
												lowercase   :       int								=reader.get_variable_to_shape_map()
												for key_name in shapes.keys():
																		lowercase   :       Any								=reader.get_tensor(__magic_name__	).astype(np.floataa	)
																		if key_name.endswith('''/adam_m'''	) or key_name.endswith('''/adam_v'''	):
																								continue
																		if key_name.startswith('''pasts/'''	):
																								if key_name.startswith('''pasts/mlp'''	):
																														lowercase   :       int								=int(key_name[9]	)
																								elif key_name.startswith('''pasts/out'''	):
																														lowercase   :       Union[str, Any]								=8
																								lowercase   :       Any								='''model.sqout.%d.weight''' % (player * 2)  # enter to nn.Sequencial with Tanh, so 2 at a time
																								lowercase   :       Dict								=vnp.transpose([1, 0]	).copy()  # Mesh-Tensorflow is a diagonal matrix
																								lowercase   :       List[str]								=torch.tensor(__magic_name__	)
																		elif key_name.startswith('''model/moe'''	):
																								lowercase   :       Union[str, Any]								=int(key_name[9:].split('''/'''	)[0]	)
																								if key_name.endswith('''/switch_gating/kernel'''	):
																														lowercase   :       Dict								='''model.blocks.%d.feed_forward.mlp.router.classifier.weight''' % player
																														lowercase   :       Any								=vnp.transpose([1, 0]	).copy()  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       List[Any]								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/softmlp/kernel'''	):
																														lowercase   :       Optional[int]								='''model.blocks.%d.feed_forward.soft_bypass_mlp.weight''' % player
																														lowercase   :       int								=vnp.transpose([1, 0]	).copy()  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       Optional[Any]								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/wo/kernel'''	) or key_name.endswith('''/wi/kernel'''	):
																														lowercase   :       Union[str, Any]								=key_name[-9:-7]
																														for i in range(16	):
																																				lowercase   :       Dict								='''model.blocks.%d.feed_forward.mlp.experts.expert_%d.%s.weight''' % (player, i, nlayer)
																																				lowercase   :       Any								=(
																																				    vnp[i].transpose([1, 0]	).copy()
																																				)  # In Mesh-Tensorflow, it is one array, so it is divided
																																				lowercase   :       List[str]								=torch.tensor(__magic_name__	)
																		elif key_name.startswith('''model/mlp'''	):
																								lowercase   :       Dict								=int(key_name[9:].split('''/'''	)[0]	)
																								if key_name.endswith('''/p1/kernel'''	):
																														lowercase   :       Any								='''model.blocks.%d.feed_forward.mlp.wi.weight''' % player
																														lowercase   :       str								=vnp.transpose([1, 0]	).copy()  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       Any								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/p1/bias'''	):
																														lowercase   :       List[Any]								='''model.blocks.%d.feed_forward.mlp.wi.bias''' % player
																														lowercase   :       Union[str, Any]								=vnp.copy()  # same because it is one dimensional
																														lowercase   :       Union[str, Any]								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/p2/kernel'''	):
																														lowercase   :       int								='''model.blocks.%d.feed_forward.mlp.wo.weight''' % player
																														lowercase   :       Tuple								=vnp.transpose([1, 0]	).copy()  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       int								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/p2/bias'''	):
																														lowercase   :       str								='''model.blocks.%d.feed_forward.mlp.wo.bias''' % player
																														lowercase   :       Optional[int]								=vnp.copy()  # same because it is one dimensional
																														lowercase   :       List[Any]								=torch.tensor(__magic_name__	)
																		elif key_name.startswith('''model/ln'''	):
																								lowercase   :       int								=int(key_name[8:].split('''/'''	)[0]	)
																								if key_name.endswith('''/b'''	):
																														lowercase   :       Any								='''model.blocks.%d.feed_forward.norm.bias''' % player
																														lowercase   :       Optional[int]								=vnp.copy()  # same because it is one dimensional
																														lowercase   :       Union[str, Any]								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/g'''	):
																														lowercase   :       Optional[Any]								='''model.blocks.%d.feed_forward.norm.weight''' % player
																														lowercase   :       Any								=vnp.copy()  # same because it is one dimensional
																														lowercase   :       List[Any]								=torch.tensor(__magic_name__	)
																		elif key_name.startswith('''model/att'''	):
																								lowercase   :       int								=int(key_name[9:].split('''/'''	)[0]	)
																								if key_name.endswith('''/qkv/kernel'''	):
																														lowercase   :       Optional[int]								=vnp.copy()  # Compute same dimension as Mesh-tensorflow using einsum
																														lowercase   :       Dict								=state[:, 0, :, :]
																														lowercase   :       Tuple								=state[:, 1, :, :]
																														lowercase   :       List[Any]								=state[:, 2, :, :]
																														lowercase   :       Optional[int]								=(
																														    state_q.reshape([state_q.shape[0], state_q.shape[1] * state_q.shape[2]]	)
																														    .transpose([1, 0]	)
																														    .copy()
																														)  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       Optional[Any]								=(
																														    state_k.reshape([state_k.shape[0], state_k.shape[1] * state_k.shape[2]]	)
																														    .transpose([1, 0]	)
																														    .copy()
																														)  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       Optional[int]								=(
																														    state_v.reshape([state_v.shape[0], state_v.shape[1] * state_v.shape[2]]	)
																														    .transpose([1, 0]	)
																														    .copy()
																														)  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       Dict								='''model.blocks.%d.self_attn.self_attn.q_proj.weight''' % player
																														lowercase   :       Dict								=torch.tensor(__magic_name__	)
																														lowercase   :       List[Any]								='''model.blocks.%d.self_attn.self_attn.k_proj.weight''' % player
																														lowercase   :       Optional[Any]								=torch.tensor(__magic_name__	)
																														lowercase   :       Optional[Any]								='''model.blocks.%d.self_attn.self_attn.v_proj.weight''' % player
																														lowercase   :       Tuple								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/o/kernel'''	):
																														lowercase   :       Dict								='''model.blocks.%d.self_attn.self_attn.out_proj.weight''' % player
																														lowercase   :       List[Any]								=(
																														    vnp.reshape([vnp.shape[0] * vnp.shape[1], vnp.shape[2]]	).transpose([1, 0]	).copy()
																														)  # Mesh-Tensorflow is a diagonal matrix
																														lowercase   :       str								=torch.tensor(__magic_name__	)
																		elif key_name.startswith('''model/an'''	):
																								lowercase   :       Optional[Any]								=int(key_name[8:].split('''/'''	)[0]	)
																								if key_name.endswith('''/b'''	):
																														lowercase   :       List[str]								='''model.blocks.%d.self_attn.norm.bias''' % player
																														lowercase   :       Union[str, Any]								=vnp.copy()  # same because it is one dimensional
																														lowercase   :       List[str]								=torch.tensor(__magic_name__	)
																								elif key_name.endswith('''/g'''	):
																														lowercase   :       Any								='''model.blocks.%d.self_attn.norm.weight''' % player
																														lowercase   :       Any								=vnp.copy()  # same because it is one dimensional
																														lowercase   :       Optional[Any]								=torch.tensor(__magic_name__	)
																		elif (
																		    key_name.startswith('''model/wte'''	)
																		    or key_name.startswith('''model/wpe'''	)
																		    or key_name.startswith('''model/ete'''	)
																		):
																								lowercase   :       Any								={'''wte''': '''embed_tokens''', '''wpe''': '''position_embeddings''', '''ete''': '''extra_position_embeddings'''}[
																								    key_name[-3:]
																								]
																								lowercase   :       Optional[Any]								='''model.%s.weight''' % nlayer
																								lowercase   :       Optional[int]								=vnp.copy()  # same in embedded
																								lowercase   :       List[Any]								=torch.tensor(__magic_name__	)
																								if key_name.startswith('''model/wte'''	):
																														lowercase   :       Tuple								='''lm_head.weight'''
																														lowercase   :       str								=vnp.copy()  # same in embedded
																														lowercase   :       Union[str, Any]								=torch.tensor(__magic_name__	)
																		elif key_name.startswith('''model/wob'''	):
																								lowercase   :       List[str]								='''final_logits_bias'''
																								lowercase   :       Dict								=vnp.copy()  # same in embedded
																								lowercase   :       Tuple								=state.reshape((1, -1)	)
																								lowercase   :       Dict								=torch.tensor(__magic_name__	)
																		elif key_name == "model/dense/kernel":
																								lowercase   :       Dict								='''model.last_project.weight'''
																								lowercase   :       int								=vnp.transpose([1, 0]	).copy()  # Mesh-Tensorflow is a diagonal matrix
																								lowercase   :       Optional[Any]								=torch.tensor(__magic_name__	)
																		elif key_name == "model/dense_1/bias":
																								lowercase   :       List[Any]								='''model.last_project.bias'''
																								lowercase   :       str								=vnp.copy()  # same because it is one dimensional
																								lowercase   :       Optional[Any]								=torch.tensor(__magic_name__	)
						torch.save(__magic_name__      ,   args.output	)


if __name__ == "__main__":
							UpperCamelCase_							=					argparse.ArgumentParser(
							    description="""model converter.""", formatter_class=argparse.ArgumentDefaultsHelpFormatter
							)
							parser.add_argument("""--tf_model_dir""", metavar="""PATH""", type=str, required=True, help="""import model""")
							parser.add_argument("""--output""", metavar="""PATH""", type=str, required=True, help="""output model""")
							UpperCamelCase_							=					parser.parse_args()
							convert_tf_gptsan_to_pt(args)





