
'''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 						_snake_case    (   A_							:     List[str] ):
  """simple docstring"""







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

  if "tiny" in model_name or "small" in model_name or "base" in model_name:
    a_   :						Any   =				[3, 3, 3, 3]
    if "lrf" in model_name:
      a_   :						int   =				[3, 3, 3, 3]
    else:
      a_   :						Any   =				[2, 2, 2, 2]

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

  # set label information
  a_   :						Any   =				"""huggingface/label-files"""
  if "large" in model_name or "huge" in model_name:
    a_   :						Dict   =				"""imagenet-22k-id2label.json"""
  else:
    a_   :						Dict   =				"""imagenet-1k-id2label.json"""

  a_   :						Optional[Any]   =				json.load(open(hf_hub_download(_A  ,		_A  ,		repo_type="""dataset""" )  ,		"""r""" ) )
  a_   :						Tuple   =				{int(_A ): v for k, v in idalabel.items()}
  a_   :						Optional[int]   =				{v: k for k, v in idalabel.items()}

  a_   :						int   =				FocalNetConfig(
      embed_dim=_A  ,		depths=_A  ,		focal_levels=_A  ,		focal_windows=_A  ,		use_conv_embed=_A  ,		idalabel=_A  ,		labelaid=_A  ,		use_post_layernorm=_A  ,		use_layerscale=_A  ,		)

  return config






def 						_snake_case    (   A_							:     Optional[Any] ):
  """simple docstring"""







  if "patch_embed.proj" in name:
    a_   :						Union[str, 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_   :						Optional[Any]   =				"""encoder.""" + name
  if "encoder.layers" in name:
    a_   :						Union[str, Any]   =				name.replace("""encoder.layers"""  ,		"""encoder.stages""" )
  if "downsample.proj" in name:
    a_   :						Optional[Any]   =				name.replace("""downsample.proj"""  ,		"""downsample.projection""" )
  if "blocks" in name:
    a_   :						List[str]   =				name.replace("""blocks"""  ,		"""layers""" )
  if "modulation.f.weight" in name or "modulation.f.bias" in name:
    a_   :						Optional[int]   =				name.replace("""modulation.f"""  ,		"""modulation.projection_in""" )
  if "modulation.h.weight" in name or "modulation.h.bias" in name:
    a_   :						str   =				name.replace("""modulation.h"""  ,		"""modulation.projection_context""" )
  if "modulation.proj.weight" in name or "modulation.proj.bias" in name:
    a_   :						Optional[Any]   =				name.replace("""modulation.proj"""  ,		"""modulation.projection_out""" )

  if name == "norm.weight":
    a_   :						Dict   =				"""layernorm.weight"""
  if name == "norm.bias":
    a_   :						str   =				"""layernorm.bias"""

  if "head" in name:
    a_   :						Dict   =				name.replace("""head"""  ,		"""classifier""" )
  else:
    a_   :						List[str]   =				"""focalnet.""" + name

  return name






def 						_snake_case    (   A_							:     List[Any]  ,		A_							:     Tuple  ,		A_							:     Optional[Any]=False ):
  """simple docstring"""







  a_   :						int   =				{
      """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_   :						int   =				model_name_to_url[model_name]
  print("""Checkpoint URL: """  ,		_A )
  a_   :						List[Any]   =				torch.hub.load_state_dict_from_url(_A  ,		map_location="""cpu""" )["""model"""]

  # rename keys
  for key in state_dict.copy().keys():
    a_   :						int   =				state_dict.pop(_A )
    a_   :						Optional[Any]   =				val

  a_   :						Tuple   =				get_focalnet_config(_A )
  a_   :						str   =				FocalNetForImageClassification(_A )
  model.eval()

  # load state dict
  model.load_state_dict(_A )

  # verify conversion
  a_   :						Optional[Any]   =				"""http://images.cocodataset.org/val2017/000000039769.jpg"""

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

  a_   :						Union[str, Any]   =				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_   :						int   =				image_transforms(_A ).unsqueeze(0 )

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

  a_   :						List[str]   =				model(**_A )

  a_   :						int   =				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_   :						Optional[Any]   =				torch.tensor([0.2166, -0.4368, 0.2191] )
  elif model_name == "focalnet-tiny-lrf":
    a_   :						Any   =				torch.tensor([1.1669, 0.0125, -0.1695] )
  elif model_name == "focalnet-small":
    a_   :						Optional[Any]   =				torch.tensor([0.4917, -0.0430, 0.1341] )
  elif model_name == "focalnet-small-lrf":
    a_   :						str   =				torch.tensor([-0.2588, -0.5342, -0.2331] )
  elif model_name == "focalnet-base":
    a_   :						str   =				torch.tensor([-0.1655, -0.4090, -0.1730] )
  elif model_name == "focalnet-base-lrf":
    a_   :						List[str]   =				torch.tensor([0.5306, -0.0483, -0.3928] )
  assert torch.allclose(outputs.logits[0, :3]  ,		_A  ,		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(_A )
    processor.save_pretrained(_A )

  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__":
    __snake_case:	Optional[int]										=      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.",
    )

    __snake_case:	Optional[int]										=      parser.parse_args()
    convert_focalnet_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)



import json
import logging
import os
import sys
from pathlib import Path

import finetune_rag

from transformers.file_utils import is_apex_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    require_ray,
    require_torch_gpu,
    require_torch_multi_gpu,
)


logging.basicConfig(level=logging.DEBUG)
SCREAMING_SNAKE_CASE			:						List[Any]				  =		logging.getLogger()

SCREAMING_SNAKE_CASE			:						List[str]				  =		logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)
class  A_			(							a_			):

  def 		_UpperCAmelCase (		self			:   str  ,				__SCREAMING_SNAKE_CASE			:   int							):
   os.makedirs(__SCREAMING_SNAKE_CASE  ,				exist_ok=__SCREAMING_SNAKE_CASE							)
   __a							=	{"source": "What is love ?", "target": "life"}
   __a							=	{"train": 12, "val": 2, "test": 2}
   for split in ["train", "test", "val"]:
    for field in ["source", "target"]:
     __a							=	"\n".join([contents[field]] * n_lines[split]							)
     with open(os.path.join(__SCREAMING_SNAKE_CASE  ,				f"""{split}.{field}"""							)  ,				"w"							) as f:
      f.write(__SCREAMING_SNAKE_CASE							)

  def 		_UpperCAmelCase (		self			:   str  ,				__SCREAMING_SNAKE_CASE			:   int  ,				__SCREAMING_SNAKE_CASE			:   str = "pytorch"							):
   __a							=	self.get_auto_remove_tmp_dir()
   __a							=	os.path.join(__SCREAMING_SNAKE_CASE  ,				"output"							)
   __a							=	os.path.join(__SCREAMING_SNAKE_CASE  ,				"data"							)
   self._create_dummy_data(data_dir=__SCREAMING_SNAKE_CASE							)

   __a							=	f"""
                --data_dir {data_dir} \
                --output_dir {output_dir} \
                --model_name_or_path facebook/rag-sequence-base \
                --model_type rag_sequence \
                --do_train \
                --do_predict \
                --n_val -1 \
                --val_check_interval 1.0 \
                --train_batch_size 2 \
                --eval_batch_size 1 \
                --max_source_length 25 \
                --max_target_length 25 \
                --val_max_target_length 25 \
                --test_max_target_length 25 \
                --label_smoothing 0.1 \
                --dropout 0.1 \
                --attention_dropout 0.1 \
                --weight_decay 0.001 \
                --adam_epsilon 1e-08 \
                --max_grad_norm 0.1 \
                --lr_scheduler polynomial \
                --learning_rate 3e-04 \
                --num_train_epochs 1 \
                --warmup_steps 4 \
                --gradient_accumulation_steps 1 \
                --distributed-port 8787 \
                --use_dummy_dataset 1 \
                --distributed_retriever {distributed_retriever} \
            """.split()

   if gpus > 0:
    testargs.append(f"""--gpus={gpus}"""							)
    if is_apex_available():
     testargs.append("--fp16"							)
   else:
    testargs.append("--gpus=0"							)
    testargs.append("--distributed_backend=ddp_cpu"							)
    testargs.append("--num_processes=2"							)

   __a							=	[sys.executable, str(Path(finetune_rag.__file__							).resolve()							)] + testargs
   execute_subprocess_async(__SCREAMING_SNAKE_CASE  ,				env=self.get_env()							)

   __a							=	os.path.join(__SCREAMING_SNAKE_CASE  ,				"metrics.json"							)
   with open(__SCREAMING_SNAKE_CASE							) as f:
    __a							=	json.load(__SCREAMING_SNAKE_CASE							)
   return result

  @require_torch_gpu
  def 		_UpperCAmelCase (		self			:   Dict							):
   __a							=	self._run_finetune(gpus=1							)
   self.assertGreaterEqual(result["test"][0]["test_avg_em"]  ,				0.2							)

  @require_torch_multi_gpu
  def 		_UpperCAmelCase (		self			:   Optional[int]							):
   __a							=	self._run_finetune(gpus=2							)
   self.assertGreaterEqual(result["test"][0]["test_avg_em"]  ,				0.2							)

  @require_torch_gpu
  @require_ray
  def 		_UpperCAmelCase (		self			:   Optional[Any]							):
   __a							=	self._run_finetune(gpus=1  ,				distributed_retriever="ray"							)
   self.assertGreaterEqual(result["test"][0]["test_avg_em"]  ,				0.2							)

  @require_torch_multi_gpu
  @require_ray
  def 		_UpperCAmelCase (		self			:   Any							):
   __a							=	self._run_finetune(gpus=1  ,				distributed_retriever="ray"							)
   self.assertGreaterEqual(result["test"][0]["test_avg_em"]  ,				0.2							)


