

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


lowerCAmelCase__             =		logging.get_logger(__name__)







class 						lowercase     (					nn.Module				):


		"""simple docstring"""




		a__       							=	4_2
		a__       							=	jnp.floataa


		def 		A__       (					self):
			_UpperCamelCase :		Optional[int]								=    self.config.text_config
			_UpperCamelCase :		int								=    self.config.vision_config

			_UpperCamelCase :		Any								=    self.config.projection_dim
			_UpperCamelCase :		str								=    text_config.hidden_size
			_UpperCamelCase :		Optional[Any]								=    vision_config.hidden_size

			_UpperCamelCase :		List[str]								=    FLAX_MODEL_MAPPING[self.config.text_config.__class__].module_class
			_UpperCamelCase :		Dict								=    FLAX_MODEL_MAPPING.get(self.config.vision_config.__class__						,							UpperCamelCase_).module_class

			_UpperCamelCase :		Dict								=    text_module(UpperCamelCase_						,							dtype=self.dtype)
			_UpperCamelCase :		List[Any]								=    vision_module(UpperCamelCase_						,							dtype=self.dtype)

			_UpperCamelCase :		str								=    nn.Dense(
			    self.projection_dim						,							dtype=self.dtype						,							kernel_init=jax.nn.initializers.normal(0.0_2)						,							use_bias=UpperCamelCase_						,							)
			_UpperCamelCase :		Dict								=    nn.Dense(
			    self.projection_dim						,							dtype=self.dtype						,							kernel_init=jax.nn.initializers.normal(0.0_2)						,							use_bias=UpperCamelCase_						,							)
			_UpperCamelCase :		List[str]								=    self.param('logit_scale'						,							jax.nn.initializers.ones						,							[])






		def __call__(					self						,							__snake_case=None						,							__snake_case=None						,							__snake_case=None						,							__snake_case=None						,							__snake_case=None						,							__snake_case = True						,							__snake_case=None						,							__snake_case=None						,							__snake_case=None						,							):
			_UpperCamelCase :		Dict								=    return_dict if return_dict is not None else self.config.return_dict

			_UpperCamelCase :		int								=    self.vision_model(
			    pixel_values=UpperCamelCase_						,							deterministic=UpperCamelCase_						,							output_attentions=UpperCamelCase_						,							output_hidden_states=UpperCamelCase_						,							return_dict=UpperCamelCase_						,							)

			_UpperCamelCase :		Dict								=    self.text_model(
			    input_ids=UpperCamelCase_						,							attention_mask=UpperCamelCase_						,							token_type_ids=UpperCamelCase_						,							position_ids=UpperCamelCase_						,							deterministic=UpperCamelCase_						,							output_attentions=UpperCamelCase_						,							output_hidden_states=UpperCamelCase_						,							return_dict=UpperCamelCase_						,							)

			_UpperCamelCase :		Any								=    vision_outputs[1]
			_UpperCamelCase :		Tuple								=    self.visual_projection(UpperCamelCase_)

			_UpperCamelCase :		Any								=    text_outputs[1]
			_UpperCamelCase :		Tuple								=    self.text_projection(UpperCamelCase_)

			# normalized features
			_UpperCamelCase :		Union[str, Any]								=    image_embeds / jnp.linalg.norm(UpperCamelCase_						,							axis=-1						,							keepdims=UpperCamelCase_)
			_UpperCamelCase :		int								=    text_embeds / jnp.linalg.norm(UpperCamelCase_						,							axis=-1						,							keepdims=UpperCamelCase_)

			# cosine similarity as logits
			_UpperCamelCase :		List[Any]								=    jnp.exp(self.logit_scale)
			_UpperCamelCase :		Any								=    jnp.matmul(UpperCamelCase_						,							image_embeds.T) * logit_scale
			_UpperCamelCase :		str								=    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=UpperCamelCase_						,							logits_per_text=UpperCamelCase_						,							text_embeds=UpperCamelCase_						,							image_embeds=UpperCamelCase_						,							text_model_output=UpperCamelCase_						,							vision_model_output=UpperCamelCase_						,							)







class 						lowercase     (					__A				):


		"""simple docstring"""




		a__       							=	HybridCLIPConfig
		a__       							=	FlaxHybridCLIPModule


		def __init__(					self						,							__snake_case						,							__snake_case = None						,							__snake_case = 0						,							__snake_case = jnp.floataa						,							**__snake_case						,							):





			if input_shape is None:
				_UpperCamelCase :		str								=    ((1, 1), (1, config.vision_config.image_size, config.vision_config.image_size, 3))

			_UpperCamelCase :		Optional[int]								=    self.module_class(config=UpperCamelCase_						,							dtype=UpperCamelCase_						,							**UpperCamelCase_)
			super().__init__(UpperCamelCase_						,							UpperCamelCase_						,							input_shape=UpperCamelCase_						,							seed=UpperCamelCase_						,							dtype=UpperCamelCase_)


		def 		A__       (					self						,							__snake_case						,							__snake_case						,							__snake_case = None):
			_UpperCamelCase :		Any								=    jnp.zeros(input_shape[0]						,							dtype='i4')
			_UpperCamelCase :		Tuple								=    jnp.broadcast_to(jnp.arange(jnp.atleast_ad(UpperCamelCase_).shape[-1])						,							input_shape[0])
			_UpperCamelCase :		int								=    jnp.ones_like(UpperCamelCase_)
			_UpperCamelCase :		Optional[Any]								=    jnp.ones_like(UpperCamelCase_)

			_UpperCamelCase :		List[str]								=    jax.random.normal(UpperCamelCase_						,							input_shape[1])

			_UpperCamelCase :		List[str]								=    jax.random.split(UpperCamelCase_)
			_UpperCamelCase :		Tuple								=    {'''params''': params_rng, '''dropout''': dropout_rng}

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


		def __call__(					self						,							__snake_case						,							__snake_case						,							__snake_case=None						,							__snake_case=None						,							__snake_case=None						,							__snake_case = None						,							__snake_case = None						,							__snake_case = False						,							__snake_case = None						,							__snake_case = None						,							__snake_case = None						,							):
			_UpperCamelCase :		int								=    output_attentions if output_attentions is not None else self.config.output_attentions
			_UpperCamelCase :		int								=    (
			    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
			)
			_UpperCamelCase :		Tuple								=    return_dict if return_dict is not None else self.config.return_dict

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

			if token_type_ids is None:
				_UpperCamelCase :		List[str]								=    jnp.zeros_like(UpperCamelCase_)

			if attention_mask is None:
				_UpperCamelCase :		Optional[Any]								=    jnp.ones_like(UpperCamelCase_)

			# Handle any PRNG if needed
			_UpperCamelCase :		Optional[Any]								=    {}
			if dropout_rng is not None:
				_UpperCamelCase :		Optional[Any]								=    dropout_rng

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


		def 		A__       (					self						,							__snake_case						,							__snake_case=None						,							__snake_case=None						,							__snake_case=None						,							__snake_case = None						,							__snake_case = None						,							__snake_case=False						,							):





			if position_ids is None:
				_UpperCamelCase :		Optional[int]								=    jnp.broadcast_to(jnp.arange(jnp.atleast_ad(UpperCamelCase_).shape[-1])						,							input_ids.shape)

			if token_type_ids is None:
				_UpperCamelCase :		Union[str, Any]								=    jnp.zeros_like(UpperCamelCase_)

			if attention_mask is None:
				_UpperCamelCase :		Any								=    jnp.ones_like(UpperCamelCase_)

			# Handle any PRNG if needed
			_UpperCamelCase :		List[Any]								=    {}
			if dropout_rng is not None:
				_UpperCamelCase :		List[Any]								=    dropout_rng

			def _get_features(__snake_case						,							__snake_case						,							__snake_case						,							__snake_case						,							__snake_case						,							__snake_case):
				_UpperCamelCase :		Any								=    module.text_model(
				    input_ids=UpperCamelCase_						,							attention_mask=UpperCamelCase_						,							position_ids=UpperCamelCase_						,							token_type_ids=UpperCamelCase_						,							deterministic=UpperCamelCase_						,							)
				_UpperCamelCase :		Union[str, Any]								=    text_outputs[1]
				_UpperCamelCase :		List[Any]								=    module.text_projection(UpperCamelCase_)
				return text_features

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


		def 		A__       (					self						,							__snake_case						,							__snake_case = None						,							__snake_case = None						,							__snake_case=False):
			_UpperCamelCase :		Optional[int]								=    {}
			if dropout_rng is not None:
				_UpperCamelCase :		int								=    dropout_rng

			def _get_features(__snake_case						,							__snake_case						,							__snake_case):
				_UpperCamelCase :		Optional[int]								=    module.vision_model(pixel_values=UpperCamelCase_						,							deterministic=UpperCamelCase_)
				_UpperCamelCase :		Optional[Any]								=    vision_outputs[1]  # pooled_output
				_UpperCamelCase :		Dict								=    module.visual_projection(UpperCamelCase_)
				return image_features

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






		@classmethod
		def 		A__       (					cls						,							__snake_case = None						,							__snake_case = None						,							*__snake_case						,							**__snake_case						,							):
			_UpperCamelCase :		Dict								=    {
			    argument[len('text_') :]: value for argument, value in kwargs.items() if argument.startswith('text_')
			}

			_UpperCamelCase :		Dict								=    {
			    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
			_UpperCamelCase :		int								=    kwargs_text.pop('model'						,							UpperCamelCase_)
			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

					_UpperCamelCase :		Any								=    AutoConfig.from_pretrained(UpperCamelCase_)
					_UpperCamelCase :		Optional[Any]								=    text_config

				_UpperCamelCase :		Any								=    FlaxAutoModel.from_pretrained(UpperCamelCase_						,							*UpperCamelCase_						,							**UpperCamelCase_)

			_UpperCamelCase :		List[str]								=    kwargs_vision.pop('model'						,							UpperCamelCase_)
			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

					_UpperCamelCase :		Any								=    AutoConfig.from_pretrained(UpperCamelCase_)
					_UpperCamelCase :		Optional[Any]								=    vision_config

				_UpperCamelCase :		str								=    FlaxAutoModel.from_pretrained(UpperCamelCase_						,							*UpperCamelCase_						,							**UpperCamelCase_)

			# instantiate config with corresponding kwargs
			_UpperCamelCase :		List[str]								=    kwargs.pop('dtype'						,							jnp.floataa)
			_UpperCamelCase :		int								=    HybridCLIPConfig.from_text_vision_configs(text_model.config						,							vision_model.config						,							**UpperCamelCase_)

			# init model
			_UpperCamelCase :		Tuple								=    cls(UpperCamelCase_						,							*UpperCamelCase_						,							dtype=UpperCamelCase_						,							**UpperCamelCase_)

			if vision_config.model_type == "clip":
				_UpperCamelCase :		List[Any]								=    vision_model.params['''vision_model''']
				_UpperCamelCase :		int								=    vision_model.params['''visual_projection''']['''kernel''']
			else:
				_UpperCamelCase :		int								=    vision_model.params

			_UpperCamelCase :		Union[str, Any]								=    text_model.params

			return model









'''simple docstring'''






import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
						import torch


lowerCamelCase__	  =		logging.get_logger(__name__)

class 	_lowerCAmelCase     :






		'''simple docstring'''
		def __init__(						self							:       Optional[int]      ,		UpperCamelCase_							:       Optional[Any]      ,		UpperCamelCase_							:       Union[str, Any]=21      ,		UpperCamelCase_							:       Dict=[]      ,		UpperCamelCase_							:       Any=None		)      ->			Any:







									'''simple docstring'''





									_lowercase							:		Optional[Any]				=   model
									_lowercase							:		Tuple				=   trace_batch_nums
									_lowercase							:		Optional[Any]				=   abort_after_batch_num

									# keep a LIFO buffer of frames to dump as soon as inf/nan is encountered to give context to the problem emergence
									_lowercase							:		int				=   collections.deque([]      ,		UpperCamelCase_		)
									_lowercase							:		List[Any]				=   []
									_lowercase							:		List[str]				=   0
									_lowercase							:		Optional[int]				=   0
									_lowercase							:		List[Any]				=   False
									_lowercase							:		List[Any]				=   '''                 '''

									self.analyse_model()

									self.register_forward_hook()
		def 						__lowercase      (						self							:       Union[str, Any]      ,		UpperCamelCase_							:       Optional[Any]=None		)      ->			Optional[Any]:







									'''simple docstring'''





									if frame is not None:
																self.expand_frame(UpperCamelCase_		)
									self.frames.append('''\n'''.join(self.frame		)		)
									_lowercase							:		Dict				=   []  # start a new frame
		def 						__lowercase      (						self							:       Union[str, Any]      ,		UpperCamelCase_							:       Optional[Any]		)      ->			Union[str, Any]:







									'''simple docstring'''





									self.frame.append(UpperCamelCase_		)
		def 						__lowercase      (						self							:       str		)      ->			List[str]:







									'''simple docstring'''





									print('''\n'''.join(self.frames		)		)
									_lowercase							:		Union[str, Any]				=   []
		def 						__lowercase      (						self							:       Optional[int]		)      ->			List[str]:







									'''simple docstring'''





									_lowercase							:		Optional[Any]				=   []
		def 						__lowercase      (						self							:       List[Any]		)      ->			Optional[int]:







									'''simple docstring'''





									print(F"\nDetected inf/nan during batch_number={self.batch_number}"		)
									print(F"Last {len(self.frames		)} forward frames:"		)
									print(F"{'abs min':8} {'abs max':8} metadata"		)
									print('''\n'''.join(self.frames		)		)
									print('''\n\n'''		)
									_lowercase							:		List[Any]				=   []
		def 						__lowercase      (						self							:       List[str]		)      ->			Tuple:







									'''simple docstring'''





									_lowercase							:		int				=   {m: name for name, m in self.model.named_modules()}
									# self.longest_module_name = max(len(v) for v in self.module_names.values())
		def 						__lowercase      (						self							:       Dict      ,		UpperCamelCase_							:       str      ,		UpperCamelCase_							:       Optional[int]		)      ->			int:







									'''simple docstring'''





									if torch.is_tensor(UpperCamelCase_		):
																self.expand_frame(get_abs_min_max(UpperCamelCase_      ,		UpperCamelCase_		)		)
																if detect_overflow(UpperCamelCase_      ,		UpperCamelCase_		):
																							_lowercase							:		Any				=   True
									elif var is None:
																self.expand_frame(F"{'None':>17} {ctx}"		)
									else:
																self.expand_frame(F"{'not a tensor':>17} {ctx}"		)
		def 						__lowercase      (						self							:       Union[str, Any]		)      ->			int:







									'''simple docstring'''





									self.expand_frame(F"\n\n{self.prefix} *** Starting batch number={self.batch_number} ***"		)
									self.expand_frame(F"{'abs min':8} {'abs max':8} metadata"		)
		def 						__lowercase      (						self							:       str		)      ->			Any:







									'''simple docstring'''





									self.expand_frame(F"{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"		)
		def 						__lowercase      (						self							:       Tuple      ,		UpperCamelCase_							:       Any      ,		UpperCamelCase_							:       Union[str, Any]      ,		UpperCamelCase_							:       Optional[int]		)      ->			Dict:







									'''simple docstring'''





									self.expand_frame(F"{self.prefix} {self.module_names[module]} {module.__class__.__name__}"		)

									# params
									for name, p in module.named_parameters(recurse=UpperCamelCase_		):
																self.analyse_variable(UpperCamelCase_      ,		UpperCamelCase_		)

									# inputs
									if isinstance(UpperCamelCase_      ,		UpperCamelCase_		):
																for i, x in enumerate(UpperCamelCase_		):
																							self.analyse_variable(UpperCamelCase_      ,		F"input[{i}]"		)
									else:
																self.analyse_variable(UpperCamelCase_      ,		'''input'''		)

									# outputs
									if isinstance(UpperCamelCase_      ,		UpperCamelCase_		):
																for i, x in enumerate(UpperCamelCase_		):
																							# possibly a tuple of tuples
																							if isinstance(UpperCamelCase_      ,		UpperCamelCase_		):
																														for j, y in enumerate(UpperCamelCase_		):
																																					self.analyse_variable(UpperCamelCase_      ,		F"output[{i}][{j}]"		)
																							else:
																														self.analyse_variable(UpperCamelCase_      ,		F"output[{i}]"		)
									else:
																self.analyse_variable(UpperCamelCase_      ,		'''output'''		)

									self.save_frame()
		def 						__lowercase      (						self							:       Optional[int]		)      ->			Dict:







									'''simple docstring'''





									self.model.apply(self._register_forward_hook		)
		def 						__lowercase      (						self							:       Optional[int]      ,		UpperCamelCase_							:       int		)      ->			List[str]:







									'''simple docstring'''





									module.register_forward_hook(self.forward_hook		)





		def 						__lowercase      (						self							:       Optional[int]      ,		UpperCamelCase_							:       Optional[int]      ,		UpperCamelCase_							:       Dict      ,		UpperCamelCase_							:       List[Any]		)      ->			List[Any]:







									'''simple docstring'''





									_lowercase							:		str				=   False

									_lowercase							:		Dict				=   True if self.batch_number in self.trace_batch_nums else False
									if trace_mode:
																self.reset_saved_frames()

									if self.total_calls == 0:
																self.batch_start_frame()
									self.total_calls += 1

									# count batch numbers - the very first forward hook of the batch will be called when the
									# batch completes - i.e. it gets called very last - we know this batch has finished
									if module == self.model:
																self.batch_number += 1
																_lowercase							:		Union[str, Any]				=   True

									self.create_frame(UpperCamelCase_      ,		UpperCamelCase_      ,		UpperCamelCase_		)

									# if last_frame_of_batch:
									#     self.batch_end_frame()

									if trace_mode:
																self.trace_frames()

									if last_frame_of_batch:
																self.batch_start_frame()

									if self.detected_overflow and not trace_mode:
																self.dump_saved_frames()

																# now we can abort, as it's pointless to continue running
																raise ValueError(
																    '''DebugUnderflowOverflow: inf/nan detected, aborting as there is no point running further. '''
																    '''Please scroll up above this traceback to see the activation values prior to this event.'''		)

									# abort after certain batch if requested to do so
									if self.abort_after_batch_num is not None and self.batch_number > self.abort_after_batch_num:
																raise ValueError(
																    F"DebugUnderflowOverflow: aborting after {self.batch_number} batches due to"
																    F" `abort_after_batch_num={self.abort_after_batch_num}` arg"		)





def 				_SCREAMING_SNAKE_CASE(					snake_case_    :     Any ,       snake_case_    :     Dict  )		->Any:






							'''simple docstring'''
							_lowercase							:		Union[str, Any]				=   var.abs()
							return F"{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"





def 				_SCREAMING_SNAKE_CASE(					snake_case_    :     int ,       snake_case_    :     Any  )		->int:






							'''simple docstring'''
							_lowercase							:		List[Any]				=   False
							if torch.isnan(snake_case_  ).any().item():
														_lowercase							:		Any				=   True
														print(F"{ctx} has nans"  )
							if torch.isinf(snake_case_  ).any().item():
														_lowercase							:		Optional[Any]				=   True
														print(F"{ctx} has infs"  )

							# if needed to monitor large elements can enable the following
							if 0:  # and detected:
														_lowercase							:		List[Any]				=   var[torch.ge(var.abs() ,       1_00  )]
														if naaa.numel() > 0:
																					print(F"{ctx}:  n100={naaa.numel()}"  )
														_lowercase							:		List[str]				=   var[torch.ge(var.abs() ,       10_00  )]
														if naaaa.numel() > 0:
																					print(F"{ctx}: n1000={naaaa.numel()}"  )
														_lowercase							:		Tuple				=   var[torch.ge(var.abs() ,       1_00_00  )]
														if naaaaa.numel() > 0:
																					print(F"{ctx}: n10000={naaaaa.numel()}"  )

							if 0:
														print(F"min={var.min():9.2e} max={var.max():9.2e}"  )

							if 0:
														print(F"min={var.min():9.2e} max={var.max():9.2e} var={var.var():9.2e} mean={var.mean():9.2e} ({ctx})"  )

							return detected




class 	_lowerCAmelCase     (    __A       ):






		'''simple docstring'''
		snake_case_							= 'underflow_overflow'
		snake_case_							= 'tpu_metrics_debug'



