






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 			A__							(      _a  :					Dict	):





 '''simple docstring'''







 snake_case__  :    List[str]							       =[2, 2, 6, 2] if """tiny""" in model_name else [2, 2, 18, 2]
 snake_case__  :    List[str]							       =True if """large""" in model_name or """huge""" in model_name else False
 snake_case__  :    List[str]							       =True if """large""" in model_name or """huge""" in model_name else False
 snake_case__  :    Optional[int]							       =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:
   snake_case__  :    List[Any]							       =[3, 3, 3, 3]
   snake_case__  :    Union[str, Any]							       =[5, 5, 5, 5]
  elif "fl4" in model_name:
   snake_case__  :    List[Any]							       =[4, 4, 4, 4]
   snake_case__  :    Tuple							       =[3, 3, 3, 3]

 if "tiny" in model_name or "small" in model_name or "base" in model_name:
  snake_case__  :    Union[str, Any]							       =[3, 3, 3, 3]
  if "lrf" in model_name:
   snake_case__  :    Tuple							       =[3, 3, 3, 3]
  else:
   snake_case__  :    Dict							       =[2, 2, 2, 2]

 if "tiny" in model_name:
  snake_case__  :    List[Any]							       =96
 elif "small" in model_name:
  snake_case__  :    Tuple							       =96
 elif "base" in model_name:
  snake_case__  :    Any							       =128
 elif "large" in model_name:
  snake_case__  :    Optional[Any]							       =192
 elif "xlarge" in model_name:
  snake_case__  :    List[str]							       =256
 elif "huge" in model_name:
  snake_case__  :    int							       =352

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

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

 snake_case__  :    List[Any]							       =FocalNetConfig(
     embed_dim=__snake_case		,  depths=__snake_case		,  focal_levels=__snake_case		,  focal_windows=__snake_case		,  use_conv_embed=__snake_case		,  idalabel=__snake_case		,  labelaid=__snake_case		,  use_post_layernorm=__snake_case		,  use_layerscale=__snake_case		,  )

 return config







def 			A__							(      _a  :					Dict	):





 '''simple docstring'''







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

 if name == "norm.weight":
  snake_case__  :    Union[str, Any]							       ="""layernorm.weight"""
 if name == "norm.bias":
  snake_case__  :    List[Any]							       ="""layernorm.bias"""

 if "head" in name:
  snake_case__  :    Tuple							       =name.replace("""head"""		,  """classifier"""	)
 else:
  snake_case__  :    Union[str, Any]							       ="""focalnet.""" + name

 return name







def 			A__							(      _a  :					List[Any]		,  _a  :					Any		,  _a  :					List[str]=False	):





 '''simple docstring'''







 snake_case__  :    str							       ={
     """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

 snake_case__  :    Optional[Any]							       =model_name_to_url[model_name]
 print("""Checkpoint URL: """		,  __snake_case	)
 snake_case__  :    int							       =torch.hub.load_state_dict_from_url(__snake_case		,  map_location="""cpu"""	)["""model"""]

 # rename keys
 for key in state_dict.copy().keys():
  snake_case__  :    Dict							       =state_dict.pop(__snake_case	)
  snake_case__  :    Dict							       =val

 snake_case__  :    List[str]							       =get_focalnet_config(__snake_case	)
 snake_case__  :    Any							       =FocalNetForImageClassification(__snake_case	)
 model.eval()

 # load state dict
 model.load_state_dict(__snake_case	)

 # verify conversion
 snake_case__  :    Dict							       ="""http://images.cocodataset.org/val2017/000000039769.jpg"""

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

 snake_case__  :    List[Any]							       =transforms.Compose(
     [
         transforms.Resize(256	),
         transforms.CenterCrop(224	),
         transforms.ToTensor(),
         transforms.Normalize(mean=[0.4_8_5, 0.4_5_6, 0.4_0_6]		,  std=[0.2_2_9, 0.2_2_4, 0.2_2_5]	),
     ]	)

 snake_case__  :    str							       =image_transforms(__snake_case	).unsqueeze(0	)

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

 snake_case__  :    Optional[int]							       =model(**__snake_case	)

 snake_case__  :    Optional[Any]							       =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":
  snake_case__  :    Optional[Any]							       =torch.tensor([0.2_1_6_6, -0.4_3_6_8, 0.2_1_9_1]	)
 elif model_name == "focalnet-tiny-lrf":
  snake_case__  :    str							       =torch.tensor([1.1_6_6_9, 0.0_1_2_5, -0.1_6_9_5]	)
 elif model_name == "focalnet-small":
  snake_case__  :    Optional[Any]							       =torch.tensor([0.4_9_1_7, -0.0_4_3_0, 0.1_3_4_1]	)
 elif model_name == "focalnet-small-lrf":
  snake_case__  :    Optional[int]							       =torch.tensor([-0.2_5_8_8, -0.5_3_4_2, -0.2_3_3_1]	)
 elif model_name == "focalnet-base":
  snake_case__  :    Tuple							       =torch.tensor([-0.1_6_5_5, -0.4_0_9_0, -0.1_7_3_0]	)
 elif model_name == "focalnet-base-lrf":
  snake_case__  :    Union[str, Any]							       =torch.tensor([0.5_3_0_6, -0.0_4_8_3, -0.3_9_2_8]	)
 assert torch.allclose(outputs.logits[0, :3]		,  __snake_case		,  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(__snake_case	)
  processor.save_pretrained(__snake_case	)

 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__":
 __lowerCamelCase	:		Any										=   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.""",
 )

 __lowerCamelCase	:		Dict										=   parser.parse_args()
 convert_focalnet_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)













import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()







class  _lowercase   (   _A						,				unittest.TestCase						):
      _a							:						Any										=							KandinskyImgaImgPipeline
      _a							:						List[Any]										=							['prompt', 'image_embeds', 'negative_image_embeds', 'image']
      _a							:						int										=							[
          'prompt',
          'negative_prompt',
          'image_embeds',
          'negative_image_embeds',
          'image',
      ]
      _a							:						int										=							[
          'generator',
          'height',
          'width',
          'strength',
          'guidance_scale',
          'negative_prompt',
          'num_inference_steps',
          'return_dict',
          'guidance_scale',
          'num_images_per_prompt',
          'output_type',
          'return_dict',
      ]
      _a							:						List[str]										=							False
      @property
      def 						lowercase__					(     self   ):
       return 3_2
      @property
      def 						lowercase__					(     self   ):
       return 3_2
      @property
      def 						lowercase__					(     self   ):
       return self.time_input_dim
      @property
      def 						lowercase__					(     self   ):
       return self.time_input_dim * 4
      @property
      def 						lowercase__					(     self   ):
       return 1_0_0
      @property
      def 						lowercase__					(     self   ):
       snake_case__  :    Dict							       =XLMRobertaTokenizerFast.from_pretrained("""YiYiXu/tiny-random-mclip-base"""   )
       return tokenizer
      @property
      def 						lowercase__					(     self   ):
       torch.manual_seed(0   )
       snake_case__  :    Optional[Any]							       =MCLIPConfig(
           numDims=self.cross_attention_dim				,			transformerDimensions=self.text_embedder_hidden_size				,			hidden_size=self.text_embedder_hidden_size				,			intermediate_size=3_7				,			num_attention_heads=4				,			num_hidden_layers=5				,			vocab_size=1_0_0_5				,			)

       snake_case__  :    Optional[int]							       =MultilingualCLIP(a   )
       snake_case__  :    Union[str, Any]							       =text_encoder.eval()

       return text_encoder
      @property
      def 						lowercase__					(     self   ):
       torch.manual_seed(0   )

       snake_case__  :    Optional[int]							       ={
           """in_channels""": 4,
           # Out channels is double in channels because predicts mean and variance
           """out_channels""": 8,
           """addition_embed_type""": """text_image""",
           """down_block_types""": ("""ResnetDownsampleBlock2D""", """SimpleCrossAttnDownBlock2D"""),
           """up_block_types""": ("""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""),
           """mid_block_type""": """UNetMidBlock2DSimpleCrossAttn""",
           """block_out_channels""": (self.block_out_channels_a, self.block_out_channels_a * 2),
           """layers_per_block""": 1,
           """encoder_hid_dim""": self.text_embedder_hidden_size,
           """encoder_hid_dim_type""": """text_image_proj""",
           """cross_attention_dim""": self.cross_attention_dim,
           """attention_head_dim""": 4,
           """resnet_time_scale_shift""": """scale_shift""",
           """class_embed_type""": None,
       }

       snake_case__  :    List[str]							       =UNetaDConditionModel(**a   )
       return model
      @property
      def 						lowercase__					(     self   ):
       return {
           "block_out_channels": [3_2, 6_4],
           "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
           "in_channels": 3,
           "latent_channels": 4,
           "layers_per_block": 1,
           "norm_num_groups": 8,
           "norm_type": "spatial",
           "num_vq_embeddings": 1_2,
           "out_channels": 3,
           "up_block_types": [
               "AttnUpDecoderBlock2D",
               "UpDecoderBlock2D",
           ],
           "vq_embed_dim": 4,
       }
      @property
      def 						lowercase__					(     self   ):
       torch.manual_seed(0   )
       snake_case__  :    List[Any]							       =VQModel(**self.dummy_movq_kwargs   )
       return model
      def 						lowercase__					(     self   ):
       snake_case__  :    str							       =self.dummy_text_encoder
       snake_case__  :    Tuple							       =self.dummy_tokenizer
       snake_case__  :    List[str]							       =self.dummy_unet
       snake_case__  :    Dict							       =self.dummy_movq

       snake_case__  :    Union[str, Any]							       ={
           """num_train_timesteps""": 1_0_0_0,
           """beta_schedule""": """linear""",
           """beta_start""": 0.00085,
           """beta_end""": 0.012,
           """clip_sample""": False,
           """set_alpha_to_one""": False,
           """steps_offset""": 0,
           """prediction_type""": """epsilon""",
           """thresholding""": False,
       }

       snake_case__  :    Union[str, Any]							       =DDIMScheduler(**a   )

       snake_case__  :    Any							       ={
           """text_encoder""": text_encoder,
           """tokenizer""": tokenizer,
           """unet""": unet,
           """scheduler""": scheduler,
           """movq""": movq,
       }

       return components
      def 						lowercase__					(     self				,			a				,			a=0   ):
       snake_case__  :    str							       =floats_tensor((1, self.cross_attention_dim)				,			rng=random.Random(a   )   ).to(a   )
       snake_case__  :    Tuple							       =floats_tensor((1, self.cross_attention_dim)				,			rng=random.Random(seed + 1   )   ).to(a   )
       # create init_image
       snake_case__  :    str							       =floats_tensor((1, 3, 6_4, 6_4)				,			rng=random.Random(a   )   ).to(a   )
       snake_case__  :    Tuple							       =image.cpu().permute(0				,			2				,			3				,			1   )[0]
       snake_case__  :    Dict							       =Image.fromarray(np.uinta(a   )   ).convert("""RGB"""   ).resize((2_5_6, 2_5_6)   )

       if str(a   ).startswith("""mps"""   ):
        snake_case__  :    Dict							       =torch.manual_seed(a   )
       else:
        snake_case__  :    Optional[Any]							       =torch.Generator(device=a   ).manual_seed(a   )
       snake_case__  :    str							       ={
           """prompt""": """horse""",
           """image""": init_image,
           """image_embeds""": image_embeds,
           """negative_image_embeds""": negative_image_embeds,
           """generator""": generator,
           """height""": 6_4,
           """width""": 6_4,
           """num_inference_steps""": 1_0,
           """guidance_scale""": 7.0,
           """strength""": 0.2,
           """output_type""": """np""",
       }
       return inputs



      def 						lowercase__					(     self   ):
       snake_case__  :    List[str]							       ="""cpu"""

       snake_case__  :    Tuple							       =self.get_dummy_components()

       snake_case__  :    Optional[int]							       =self.pipeline_class(**a   )
       snake_case__  :    Union[str, Any]							       =pipe.to(a   )

       pipe.set_progress_bar_config(disable=a   )

       snake_case__  :    List[Any]							       =pipe(**self.get_dummy_inputs(a   )   )
       snake_case__  :    List[str]							       =output.images

       snake_case__  :    int							       =pipe(
           **self.get_dummy_inputs(a   )				,			return_dict=a				,			)[0]

       snake_case__  :    str							       =image[0, -3:, -3:, -1]
       snake_case__  :    Tuple							       =image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (1, 6_4, 6_4, 3)

       snake_case__  :    List[str]							       =np.array(
           [0.61474943, 0.6073539, 0.43308544, 0.5928269, 0.47493595, 0.46755973, 0.4613838, 0.45368797, 0.50119233]   )
       assert (
           np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2
       ), F" expected_slice {expected_slice}, but got {image_slice.flatten()}"
       assert (
           np.abs(image_from_tuple_slice.flatten() - expected_slice   ).max() < 1e-2
       ), F" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"







@slow
@require_torch_gpu
class  _lowercase   (   unittest.TestCase						):
      def 						lowercase__					(     self   ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()



      def 						lowercase__					(     self   ):
       snake_case__  :    Any							       =load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
           """/kandinsky/kandinsky_img2img_frog.npy"""   )

       snake_case__  :    Dict							       =load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/kandinsky/cat.png"""   )
       snake_case__  :    Any							       ="""A red cartoon frog, 4k"""

       snake_case__  :    Dict							       =KandinskyPriorPipeline.from_pretrained(
           """kandinsky-community/kandinsky-2-1-prior"""				,			torch_dtype=torch.floataa   )
       pipe_prior.to(a   )

       snake_case__  :    Tuple							       =KandinskyImgaImgPipeline.from_pretrained(
           """kandinsky-community/kandinsky-2-1"""				,			torch_dtype=torch.floataa   )
       snake_case__  :    str							       =pipeline.to(a   )

       pipeline.set_progress_bar_config(disable=a   )

       snake_case__  :    Optional[Any]							       =torch.Generator(device="""cpu"""   ).manual_seed(0   )
       snake_case__       ,   snake_case__  :    Any							       =pipe_prior(
           a				,			generator=a				,			num_inference_steps=5				,			negative_prompt=""""""				,			).to_tuple()

       snake_case__  :    str							       =pipeline(
           a				,			image=a				,			image_embeds=a				,			negative_image_embeds=a				,			generator=a				,			num_inference_steps=1_0_0				,			height=7_6_8				,			width=7_6_8				,			strength=0.2				,			output_type="""np"""				,			)

       snake_case__  :    Optional[int]							       =output.images[0]

       assert image.shape == (7_6_8, 7_6_8, 3)

       assert_mean_pixel_difference(a				,			a   )





