



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_tf_available():
				from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_VISION_2_SEQ_MAPPING

if is_torch_available():
				import torch

				from ..models.auto.modeling_auto import MODEL_FOR_VISION_2_SEQ_MAPPING

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





			def __init__(self							,    *__magic_name__							,    **__magic_name__				):
						super().__init__(*__magic_name__							,    **__magic_name__				)
						requires_backends(self							,    """vision"""				)
						self.check_model_type(
						    TF_MODEL_FOR_VISION_2_SEQ_MAPPING if self.framework == """tf""" else MODEL_FOR_VISION_2_SEQ_MAPPING				)





			def        _snake_case    (self							,    __magic_name__=None							,    __magic_name__=None							,    __magic_name__=None				):
						lowerCamelCase__ :      Union[str, Any]								=		{}
						lowerCamelCase__ :      Union[str, Any]								=		{}

						if prompt is not None:
									lowerCamelCase__ :      List[str]								=		prompt

						if generate_kwargs is not None:
									lowerCamelCase__ :      int								=		generate_kwargs
						if max_new_tokens is not None:
									if "generate_kwargs" not in forward_kwargs:
												lowerCamelCase__ :      str								=		{}
									if "max_new_tokens" in forward_kwargs["generate_kwargs"]:
												raise ValueError(
												    """'max_new_tokens' is defined twice, once in 'generate_kwargs' and once as a direct parameter,"""
												    """ please use only one"""				)
									lowerCamelCase__ :      List[str]								=		max_new_tokens
						return preprocess_params, forward_kwargs, {}





			def __call__(self							,    __magic_name__							,    **__magic_name__				):
						return super().__call__(__magic_name__							,    **__magic_name__				)





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__=None				):
						lowerCamelCase__ :      Optional[int]								=		load_image(__magic_name__				)

						if prompt is not None:
									if not isinstance(__magic_name__							,    __magic_name__				):
												raise ValueError(
												    f"Received an invalid text input, got - {type(__magic_name__				)} - but expected a single string. "
												    """Note also that one single text can be provided for conditional image to text generation."""				)

									lowerCamelCase__ :      int								=		self.model.config.model_type

									if model_type == "git":
												lowerCamelCase__ :      List[str]								=		self.image_processor(images=__magic_name__							,    return_tensors=self.framework				)
												lowerCamelCase__ :      Any								=		self.tokenizer(text=__magic_name__							,    add_special_tokens=__magic_name__				).input_ids
												lowerCamelCase__ :      List[str]								=		[self.tokenizer.cls_token_id] + input_ids
												lowerCamelCase__ :      Optional[Any]								=		torch.tensor(__magic_name__				).unsqueeze(0				)
												model_inputs.update({"""input_ids""": input_ids}				)

									elif model_type == "pix2struct":
												lowerCamelCase__ :      Any								=		self.image_processor(images=__magic_name__							,    header_text=__magic_name__							,    return_tensors=self.framework				)

									elif model_type != "vision-encoder-decoder":
												# vision-encoder-decoder does not support conditional generation
												lowerCamelCase__ :      str								=		self.image_processor(images=__magic_name__							,    return_tensors=self.framework				)
												lowerCamelCase__ :      Dict								=		self.tokenizer(__magic_name__							,    return_tensors=self.framework				)
												model_inputs.update(__magic_name__				)

									else:
												raise ValueError(f"Model type {model_type} does not support conditional text generation"				)

						else:
									lowerCamelCase__ :      List[Any]								=		self.image_processor(images=__magic_name__							,    return_tensors=self.framework				)

						if self.model.config.model_type == "git" and prompt is None:
									lowerCamelCase__ :      Optional[Any]								=		None

						return model_inputs





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__=None				):
						# Git model sets `model_inputs["input_ids"] = None` in `preprocess` (when `prompt=None`). In batch model, the
						# pipeline will group them into a list of `None`, which fail `_forward`. Avoid this by checking it first.
						if (
						    "input_ids" in model_inputs
						    and isinstance(model_inputs["""input_ids"""]							,    __magic_name__				)
						    and all(x is None for x in model_inputs["""input_ids"""]				)
						):
									lowerCamelCase__ :      Optional[Any]								=		None

						if generate_kwargs is None:
									lowerCamelCase__ :      int								=		{}
						# FIXME: We need to pop here due to a difference in how `generation.py` and `generation.tf_utils.py`
						#  parse inputs. In the Tensorflow version, `generate` raises an error if we don't use `input_ids` whereas
						#  the PyTorch version matches it with `self.model.main_input_name` or `self.model.encoder.main_input_name`
						#  in the `_prepare_model_inputs` method.
						lowerCamelCase__ :      List[str]								=		model_inputs.pop(self.model.main_input_name				)
						lowerCamelCase__ :      Union[str, Any]								=		self.model.generate(__magic_name__							,    **__magic_name__							,    **__magic_name__				)
						return model_outputs





			def        _snake_case    (self							,    __magic_name__				):
						lowerCamelCase__ :      str								=		[]
						for output_ids in model_outputs:
									lowerCamelCase__ :      int								=		{
									    """generated_text""": self.tokenizer.decode(
									        __magic_name__							,    skip_special_tokens=__magic_name__							,    )
									}
									records.append(__magic_name__				)
						return records








import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


_lowercase     							=						logging.get_logger(__name__)






def      _A				(UpperCamelCase		:      Dict      ,			UpperCamelCase		:      Optional[int] )  ->Tuple:



			'''simple docstring'''







			lowerCamelCase__ :      List[str]								=		nn.functional.normalize(UpperCamelCase )
			lowerCamelCase__ :      Tuple								=		nn.functional.normalize(UpperCamelCase )
			return torch.mm(UpperCamelCase      ,			normalized_text_embeds.t() )
class 				__A					( A_      ):
			UpperCamelCase					:Optional[Any]     =			CLIPConfig

			UpperCamelCase					:Optional[Any]     =			['''CLIPEncoderLayer''']





			def __init__(self							,    __magic_name__				):
						super().__init__(__magic_name__				)

						lowerCamelCase__ :      Union[str, Any]								=		CLIPVisionModel(config.vision_config				)
						lowerCamelCase__ :      Any								=		nn.Linear(config.vision_config.hidden_size							,    config.projection_dim							,    bias=__magic_name__				)

						lowerCamelCase__ :      Any								=		nn.Parameter(torch.ones(17							,    config.projection_dim				)							,    requires_grad=__magic_name__				)
						lowerCamelCase__ :      Union[str, Any]								=		nn.Parameter(torch.ones(3							,    config.projection_dim				)							,    requires_grad=__magic_name__				)

						lowerCamelCase__ :      Any								=		nn.Parameter(torch.ones(17				)							,    requires_grad=__magic_name__				)
						lowerCamelCase__ :      Optional[Any]								=		nn.Parameter(torch.ones(3				)							,    requires_grad=__magic_name__				)





			@torch.no_grad()
			def        _snake_case    (self							,    __magic_name__							,    __magic_name__				):
						lowerCamelCase__ :      Any								=		self.vision_model(__magic_name__				)[1]  # pooled_output
						lowerCamelCase__ :      Tuple								=		self.visual_projection(__magic_name__				)

						# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
						lowerCamelCase__ :      Any								=		cosine_distance(__magic_name__							,    self.special_care_embeds				).cpu().float().numpy()
						lowerCamelCase__ :      str								=		cosine_distance(__magic_name__							,    self.concept_embeds				).cpu().float().numpy()

						lowerCamelCase__ :      int								=		[]
						lowerCamelCase__ :      str								=		image_embeds.shape[0]
						for i in range(__magic_name__				):
									lowerCamelCase__ :      str								=		{"""special_scores""": {}, """special_care""": [], """concept_scores""": {}, """bad_concepts""": []}

									# increase this value to create a stronger `nfsw` filter
									# at the cost of increasing the possibility of filtering benign images
									lowerCamelCase__ :      Optional[Any]								=		0.0

									for concept_idx in range(len(special_cos_dist[0]				)				):
												lowerCamelCase__ :      List[str]								=		special_cos_dist[i][concept_idx]
												lowerCamelCase__ :      Tuple								=		self.special_care_embeds_weights[concept_idx].item()
												lowerCamelCase__ :      str								=		round(concept_cos - concept_threshold + adjustment							,    3				)
												if result_img["special_scores"][concept_idx] > 0:
															result_img["special_care"].append({concept_idx, result_img["""special_scores"""][concept_idx]}				)
															lowerCamelCase__ :      List[str]								=		0.01

									for concept_idx in range(len(cos_dist[0]				)				):
												lowerCamelCase__ :      str								=		cos_dist[i][concept_idx]
												lowerCamelCase__ :      Dict								=		self.concept_embeds_weights[concept_idx].item()
												lowerCamelCase__ :      int								=		round(concept_cos - concept_threshold + adjustment							,    3				)
												if result_img["concept_scores"][concept_idx] > 0:
															result_img["bad_concepts"].append(__magic_name__				)

									result.append(__magic_name__				)

						lowerCamelCase__ :      Tuple								=		[len(res["""bad_concepts"""]				) > 0 for res in result]

						return images, has_nsfw_concepts





			@torch.no_grad()
			def        _snake_case    (self							,    __magic_name__							,    __magic_name__				):
						lowerCamelCase__ :      Union[str, Any]								=		self.vision_model(__magic_name__				)[1]  # pooled_output
						lowerCamelCase__ :      str								=		self.visual_projection(__magic_name__				)

						lowerCamelCase__ :      Optional[Any]								=		cosine_distance(__magic_name__							,    self.special_care_embeds				)
						lowerCamelCase__ :      Dict								=		cosine_distance(__magic_name__							,    self.concept_embeds				)

						# increase this value to create a stronger `nsfw` filter
						# at the cost of increasing the possibility of filtering benign images
						lowerCamelCase__ :      Any								=		0.0

						lowerCamelCase__ :      Dict								=		special_cos_dist - self.special_care_embeds_weights + adjustment
						# special_scores = special_scores.round(decimals=3)
						lowerCamelCase__ :      Union[str, Any]								=		torch.any(special_scores > 0							,    dim=1				)
						lowerCamelCase__ :      Optional[Any]								=		special_care * 0.01
						lowerCamelCase__ :      Dict								=		special_adjustment.unsqueeze(1				).expand(-1							,    cos_dist.shape[1]				)

						lowerCamelCase__ :      Tuple								=		(cos_dist - self.concept_embeds_weights) + special_adjustment
						# concept_scores = concept_scores.round(decimals=3)
						lowerCamelCase__ :      Any								=		torch.any(concept_scores > 0							,    dim=1				)

						return images, has_nsfw_concepts



