



import gc
import unittest

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

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

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


enable_full_determinism()







class     __lowerCAmelCase					(    A       ,		A       ,		unittest.TestCase		):
			UpperCamelCase           =  StableDiffusionSAGPipeline
			UpperCamelCase           =  TEXT_TO_IMAGE_PARAMS
			UpperCamelCase           =  TEXT_TO_IMAGE_BATCH_PARAMS
			UpperCamelCase           =  TEXT_TO_IMAGE_IMAGE_PARAMS
			UpperCamelCase           =  TEXT_TO_IMAGE_IMAGE_PARAMS
			UpperCamelCase           =  False



			def 					_lowerCamelCase (		self     :						Union[str, Any])     ->					List[Any]:



						"""simple docstring"""






						torch.manual_seed(0)
						_UpperCAmelCase	 =     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				,	)
						_UpperCAmelCase	 =     DDIMScheduler(
						    beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	beta_schedule='scaled_linear'				,	clip_sample=A				,	set_alpha_to_one=A				,	)
						torch.manual_seed(0)
						_UpperCAmelCase	 =     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)
						_UpperCAmelCase	 =     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=10_00				,	)
						_UpperCAmelCase	 =     CLIPTextModel(A)
						_UpperCAmelCase	 =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

						_UpperCAmelCase	 =     {
						    'unet': unet,
						    'scheduler': scheduler,
						    'vae': vae,
						    'text_encoder': text_encoder,
						    'tokenizer': tokenizer,
						    'safety_checker': None,
						    'feature_extractor': None,
						}
						return components



			def 					_lowerCamelCase (		self     :						Optional[int]				,	A     :						Tuple				,	A     :						List[str]=0)     ->					Tuple:



						"""simple docstring"""






						if str(A).startswith('mps'):
									_UpperCAmelCase	 =     torch.manual_seed(A)
						else:
									_UpperCAmelCase	 =     torch.Generator(device=A).manual_seed(A)
						_UpperCAmelCase	 =     {
						    'prompt': '.',
						    'generator': generator,
						    'num_inference_steps': 2,
						    'guidance_scale': 1.0,
						    'sag_scale': 1.0,
						    'output_type': 'numpy',
						}
						return inputs



			def 					_lowerCamelCase (		self     :						int)     ->					List[str]:



						"""simple docstring"""






						super().test_inference_batch_single_identical(expected_max_diff=3E-3)







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



			def 					_lowerCamelCase (		self     :						str)     ->					Union[str, Any]:



						"""simple docstring"""






						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()



			def 					_lowerCamelCase (		self     :						Tuple)     ->					str:



						"""simple docstring"""






						_UpperCAmelCase	 =     StableDiffusionSAGPipeline.from_pretrained('CompVis/stable-diffusion-v1-4')
						_UpperCAmelCase	 =     sag_pipe.to(A)
						sag_pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase	 =     '.'
						_UpperCAmelCase	 =     torch.manual_seed(0)
						_UpperCAmelCase	 =     sag_pipe(
						    [prompt]				,	generator=A				,	guidance_scale=7.5				,	sag_scale=1.0				,	num_inference_steps=20				,	output_type='np')

						_UpperCAmelCase	 =     output.images

						_UpperCAmelCase	 =     image[0, -3:, -3:, -1]

						assert image.shape == (1, 5_12, 5_12, 3)
						_UpperCAmelCase	 =     np.array([0.1_5_6_8, 0.1_7_3_8, 0.1_6_9_5, 0.1_6_9_3, 0.1_5_0_7, 0.1_7_0_5, 0.1_5_4_7, 0.1_7_5_1, 0.1_9_4_9])

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



			def 					_lowerCamelCase (		self     :						Tuple)     ->					Any:



						"""simple docstring"""






						_UpperCAmelCase	 =     StableDiffusionSAGPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base')
						_UpperCAmelCase	 =     sag_pipe.to(A)
						sag_pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase	 =     '.'
						_UpperCAmelCase	 =     torch.manual_seed(0)
						_UpperCAmelCase	 =     sag_pipe(
						    [prompt]				,	generator=A				,	guidance_scale=7.5				,	sag_scale=1.0				,	num_inference_steps=20				,	output_type='np')

						_UpperCAmelCase	 =     output.images

						_UpperCAmelCase	 =     image[0, -3:, -3:, -1]

						assert image.shape == (1, 5_12, 5_12, 3)
						_UpperCAmelCase	 =     np.array([0.3_4_5_9, 0.2_8_7_6, 0.2_5_3_7, 0.3_0_0_2, 0.2_6_7_1, 0.2_1_6_0, 0.3_0_2_6, 0.2_2_6_2, 0.2_3_7_1])

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



			def 					_lowerCamelCase (		self     :						List[Any])     ->					int:



						"""simple docstring"""






						_UpperCAmelCase	 =     StableDiffusionSAGPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base')
						_UpperCAmelCase	 =     sag_pipe.to(A)
						sag_pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase	 =     '.'
						_UpperCAmelCase	 =     torch.manual_seed(0)
						_UpperCAmelCase	 =     sag_pipe(
						    [prompt]				,	width=7_68				,	height=5_12				,	generator=A				,	guidance_scale=7.5				,	sag_scale=1.0				,	num_inference_steps=20				,	output_type='np'				,	)

						_UpperCAmelCase	 =     output.images

						assert image.shape == (1, 5_12, 7_68, 3)





import argparse
import json
import os
from pathlib import Path

import requests
import torch

from transformers import JukeboxConfig, JukeboxModel
from transformers.utils import logging


logging.set_verbosity_info()
UpperCAmelCase__     = logging.get_logger(__name__)


UpperCAmelCase__     = "https://openaipublic.azureedge.net/jukebox/models/"
UpperCAmelCase__     = {
    "jukebox-1b-lyrics": [
        "5b/vqvae.pth.tar",
        "5b/prior_level_0.pth.tar",
        "5b/prior_level_1.pth.tar",
        "1b_lyrics/prior_level_2.pth.tar",
    ],
    "jukebox-5b-lyrics": [
        "5b/vqvae.pth.tar",
        "5b/prior_level_0.pth.tar",
        "5b/prior_level_1.pth.tar",
        "5b_lyrics/prior_level_2.pth.tar",
    ],
}





def 							A			(  _UpperCAmelCase       :     List[str]					)       ->				Tuple:







			'''simple docstring'''
			if key.endswith('.model.1.bias'					) and len(key.split('.'					)					) > 10:
						_UpperCAmelCase	 =     key.replace('.model.1.bias'							,     '.conv1d_1.bias'					)
			elif key.endswith('.model.1.weight'					) and len(key.split('.'					)					) > 10:
						_UpperCAmelCase	 =     key.replace('.model.1.weight'							,     '.conv1d_1.weight'					)
			elif key.endswith('.model.3.bias'					) and len(key.split('.'					)					) > 10:
						_UpperCAmelCase	 =     key.replace('.model.3.bias'							,     '.conv1d_2.bias'					)
			elif key.endswith('.model.3.weight'					) and len(key.split('.'					)					) > 10:
						_UpperCAmelCase	 =     key.replace('.model.3.weight'							,     '.conv1d_2.weight'					)

			if "conditioner_blocks.0." in key:
						_UpperCAmelCase	 =     key.replace('conditioner_blocks.0'							,     'conditioner_blocks'					)

			if "prime_prior" in key:
						_UpperCAmelCase	 =     key.replace('prime_prior'							,     'encoder'					)

			if ".emb." in key and "total" not in key and "absolute" not in key and "relative" not in key:
						_UpperCAmelCase	 =     key.replace('.emb.'							,     '.'					)

			if key.endswith('k'					):  # replace vqvae.X.k with vqvae.X.codebook
						return key.replace('.k'							,     '.codebook'					)
			if "y_emb." in key:
						return key.replace('y_emb.'							,     'metadata_embedding.'					)

			if "x_emb.emb." in key:
						_UpperCAmelCase	 =     key.replace('0.x_emb.emb'							,     'embed_tokens'					)

			if "prime_state_ln" in key:
						return key.replace('prime_state_ln'							,     'encoder.final_layer_norm'					)
			if ".ln" in key:
						return key.replace('.ln'							,     '.layer_norm'					)
			if "_ln" in key:
						return key.replace('_ln'							,     '_layer_norm'					)

			if "prime_state_proj" in key:
						return key.replace('prime_state_proj'							,     'encoder.proj_in'					)
			if "prime_x_out" in key:
						return key.replace('prime_x_out'							,     'encoder.lm_head'					)
			if "prior.x_out" in key:
						return key.replace('x_out'							,     'fc_proj_out'					)
			if "x_emb" in key:
						return key.replace('x_emb'							,     'embed_tokens'					)

			return key





def 							A			(  _UpperCAmelCase       :     str							,     _UpperCAmelCase       :     str							,     _UpperCAmelCase       :     Tuple							,     _UpperCAmelCase       :     List[Any]					)       ->				Tuple:







			'''simple docstring'''
			_UpperCAmelCase	 =     {}
			import re

			_UpperCAmelCase	 =     re.compile(R'encoders.(\d*).level_blocks.(\d*).model.(\d*).(\d).(bias|weight)'					)
			_UpperCAmelCase	 =     re.compile(
			    R'encoders.(\d*).level_blocks.(\d*).model.(\d*).(\d).model.(\d*).model.(\d*).(bias|weight)'					)
			_UpperCAmelCase	 =     re.compile(R'encoders.(\d*).level_blocks.(\d*).model.(\d*).(bias|weight)'					)

			_UpperCAmelCase	 =     re.compile(R'decoders.(\d*).level_blocks.(\d*).model.(\d*).(\d).(bias|weight)'					)
			_UpperCAmelCase	 =     re.compile(
			    R'decoders.(\d*).level_blocks.(\d*).model.(\d*).(\d).model.(\d*).model.(\d*).(bias|weight)'					)
			_UpperCAmelCase	 =     re.compile(R'decoders.(\d*).level_blocks.(\d*).model.(\d*).(bias|weight)'					)

			_UpperCAmelCase	 =     re.compile(R'conditioner_blocks.(\d*).cond.model.(\d*).(\d).(bias|weight)'					)
			_UpperCAmelCase	 =     re.compile(
			    R'conditioner_blocks.(\d*).cond.model.(\d*).(\d).model.(\d*).model.(\d*).(bias|weight)'					)
			_UpperCAmelCase	 =     re.compile(R'conditioner_blocks.(\d*).cond.model.(\d*).(bias|weight)'					)

			for original_key, value in state_dict.items():
						# rename vqvae.encoder keys
						if re_encoder_block_conv_in.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_encoder_block_conv_in.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     int(groups[2]					) * 2 + int(groups[3]					)
									_UpperCAmelCase	 =     F"encoders.{groups[0]}.level_blocks.{groups[1]}.downsample_block.{block_index}.{groups[-1]}"
									_UpperCAmelCase	 =     re_encoder_block_conv_in.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						elif re_encoder_block_resnet.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_encoder_block_resnet.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     int(groups[2]					) * 2 + int(groups[3]					)
									_UpperCAmelCase	 =     {'1': 1, '3': 2}[groups[-2]]
									_UpperCAmelCase	 =     F"encoders.{groups[0]}.level_blocks.{groups[1]}.downsample_block.{block_index}."
									_UpperCAmelCase	 =     F"resnet_block.{groups[-3]}.conv1d_{conv_index}.{groups[-1]}"
									_UpperCAmelCase	 =     prefix + resnet_block
									_UpperCAmelCase	 =     re_encoder_block_resnet.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						elif re_encoder_block_proj_out.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_encoder_block_proj_out.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     F"encoders.{groups[0]}.level_blocks.{groups[1]}.proj_out.{groups[-1]}"
									_UpperCAmelCase	 =     re_encoder_block_proj_out.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						# rename vqvae.decoder keys
						elif re_decoder_block_conv_out.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_decoder_block_conv_out.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     int(groups[2]					) * 2 + int(groups[3]					) - 2
									_UpperCAmelCase	 =     F"decoders.{groups[0]}.level_blocks.{groups[1]}.upsample_block.{block_index}.{groups[-1]}"
									_UpperCAmelCase	 =     re_decoder_block_conv_out.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						elif re_decoder_block_resnet.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_decoder_block_resnet.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     int(groups[2]					) * 2 + int(groups[3]					) - 2
									_UpperCAmelCase	 =     {'1': 1, '3': 2}[groups[-2]]
									_UpperCAmelCase	 =     F"decoders.{groups[0]}.level_blocks.{groups[1]}.upsample_block.{block_index}."
									_UpperCAmelCase	 =     F"resnet_block.{groups[-3]}.conv1d_{conv_index}.{groups[-1]}"
									_UpperCAmelCase	 =     prefix + resnet_block
									_UpperCAmelCase	 =     re_decoder_block_resnet.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						elif re_decoder_block_proj_in.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_decoder_block_proj_in.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     F"decoders.{groups[0]}.level_blocks.{groups[1]}.proj_in.{groups[-1]}"
									_UpperCAmelCase	 =     re_decoder_block_proj_in.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						# rename prior cond.model to upsampler.upsample_block and resnet
						elif re_prior_cond_conv_out.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_prior_cond_conv_out.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     int(groups[1]					) * 2 + int(groups[2]					) - 2
									_UpperCAmelCase	 =     F"conditioner_blocks.upsampler.upsample_block.{block_index}.{groups[-1]}"
									_UpperCAmelCase	 =     re_prior_cond_conv_out.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						elif re_prior_cond_resnet.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_prior_cond_resnet.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     int(groups[1]					) * 2 + int(groups[2]					) - 2
									_UpperCAmelCase	 =     {'1': 1, '3': 2}[groups[-2]]
									_UpperCAmelCase	 =     F"conditioner_blocks.upsampler.upsample_block.{block_index}."
									_UpperCAmelCase	 =     F"resnet_block.{groups[-3]}.conv1d_{conv_index}.{groups[-1]}"
									_UpperCAmelCase	 =     prefix + resnet_block
									_UpperCAmelCase	 =     re_prior_cond_resnet.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						elif re_prior_cond_proj_in.fullmatch(_UpperCAmelCase					):
									_UpperCAmelCase	 =     re_prior_cond_proj_in.match(_UpperCAmelCase					)
									_UpperCAmelCase	 =     regex_match.groups()
									_UpperCAmelCase	 =     F"conditioner_blocks.upsampler.proj_in.{groups[-1]}"
									_UpperCAmelCase	 =     re_prior_cond_proj_in.sub(_UpperCAmelCase							,     _UpperCAmelCase					)

						# keep original key
						else:
									_UpperCAmelCase	 =     original_key

						_UpperCAmelCase	 =     replace_key(_UpperCAmelCase					)

						if F"{key_prefix}.{key}" not in model_state_dict or key is None:
									print(F"failed converting {original_key} to {key}, does not match"					)

						# handle missmatched shape
						elif value.shape != model_state_dict[F"{key_prefix}.{key}"].shape:
									_UpperCAmelCase	 =     model_state_dict[F"{key_prefix}.{key}"]
									print(F"{original_key}-> {key} : \nshape {val.shape} and { value.shape}, do not match"					)
									_UpperCAmelCase	 =     original_key

						_UpperCAmelCase	 =     original_key
						_UpperCAmelCase	 =     value

			return new_dict





@torch.no_grad()
def 							A			(  _UpperCAmelCase       :     List[str]=None							,     _UpperCAmelCase       :     Dict=None					)       ->				Dict:







			'''simple docstring'''
			for file in MODEL_MAPPING[model_name]:
						if not os.path.isfile(F"{pytorch_dump_folder_path}/{file.split('/'					)[-1]}"					):
									_UpperCAmelCase	 =     requests.get(F"{PREFIX}{file}"							,     allow_redirects=_UpperCAmelCase					)
									os.makedirs(F"{pytorch_dump_folder_path}/"							,     exist_ok=_UpperCAmelCase					)
									open(F"{pytorch_dump_folder_path}/{file.split('/'					)[-1]}"							,     'wb'					).write(r.content					)

			_UpperCAmelCase	 =     MODEL_MAPPING[model_name.split('/'					)[-1]]

			_UpperCAmelCase	 =     JukeboxConfig.from_pretrained(_UpperCAmelCase					)
			_UpperCAmelCase	 =     JukeboxModel(_UpperCAmelCase					)

			_UpperCAmelCase	 =     []
			_UpperCAmelCase	 =     {}
			for i, dict_name in enumerate(_UpperCAmelCase					):
						_UpperCAmelCase	 =     torch.load(F"{pytorch_dump_folder_path}/{dict_name.split('/'					)[-1]}"					)['model']

						_UpperCAmelCase	 =     {}
						for k in old_dic.keys():
									if k.endswith('.b'					):
												_UpperCAmelCase	 =     old_dic[k]
									elif k.endswith('.w'					):
												_UpperCAmelCase	 =     old_dic[k]
									elif "level_2" not in dict_name and "cond.model." in k:
												_UpperCAmelCase	 =     old_dic[k]
									else:
												_UpperCAmelCase	 =     old_dic[k]

						_UpperCAmelCase	 =     'vqvae' if i == 0 else F"priors.{3 - i}"
						_UpperCAmelCase	 =     fix_jukebox_keys(_UpperCAmelCase							,     model.state_dict()							,     _UpperCAmelCase							,     _UpperCAmelCase					)
						weight_dict.append(_UpperCAmelCase					)

			_UpperCAmelCase	 =     weight_dict.pop(0					)
			model.vqvae.load_state_dict(_UpperCAmelCase					)
			for i in range(len(_UpperCAmelCase					)					):
						model.priors[i].load_state_dict(weight_dict[2 - i]					)

			Path(_UpperCAmelCase					).mkdir(exist_ok=_UpperCAmelCase					)
			with open(F"{pytorch_dump_folder_path}/mapping.json"							,     'w'					) as txtfile:
						json.dump(_UpperCAmelCase							,     _UpperCAmelCase					)

			print(F"Saving model {model_name} to {pytorch_dump_folder_path}"					)
			model.save_pretrained(_UpperCAmelCase					)

			return weight_dict


if __name__ == "__main__":
							UpperCAmelCase__     = argparse.ArgumentParser()
							# Required parameters
							parser.add_argument(
							    "--model_name",
							    default="jukebox-5b-lyrics",
							    type=str,
							    help="Name of the model you'd like to convert.",
							)
							parser.add_argument(
							    "--pytorch_dump_folder_path",
							    default="jukebox-5b-lyrics-converted",
							    type=str,
							    help="Path to the output PyTorch model directory.",
							)
							UpperCAmelCase__     = parser.parse_args()
							convert_openai_checkpoint(args.model_name, args.pytorch_dump_folder_path)
