

from typing import Optional, Tuple

import flax.linen as nn
import jax
import jax.numpy as jnp
from configuration_hybrid_clip import HybridCLIPConfig
from flax.core.frozen_dict import FrozenDict

from transformers import FLAX_MODEL_MAPPING, FlaxCLIPVisionModel
from transformers.modeling_flax_utils import FlaxPreTrainedModel
from transformers.models.clip.modeling_flax_clip import FlaxCLIPOutput
from transformers.utils import logging


__UpperCAmelCase     					=  logging.get_logger(__name__)





class A__     (			nn.Module			):



 """simple docstring"""







 _lowercase  :       HybridCLIPConfig
 _lowercase  :       jnp.dtype  			=    jnp.floataa

 def  __magic_name__							(		self			:						Union[str, Any]							):






      '''simple docstring'''




      _lowerCAmelCase :       int	   =     self.config.text_config
      _lowerCAmelCase :       Dict	   =     self.config.vision_config

      _lowerCAmelCase :       Union[str, Any]	   =     self.config.projection_dim
      _lowerCAmelCase :       List[str]	   =     text_config.hidden_size
      _lowerCAmelCase :       Dict	   =     vision_config.hidden_size

      _lowerCAmelCase :       Dict	   =     FLAX_MODEL_MAPPING[self.config.text_config.__class__].module_class
      _lowerCAmelCase :       Tuple	   =     FLAX_MODEL_MAPPING.get(self.config.vision_config.__class__  ,    A_							).module_class

      _lowerCAmelCase :       Tuple	   =     text_module(A_  ,    dtype=self.dtype							)
      _lowerCAmelCase :       List[Any]	   =     vision_module(A_  ,    dtype=self.dtype							)

      _lowerCAmelCase :       Union[str, Any]	   =     nn.Dense(
          self.projection_dim  ,    dtype=self.dtype  ,    kernel_init=jax.nn.initializers.normal(0.02							)  ,    use_bias=A_  ,    )
      _lowerCAmelCase :       str	   =     nn.Dense(
          self.projection_dim  ,    dtype=self.dtype  ,    kernel_init=jax.nn.initializers.normal(0.02							)  ,    use_bias=A_  ,    )
      _lowerCAmelCase :       List[Any]	   =     self.param("logit_scale"  ,    jax.nn.initializers.ones  ,    []							)


 def __call__(		self			:						str  ,    A_			:						int=None  ,    A_			:						Tuple=None  ,    A_			:						str=None  ,    A_			:						Dict=None  ,    A_			:						Any=None  ,    A_			:						bool = True  ,    A_			:						Optional[int]=None  ,    A_			:						List[Any]=None  ,    A_			:						List[Any]=None  ,    ):






      '''simple docstring'''




      _lowerCAmelCase :       Optional[Any]	   =     return_dict if return_dict is not None else self.config.return_dict

      _lowerCAmelCase :       int	   =     self.vision_model(
          pixel_values=A_  ,    deterministic=A_  ,    output_attentions=A_  ,    output_hidden_states=A_  ,    return_dict=A_  ,    )

      _lowerCAmelCase :       Optional[int]	   =     self.text_model(
          input_ids=A_  ,    attention_mask=A_  ,    token_type_ids=A_  ,    position_ids=A_  ,    deterministic=A_  ,    output_attentions=A_  ,    output_hidden_states=A_  ,    return_dict=A_  ,    )

      _lowerCAmelCase :       Tuple	   =     vision_outputs[1]
      _lowerCAmelCase :       Optional[int]	   =     self.visual_projection(A_							)

      _lowerCAmelCase :       Optional[int]	   =     text_outputs[1]
      _lowerCAmelCase :       int	   =     self.text_projection(A_							)

      # normalized features
      _lowerCAmelCase :       str	   =     image_embeds / jnp.linalg.norm(A_  ,    axis=-1  ,    keepdims=A_							)
      _lowerCAmelCase :       Any	   =     text_embeds / jnp.linalg.norm(A_  ,    axis=-1  ,    keepdims=A_							)

      # cosine similarity as logits
      _lowerCAmelCase :       str	   =     jnp.exp(self.logit_scale							)
      _lowerCAmelCase :       List[str]	   =     jnp.matmul(A_  ,    image_embeds.T							) * logit_scale
      _lowerCAmelCase :       Any	   =     logits_per_text.T

      if not return_dict:
           return (logits_per_image, logits_per_text, text_embeds, image_embeds, text_outputs, vision_outputs)

      return FlaxCLIPOutput(
          logits_per_image=A_  ,    logits_per_text=A_  ,    text_embeds=A_  ,    image_embeds=A_  ,    text_model_output=A_  ,    vision_model_output=A_  ,    )





class A__     (			A			):



 """simple docstring"""







 _lowercase  :       Any  			=    HybridCLIPConfig
 _lowercase  :       Union[str, Any]  			=    FlaxHybridCLIPModule

 def __init__(		self			:						Dict  ,    A_			:						HybridCLIPConfig  ,    A_			:						Optional[Tuple] = None  ,    A_			:						int = 0  ,    A_			:						jnp.dtype = jnp.floataa  ,    **A_			:						str  ,    ):






      '''simple docstring'''




      if input_shape is None:
           _lowerCAmelCase :       Dict	   =     ((1, 1), (1, config.vision_config.image_size, config.vision_config.image_size, 3))

      _lowerCAmelCase :       List[Any]	   =     self.module_class(config=A_  ,    dtype=A_  ,    **A_							)
      super().__init__(A_  ,    A_  ,    input_shape=A_  ,    seed=A_  ,    dtype=A_							)

 def  __magic_name__							(		self			:						Union[str, Any]  ,    A_			:						jax.random.PRNGKey  ,    A_			:						Tuple  ,    A_			:						FrozenDict = None							):






      '''simple docstring'''




      _lowerCAmelCase :       int	   =     jnp.zeros(input_shape[0]  ,    dtype="i4"							)
      _lowerCAmelCase :       Optional[Any]	   =     jnp.broadcast_to(jnp.arange(jnp.atleast_ad(A_							).shape[-1]							)  ,    input_shape[0]							)
      _lowerCAmelCase :       List[str]	   =     jnp.ones_like(A_							)
      _lowerCAmelCase :       List[str]	   =     jnp.ones_like(A_							)

      _lowerCAmelCase :       List[Any]	   =     jax.random.normal(A_  ,    input_shape[1]							)

      _lowerCAmelCase   ,      _lowerCAmelCase :       Optional[int]	   =     jax.random.split(A_							)
      _lowerCAmelCase :       List[Any]	   =     {"params": params_rng, "dropout": dropout_rng}

      return self.module.init(A_  ,    A_  ,    A_  ,    A_  ,    A_  ,    A_							)["params"]

 def __call__(		self			:						Dict  ,    A_			:						Tuple  ,    A_			:						Any  ,    A_			:						List[str]=None  ,    A_			:						Any=None  ,    A_			:						Union[str, Any]=None  ,    A_			:						dict = None  ,    A_			:						jax.random.PRNGKey = None  ,    A_			:						bool = False  ,    A_			:						Optional[bool] = None  ,    A_			:						Optional[bool] = None  ,    A_			:						Optional[bool] = None  ,    ):






      '''simple docstring'''




      _lowerCAmelCase :       List[str]	   =     output_attentions if output_attentions is not None else self.config.output_attentions
      _lowerCAmelCase :       int	   =     (
          output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
      )
      _lowerCAmelCase :       int	   =     return_dict if return_dict is not None else self.config.return_dict

      if position_ids is None:
           _lowerCAmelCase :       List[str]	   =     jnp.broadcast_to(jnp.arange(jnp.atleast_ad(A_							).shape[-1]							)  ,    input_ids.shape							)

      if token_type_ids is None:
           _lowerCAmelCase :       Any	   =     jnp.zeros_like(A_							)

      if attention_mask is None:
           _lowerCAmelCase :       List[Any]	   =     jnp.ones_like(A_							)

      # Handle any PRNG if needed
      _lowerCAmelCase :       Tuple	   =     {}
      if dropout_rng is not None:
           _lowerCAmelCase :       Tuple	   =     dropout_rng

      return self.module.apply(
          {"params": params or self.params}  ,    jnp.array(A_  ,    dtype="i4"							)  ,    jnp.array(A_  ,    dtype=jnp.floataa							)  ,    jnp.array(A_  ,    dtype="i4"							)  ,    jnp.array(A_  ,    dtype="i4"							)  ,    jnp.array(A_  ,    dtype="i4"							)  ,    not train  ,    A_  ,    A_  ,    A_  ,    rngs=A_  ,    )

 def  __magic_name__							(		self			:						Optional[Any]  ,    A_			:						Tuple  ,    A_			:						Optional[Any]=None  ,    A_			:						Union[str, Any]=None  ,    A_			:						int=None  ,    A_			:						dict = None  ,    A_			:						jax.random.PRNGKey = None  ,    A_			:						Optional[Any]=False  ,    ):






      '''simple docstring'''




      if position_ids is None:
           _lowerCAmelCase :       Optional[Any]	   =     jnp.broadcast_to(jnp.arange(jnp.atleast_ad(A_							).shape[-1]							)  ,    input_ids.shape							)

      if token_type_ids is None:
           _lowerCAmelCase :       Optional[Any]	   =     jnp.zeros_like(A_							)

      if attention_mask is None:
           _lowerCAmelCase :       Any	   =     jnp.ones_like(A_							)

      # Handle any PRNG if needed
      _lowerCAmelCase :       Dict	   =     {}
      if dropout_rng is not None:
           _lowerCAmelCase :       Optional[int]	   =     dropout_rng

      def _get_features(A_			:						Optional[int]  ,    A_			:						int  ,    A_			:						Dict  ,    A_			:						str  ,    A_			:						Union[str, Any]  ,    A_			:						Any							):
           _lowerCAmelCase :       Optional[Any]	   =     module.text_model(
               input_ids=A_  ,    attention_mask=A_  ,    position_ids=A_  ,    token_type_ids=A_  ,    deterministic=A_  ,    )
           _lowerCAmelCase :       Optional[int]	   =     text_outputs[1]
           _lowerCAmelCase :       Optional[int]	   =     module.text_projection(A_							)
           return text_features

      return self.module.apply(
          {"params": params or self.params}  ,    jnp.array(A_  ,    dtype="i4"							)  ,    jnp.array(A_  ,    dtype="i4"							)  ,    jnp.array(A_  ,    dtype="i4"							)  ,    jnp.array(A_  ,    dtype="i4"							)  ,    not train  ,    method=_get_features  ,    rngs=A_  ,    )

 def  __magic_name__							(		self			:						Dict  ,    A_			:						Optional[Any]  ,    A_			:						dict = None  ,    A_			:						jax.random.PRNGKey = None  ,    A_			:						Optional[Any]=False							):






      '''simple docstring'''




      _lowerCAmelCase :       List[Any]	   =     {}
      if dropout_rng is not None:
           _lowerCAmelCase :       List[str]	   =     dropout_rng

      def _get_features(A_			:						List[str]  ,    A_			:						Union[str, Any]  ,    A_			:						str							):
           _lowerCAmelCase :       Tuple	   =     module.vision_model(pixel_values=A_  ,    deterministic=A_							)
           _lowerCAmelCase :       Optional[Any]	   =     vision_outputs[1]  # pooled_output
           _lowerCAmelCase :       List[str]	   =     module.visual_projection(A_							)
           return image_features

      return self.module.apply(
          {"params": params or self.params}  ,    jnp.array(A_  ,    dtype=jnp.floataa							)  ,    not train  ,    method=_get_features  ,    rngs=A_  ,    )


 @classmethod
 def  __magic_name__							(		cls			:						List[str]  ,    A_			:						str = None  ,    A_			:						str = None  ,    *A_			:						Any  ,    **A_			:						Dict  ,    ):






      '''simple docstring'''




      _lowerCAmelCase :       Optional[int]	   =     {
          argument[len("text_"							) :]: value for argument, value in kwargs.items() if argument.startswith("text_"							)
      }

      _lowerCAmelCase :       Optional[Any]	   =     {
          argument[len("vision_"							) :]: value for argument, value in kwargs.items() if argument.startswith("vision_"							)
      }

      # remove text, vision kwargs from kwargs
      for key in kwargs_text.keys():
           del kwargs["text_" + key]
      for key in kwargs_vision.keys():
           del kwargs["vision_" + key]

      # Load and initialize the text and vision model
      _lowerCAmelCase :       Tuple	   =     kwargs_text.pop("model"  ,    A_							)
      if text_model is None:
           assert (
               text_model_name_or_path is not None
           ), "If `model` is not defined as an argument, a `text_model_name_or_path` has to be defined"
           from transformers import FlaxAutoModel

           if "config" not in kwargs_text:
                from transformers import AutoConfig

                _lowerCAmelCase :       List[Any]	   =     AutoConfig.from_pretrained(A_							)
                _lowerCAmelCase :       Optional[int]	   =     text_config

           _lowerCAmelCase :       Union[str, Any]	   =     FlaxAutoModel.from_pretrained(A_  ,    *A_  ,    **A_							)

      _lowerCAmelCase :       int	   =     kwargs_vision.pop("model"  ,    A_							)
      if vision_model is None:
           assert (
               vision_model_name_or_path is not None
           ), "If `model` is not defined as an argument, a `vision_model_name_or_path` has to be defined"
           from transformers import FlaxAutoModel

           if "config" not in kwargs_vision:
                from transformers import AutoConfig

                _lowerCAmelCase :       Optional[int]	   =     AutoConfig.from_pretrained(A_							)
                _lowerCAmelCase :       List[Any]	   =     vision_config

           _lowerCAmelCase :       List[Any]	   =     FlaxAutoModel.from_pretrained(A_  ,    *A_  ,    **A_							)

      # instantiate config with corresponding kwargs
      _lowerCAmelCase :       Union[str, Any]	   =     kwargs.pop("dtype"  ,    jnp.floataa							)
      _lowerCAmelCase :       Union[str, Any]	   =     HybridCLIPConfig.from_text_vision_configs(text_model.config  ,    vision_model.config  ,    **A_							)

      # init model
      _lowerCAmelCase :       str	   =     cls(A_  ,    *A_  ,    dtype=A_  ,    **A_							)

      if vision_config.model_type == "clip":
           _lowerCAmelCase :       Optional[Any]	   =     vision_model.params["vision_model"]
           _lowerCAmelCase :       Any	   =     vision_model.params["visual_projection"]["kernel"]
      else:
           _lowerCAmelCase :       Optional[Any]	   =     vision_model.params

      _lowerCAmelCase :       List[str]	   =     text_model.params

      return model









import numpy as np
import PIL
import torch
import torchvision.transforms as T
import torchvision.transforms.functional as TF
from PIL import Image







def 						_snake_case		(     SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=256					)							->   Optional[int]:

     """simple docstring"""







     _lowerCAmelCase :       Any	   =     min(img.size					)

     if s < target_image_size:
          raise ValueError(f'''min dim for image {s} < {target_image_size}'''					)

     _lowerCAmelCase :       Dict	   =     target_image_size / s
     _lowerCAmelCase :       Dict	   =     (round(r * img.size[1]					), round(r * img.size[0]					))
     _lowerCAmelCase :       Any	   =     TF.resize(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				interpolation=PIL.Image.LANCZOS					)
     _lowerCAmelCase :       str	   =     TF.center_crop(SCREAMING_SNAKE_CASE		,				output_size=2 * [target_image_size]					)
     _lowerCAmelCase :       int	   =     torch.unsqueeze(T.ToTensor()(SCREAMING_SNAKE_CASE					)		,				0					)
     return img







def 						_snake_case		(     SCREAMING_SNAKE_CASE					)							->   List[str]:

     """simple docstring"""







     _lowerCAmelCase :       List[Any]	   =     2.0 * x - 1.0
     return x







def 						_snake_case		(     SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=True		,				SCREAMING_SNAKE_CASE="RGB"					)							->   int:

     """simple docstring"""







     _lowerCAmelCase :       Optional[int]	   =     x.detach().cpu()
     if process:
          _lowerCAmelCase :       List[Any]	   =     post_process_tensor(SCREAMING_SNAKE_CASE					)
     _lowerCAmelCase :       Optional[int]	   =     x.numpy()
     if process:
          _lowerCAmelCase :       List[Any]	   =     (255 * x).astype(np.uinta					)
     _lowerCAmelCase :       Optional[int]	   =     Image.fromarray(SCREAMING_SNAKE_CASE					)
     if not x.mode == mode:
          _lowerCAmelCase :       str	   =     x.convert(SCREAMING_SNAKE_CASE					)
     return x







def 						_snake_case		(     SCREAMING_SNAKE_CASE					)							->   Union[str, Any]:

     """simple docstring"""







     _lowerCAmelCase :       Any	   =     torch.clamp(SCREAMING_SNAKE_CASE		,				-1.0		,				1.0					)
     _lowerCAmelCase :       Any	   =     (x + 1.0) / 2.0
     _lowerCAmelCase :       List[Any]	   =     x.permute(1		,				2		,				0					)
     return x







def 						_snake_case		(     SCREAMING_SNAKE_CASE					)							->   Tuple:

     """simple docstring"""







     _lowerCAmelCase :       str	   =     post_process_tensor(x.squeeze()					)
     return x.permute(2		,				0		,				1					).unsqueeze(0					)






