






import argparse
import collections

import jax
import jax.numpy as jnp
import torch
import torch.nn as nn
from clip.model import CLIP
from flax.training import checkpoints
from huggingface_hub import Repository

from transformers import (
    CLIPTokenizer,
    OwlViTConfig,
    OwlViTForObjectDetection,
    OwlViTImageProcessor,
    OwlViTModel,
    OwlViTProcessor,
)


UpperCAmelCase__            =	{
    "vit_b32": {
        "embed_dim": 512,
        "image_resolution": 768,
        "context_length": 16,
        "vocab_size": 49408,
        "vision_layers": 12,
        "vision_width": 768,
        "vision_patch_size": 32,
        "transformer_width": 512,
        "transformer_heads": 8,
        "transformer_layers": 12,
    },
    "vit_b16": {
        "embed_dim": 512,
        "image_resolution": 768,
        "context_length": 16,
        "vocab_size": 49408,
        "vision_layers": 12,
        "vision_width": 768,
        "vision_patch_size": 16,
        "transformer_width": 512,
        "transformer_heads": 8,
        "transformer_layers": 12,
    },
    "vit_l14": {
        "embed_dim": 768,
        "image_resolution": 840,
        "context_length": 16,
        "vocab_size": 49408,
        "vision_layers": 24,
        "vision_width": 1024,
        "vision_patch_size": 14,
        "transformer_width": 768,
        "transformer_heads": 12,
        "transformer_layers": 12,
    },
}
def 		_a					(   a				:Dict			,							a				:List[Any]=""			,							a				:Tuple="/"				)							->    Union[str, Any]:
 a							=   []

 for k, v in params.items():
  a							=   parent_key + sep + k if parent_key else k

  if isinstance(a			,							collections.MutableMapping				):
   items.extend(flatten_nested_dict(a			,							a			,							sep=a				).items()				)
  else:
   items.append((new_key, v)				)
 return dict(a				)
def 		_a					(   a				:List[Any]				)							->    str:
 return jax.tree_util.tree_map(lambda a				: x.astype(jnp.floataa				) if x.dtype == jnp.bfloataa else x			,							a				)
def 		_a					(   a				:List[str]			,							a				:str				)							->    Optional[int]:
 a						,  a						,  a							=   pt_attn_layer.in_proj_weight.chunk(3			,							dim=0				)
 a						,  a						,  a							=   pt_attn_layer.in_proj_bias.chunk(3			,							dim=0				)

 a							=   pt_attn_layer.out_proj.weight
 a							=   pt_attn_layer.out_proj.bias

 a							=   q_proj
 a							=   q_proj_bias

 a							=   k_proj
 a							=   k_proj_bias

 a							=   v_proj
 a							=   v_proj_bias

 a							=   out_proj_weights
 a							=   out_proj_bias
def 		_a					(   a				:str			,							a				:Union[str, Any]				)							->    Union[str, Any]:
 copy_linear(hf_mlp.fca			,							pt_mlp.c_fc				)
 copy_linear(hf_mlp.fca			,							pt_mlp.c_proj				)
def 		_a					(   a				:Optional[Any]			,							a				:Any				)							->    str:
 a							=   pt_linear.weight
 a							=   pt_linear.bias
def 		_a					(   a				:Any			,							a				:List[str]				)							->    Dict:
 # copy layer norms
 copy_linear(hf_layer.layer_norma			,							pt_layer.ln_a				)
 copy_linear(hf_layer.layer_norma			,							pt_layer.ln_a				)

 # copy MLP
 copy_mlp(hf_layer.mlp			,							pt_layer.mlp				)

 # copy attn
 copy_attn_layer(hf_layer.self_attn			,							pt_layer.attn				)
def 		_a					(   a				:List[Any]			,							a				:Union[str, Any]				)							->    Dict:
 for hf_layer, pt_layer in zip(a			,							a				):
  copy_layer(a			,							a				)
def 		_a					(   a				:Union[str, Any]			,							a				:List[str]				)							->    str:
 # copy  embeds
 a							=   pt_model.token_embedding.weight
 a							=   pt_model.positional_embedding

 # copy layer norm
 copy_linear(hf_encoder.final_layer_norm			,							pt_model.ln_final				)

 # copy hidden layers
 copy_layers(hf_encoder.encoder.layers			,							pt_model.transformer.resblocks				)
def 		_a					(   a				:Optional[int]			,							a				:List[str]				)							->    Dict:
 # copy projection
 a							=   pt_model.text_projection.data.T

 # copy text encoder
 copy_encoder(hf_model.text_model			,							a				)
def 		_a					(   a				:Any			,							a				:Dict				)							->    Optional[Any]:
 # copy projection
 a							=   pt_model.visual.proj.data.T

 # copy layer norms
 copy_linear(hf_model.vision_model.pre_layernorm			,							pt_model.visual.ln_pre				)
 copy_linear(hf_model.vision_model.post_layernorm			,							pt_model.visual.ln_post				)

 # copy embeds
 a							=   pt_model.visual.conva.weight.data
 a							=   pt_model.visual.class_embedding
 a							=   pt_model.visual.positional_embedding.data

 # copy encoder
 copy_layers(hf_model.vision_model.encoder.layers			,							pt_model.visual.transformer.resblocks				)
def 		_a					(   a				:Tuple			,							a				:Tuple				)							->    str:
 a							=   flatten_nested_dict(flax_params['''backbone''']['''merged_class_token''']				)

 a							=   torch.from_numpy(flax_class_token_params['''scale''']				)
 a							=   torch.from_numpy(flax_class_token_params['''bias''']				)
 a							=   nn.Parameter(a				)
 a							=   nn.Parameter(a				)
def 		_a					(   a				:Union[str, Any]			,							a				:List[str]				)							->    Optional[int]:
 a							=   hf_model.state_dict()
 a							=   {}

 # Rename class prediction head flax params to pytorch HF
 a							=   flatten_nested_dict(flax_params['''class_head''']				)

 for flax_key, v in flax_class_params.items():
  a							=   flax_key.replace('''/'''			,							'''.'''				)
  a							=   torch_key.replace('''.kernel'''			,							'''.weight'''				)
  a							=   torch_key.replace('''Dense_0'''			,							'''dense0'''				)
  a							=   '''class_head.''' + torch_key

  if "weight" in torch_key and v.ndim == 2:
   a							=   v.T

  a							=   nn.Parameter(torch.from_numpy(a				)				)

 # Rename box prediction box flax params to pytorch HF
 a							=   flatten_nested_dict(flax_params['''obj_box_head''']				)

 for flax_key, v in flax_box_params.items():
  a							=   flax_key.replace('''/'''			,							'''.'''				)
  a							=   torch_key.replace('''.kernel'''			,							'''.weight'''				)
  a							=   torch_key.replace('''_'''			,							''''''				).lower()
  a							=   '''box_head.''' + torch_key

  if "weight" in torch_key and v.ndim == 2:
   a							=   v.T

  a							=   nn.Parameter(torch.from_numpy(a				)				)

 # Copy flax params to PyTorch params
 for name, param in new_params.items():
  if name in pt_params.keys():
   pt_params[name].copy_(a				)
def 		_a					(   a				:int			,							a				:List[str]				)							->    Optional[int]:
 for k, v in flax_attn_params.items():
  if k.startswith('''transformer'''				):
   a							=   k.replace('''transformer.resblocks'''			,							'''text_model.encoder.layers'''				)
  else:
   a							=   k.replace('''visual.transformer.resblocks'''			,							'''vision_model.encoder.layers'''				)

  a							=   torch_key.replace('''attn'''			,							'''self_attn'''				)
  a							=   torch_key.replace('''key'''			,							'''k_proj'''				)
  a							=   torch_key.replace('''value'''			,							'''v_proj'''				)
  a							=   torch_key.replace('''query'''			,							'''q_proj'''				)
  a							=   torch_key.replace('''out'''			,							'''out_proj'''				)

  if "bias" in torch_key and v.ndim == 2:
   a							=   v.shape[0] * v.shape[1]
   a							=   v.reshape(a				)

  if "weight" in torch_key and "out" in torch_key:
   a							=   (v.shape[0] * v.shape[1], v.shape[2])
   a							=   v.reshape(a				).T

  if "weight" in torch_key and "out" not in torch_key:
   a							=   (v.shape[0], v.shape[1] * v.shape[2])
   a							=   v.reshape(a				).T

  # Copy flax CLIP attn params to HF PyTorch params
  a							=   torch.from_numpy(a				)
  hf_backbone.state_dict()[torch_key].copy_(a				)
def 		_a					(   a				:Any				)							->    int:
 a							=   {}
 a							=   []

 for k, v in params.items():
  if "attn." in k:
   a							=   k[: k.rindex('''attn.'''				) + 5]
   if base in processed_attn_layers:
    continue

   processed_attn_layers.append(a				)
   a							=   params[base + '''out.weight'''].shape[-1]
   a							=   params[base + '''out.weight'''].reshape(a			,							a				).T
   a							=   params[base + '''out.bias''']
  else:
   a							=   v
 return new_params
def 		_a					(   a				:Optional[Any]			,							a				:Union[str, Any]				)							->    Dict:
 a							=   CLIP(**a				)
 torch_model.eval()
 a							=   torch_model.state_dict()

 a							=   flatten_nested_dict(flax_params['''backbone''']['''clip''']				)
 a							=   {}

 for flax_key, v in flax_clip_params.items():
  a							=   flax_key.replace('''/'''			,							'''.'''				)
  a							=   torch_key.replace('''text.token_embedding.embedding'''			,							'''token_embedding.kernel'''				)

  if (
      torch_key.startswith('''text.transformer'''				)
      or torch_key.startswith('''text.text_projection'''				)
      or torch_key.startswith('''text.ln_final'''				)
      or torch_key.startswith('''text.positional_embedding'''				)
  ):
   a							=   torch_key[5:]

  a							=   torch_key.replace('''text_projection.kernel'''			,							'''text_projection'''				)
  a							=   torch_key.replace('''visual.proj.kernel'''			,							'''visual.proj'''				)
  a							=   torch_key.replace('''.scale'''			,							'''.weight'''				)
  a							=   torch_key.replace('''.kernel'''			,							'''.weight'''				)

  if "conv" in torch_key or "downsample.0.weight" in torch_key:
   a							=   v.transpose(3			,							2			,							0			,							1				)

  elif "weight" in torch_key and v.ndim == 2 and "embedding" not in torch_key:
   # Fully connected layers are transposed, embeddings are not
   a							=   v.T

  a							=   v

 a							=   _convert_attn_layers(a				)
 new_torch_params.update(a				)
 a							=   {}

 # Copy flax CLIP backbone params to PyTorch params
 for name, param in new_torch_params.items():
  if name in torch_clip_params.keys():
   a							=   torch.from_numpy(new_torch_params[name]				)
   torch_clip_params[name].copy_(a				)
  else:
   a							=   param

 return torch_clip_params, torch_model, attn_params
@torch.no_grad()
def 		_a					(   a				:Union[str, Any]			,							a				:int			,							a				:Tuple			,							a				:List[str]			,							a				:Tuple=None				)							->    Any:
 a							=   Repository(a			,							clone_from=F"""google/{pytorch_dump_folder_path}"""				)
 repo.git_pull()

 if config_path is not None:
  a							=   OwlViTConfig.from_pretrained(a				)
 else:
  a							=   OwlViTConfig()

 a							=   OwlViTModel(a				).eval()
 a							=   OwlViTForObjectDetection(a				).eval()

 copy_text_model_and_projection(a			,							a				)
 copy_vision_model_and_projection(a			,							a				)
 a							=   pt_backbone.logit_scale
 copy_flax_attn_params(a			,							a				)

 a							=   hf_backbone
 copy_class_merge_token(a			,							a				)
 copy_class_box_heads(a			,							a				)

 # Save HF model
 hf_model.save_pretrained(repo.local_dir				)

 # Initialize image processor
 a							=   OwlViTImageProcessor(
     size=config.vision_config.image_size			,							crop_size=config.vision_config.image_size				)
 # Initialize tokenizer
 a							=   CLIPTokenizer.from_pretrained('''openai/clip-vit-base-patch32'''			,							pad_token='''!'''			,							model_max_length=16				)

 # Initialize processor
 a							=   OwlViTProcessor(image_processor=a			,							tokenizer=a				)
 image_processor.save_pretrained(repo.local_dir				)
 processor.save_pretrained(repo.local_dir				)

 repo.git_add()
 repo.git_commit('''Upload model and processor'''				)
 repo.git_push()


if __name__ == "__main__":
     UpperCAmelCase__            =	argparse.ArgumentParser()
     # Required parameters
     parser.add_argument(
         "--owlvit_version",
         default=None,
         type=str,
         required=True,
         help="OWL-ViT model name [clip_b16, clip_b32, clip_l14].",
     )
     parser.add_argument(
         "--owlvit_checkpoint", default=None, type=str, required=True, help="Path to flax model checkpoint."
     )
     parser.add_argument("--hf_config", default=None, type=str, required=True, help="Path to HF model config.")
     parser.add_argument(
         "--pytorch_dump_folder_path", default="hf_model", type=str, help="Path to the output PyTorch model."
     )
     UpperCAmelCase__            =	parser.parse_args()

     # Initialize PyToch clip model
     UpperCAmelCase__            =	args.owlvit_version
     if model_name == "clip_b16":
          UpperCAmelCase__            =	CONFIGS["vit_b16"]
     elif model_name == "clip_b32":
          UpperCAmelCase__            =	CONFIGS["vit_b32"]
     elif model_name == "clip_l14":
          UpperCAmelCase__            =	CONFIGS["vit_l14"]

     # Load from checkpoint and convert params to float-32
     UpperCAmelCase__            =	checkpoints.restore_checkpoint(args.owlvit_checkpoint, target=None)["optimizer"]["target"]
     UpperCAmelCase__            =	jax.tree_util.tree_map(lambda x: x.astype(jnp.floataa) if x.dtype == jnp.bfloataa else x, variables)
     del variables

     # Convert CLIP backbone
     UpperCAmelCase__     ,   UpperCAmelCase__     ,   UpperCAmelCase__            =	convert_clip_backbone(flax_params, torch_config)

     convert_owlvit_checkpoint(clip_pt, flax_params, attn_params, args.pytorch_dump_folder_path, args.hf_config)










import enum
import inspect
from typing import Iterable, List, Optional, Tuple, Union






class 	lowercase_      (			enum.Enum     ):






      '''simple docstring'''




      __snake_case  						=						'''timm'''
      __snake_case  						=						'''transformers'''
def 		_a					(   a				:Optional[Iterable[str]]			,							a				:Optional[Iterable[int]]			,							a				:Optional[Iterable[str]]				)							->    Optional[Any]:
 if stage_names is None:
  raise ValueError('''Stage_names must be set for transformers backbones'''				)

 if out_features is not None:
  if not isinstance(a			,							(list,)				):
   raise ValueError(F"""out_features must be a list {type(a				)}"""				)
  if any(feat not in stage_names for feat in out_features				):
   raise ValueError(F"""out_features must be a subset of stage_names: {stage_names} got {out_features}"""				)

 if out_indices is not None:
  if not isinstance(a			,							(list, tuple)				):
   raise ValueError(F"""out_indices must be a list or tuple, got {type(a				)}"""				)
  if any(idx >= len(a				) for idx in out_indices				):
   raise ValueError('''out_indices must be valid indices for stage_names {stage_names}, got {out_indices}'''				)

 if out_features is not None and out_indices is not None:
  if len(a				) != len(a				):
   raise ValueError('''out_features and out_indices should have the same length if both are set'''				)
  if out_features != [stage_names[idx] for idx in out_indices]:
   raise ValueError('''out_features and out_indices should correspond to the same stages if both are set'''				)
def 		_a					(   a				:Optional[List[str]]			,							a				:Optional[Union[List[int], Tuple[int]]]			,							a				:List[str]			,							)							->    Union[str, Any]:
 if out_indices is None and out_features is None:
  a							=   [len(a				) - 1]
  a							=   [stage_names[-1]]
 elif out_indices is None and out_features is not None:
  a							=   [stage_names.index(a				) for layer in out_features]
 elif out_features is None and out_indices is not None:
  a							=   [stage_names[idx] for idx in out_indices]
 return out_features, out_indices
def 		_a					(   a				:Optional[List[str]]			,							a				:Optional[Union[List[int], Tuple[int]]]			,							a				:List[str]			,							)							->    Tuple[List[str], List[int]]:
 verify_out_features_out_indices(out_features=a			,							out_indices=a			,							stage_names=a				)
 a						,  a							=   _align_output_features_output_indices(
     out_features=a			,							out_indices=a			,							stage_names=a				)
 # Verify that the aligned out_features and out_indices are valid
 verify_out_features_out_indices(out_features=a			,							out_indices=a			,							stage_names=a				)
 return output_features, output_indices






class 	lowercase_      :






      '''simple docstring'''




      __snake_case  						=						None







      def    __lowerCAmelCase      (   self   :		Any				,      __UpperCAmelCase   :		Tuple						)					->None:




       """simple docstring"""


       if getattr(self				,      '''_backbone'''				,      __UpperCAmelCase						) is None:
        raise ValueError('''self._backbone must be set before calling _init_timm_backbone'''						)

       # These will diagree with the defaults for the transformers models e.g. for resnet50
       # the transformer model has out_features = ['stem', 'stage1', 'stage2', 'stage3', 'stage4']
       # the timm model has out_features = ['act', 'layer1', 'layer2', 'layer3', 'layer4']
       a							=   [stage['''module'''] for stage in self._backbone.feature_info.info]
       a							=   [stage['''num_chs'''] for stage in self._backbone.feature_info.info]
       a							=   self._backbone.feature_info.out_indices
       a							=   self._backbone.feature_info.module_name()

       # We verify the out indices and out features are valid
       verify_out_features_out_indices(
           out_features=__UpperCAmelCase				,      out_indices=__UpperCAmelCase				,      stage_names=self.stage_names						)
       a						,  a							=   out_features, out_indices







      def    __lowerCAmelCase      (   self   :		Any				,      __UpperCAmelCase   :		Tuple						)					->None:




       """simple docstring"""


       a							=   getattr(__UpperCAmelCase				,      '''stage_names'''						)
       a							=   getattr(__UpperCAmelCase				,      '''out_features'''				,      __UpperCAmelCase						)
       a							=   getattr(__UpperCAmelCase				,      '''out_indices'''				,      __UpperCAmelCase						)

       a							=   stage_names
       a						,  a							=   get_aligned_output_features_output_indices(
           out_features=__UpperCAmelCase				,      out_indices=__UpperCAmelCase				,      stage_names=__UpperCAmelCase						)
       # Number of channels for each stage. This is set in the transformer backbone model init
       a							=   None







      def    __lowerCAmelCase      (   self   :		str				,      __UpperCAmelCase   :		Union[str, Any]						)					->None:




       """simple docstring"""


       a							=   config

       a							=   getattr(__UpperCAmelCase				,      '''use_timm_backbone'''				,      __UpperCAmelCase						)
       a							=   BackboneType.TIMM if self.use_timm_backbone else BackboneType.TRANSFORMERS

       if self.backbone_type == BackboneType.TIMM:
        self._init_timm_backbone(__UpperCAmelCase						)
       elif self.backbone_type == BackboneType.TRANSFORMERS:
        self._init_transformers_backbone(__UpperCAmelCase						)
       else:
        raise ValueError(F"""backbone_type {self.backbone_type} not supported."""						)







      @property
      def    __lowerCAmelCase      (   self   :		Dict						)					->Optional[Any]:




       """simple docstring"""


       return self._out_features







      @out_features.setter
      def    __lowerCAmelCase      (   self   :		Tuple				,      __UpperCAmelCase   :		List[str]						)					->List[Any]:




       """simple docstring"""


       a						,  a							=   get_aligned_output_features_output_indices(
           out_features=__UpperCAmelCase				,      out_indices=__UpperCAmelCase				,      stage_names=self.stage_names						)







      @property
      def    __lowerCAmelCase      (   self   :		Dict						)					->int:




       """simple docstring"""


       return self._out_indices







      @out_indices.setter
      def    __lowerCAmelCase      (   self   :		Tuple				,      __UpperCAmelCase   :		Union[Tuple[int], List[int]]						)					->List[str]:




       """simple docstring"""


       a						,  a							=   get_aligned_output_features_output_indices(
           out_features=__UpperCAmelCase				,      out_indices=__UpperCAmelCase				,      stage_names=self.stage_names						)







      @property
      def    __lowerCAmelCase      (   self   :		Tuple						)					->str:




       """simple docstring"""


       return {stage: self.num_features[i] for i, stage in enumerate(self.stage_names						)}







      @property
      def    __lowerCAmelCase      (   self   :		int						)					->Optional[int]:




       """simple docstring"""


       return [self.out_feature_channels[name] for name in self.out_features]







      def    __lowerCAmelCase      (   self   :		Optional[Any]				,      *__UpperCAmelCase   :		List[str]				,      **__UpperCAmelCase   :		Optional[Any]						)					->Optional[int]:




       """simple docstring"""


       a							=   dict(inspect.signature(self.forward						).parameters						)
       a							=   {k: v for k, v in kwargs.items() if k in signature}
       return self(*__UpperCAmelCase				,      **__UpperCAmelCase						)







      def    __lowerCAmelCase      (   self   :		List[str]				,      __UpperCAmelCase   :		Dict				,      __UpperCAmelCase   :		Optional[bool] = None				,      __UpperCAmelCase   :		Optional[bool] = None				,      __UpperCAmelCase   :		Optional[bool] = None				,      )					->int:




       """simple docstring"""


       raise NotImplementedError('''This method should be implemented by the derived class.'''						)







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




       """simple docstring"""


       a							=   super().to_dict()
       a							=   output.pop('''_out_features'''						)
       a							=   output.pop('''_out_indices'''						)
       return output






class 	lowercase_      :






      '''simple docstring'''




      @property
      def    __lowerCAmelCase      (   self   :		List[str]						)					->Optional[int]:




       """simple docstring"""


       return self._out_features







      @out_features.setter
      def    __lowerCAmelCase      (   self   :		int				,      __UpperCAmelCase   :		List[str]						)					->int:




       """simple docstring"""


       a						,  a							=   get_aligned_output_features_output_indices(
           out_features=__UpperCAmelCase				,      out_indices=__UpperCAmelCase				,      stage_names=self.stage_names						)







      @property
      def    __lowerCAmelCase      (   self   :		str						)					->str:




       """simple docstring"""


       return self._out_indices







      @out_indices.setter
      def    __lowerCAmelCase      (   self   :		str				,      __UpperCAmelCase   :		Union[Tuple[int], List[int]]						)					->int:




       """simple docstring"""


       a						,  a							=   get_aligned_output_features_output_indices(
           out_features=__UpperCAmelCase				,      out_indices=__UpperCAmelCase				,      stage_names=self.stage_names						)







      def    __lowerCAmelCase      (   self   :		Dict						)					->List[Any]:




       """simple docstring"""


       a							=   super().to_dict()
       a							=   output.pop('''_out_features'''						)
       a							=   output.pop('''_out_indices'''						)
       return output


