






import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()


class 							__UpperCamelCase     (  _lowerCAmelCase ,      _lowerCAmelCase ,      _lowerCAmelCase ,      unittest.TestCase ):
	__snake_case						:Dict =							StableUnCLIPPipeline
	__snake_case						:List[Any] =							TEXT_TO_IMAGE_PARAMS
	__snake_case						:Dict =							TEXT_TO_IMAGE_BATCH_PARAMS
	__snake_case						:Union[str, Any] =							TEXT_TO_IMAGE_IMAGE_PARAMS
	__snake_case						:Any =							TEXT_TO_IMAGE_IMAGE_PARAMS

	# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
	__snake_case						:Tuple =							False







	def 			_a (	self   :				str      )		->			Any:
		"""simple docstring"""







		__lowercase   							=	32
		__lowercase   							=	embedder_hidden_size

		# prior components

		torch.manual_seed(0      )
		__lowercase   							=	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""      )

		torch.manual_seed(0      )
		__lowercase   							=	CLIPTextModelWithProjection(
		    CLIPTextConfig(
		        bos_token_id=0							,					eos_token_id=2							,					hidden_size=_lowerCAmelCase							,					projection_dim=_lowerCAmelCase							,					intermediate_size=37							,					layer_norm_eps=1e-05							,					num_attention_heads=4							,					num_hidden_layers=5							,					pad_token_id=1							,					vocab_size=1000							,					)      )

		torch.manual_seed(0      )
		__lowercase   							=	PriorTransformer(
		    num_attention_heads=2							,					attention_head_dim=12							,					embedding_dim=_lowerCAmelCase							,					num_layers=1							,					)

		torch.manual_seed(0      )
		__lowercase   							=	DDPMScheduler(
		    variance_type="""fixed_small_log"""							,					prediction_type="""sample"""							,					num_train_timesteps=1000							,					clip_sample=_lowerCAmelCase							,					clip_sample_range=5.0							,					beta_schedule="""squaredcos_cap_v2"""							,					)

		# regular denoising components

		torch.manual_seed(0      )
		__lowercase   							=	StableUnCLIPImageNormalizer(embedding_dim=_lowerCAmelCase      )
		__lowercase   							=	DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""      )

		torch.manual_seed(0      )
		__lowercase   							=	CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""      )

		torch.manual_seed(0      )
		__lowercase   							=	CLIPTextModel(
		    CLIPTextConfig(
		        bos_token_id=0							,					eos_token_id=2							,					hidden_size=_lowerCAmelCase							,					projection_dim=32							,					intermediate_size=37							,					layer_norm_eps=1e-05							,					num_attention_heads=4							,					num_hidden_layers=5							,					pad_token_id=1							,					vocab_size=1000							,					)      )

		torch.manual_seed(0      )
		__lowercase   							=	UNetaDConditionModel(
		    sample_size=32							,					in_channels=4							,					out_channels=4							,					down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")							,					up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")							,					block_out_channels=(32, 64)							,					attention_head_dim=(2, 4)							,					class_embed_type="""projection"""							,					projection_class_embeddings_input_dim=embedder_projection_dim * 2							,					cross_attention_dim=_lowerCAmelCase							,					layers_per_block=1							,					upcast_attention=_lowerCAmelCase							,					use_linear_projection=_lowerCAmelCase							,					)

		torch.manual_seed(0      )
		__lowercase   							=	DDIMScheduler(
		    beta_schedule="""scaled_linear"""							,					beta_start=0.00_085							,					beta_end=0.012							,					prediction_type="""v_prediction"""							,					set_alpha_to_one=_lowerCAmelCase							,					steps_offset=1							,					)

		torch.manual_seed(0      )
		__lowercase   							=	AutoencoderKL()

		__lowercase   							=	{
		    # prior components
		    """prior_tokenizer""": prior_tokenizer,
		    """prior_text_encoder""": prior_text_encoder,
		    """prior""": prior,
		    """prior_scheduler""": prior_scheduler,
		    # image noising components
		    """image_normalizer""": image_normalizer,
		    """image_noising_scheduler""": image_noising_scheduler,
		    # regular denoising components
		    """tokenizer""": tokenizer,
		    """text_encoder""": text_encoder,
		    """unet""": unet,
		    """scheduler""": scheduler,
		    """vae""": vae,
		}

		return components







	def 			_a (	self   :				int							,					_lowerCAmelCase   :				str							,					_lowerCAmelCase   :				Union[str, Any]=0      )		->			List[Any]:
		"""simple docstring"""







		if str(_lowerCAmelCase      ).startswith("""mps"""      ):
			__lowercase   							=	torch.manual_seed(_lowerCAmelCase      )
		else:
			__lowercase   							=	torch.Generator(device=_lowerCAmelCase      ).manual_seed(_lowerCAmelCase      )
		__lowercase   							=	{
		    """prompt""": """A painting of a squirrel eating a burger""",
		    """generator""": generator,
		    """num_inference_steps""": 2,
		    """prior_num_inference_steps""": 2,
		    """output_type""": """numpy""",
		}
		return inputs







	def 			_a (	self   :				Tuple      )		->			Tuple:
		"""simple docstring"""







		__lowercase   							=	torch_device == """cpu"""

		self._test_attention_slicing_forward_pass(test_max_difference=_lowerCAmelCase      )







	def 			_a (	self   :				Optional[Any]      )		->			str:
		"""simple docstring"""







		__lowercase   							=	torch_device in ["""cpu""", """mps"""]

		self._test_inference_batch_single_identical(test_max_difference=_lowerCAmelCase      )


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







	def 			_a (	self   :				Dict      )		->			Dict:
		"""simple docstring"""







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







	def 			_a (	self   :				Tuple      )		->			Any:
		"""simple docstring"""







		__lowercase   							=	load_numpy(
		    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""      )

		__lowercase   							=	StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""							,					torch_dtype=torch.floataa      )
		pipe.to(_lowerCAmelCase      )
		pipe.set_progress_bar_config(disable=_lowerCAmelCase      )
		# stable unclip will oom when integration tests are run on a V100,
		# so turn on memory savings
		pipe.enable_attention_slicing()
		pipe.enable_sequential_cpu_offload()

		__lowercase   							=	torch.Generator(device="""cpu"""      ).manual_seed(0      )
		__lowercase   							=	pipe("""anime turle"""							,					generator=_lowerCAmelCase							,					output_type="""np"""      )

		__lowercase   							=	output.images[0]

		assert image.shape == (768, 768, 3)

		assert_mean_pixel_difference(_lowerCAmelCase							,					_lowerCAmelCase      )







	def 			_a (	self   :				List[str]      )		->			int:
		"""simple docstring"""







		torch.cuda.empty_cache()
		torch.cuda.reset_max_memory_allocated()
		torch.cuda.reset_peak_memory_stats()

		__lowercase   							=	StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""							,					torch_dtype=torch.floataa      )
		__lowercase   							=	pipe.to(_lowerCAmelCase      )
		pipe.set_progress_bar_config(disable=_lowerCAmelCase      )
		pipe.enable_attention_slicing()
		pipe.enable_sequential_cpu_offload()

		__lowercase   							=	pipe(
		    """anime turtle"""							,					prior_num_inference_steps=2							,					num_inference_steps=2							,					output_type="""np"""							,					)

		__lowercase   							=	torch.cuda.max_memory_allocated()
		# make sure that less than 7 GB is allocated
		assert mem_bytes < 7 * 10**9








"""simple docstring"""




import argparse
import json

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image
from torchvision import transforms

from transformers import BitImageProcessor, FocalNetConfig, FocalNetForImageClassification
from transformers.image_utils import IMAGENET_DEFAULT_MEAN, IMAGENET_DEFAULT_STD, PILImageResampling






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       List[Any]   )					->Dict:






	'''simple docstring'''






	a :						Any      =							[2, 2, 6, 2] if "tiny" in model_name else [2, 2, 18, 2]
	a :						str      =							True if "large" in model_name or "huge" in model_name else False
	a :						Optional[Any]      =							True if "large" in model_name or "huge" in model_name else False
	a :						Dict      =							True if "large" in model_name or "huge" in model_name else False

	if "large" in model_name or "xlarge" in model_name or "huge" in model_name:
		if "fl3" in model_name:
			a :						Union[str, Any]      =							[3, 3, 3, 3]
			a :						List[str]      =							[5, 5, 5, 5]
		elif "fl4" in model_name:
			a :						Any      =							[4, 4, 4, 4]
			a :						Optional[Any]      =							[3, 3, 3, 3]

	if "tiny" in model_name or "small" in model_name or "base" in model_name:
		a :						Dict      =							[3, 3, 3, 3]
		if "lrf" in model_name:
			a :						Optional[int]      =							[3, 3, 3, 3]
		else:
			a :						Tuple      =							[2, 2, 2, 2]

	if "tiny" in model_name:
		a :						List[str]      =							96
	elif "small" in model_name:
		a :						Union[str, Any]      =							96
	elif "base" in model_name:
		a :						Dict      =							128
	elif "large" in model_name:
		a :						Union[str, Any]      =							192
	elif "xlarge" in model_name:
		a :						Tuple      =							256
	elif "huge" in model_name:
		a :						List[str]      =							352

	# set label information
	a :						List[Any]      =							"huggingface/label-files"
	if "large" in model_name or "huge" in model_name:
		a :						Optional[int]      =							"imagenet-22k-id2label.json"
	else:
		a :						List[str]      =							"imagenet-1k-id2label.json"

	a :						Optional[int]      =							json.load(open(hf_hub_download(_lowercase       ,			_lowercase       ,			repo_type="dataset"   )       ,			"r"   )   )
	a :						str      =							{int(_lowercase   ): v for k, v in idalabel.items()}
	a :						List[str]      =							{v: k for k, v in idalabel.items()}

	a :						Dict      =							FocalNetConfig(
	    embed_dim=_lowercase       ,			depths=_lowercase       ,			focal_levels=_lowercase       ,			focal_windows=_lowercase       ,			use_conv_embed=_lowercase       ,			idalabel=_lowercase       ,			labelaid=_lowercase       ,			use_post_layernorm=_lowercase       ,			use_layerscale=_lowercase       ,			)

	return config






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       List[str]   )					->List[Any]:






	'''simple docstring'''






	if "patch_embed.proj" in name:
		a :						Any      =							name.replace("patch_embed.proj"       ,			"embeddings.patch_embeddings.projection"   )
	if "patch_embed.norm" in name:
		a :						List[str]      =							name.replace("patch_embed.norm"       ,			"embeddings.norm"   )
	if "layers" in name:
		a :						List[Any]      =							"encoder." + name
	if "encoder.layers" in name:
		a :						int      =							name.replace("encoder.layers"       ,			"encoder.stages"   )
	if "downsample.proj" in name:
		a :						Any      =							name.replace("downsample.proj"       ,			"downsample.projection"   )
	if "blocks" in name:
		a :						str      =							name.replace("blocks"       ,			"layers"   )
	if "modulation.f.weight" in name or "modulation.f.bias" in name:
		a :						Union[str, Any]      =							name.replace("modulation.f"       ,			"modulation.projection_in"   )
	if "modulation.h.weight" in name or "modulation.h.bias" in name:
		a :						Dict      =							name.replace("modulation.h"       ,			"modulation.projection_context"   )
	if "modulation.proj.weight" in name or "modulation.proj.bias" in name:
		a :						Any      =							name.replace("modulation.proj"       ,			"modulation.projection_out"   )

	if name == "norm.weight":
		a :						str      =							"layernorm.weight"
	if name == "norm.bias":
		a :						Optional[Any]      =							"layernorm.bias"

	if "head" in name:
		a :						Tuple      =							name.replace("head"       ,			"classifier"   )
	else:
		a :						int      =							"focalnet." + name

	return name






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       Tuple       ,			_lowercase		:       Optional[Any]       ,			_lowercase		:       Tuple=False   )					->str:






	'''simple docstring'''






	a :						List[Any]      =							{
	    "focalnet-tiny": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_tiny_srf.pth",
	    "focalnet-tiny-lrf": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_tiny_lrf.pth",
	    "focalnet-small": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_small_srf.pth",
	    "focalnet-small-lrf": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_small_lrf.pth",
	    "focalnet-base": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_base_srf.pth",
	    "focalnet-base-lrf": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_base_lrf.pth",
	    "focalnet-large-lrf-fl3": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_large_lrf_384.pth",
	    "focalnet-large-lrf-fl4": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_large_lrf_384_fl4.pth",
	    "focalnet-xlarge-lrf-fl3": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_xlarge_lrf_384.pth",
	    "focalnet-xlarge-lrf-fl4": "https://projects4jw.blob.core.windows.net/focalnet/release/classification/focalnet_xlarge_lrf_384_fl4.pth",
	}
	# fmt: on

	a :						str      =							model_name_to_url[model_name]
	print("Checkpoint URL: "       ,			_lowercase   )
	a :						Any      =							torch.hub.load_state_dict_from_url(_lowercase       ,			map_location="cpu"   )["model"]

	# rename keys
	for key in state_dict.copy().keys():
		a :						Any      =							state_dict.pop(_lowercase   )
		a :						Any      =							val

	a :						Any      =							get_focalnet_config(_lowercase   )
	a :						Optional[int]      =							FocalNetForImageClassification(_lowercase   )
	model.eval()

	# load state dict
	model.load_state_dict(_lowercase   )

	# verify conversion
	a :						Union[str, Any]      =							"http://images.cocodataset.org/val2017/000000039769.jpg"

	a :						Optional[int]      =							BitImageProcessor(
	    do_resize=_lowercase       ,			size={"shortest_edge": 256}       ,			resample=PILImageResampling.BILINEAR       ,			do_center_crop=_lowercase       ,			crop_size=224       ,			do_normalize=_lowercase       ,			image_mean=_lowercase       ,			image_std=_lowercase       ,			)
	a :						int      =							Image.open(requests.get(_lowercase       ,			stream=_lowercase   ).raw   )
	a :						Dict      =							processor(images=_lowercase       ,			return_tensors="pt"   )

	a :						Optional[int]      =							transforms.Compose(
	    [
	        transforms.Resize(256   ),
	        transforms.CenterCrop(224   ),
	        transforms.ToTensor(),
	        transforms.Normalize(mean=[0.485, 0.456, 0.406]       ,			std=[0.229, 0.224, 0.225]   ),
	    ]   )

	a :						str      =							image_transforms(_lowercase   ).unsqueeze(0   )

	# verify pixel_values
	assert torch.allclose(inputs.pixel_values       ,			_lowercase       ,			atol=1E-4   )

	a :						Dict      =							model(**_lowercase   )

	a :						List[str]      =							outputs.logits.argmax(-1   ).item()
	print("Predicted class:"       ,			model.config.idalabel[predicted_class_idx]   )

	print("First values of logits:"       ,			outputs.logits[0, :3]   )

	if model_name == "focalnet-tiny":
		a :						Union[str, Any]      =							torch.tensor([0.2166, -0.4368, 0.2191]   )
	elif model_name == "focalnet-tiny-lrf":
		a :						Union[str, Any]      =							torch.tensor([1.1669, 0.0125, -0.1695]   )
	elif model_name == "focalnet-small":
		a :						Dict      =							torch.tensor([0.4917, -0.0430, 0.1341]   )
	elif model_name == "focalnet-small-lrf":
		a :						Dict      =							torch.tensor([-0.2588, -0.5342, -0.2331]   )
	elif model_name == "focalnet-base":
		a :						Any      =							torch.tensor([-0.1655, -0.4090, -0.1730]   )
	elif model_name == "focalnet-base-lrf":
		a :						str      =							torch.tensor([0.5306, -0.0483, -0.3928]   )
	assert torch.allclose(outputs.logits[0, :3]       ,			_lowercase       ,			atol=1E-4   )
	print("Looks ok!"   )

	if pytorch_dump_folder_path is not None:
		print(F"""Saving model and processor of {model_name} to {pytorch_dump_folder_path}"""   )
		model.save_pretrained(_lowercase   )
		processor.save_pretrained(_lowercase   )

	if push_to_hub:
		print(F"""Pushing model and processor of {model_name} to the hub..."""   )
		model.push_to_hub(F"""{model_name}"""   )
		processor.push_to_hub(F"""{model_name}"""   )


if __name__ == "__main__":
		a		:					str	      =     argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    '''--model_name''',
		    default='''focalnet-tiny''',
		    type=str,
		    help='''Name of the FocalNet model you\'d like to convert.''',
		)
		parser.add_argument(
		    '''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the output PyTorch model directory.'''
		)
		parser.add_argument(
		    '''--push_to_hub''',
		    action='''store_true''',
		    help='''Whether to push the model and processor to the hub.''',
		)

		a		:					Tuple	      =     parser.parse_args()
		convert_focalnet_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)


