


"""simple docstring"""


import argparse
import os

import torch

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)


A     :   List[str]			    =			{
    "sample_size": 3_2,
    "in_channels": 3,
    "out_channels": 3,
    "layers_per_block": 2,
    "num_class_embeds": 1_0_0_0,
    "block_out_channels": [3_2, 6_4],
    "attention_head_dim": 8,
    "down_block_types": [
        "ResnetDownsampleBlock2D",
        "AttnDownBlock2D",
    ],
    "up_block_types": [
        "AttnUpBlock2D",
        "ResnetUpsampleBlock2D",
    ],
    "resnet_time_scale_shift": "scale_shift",
    "upsample_type": "resnet",
    "downsample_type": "resnet",
}

A     :   Tuple			    =			{
    "sample_size": 6_4,
    "in_channels": 3,
    "out_channels": 3,
    "layers_per_block": 3,
    "num_class_embeds": 1_0_0_0,
    "block_out_channels": [1_9_2, 1_9_2 * 2, 1_9_2 * 3, 1_9_2 * 4],
    "attention_head_dim": 6_4,
    "down_block_types": [
        "ResnetDownsampleBlock2D",
        "AttnDownBlock2D",
        "AttnDownBlock2D",
        "AttnDownBlock2D",
    ],
    "up_block_types": [
        "AttnUpBlock2D",
        "AttnUpBlock2D",
        "AttnUpBlock2D",
        "ResnetUpsampleBlock2D",
    ],
    "resnet_time_scale_shift": "scale_shift",
    "upsample_type": "resnet",
    "downsample_type": "resnet",
}

A     :   Dict			    =			{
    "sample_size": 2_5_6,
    "in_channels": 3,
    "out_channels": 3,
    "layers_per_block": 2,
    "num_class_embeds": None,
    "block_out_channels": [2_5_6, 2_5_6, 2_5_6 * 2, 2_5_6 * 2, 2_5_6 * 4, 2_5_6 * 4],
    "attention_head_dim": 6_4,
    "down_block_types": [
        "ResnetDownsampleBlock2D",
        "ResnetDownsampleBlock2D",
        "ResnetDownsampleBlock2D",
        "AttnDownBlock2D",
        "AttnDownBlock2D",
        "AttnDownBlock2D",
    ],
    "up_block_types": [
        "AttnUpBlock2D",
        "AttnUpBlock2D",
        "AttnUpBlock2D",
        "ResnetUpsampleBlock2D",
        "ResnetUpsampleBlock2D",
        "ResnetUpsampleBlock2D",
    ],
    "resnet_time_scale_shift": "default",
    "upsample_type": "resnet",
    "downsample_type": "resnet",
}

A     :   Optional[Any]			    =			{
    "num_train_timesteps": 4_0,
    "sigma_min": 0.002,
    "sigma_max": 80.0,
}

A     :   List[str]			    =			{
    "num_train_timesteps": 2_0_1,
    "sigma_min": 0.002,
    "sigma_max": 80.0,
}

A     :   Optional[Any]			    =			{
    "num_train_timesteps": 1_5_1,
    "sigma_min": 0.002,
    "sigma_max": 80.0,
}


def _lowerCamelCase	(		_UpperCamelCase      ):

			'''simple docstring'''




			if isinstance(_UpperCamelCase    , _UpperCamelCase      ):
						return v
			if v.lower() in ("yes", "true", "t", "y", "1"):
						return True
			elif v.lower() in ("no", "false", "f", "n", "0"):
						return False
			else:
						raise argparse.ArgumentTypeError("boolean value expected"      )


def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase=False      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.in_layers.0.weight"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.in_layers.0.bias"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.in_layers.2.weight"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.in_layers.2.bias"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.emb_layers.1.weight"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.emb_layers.1.bias"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.out_layers.0.weight"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.out_layers.0.bias"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.out_layers.3.weight"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.out_layers.3.bias"]

			if has_skip:
						__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.skip_connection.weight"]
						__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.skip_connection.bias"]

			return new_checkpoint


def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase=None      ):

			'''simple docstring'''




			__lowerCAmelCase    ,	__lowerCAmelCase    ,	__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.qkv.weight"].chunk(3    , dim=0      )
			__lowerCAmelCase    ,	__lowerCAmelCase    ,	__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.qkv.bias"].chunk(3    , dim=0      )

			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.norm.weight"]
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.norm.bias"]

			__lowerCAmelCase				    =			weight_q.squeeze(-1      ).squeeze(-1      )
			__lowerCAmelCase				    =			bias_q.squeeze(-1      ).squeeze(-1      )
			__lowerCAmelCase				    =			weight_k.squeeze(-1      ).squeeze(-1      )
			__lowerCAmelCase				    =			bias_k.squeeze(-1      ).squeeze(-1      )
			__lowerCAmelCase				    =			weight_v.squeeze(-1      ).squeeze(-1      )
			__lowerCAmelCase				    =			bias_v.squeeze(-1      ).squeeze(-1      )

			__lowerCAmelCase				    =			(
			    checkpoint[f"{old_prefix}.proj_out.weight"].squeeze(-1      ).squeeze(-1      )
			)
			__lowerCAmelCase				    =			checkpoint[f"{old_prefix}.proj_out.bias"].squeeze(-1      ).squeeze(-1      )

			return new_checkpoint


def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			torch.load(_UpperCamelCase    , map_location="cpu"      )
			__lowerCAmelCase				    =			{}

			__lowerCAmelCase				    =			checkpoint["time_embed.0.weight"]
			__lowerCAmelCase				    =			checkpoint["time_embed.0.bias"]
			__lowerCAmelCase				    =			checkpoint["time_embed.2.weight"]
			__lowerCAmelCase				    =			checkpoint["time_embed.2.bias"]

			if unet_config["num_class_embeds"] is not None:
						__lowerCAmelCase				    =			checkpoint["label_emb.weight"]

			__lowerCAmelCase				    =			checkpoint["input_blocks.0.0.weight"]
			__lowerCAmelCase				    =			checkpoint["input_blocks.0.0.bias"]

			__lowerCAmelCase				    =			unet_config["down_block_types"]
			__lowerCAmelCase				    =			unet_config["layers_per_block"]
			__lowerCAmelCase				    =			unet_config["attention_head_dim"]
			__lowerCAmelCase				    =			unet_config["block_out_channels"]
			__lowerCAmelCase				    =			1
			__lowerCAmelCase				    =			channels_list[0]

			for i, layer_type in enumerate(_UpperCamelCase      ):
						__lowerCAmelCase				    =			channels_list[i]
						__lowerCAmelCase				    =			current_channels != prev_channels
						if layer_type == "ResnetDownsampleBlock2D":
									for j in range(_UpperCamelCase      ):
												__lowerCAmelCase				    =			f"down_blocks.{i}.resnets.{j}"
												__lowerCAmelCase				    =			f"input_blocks.{current_layer}.0"
												__lowerCAmelCase				    =			True if j == 0 and downsample_block_has_skip else False
												__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , has_skip=_UpperCamelCase      )
												current_layer += 1

						elif layer_type == "AttnDownBlock2D":
									for j in range(_UpperCamelCase      ):
												__lowerCAmelCase				    =			f"down_blocks.{i}.resnets.{j}"
												__lowerCAmelCase				    =			f"input_blocks.{current_layer}.0"
												__lowerCAmelCase				    =			True if j == 0 and downsample_block_has_skip else False
												__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , has_skip=_UpperCamelCase      )
												__lowerCAmelCase				    =			f"down_blocks.{i}.attentions.{j}"
												__lowerCAmelCase				    =			f"input_blocks.{current_layer}.1"
												__lowerCAmelCase				    =			convert_attention(
												    _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )
												current_layer += 1

						if i != len(_UpperCamelCase      ) - 1:
									__lowerCAmelCase				    =			f"down_blocks.{i}.downsamplers.0"
									__lowerCAmelCase				    =			f"input_blocks.{current_layer}.0"
									__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )
									current_layer += 1

						__lowerCAmelCase				    =			current_channels

			# hardcoded the mid-block for now
			__lowerCAmelCase				    =			"mid_block.resnets.0"
			__lowerCAmelCase				    =			"middle_block.0"
			__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )
			__lowerCAmelCase				    =			"mid_block.attentions.0"
			__lowerCAmelCase				    =			"middle_block.1"
			__lowerCAmelCase				    =			convert_attention(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )
			__lowerCAmelCase				    =			"mid_block.resnets.1"
			__lowerCAmelCase				    =			"middle_block.2"
			__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )

			__lowerCAmelCase				    =			0
			__lowerCAmelCase				    =			unet_config["up_block_types"]

			for i, layer_type in enumerate(_UpperCamelCase      ):
						if layer_type == "ResnetUpsampleBlock2D":
									for j in range(layers_per_block + 1      ):
												__lowerCAmelCase				    =			f"up_blocks.{i}.resnets.{j}"
												__lowerCAmelCase				    =			f"output_blocks.{current_layer}.0"
												__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , has_skip=_UpperCamelCase      )
												current_layer += 1

									if i != len(_UpperCamelCase      ) - 1:
												__lowerCAmelCase				    =			f"up_blocks.{i}.upsamplers.0"
												__lowerCAmelCase				    =			f"output_blocks.{current_layer-1}.1"
												__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )
						elif layer_type == "AttnUpBlock2D":
									for j in range(layers_per_block + 1      ):
												__lowerCAmelCase				    =			f"up_blocks.{i}.resnets.{j}"
												__lowerCAmelCase				    =			f"output_blocks.{current_layer}.0"
												__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , has_skip=_UpperCamelCase      )
												__lowerCAmelCase				    =			f"up_blocks.{i}.attentions.{j}"
												__lowerCAmelCase				    =			f"output_blocks.{current_layer}.1"
												__lowerCAmelCase				    =			convert_attention(
												    _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )
												current_layer += 1

									if i != len(_UpperCamelCase      ) - 1:
												__lowerCAmelCase				    =			f"up_blocks.{i}.upsamplers.0"
												__lowerCAmelCase				    =			f"output_blocks.{current_layer-1}.2"
												__lowerCAmelCase				    =			convert_resnet(_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      )

			__lowerCAmelCase				    =			checkpoint["out.0.weight"]
			__lowerCAmelCase				    =			checkpoint["out.0.bias"]
			__lowerCAmelCase				    =			checkpoint["out.2.weight"]
			__lowerCAmelCase				    =			checkpoint["out.2.bias"]

			return new_checkpoint


if __name__ == "__main__":
	A     :   Dict			    =			argparse.ArgumentParser()

	parser.add_argument("--unet_path", default=None, type=str, required=True, help="Path to the unet.pt to convert.")
	parser.add_argument(
	    "--dump_path", default=None, type=str, required=True, help="Path to output the converted UNet model."
	)
	parser.add_argument("--class_cond", default=True, type=str, help="Whether the model is class-conditional.")

	A     :   Optional[Any]			    =			parser.parse_args()
	A     :   int			    =			strabool(args.class_cond)

	A     :   Dict			    =			os.path.basename(args.unet_path)
	print(f'''Checkpoint: {ckpt_name}''')

	# Get U-Net config
	if "imagenet64" in ckpt_name:
		A     :   Union[str, Any]			    =			IMAGENET_64_UNET_CONFIG
	elif "256" in ckpt_name and (("bedroom" in ckpt_name) or ("cat" in ckpt_name)):
		A     :   Any			    =			LSUN_256_UNET_CONFIG
	elif "test" in ckpt_name:
		A     :   Union[str, Any]			    =			TEST_UNET_CONFIG
	else:
		raise ValueError(f'''Checkpoint type {ckpt_name} is not currently supported.''')

	if not args.class_cond:
		A     :   Union[str, Any]			    =			None

	A     :   int			    =			con_pt_to_diffuser(args.unet_path, unet_config)

	A     :   Tuple			    =			UNetaDModel(**unet_config)
	image_unet.load_state_dict(converted_unet_ckpt)

	# Get scheduler config
	if "cd" in ckpt_name or "test" in ckpt_name:
		A     :   Union[str, Any]			    =			CD_SCHEDULER_CONFIG
	elif "ct" in ckpt_name and "imagenet64" in ckpt_name:
		A     :   Any			    =			CT_IMAGENET_64_SCHEDULER_CONFIG
	elif "ct" in ckpt_name and "256" in ckpt_name and (("bedroom" in ckpt_name) or ("cat" in ckpt_name)):
		A     :   Any			    =			CT_LSUN_256_SCHEDULER_CONFIG
	else:
		raise ValueError(f'''Checkpoint type {ckpt_name} is not currently supported.''')

	A     :   Optional[int]			    =			CMStochasticIterativeScheduler(**scheduler_config)

	A     :   Optional[Any]			    =			ConsistencyModelPipeline(unet=image_unet, scheduler=cm_scheduler)
	consistency_model.save_pretrained(args.dump_path)







import gc
import unittest

from parameterized import parameterized

from diffusers import FlaxUNetaDConditionModel
from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import load_hf_numpy, require_flax, slow


if is_flax_available():
					import jax
					import jax.numpy as jnp





@slow
@require_flax
class 	snake_case_   (		unittest.TestCase					):







			'''simple docstring'''

			def 				snake_case__(     self :				Any							,   _UpperCamelCase :				Any							,   _UpperCamelCase :				Tuple							)      ->List[Any]:
									return f'''gaussian_noise_s={seed}_shape={'_'.join([str(_UpperCamelCase							) for s in shape]							)}.npy'''







			def 				snake_case__(     self :				Any							)      ->List[str]:
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()







			def 				snake_case__(     self :				int							,   _UpperCamelCase :				Union[str, Any]=0							,   _UpperCamelCase :				int=(4, 4, 6_4, 6_4)							,   _UpperCamelCase :				Optional[int]=False							)      ->Tuple:
									snake_case_   =	jnp.bfloataa if fpaa else jnp.floataa
									snake_case_   =	jnp.array(load_hf_numpy(self.get_file_format(_UpperCamelCase							,   _UpperCamelCase							)							)							,   dtype=_UpperCamelCase							)
									return image







			def 				snake_case__(     self :				List[Any]							,   _UpperCamelCase :				Optional[Any]=False							,   _UpperCamelCase :				Optional[int]="CompVis/stable-diffusion-v1-4"							)      ->Optional[Any]:
									snake_case_   =	jnp.bfloataa if fpaa else jnp.floataa
									snake_case_   =	'''bf16''' if fpaa else None

									snake_case_,    snake_case_   =	FlaxUNetaDConditionModel.from_pretrained(
									    _UpperCamelCase							,   subfolder='''unet'''							,   dtype=_UpperCamelCase							,   revision=_UpperCamelCase							)
									return model, params







			def 				snake_case__(     self :				Dict							,   _UpperCamelCase :				List[Any]=0							,   _UpperCamelCase :				Tuple=(4, 7_7, 7_6_8)							,   _UpperCamelCase :				List[Any]=False							)      ->int:
									snake_case_   =	jnp.bfloataa if fpaa else jnp.floataa
									snake_case_   =	jnp.array(load_hf_numpy(self.get_file_format(_UpperCamelCase							,   _UpperCamelCase							)							)							,   dtype=_UpperCamelCase							)
									return hidden_states







			@parameterized.expand(
			    [
			        # fmt: off
			        [8_3, 4, [-0.2323, -0.1304, 0.0813, -0.3093, -0.0919, -0.1571, -0.1125, -0.5806]],
			        [1_7, 0.55, [-0.0831, -0.2443, 0.0901, -0.0919, 0.3396, 0.0103, -0.3743, 0.0701]],
			        [8, 0.89, [-0.4863, 0.0859, 0.0875, -0.1658, 0.9199, -0.0114, 0.4839, 0.4639]],
			        [3, 1_0_0_0, [-0.5649, 0.2402, -0.5518, 0.1248, 1.1328, -0.2443, -0.0325, -1.0078]],
			        # fmt: on
			    ]							)
			def 				snake_case__(     self :				Optional[Any]							,   _UpperCamelCase :				List[Any]							,   _UpperCamelCase :				List[str]							,   _UpperCamelCase :				Optional[int]							)      ->Union[str, Any]:
									snake_case_,    snake_case_   =	self.get_unet_model(model_id='''CompVis/stable-diffusion-v1-4'''							,   fpaa=_UpperCamelCase							)
									snake_case_   =	self.get_latents(_UpperCamelCase							,   fpaa=_UpperCamelCase							)
									snake_case_   =	self.get_encoder_hidden_states(_UpperCamelCase							,   fpaa=_UpperCamelCase							)

									snake_case_   =	model.apply(
									    {'''params''': params}							,   _UpperCamelCase							,   jnp.array(_UpperCamelCase							,   dtype=jnp.intaa							)							,   encoder_hidden_states=_UpperCamelCase							,   ).sample

									assert sample.shape == latents.shape

									snake_case_   =	jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())							)							,   dtype=jnp.floataa							)
									snake_case_   =	jnp.array(_UpperCamelCase							,   dtype=jnp.floataa							)

									# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, in the same hardware
									assert jnp.allclose(_UpperCamelCase							,   _UpperCamelCase							,   atol=1e-2							)







			@parameterized.expand(
			    [
			        # fmt: off
			        [8_3, 4, [0.1514, 0.0807, 0.1624, 0.1016, -0.1896, 0.0263, 0.0677, 0.2310]],
			        [1_7, 0.55, [0.1164, -0.0216, 0.0170, 0.1589, -0.3120, 0.1005, -0.0581, -0.1458]],
			        [8, 0.89, [-0.1758, -0.0169, 0.1004, -0.1411, 0.1312, 0.1103, -0.1996, 0.2139]],
			        [3, 1_0_0_0, [0.1214, 0.0352, -0.0731, -0.1562, -0.0994, -0.0906, -0.2340, -0.0539]],
			        # fmt: on
			    ]							)
			def 				snake_case__(     self :				Optional[int]							,   _UpperCamelCase :				Optional[Any]							,   _UpperCamelCase :				Union[str, Any]							,   _UpperCamelCase :				str							)      ->Dict:
									snake_case_,    snake_case_   =	self.get_unet_model(model_id='''stabilityai/stable-diffusion-2'''							,   fpaa=_UpperCamelCase							)
									snake_case_   =	self.get_latents(_UpperCamelCase							,   shape=(4, 4, 9_6, 9_6)							,   fpaa=_UpperCamelCase							)
									snake_case_   =	self.get_encoder_hidden_states(_UpperCamelCase							,   shape=(4, 7_7, 1_0_2_4)							,   fpaa=_UpperCamelCase							)

									snake_case_   =	model.apply(
									    {'''params''': params}							,   _UpperCamelCase							,   jnp.array(_UpperCamelCase							,   dtype=jnp.intaa							)							,   encoder_hidden_states=_UpperCamelCase							,   ).sample

									assert sample.shape == latents.shape

									snake_case_   =	jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())							)							,   dtype=jnp.floataa							)
									snake_case_   =	jnp.array(_UpperCamelCase							,   dtype=jnp.floataa							)

									# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, on the same hardware
									assert jnp.allclose(_UpperCamelCase							,   _UpperCamelCase							,   atol=1e-2							)