
import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from ...utils import logging
from ..auto import CONFIG_MAPPING


lowercase    :   Tuple						=			logging.get_logger(__name__)

lowercase    :   List[Any]						=			{
    'salesforce/blip2-opt-2.7b': 'https://huggingface.co/salesforce/blip2-opt-2.7b/resolve/main/config.json',
}




class 					lowerCamelCase__				(							__lowercase):




	'''simple docstring'''




	_A								=	'blip_2_vision_model'







	def __init__(   self						:List[Any]	,  a						:int=1_4_0_8	,  a						:Any=6_1_4_4	,  a						:Tuple=3_9	,  a						:Union[str, Any]=1_6	,  a						:Optional[int]=2_2_4	,  a						:List[Any]=1_4	,  a						:Optional[Any]="gelu"	,  a						:Tuple=0.00001	,  a						:int=0.0	,  a						:Any=1E-1_0	,  a						:Dict=True	,  **a						:Union[str, Any]	,  )						->					List[Any]:
								super().__init__(**a		)

								__UpperCamelCase      :  Tuple       =   hidden_size
								__UpperCamelCase      :  Union[str, Any]       =   intermediate_size
								__UpperCamelCase      :  List[Any]       =   num_hidden_layers
								__UpperCamelCase      :  Union[str, Any]       =   num_attention_heads
								__UpperCamelCase      :  int       =   patch_size
								__UpperCamelCase      :  str       =   image_size
								__UpperCamelCase      :  Tuple       =   initializer_range
								__UpperCamelCase      :  Dict       =   attention_dropout
								__UpperCamelCase      :  Tuple       =   layer_norm_eps
								__UpperCamelCase      :  int       =   hidden_act
								__UpperCamelCase      :  Union[str, Any]       =   qkv_bias







	@classmethod
	def   _lowerCamelCase							(   cls						:Dict	,  a						:Union[str, os.PathLike]	,  **a						:Dict		)						->					"PretrainedConfig":
								cls._set_token_in_kwargs(a		)

								__UpperCamelCase  ,		__UpperCamelCase      :  Optional[Any]       =   cls.get_config_dict(a	,  **a		)

								# get the vision config dict if we are loading from Blip2Config
								if config_dict.get("model_type"		) == "blip-2":
															__UpperCamelCase      :  Tuple       =   config_dict["vision_config"]

								if "model_type" in config_dict and hasattr(cls	,  "model_type"		) and config_dict["model_type"] != cls.model_type:
															logger.warning(
															    f'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
															    f'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'		)

								return cls.from_dict(a	,  **a		)




class 					lowerCamelCase__				(							__lowercase):




	'''simple docstring'''




	_A								=	'blip_2_qformer'







	def __init__(   self						:Tuple	,  a						:Optional[int]=3_0_5_2_2	,  a						:Union[str, Any]=7_6_8	,  a						:Any=1_2	,  a						:Optional[Any]=1_2	,  a						:Tuple=3_0_7_2	,  a						:Dict="gelu"	,  a						:List[str]=0.1	,  a						:Any=0.1	,  a						:List[Any]=5_1_2	,  a						:Dict=0.02	,  a						:Tuple=1E-1_2	,  a						:Union[str, Any]=0	,  a						:Optional[int]="absolute"	,  a						:Union[str, Any]=2	,  a						:Optional[Any]=1_4_0_8	,  **a						:Dict	,  )						->					str:
								super().__init__(pad_token_id=a	,  **a		)

								__UpperCamelCase      :  str       =   vocab_size
								__UpperCamelCase      :  Any       =   hidden_size
								__UpperCamelCase      :  Tuple       =   num_hidden_layers
								__UpperCamelCase      :  List[str]       =   num_attention_heads
								__UpperCamelCase      :  Dict       =   hidden_act
								__UpperCamelCase      :  Any       =   intermediate_size
								__UpperCamelCase      :  Union[str, Any]       =   hidden_dropout_prob
								__UpperCamelCase      :  int       =   attention_probs_dropout_prob
								__UpperCamelCase      :  Optional[int]       =   max_position_embeddings
								__UpperCamelCase      :  Tuple       =   initializer_range
								__UpperCamelCase      :  Tuple       =   layer_norm_eps
								__UpperCamelCase      :  Any       =   position_embedding_type
								__UpperCamelCase      :  Tuple       =   cross_attention_frequency
								__UpperCamelCase      :  int       =   encoder_hidden_size







	@classmethod
	def   _lowerCamelCase							(   cls						:Optional[int]	,  a						:Union[str, os.PathLike]	,  **a						:str		)						->					"PretrainedConfig":
								cls._set_token_in_kwargs(a		)

								__UpperCamelCase  ,		__UpperCamelCase      :  List[str]       =   cls.get_config_dict(a	,  **a		)

								# get the qformer config dict if we are loading from Blip2Config
								if config_dict.get("model_type"		) == "blip-2":
															__UpperCamelCase      :  Dict       =   config_dict["qformer_config"]

								if "model_type" in config_dict and hasattr(cls	,  "model_type"		) and config_dict["model_type"] != cls.model_type:
															logger.warning(
															    f'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
															    f'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'		)

								return cls.from_dict(a	,  **a		)




class 					lowerCamelCase__				(							__lowercase):




	'''simple docstring'''




	_A								=	'blip-2'
	_A								=	True







	def __init__(   self						:str	,  a						:Optional[int]=None	,  a						:int=None	,  a						:Tuple=None	,  a						:str=3_2	,  **a						:List[str]		)						->					Optional[int]:
								super().__init__(**a		)

								if vision_config is None:
															__UpperCamelCase      :  str       =   {}
															logger.info("vision_config is None. initializing the Blip2VisionConfig with default values."		)

								if qformer_config is None:
															__UpperCamelCase      :  Any       =   {}
															logger.info("qformer_config is None. Initializing the Blip2QFormerConfig with default values."		)

								if text_config is None:
															__UpperCamelCase      :  str       =   {}
															logger.info("text_config is None. Initializing the text config with default values (`OPTConfig`)."		)

								__UpperCamelCase      :  Dict       =   BlipaVisionConfig(**a		)
								__UpperCamelCase      :  List[str]       =   BlipaQFormerConfig(**a		)
								__UpperCamelCase      :  Tuple       =   text_config["model_type"] if "model_type" in text_config else "opt"
								__UpperCamelCase      :  Any       =   CONFIG_MAPPING[text_model_type](**a		)

								__UpperCamelCase      :  List[str]       =   self.text_config.tie_word_embeddings
								__UpperCamelCase      :  str       =   self.text_config.is_encoder_decoder

								__UpperCamelCase      :  List[str]       =   num_query_tokens
								__UpperCamelCase      :  int       =   self.vision_config.hidden_size
								__UpperCamelCase      :  Any       =   self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
								__UpperCamelCase      :  Any       =   1.0
								__UpperCamelCase      :  Any       =   0.02







	@classmethod
	def   _lowerCamelCase							(   cls						:Any	,  a						:BlipaVisionConfig	,  a						:BlipaQFormerConfig	,  a						:PretrainedConfig	,  **a						:Tuple	,  )						->					List[str]:
								return cls(
								    vision_config=vision_config.to_dict()	,  qformer_config=qformer_config.to_dict()	,  text_config=text_config.to_dict()	,  **a	,  )







	def   _lowerCamelCase							(   self						:Union[str, Any]		)						->					Tuple:
								__UpperCamelCase      :  Optional[int]       =   copy.deepcopy(self.__dict__		)
								__UpperCamelCase      :  str       =   self.vision_config.to_dict()
								__UpperCamelCase      :  Tuple       =   self.qformer_config.to_dict()
								__UpperCamelCase      :  List[str]       =   self.text_config.to_dict()
								__UpperCamelCase      :  Any       =   self.__class__.model_type
								return output

from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
		from ..models.auto import AutoTokenizer




class 					lowerCamelCase__				:




	'''simple docstring'''




	def   _lowerCamelCase							(   self						:Optional[Any]	,  a						:int		)						->					Any:
								raise NotImplementedError()







	def   _lowerCamelCase							(   self						:Any		)						->					Optional[int]:
								raise NotImplementedError()




class 					lowerCamelCase__				(							__lowercase):




	'''simple docstring'''




	def __init__(   self						:List[Any]	,  a						:"AutoTokenizer"	,  a						:bool = False	,  **a						:Tuple		)						->					Union[str, Any]:
								__UpperCamelCase      :  int       =   tokenizer
								__UpperCamelCase      :  Any       =   skip_prompt
								__UpperCamelCase      :  Union[str, Any]       =   decode_kwargs

								# variables used in the streaming process
								__UpperCamelCase      :  Union[str, Any]       =   []
								__UpperCamelCase      :  str       =   0
								__UpperCamelCase      :  str       =   True







	def   _lowerCamelCase							(   self						:List[str]	,  a						:Union[str, Any]		)						->					Any:
								if len(value.shape		) > 1 and value.shape[0] > 1:
															raise ValueError("TextStreamer only supports batch size 1"		)
								elif len(value.shape		) > 1:
															__UpperCamelCase      :  Optional[int]       =   value[0]

								if self.skip_prompt and self.next_tokens_are_prompt:
															__UpperCamelCase      :  Optional[Any]       =   False
															return

								# Add the new token to the cache and decodes the entire thing.
								self.token_cache.extend(value.tolist()		)
								__UpperCamelCase      :  Optional[int]       =   self.tokenizer.decode(self.token_cache	,  **self.decode_kwargs		)

								# After the symbol for a new line, we flush the cache.
								if text.endswith("\n"		):
															__UpperCamelCase      :  Optional[int]       =   text[self.print_len :]
															__UpperCamelCase      :  Dict       =   []
															__UpperCamelCase      :  int       =   0
								# If the last token is a CJK character, we print the characters.
								elif len(a		) > 0 and self._is_chinese_char(ord(text[-1]		)		):
															__UpperCamelCase      :  Optional[int]       =   text[self.print_len :]
															self.print_len += len(a		)
								# Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
								# which may change with the subsequent token -- there are probably smarter ways to do this!)
								else:
															__UpperCamelCase      :  Any       =   text[self.print_len : text.rfind(" "		) + 1]
															self.print_len += len(a		)

								self.on_finalized_text(a		)







	def   _lowerCamelCase							(   self						:List[str]		)						->					List[str]:
								# Flush the cache, if it exists
								if len(self.token_cache		) > 0:
															__UpperCamelCase      :  str       =   self.tokenizer.decode(self.token_cache	,  **self.decode_kwargs		)
															__UpperCamelCase      :  str       =   text[self.print_len :]
															__UpperCamelCase      :  str       =   []
															__UpperCamelCase      :  int       =   0
								else:
															__UpperCamelCase      :  int       =   ""

								__UpperCamelCase      :  Optional[Any]       =   True
								self.on_finalized_text(a	,  stream_end=a		)







	def   _lowerCamelCase							(   self						:Union[str, Any]	,  a						:str	,  a						:bool = False		)						->					str:
								print(a	,  flush=a	,  end="" if not stream_end else None		)







	def   _lowerCamelCase							(   self						:Optional[int]	,  a						:Optional[Any]		)						->					str:
								# This defines a "chinese character" as anything in the CJK Unicode block:
								#   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
								#
								# Note that the CJK Unicode block is NOT all Japanese and Korean characters,
								# despite its name. The modern Korean Hangul alphabet is a different block,
								# as is Japanese Hiragana and Katakana. Those alphabets are used to write
								# space-separated words, so they are not treated specially and handled
								# like the all of the other languages.
								if (
								    (cp >= 0x4e_00 and cp <= 0x9f_ff)
								    or (cp >= 0x34_00 and cp <= 0x4d_bf)  #
								    or (cp >= 0x2_00_00 and cp <= 0x2_a6_df)  #
								    or (cp >= 0x2_a7_00 and cp <= 0x2_b7_3f)  #
								    or (cp >= 0x2_b7_40 and cp <= 0x2_b8_1f)  #
								    or (cp >= 0x2_b8_20 and cp <= 0x2_ce_af)  #
								    or (cp >= 0xf9_00 and cp <= 0xfa_ff)
								    or (cp >= 0x2_f8_00 and cp <= 0x2_fa_1f)  #
								):  #
															return True

								return False




class 					lowerCamelCase__				(							__lowercase):




	'''simple docstring'''




	def __init__(   self						:Any	,  a						:"AutoTokenizer"	,  a						:bool = False	,  a						:Optional[float] = None	,  **a						:Dict		)						->					str:
								super().__init__(a	,  a	,  **a		)
								__UpperCamelCase      :  str       =   Queue()
								__UpperCamelCase      :  Dict       =   None
								__UpperCamelCase      :  Optional[int]       =   timeout







	def   _lowerCamelCase							(   self						:List[Any]	,  a						:str	,  a						:bool = False		)						->					Dict:
								self.text_queue.put(a	,  timeout=self.timeout		)
								if stream_end:
															self.text_queue.put(self.stop_signal	,  timeout=self.timeout		)







	def __iter__(   self						:Tuple		)						->					Tuple:
								return self







	def   _lowerCamelCase							(   self						:List[Any]		)						->					Dict:
								__UpperCamelCase      :  List[Any]       =   self.text_queue.get(timeout=self.timeout		)
								if value == self.stop_signal:
															raise StopIteration()
								else:
															return value