


'''simple docstring'''



import importlib.metadata
import warnings
from copy import deepcopy

from packaging import version

from ..utils import logging
from .import_utils import is_accelerate_available, is_bitsandbytes_available


if is_bitsandbytes_available():
	import bitsandbytes as bnb
	import torch
	import torch.nn as nn

	from ..pytorch_utils import ConvaD

if is_accelerate_available():
	from accelerate import init_empty_weights
	from accelerate.utils import find_tied_parameters

__lowercase	:						Any       =	logging.get_logger(__name__)






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   Union[str, Any]						,      _SCREAMING_SNAKE_CASE      :   Any						,      _SCREAMING_SNAKE_CASE      :   str						,      _SCREAMING_SNAKE_CASE      :   Optional[Any]=None						,      _SCREAMING_SNAKE_CASE      :   List[str]=None   ):
					# Recurse if needed
					if "." in tensor_name:
										__a			:       List[str]					    =  tensor_name.split('.'   )
										for split in splits[:-1]:
															__a			:       List[Any]					    =  getattr(__lowerCamelCase						,      __lowerCamelCase   )
															if new_module is None:
																				raise ValueError(F"""{module} has no attribute {split}."""   )
															__a			:       Optional[int]					    =  new_module
										__a			:       Tuple					    =  splits[-1]

					if tensor_name not in module._parameters and tensor_name not in module._buffers:
										raise ValueError(F"""{module} does not have a parameter or a buffer named {tensor_name}."""   )
					__a			:       List[Any]					    =  tensor_name in module._buffers
					__a			:       Any					    =  getattr(__lowerCamelCase						,      __lowerCamelCase   )

					if old_value.device == torch.device('meta'   ) and device not in ["meta", torch.device('meta'   )] and value is None:
										raise ValueError(F"""{tensor_name} is on the meta device, we need a `value` to put in on {device}."""   )

					__a			:       Any					    =  False
					__a			:       List[str]					    =  False
					if is_buffer or not is_bitsandbytes_available():
										__a			:       Union[str, Any]					    =  False
										__a			:       str					    =  False
					else:
										__a			:       Any					    =  hasattr(bnb.nn						,      'Params4bit'   ) and isinstance(module._parameters[tensor_name]						,      bnb.nn.Paramsabit   )
										__a			:       int					    =  isinstance(module._parameters[tensor_name]						,      bnb.nn.IntaParams   )

					if is_abit or is_abit:
										__a			:       Tuple					    =  module._parameters[tensor_name]
										if param.device.type != "cuda":
															if value is None:
																				__a			:       List[Any]					    =  old_value.to(__lowerCamelCase   )
															elif isinstance(__lowerCamelCase						,      torch.Tensor   ):
																				__a			:       str					    =  value.to('cpu'   )
																				if value.dtype == torch.inta:
																									__a			:       str					    =  version.parse(importlib.metadata.version('bitsandbytes'   )   ) > version.parse(
																									    '0.37.2'   )
																									if not is_abit_serializable:
																														raise ValueError(
																														    'Detected int8 weights but the version of bitsandbytes is not compatible with int8 serialization. '
																														    'Make sure to download the latest `bitsandbytes` version. `pip install --upgrade bitsandbytes`.'   )
															else:
																				__a			:       Tuple					    =  torch.tensor(__lowerCamelCase						,      device='cpu'   )

															# Support models using `Conv1D` in place of `nn.Linear` (e.g. gpt2) by transposing the weight matrix prior to quantization.
															# Since weights are saved in the correct "orientation", we skip transposing when loading.
															if issubclass(module.source_cls						,      __lowerCamelCase   ) and fpaa_statistics is None:
																				__a			:       str					    =  new_value.T

															__a			:       str					    =  old_value.__dict__
															if is_abit:
																				__a			:       Union[str, Any]					    =  bnb.nn.IntaParams(__lowerCamelCase						,      requires_grad=__lowerCamelCase						,      **__lowerCamelCase   ).to(__lowerCamelCase   )
															elif is_abit:
																				__a			:       Optional[int]					    =  bnb.nn.Paramsabit(__lowerCamelCase						,      requires_grad=__lowerCamelCase						,      **__lowerCamelCase   ).to(__lowerCamelCase   )

															__a			:       Tuple					    =  new_value
															if fpaa_statistics is not None:
																				setattr(module.weight						,      'SCB'						,      fpaa_statistics.to(__lowerCamelCase   )   )

					else:
										if value is None:
															__a			:       Union[str, Any]					    =  old_value.to(__lowerCamelCase   )
										elif isinstance(__lowerCamelCase						,      torch.Tensor   ):
															__a			:       List[Any]					    =  value.to(__lowerCamelCase   )
										else:
															__a			:       Optional[Any]					    =  torch.tensor(__lowerCamelCase						,      device=__lowerCamelCase   )

										if is_buffer:
															__a			:       Tuple					    =  new_value
										else:
															__a			:       List[Any]					    =  nn.Parameter(__lowerCamelCase						,      requires_grad=old_value.requires_grad   )
															__a			:       int					    =  new_value






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   str						,      _SCREAMING_SNAKE_CASE      :   List[str]=None						,      _SCREAMING_SNAKE_CASE      :   Any=None						,      _SCREAMING_SNAKE_CASE      :   Dict=None						,      _SCREAMING_SNAKE_CASE      :   Any=False   ):
					for name, module in model.named_children():
										if current_key_name is None:
															__a			:       Tuple					    =  []
										current_key_name.append(__lowerCamelCase   )

										if (isinstance(__lowerCamelCase						,      nn.Linear   ) or isinstance(__lowerCamelCase						,      __lowerCamelCase   )) and name not in modules_to_not_convert:
															# Check if the current key is not in the `modules_to_not_convert`
															if not any(key in '.'.join(__lowerCamelCase   ) for key in modules_to_not_convert   ):
																				with init_empty_weights():
																									if isinstance(__lowerCamelCase						,      __lowerCamelCase   ):
																														__a			:       str					    =  module.weight.shape
																									else:
																														__a			:       str					    =  module.in_features
																														__a			:       Union[str, Any]					    =  module.out_features

																									if quantization_config.quantization_method() == "llm_int8":
																														__a			:       Union[str, Any]					    =  bnb.nn.LinearabitLt(
																														    __lowerCamelCase						,      __lowerCamelCase						,      module.bias is not None						,      has_fpaa_weights=quantization_config.llm_inta_has_fpaa_weight						,      threshold=quantization_config.llm_inta_threshold						,      )
																														__a			:       List[Any]					    =  True
																									else:
																														if (
																														    quantization_config.llm_inta_skip_modules is not None
																														    and name in quantization_config.llm_inta_skip_modules
																														):
																																			pass
																														else:
																																			__a			:       Tuple					    =  bnb.nn.Linearabit(
																																			    __lowerCamelCase						,      __lowerCamelCase						,      module.bias is not None						,      quantization_config.bnb_abit_compute_dtype						,      compress_statistics=quantization_config.bnb_abit_use_double_quant						,      quant_type=quantization_config.bnb_abit_quant_type						,      )
																																			__a			:       Optional[int]					    =  True
                    # Store the module class in case we need to transpose the weight later
																									__a			:       str					    =  type(__lowerCamelCase   )
																									# Force requires grad to False to avoid unexpected errors
																									model._modules[name].requires_grad_(__lowerCamelCase   )
										if len(list(module.children()   )   ) > 0:
															__a			:       Any					    =  _replace_with_bnb_linear(
															    __lowerCamelCase						,      __lowerCamelCase						,      __lowerCamelCase						,      __lowerCamelCase						,      has_been_replaced=__lowerCamelCase						,      )
										# Remove the last key for recursion
										current_key_name.pop(-1   )
					return model, has_been_replaced






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   Tuple						,      _SCREAMING_SNAKE_CASE      :   Tuple=None						,      _SCREAMING_SNAKE_CASE      :   Optional[int]=None						,      _SCREAMING_SNAKE_CASE      :   Dict=None   ):
					__a			:       Dict					    =  ['''lm_head'''] if modules_to_not_convert is None else modules_to_not_convert
					__a			:       str					    =  _replace_with_bnb_linear(
					    __lowerCamelCase						,      __lowerCamelCase						,      __lowerCamelCase						,      __lowerCamelCase   )

					if not has_been_replaced:
										logger.warning(
										    'You are loading your model in 8bit or 4bit but no linear modules were found in your model.'
										    ' Please double check your model architecture, or submit an issue on github if you think this is'
										    ' a bug.'   )

					return model






def   lowerCamelCase							(*_SCREAMING_SNAKE_CASE      :   List[str]						,      **_SCREAMING_SNAKE_CASE      :   int   ):
					warnings.warn(
					    '`replace_8bit_linear` will be deprecated in a future version, please use `replace_with_bnb_linear` instead'						,      __lowerCamelCase						,      )
					return replace_with_bnb_linear(*__lowerCamelCase						,      **__lowerCamelCase   )






def   lowerCamelCase							(*_SCREAMING_SNAKE_CASE      :   Any						,      **_SCREAMING_SNAKE_CASE      :   int   ):
					warnings.warn(
					    '`set_module_8bit_tensor_to_device` will be deprecated in a future version, please use `set_module_quantized_tensor_to_device` instead'						,      __lowerCamelCase						,      )
					return set_module_quantized_tensor_to_device(*__lowerCamelCase						,      **__lowerCamelCase   )






def   lowerCamelCase							(_SCREAMING_SNAKE_CASE      :   str   ):
					__a			:       List[str]					    =  deepcopy(__lowerCamelCase   )  # this has 0 cost since it is done inside `init_empty_weights` context manager`
					tied_model.tie_weights()

					__a			:       Union[str, Any]					    =  find_tied_parameters(__lowerCamelCase   )
					# For compatibility with Accelerate < 0.18
					if isinstance(__lowerCamelCase						,      __lowerCamelCase   ):
										__a			:       Optional[int]					    =  sum(list(tied_params.values()   )						,      []   ) + list(tied_params.keys()   )
					else:
										__a			:       Union[str, Any]					    =  sum(__lowerCamelCase						,      []   )
					__a			:       Dict					    =  len(__lowerCamelCase   ) > 0

					# Check if it is a base model
					__a			:       Optional[Any]					    =  not hasattr(__lowerCamelCase						,      model.base_model_prefix   )

					# Ignore this for base models (BertModel, GPT2Model, etc.)
					if (not has_tied_params) and is_base_model:
										return []

					# otherwise they have an attached head
					__a			:       Optional[int]					    =  list(model.named_children()   )
					__a			:       int					    =  [list_modules[-1][0]]

					# add last module together with tied weights
					__a			:       int					    =  set(__lowerCamelCase   ) - set(__lowerCamelCase   )
					__a			:       Dict					    =  list(set(__lowerCamelCase   )   ) + list(__lowerCamelCase   )

					# remove ".weight" from the keys
					__a			:       Dict					    =  ['''.weight''', '''.bias''']
					__a			:       Dict					    =  []
					for name in list_untouched:
										for name_to_remove in names_to_remove:
															if name_to_remove in name:
																				__a			:       List[str]					    =  name.replace(__lowerCamelCase						,      ''   )
										filtered_module_names.append(__lowerCamelCase   )

					return filtered_module_names








"""simple docstring"""







from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
							from PIL import Image

							from ..image_utils import load_image

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
							from ..tf_utils import stable_softmax

lowerCAmelCase_   						=   logging.get_logger(__name__)
@add_end_docstrings(A_    )
class        __A       (	A_    ):







		'''simple docstring'''





		def __init__(	self						:  List[str]			,**_snake_case						:  Dict						)					->		List[Any]:






				"""simple docstring"""



				super().__init__(**_snake_case						)

				requires_backends(self			,'''vision'''						)
				self.check_model_type(
				    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
				    if self.framework == '''tf'''
				    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING						)







		def __call__(	self						:  Optional[int]			,_snake_case						:  Union[str, List[str], "Image", List["Image"]]			,**_snake_case						:  int						)					->		Optional[Any]:






				"""simple docstring"""



				return super().__call__(_snake_case			,**_snake_case						)







		def 					UpperCAmelCase    (	self						:  Dict			,**_snake_case						:  Optional[int]						)					->		List[Any]:






				"""simple docstring"""



				lowercase__   :    List[str]				=			{}
				if "candidate_labels" in kwargs:
						lowercase__   :    Any				=			kwargs['''candidate_labels''']
				if "hypothesis_template" in kwargs:
						lowercase__   :    Optional[Any]				=			kwargs['''hypothesis_template''']

				return preprocess_params, {}, {}







		def 					UpperCAmelCase    (	self						:  Optional[Any]			,_snake_case						:  Optional[int]			,_snake_case						:  Dict=None			,_snake_case						:  Union[str, Any]="This is a photo of {}."						)					->		List[str]:






				"""simple docstring"""



				lowercase__   :    List[Any]				=			load_image(_snake_case						)
				lowercase__   :    int				=			self.image_processor(images=[image]			,return_tensors=self.framework						)
				lowercase__   :    str				=			candidate_labels
				lowercase__   :    Dict				=			[hypothesis_template.format(_snake_case						) for x in candidate_labels]
				lowercase__   :    Any				=			self.tokenizer(_snake_case			,return_tensors=self.framework			,padding=_snake_case						)
				lowercase__   :    Optional[int]				=			[text_inputs]
				return inputs







		def 					UpperCAmelCase    (	self						:  Optional[Any]			,_snake_case						:  Optional[int]						)					->		List[Any]:






				"""simple docstring"""



				lowercase__   :    Optional[int]				=			model_inputs.pop('''candidate_labels'''						)
				lowercase__   :    Union[str, Any]				=			model_inputs.pop('''text_inputs'''						)
				if isinstance(text_inputs[0]			,_snake_case						):
						lowercase__   :    List[str]				=			text_inputs[0]
				else:
						# Batching case.
						lowercase__   :    int				=			text_inputs[0][0]

				lowercase__   :    Tuple				=			self.model(**_snake_case			,**_snake_case						)

				lowercase__   :    Union[str, Any]				=			{
				    '''candidate_labels''': candidate_labels,
				    '''logits''': outputs.logits_per_image,
				}
				return model_outputs







		def 					UpperCAmelCase    (	self						:  Any			,_snake_case						:  Tuple						)					->		Any:






				"""simple docstring"""



				lowercase__   :    Dict				=			model_outputs.pop('''candidate_labels'''						)
				lowercase__   :    Optional[Any]				=			model_outputs['''logits'''][0]
				if self.framework == "pt":
						lowercase__   :    Optional[int]				=			logits.softmax(dim=-1						).squeeze(-1						)
						lowercase__   :    Tuple				=			probs.tolist()
						if not isinstance(_snake_case			,_snake_case						):
								lowercase__   :    Any				=			[scores]
				elif self.framework == "tf":
						lowercase__   :    List[str]				=			stable_softmax(_snake_case			,axis=-1						)
						lowercase__   :    Optional[Any]				=			probs.numpy().tolist()
				else:
						raise ValueError(f"""Unsupported framework: {self.framework}"""						)

				lowercase__   :    Union[str, Any]				=			[
				    {'''score''': score, '''label''': candidate_label}
				    for score, candidate_label in sorted(zip(_snake_case			,_snake_case						)			,key=lambda _snake_case						: -x[0]						)
				]
				return result
