




'''simple docstring'''
from typing import List, Union

import numpy as np

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, logging
from .base import PIPELINE_INIT_ARGS, ArgumentHandler, ChunkPipeline


_SCREAMING_SNAKE_CASE			=						logging.get_logger(__name__)
class   __UpperCAmelCase				(     lowerCAmelCase				):



					'''simple docstring'''





					def     __snake_case						(	self     :						Optional[int]      ,       _lowercase     :						List[Any])    ->							Any:
								if isinstance(_lowercase      ,       _lowercase):
											A_									= [label.strip() for label in labels.split(',') if label.strip()]
								return labels



					def __call__(	self     :						Any      ,       _lowercase     :						List[str]      ,       _lowercase     :						List[Any]      ,       _lowercase     :						Tuple)    ->							str:
								if len(_lowercase) == 0 or len(_lowercase) == 0:
											raise ValueError('You must include at least one label and at least one sequence.')
								if hypothesis_template.format(labels[0]) == hypothesis_template:
											raise ValueError(
											    (
											        'The provided hypothesis_template "{}" was not able to be formatted with the target labels. '
											        'Make sure the passed template includes formatting syntax such as {{}} where the label should go.'
											    ).format(_lowercase))

								if isinstance(_lowercase      ,       _lowercase):
											A_									= [sequences]

								A_									= []
								for sequence in sequences:
											sequence_pairs.extend([[sequence, hypothesis_template.format(_lowercase)] for label in labels])

								return sequence_pairs, sequences





@add_end_docstrings(lowerCAmelCase				)
class   __UpperCAmelCase				(     lowerCAmelCase				):



					'''simple docstring'''





					def __init__(	self     :						Dict      ,       _lowercase     :						Any=ZeroShotClassificationArgumentHandler()      ,       *_lowercase     :						Tuple      ,       **_lowercase     :						Optional[Any])    ->							List[str]:
								A_									= args_parser
								super().__init__(*_lowercase      ,       **_lowercase)
								if self.entailment_id == -1:
											logger.warning(
											    'Failed to determine \'entailment\' label id from the label2id mapping in the model config. Setting to '
											    '-1. Define a descriptive label2id mapping in the model config to ensure correct outputs.')



					@property
					def     __snake_case						(	self     :						Any)    ->							Union[str, Any]:
								for label, ind in self.model.config.labelaid.items():
											if label.lower().startswith('entail'):
														return ind
								return -1



					def     __snake_case						(	self     :						List[str]      ,       _lowercase     :						Optional[Any]      ,       _lowercase     :						List[Any]=True      ,       _lowercase     :						List[str]=True      ,       _lowercase     :						Union[str, Any]=TruncationStrategy.ONLY_FIRST      ,       **_lowercase     :						int)    ->							str:
								A_									= self.framework
								if self.tokenizer.pad_token is None:
											# Override for tokenizers not supporting padding
											logger.error(
											    'Tokenizer was not supporting padding necessary for zero-shot, attempting to use '
											    ' `pad_token=eos_token`')
											A_									= self.tokenizer.eos_token
								try:
											A_									= self.tokenizer(
											    _lowercase      ,       add_special_tokens=_lowercase      ,       return_tensors=_lowercase      ,       padding=_lowercase      ,       truncation=_lowercase      ,       )
								except Exception as e:
											if "too short" in str(_lowercase):
														# tokenizers might yell that we want to truncate
														# to a value that is not even reached by the input.
														# In that case we don't want to truncate.
														# It seems there's not a really better way to catch that
														# exception.

														A_									= self.tokenizer(
														    _lowercase      ,       add_special_tokens=_lowercase      ,       return_tensors=_lowercase      ,       padding=_lowercase      ,       truncation=TruncationStrategy.DO_NOT_TRUNCATE      ,       )
											else:
														raise e

								return inputs



					def     __snake_case						(	self     :						List[str]      ,       **_lowercase     :						List[Any])    ->							List[str]:
								if kwargs.get('multi_class'      ,       _lowercase) is not None:
											A_									= kwargs['multi_class']
											logger.warning(
											    'The `multi_class` argument has been deprecated and renamed to `multi_label`. '
											    '`multi_class` will be removed in a future version of Transformers.')
								A_									= {}
								if "candidate_labels" in kwargs:
											A_									= self._args_parser._parse_labels(kwargs['candidate_labels'])
								if "hypothesis_template" in kwargs:
											A_									= kwargs['hypothesis_template']

								A_									= {}
								if "multi_label" in kwargs:
											A_									= kwargs['multi_label']
								return preprocess_params, {}, postprocess_params



					def __call__(	self     :						Optional[int]      ,       _lowercase     :						Union[str, List[str]]      ,       *_lowercase     :						Optional[int]      ,       **_lowercase     :						List[Any]      ,       )    ->							List[str]:
								if len(_lowercase) == 0:
											pass
								elif len(_lowercase) == 1 and "candidate_labels" not in kwargs:
											A_									= args[0]
								else:
											raise ValueError(F'Unable to understand extra arguments {args}')

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



					def     __snake_case						(	self     :						int      ,       _lowercase     :						List[Any]      ,       _lowercase     :						Tuple=None      ,       _lowercase     :						Optional[int]="This example is {}.")    ->							Any:
								A_		, A_									= self._args_parser(_lowercase      ,       _lowercase      ,       _lowercase)

								for i, (candidate_label, sequence_pair) in enumerate(zip(_lowercase      ,       _lowercase)):
											A_									= self._parse_and_tokenize([sequence_pair])

											yield {
											    "candidate_label": candidate_label,
											    "sequence": sequences[0],
											    "is_last": i == len(_lowercase) - 1,
											    **model_input,
											}



					def     __snake_case						(	self     :						int      ,       _lowercase     :						Optional[Any])    ->							Union[str, Any]:
								A_									= inputs['candidate_label']
								A_									= inputs['sequence']
								A_									= {k: inputs[k] for k in self.tokenizer.model_input_names}
								A_									= self.model(**_lowercase)

								A_									= {
								    'candidate_label': candidate_label,
								    'sequence': sequence,
								    'is_last': inputs['is_last'],
								    **outputs,
								}
								return model_outputs



					def     __snake_case						(	self     :						Any      ,       _lowercase     :						List[Any]      ,       _lowercase     :						int=False)    ->							Union[str, Any]:
								A_									= [outputs['candidate_label'] for outputs in model_outputs]
								A_									= [outputs['sequence'] for outputs in model_outputs]
								A_									= np.concatenate([output['logits'].numpy() for output in model_outputs])
								A_									= logits.shape[0]
								A_									= len(_lowercase)
								A_									= N // n
								A_									= logits.reshape((num_sequences, n, -1))

								if multi_label or len(_lowercase) == 1:
											# softmax over the entailment vs. contradiction dim for each label independently
											A_									= self.entailment_id
											A_									= -1 if entailment_id == 0 else 0
											A_									= reshaped_outputs[..., [contradiction_id, entailment_id]]
											A_									= np.exp(_lowercase) / np.exp(_lowercase).sum(-1      ,       keepdims=_lowercase)
											A_									= scores[..., 1]
								else:
											# softmax the "entailment" logits over all candidate labels
											A_									= reshaped_outputs[..., self.entailment_id]
											A_									= np.exp(_lowercase) / np.exp(_lowercase).sum(-1      ,       keepdims=_lowercase)

								A_									= list(reversed(scores[0].argsort()))
								return {
								    "sequence": sequences[0],
								    "labels": [candidate_labels[i] for i in top_inds],
								    "scores": scores[0, top_inds].tolist(),
								}










'''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

_SCREAMING_SNAKE_CASE			=						logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase				)
class   __UpperCAmelCase				(     lowerCAmelCase				):



					'''simple docstring'''





					def __init__(	self     :						int      ,       **_lowercase     :						List[Any])    ->							Union[str, Any]:
								super().__init__(**_lowercase)

								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[Any]      ,       _lowercase     :						Union[str, List[str], "Image", List["Image"]]      ,       **_lowercase     :						List[Any])    ->							Any:
								return super().__call__(_lowercase      ,       **_lowercase)



					def     __snake_case						(	self     :						int      ,       **_lowercase     :						Union[str, Any])    ->							Any:
								A_									= {}
								if "candidate_labels" in kwargs:
											A_									= kwargs['candidate_labels']
								if "hypothesis_template" in kwargs:
											A_									= kwargs['hypothesis_template']

								return preprocess_params, {}, {}



					def     __snake_case						(	self     :						List[str]      ,       _lowercase     :						Tuple      ,       _lowercase     :						Any=None      ,       _lowercase     :						Optional[int]="This is a photo of {}.")    ->							Union[str, Any]:
								A_									= load_image(_lowercase)
								A_									= self.image_processor(images=[image]      ,       return_tensors=self.framework)
								A_									= candidate_labels
								A_									= [hypothesis_template.format(_lowercase) for x in candidate_labels]
								A_									= self.tokenizer(_lowercase      ,       return_tensors=self.framework      ,       padding=_lowercase)
								A_									= [text_inputs]
								return inputs



					def     __snake_case						(	self     :						Optional[int]      ,       _lowercase     :						Tuple)    ->							Optional[int]:
								A_									= model_inputs.pop('candidate_labels')
								A_									= model_inputs.pop('text_inputs')
								if isinstance(text_inputs[0]      ,       _lowercase):
											A_									= text_inputs[0]
								else:
											# Batching case.
											A_									= text_inputs[0][0]

								A_									= self.model(**_lowercase      ,       **_lowercase)

								A_									= {
								    'candidate_labels': candidate_labels,
								    'logits': outputs.logits_per_image,
								}
								return model_outputs



					def     __snake_case						(	self     :						List[str]      ,       _lowercase     :						int)    ->							Optional[int]:
								A_									= model_outputs.pop('candidate_labels')
								A_									= model_outputs['logits'][0]
								if self.framework == "pt":
											A_									= logits.softmax(dim=-1).squeeze(-1)
											A_									= probs.tolist()
											if not isinstance(_lowercase      ,       _lowercase):
														A_									= [scores]
								elif self.framework == "tf":
											A_									= stable_softmax(_lowercase      ,       axis=-1)
											A_									= probs.numpy().tolist()
								else:
											raise ValueError(F'Unsupported framework: {self.framework}')

								A_									= [
								    {'score': score, 'label': candidate_label}
								    for score, candidate_label in sorted(zip(_lowercase      ,       _lowercase)      ,       key=lambda _lowercase: -x[0])
								]
								return result




