'''simple docstring'''

import warnings
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__a:			int	       =       logging.get_logger(__name__)



class       UpperCAmelCase  (						a__ ):





       '''simple docstring'''



       SCREAMING_SNAKE_CASE 		=   ["input_ids", "attention_mask"]

       def __init__(						self     ,       __lowerCAmelCase="</s>"     ,       __lowerCAmelCase="<unk>"     ,       __lowerCAmelCase="<pad>"     ,       __lowerCAmelCase=125     ,       __lowerCAmelCase=None     ,       **__lowerCAmelCase     ,       )  ->  None:
        # Add extra_ids to the special token list
        if extra_ids > 0 and additional_special_tokens is None:
         lowercase__    :   Tuple						= [F"""<extra_id_{i}>""" for i in range(__lowerCAmelCase			)]
        elif extra_ids > 0 and additional_special_tokens is not None:
         # Check that we have the right number of extra_id special tokens
         lowercase__    :   Dict						= len(set(filter(lambda __lowerCAmelCase			: bool('''extra_id''' in str(__lowerCAmelCase			)			)     ,       __lowerCAmelCase			)			)			)
         if extra_tokens != extra_ids:
          raise ValueError(
              F"""Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are"""
              ''' provided to ByT5Tokenizer. In this case the additional_special_tokens must include the'''
              ''' extra_ids tokens'''			)

        lowercase__    :   List[str]						= AddedToken(__lowerCAmelCase     ,       lstrip=__lowerCAmelCase     ,       rstrip=__lowerCAmelCase			) if isinstance(__lowerCAmelCase     ,       __lowerCAmelCase			) else pad_token
        lowercase__    :   Any						= AddedToken(__lowerCAmelCase     ,       lstrip=__lowerCAmelCase     ,       rstrip=__lowerCAmelCase			) if isinstance(__lowerCAmelCase     ,       __lowerCAmelCase			) else eos_token
        lowercase__    :   List[str]						= AddedToken(__lowerCAmelCase     ,       lstrip=__lowerCAmelCase     ,       rstrip=__lowerCAmelCase			) if isinstance(__lowerCAmelCase     ,       __lowerCAmelCase			) else unk_token

        super().__init__(
            eos_token=__lowerCAmelCase     ,       unk_token=__lowerCAmelCase     ,       pad_token=__lowerCAmelCase     ,       extra_ids=__lowerCAmelCase     ,       additional_special_tokens=__lowerCAmelCase     ,       **__lowerCAmelCase     ,       )

        lowercase__    :   Any						= extra_ids

        lowercase__    :   Any						= 2**8  # utf is 8 bits

        # define special tokens dict
        lowercase__    :   Dict[int, str]						= {
            self.pad_token: 0,
            self.eos_token: 1,
            self.unk_token: 2,
        }
        lowercase__    :   Tuple						= len(self.special_tokens_encoder			)
        lowercase__    :   Tuple						= len(__lowerCAmelCase			)
        for i, token in enumerate(__lowerCAmelCase			):
         lowercase__    :   Optional[Any]						= self.vocab_size + i - n
        lowercase__    :   Dict[str, int]						= {v: k for k, v in self.special_tokens_encoder.items()}

       @property
       def _lowerCAmelCase(						self			)  ->  List[str]:
        return self._utf_vocab_size + self._num_special_tokens + self._extra_ids

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase     ,       __lowerCAmelCase = None     ,       __lowerCAmelCase = False			)  ->  List[int]:
        if already_has_special_tokens:
         return super().get_special_tokens_mask(
             token_ids_a=__lowerCAmelCase     ,       token_ids_a=__lowerCAmelCase     ,       already_has_special_tokens=__lowerCAmelCase			)

        # normal case: some special tokens
        if token_ids_a is None:
         return ([0] * len(__lowerCAmelCase			)) + [1]
        return ([0] * len(__lowerCAmelCase			)) + [1] + ([0] * len(__lowerCAmelCase			)) + [1]

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase			)  ->  List[int]:
        if len(__lowerCAmelCase			) > 0 and token_ids[-1] == self.eos_token_id:
         warnings.warn(
             F"""This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated"""
             ''' eos tokens being added.'''			)
         return token_ids
        else:
         return token_ids + [self.eos_token_id]

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase     ,       __lowerCAmelCase = None			)  ->  List[int]:
        lowercase__    :   Optional[Any]						= [self.eos_token_id]

        if token_ids_a is None:
         return len(token_ids_a + eos			) * [0]
        return len(token_ids_a + eos + token_ids_a + eos			) * [0]

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase     ,       __lowerCAmelCase = None			)  ->  List[int]:
        lowercase__    :   Optional[Any]						= self._add_eos_if_not_present(__lowerCAmelCase			)
        if token_ids_a is None:
         return token_ids_a
        else:
         lowercase__    :   Dict						= self._add_eos_if_not_present(__lowerCAmelCase			)
         return token_ids_a + token_ids_a

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase			)  ->  List[str]:
        lowercase__    :   Tuple						= [chr(__lowerCAmelCase			) for i in text.encode('''utf-8'''			)]
        return tokens

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase			)  ->  Any:
        if token in self.special_tokens_encoder:
         lowercase__    :   Optional[int]						= self.special_tokens_encoder[token]
        elif token in self.added_tokens_encoder:
         lowercase__    :   Dict						= self.added_tokens_encoder[token]
        elif len(__lowerCAmelCase			) != 1:
         lowercase__    :   Optional[int]						= self.unk_token_id
        else:
         lowercase__    :   Tuple						= ord(__lowerCAmelCase			) + self._num_special_tokens
        return token_id

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase			)  ->  Any:
        if index in self.special_tokens_decoder:
         lowercase__    :   Any						= self.special_tokens_decoder[index]
        else:
         lowercase__    :   List[str]						= chr(index - self._num_special_tokens			)
        return token

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase			)  ->  List[Any]:
        lowercase__    :   Optional[Any]						= b''''''
        for token in tokens:
         if token in self.special_tokens_decoder:
          lowercase__    :   Union[str, Any]						= self.special_tokens_decoder[token].encode('''utf-8'''			)
         elif token in self.added_tokens_decoder:
          lowercase__    :   List[Any]						= self.special_tokens_decoder[token].encode('''utf-8'''			)
         elif token in self.special_tokens_encoder:
          lowercase__    :   Dict						= token.encode('''utf-8'''			)
         elif token in self.added_tokens_encoder:
          lowercase__    :   List[str]						= token.encode('''utf-8'''			)
         else:
          lowercase__    :   Dict						= bytes([ord(__lowerCAmelCase			)]			)
         bstring += tok_string
        lowercase__    :   Optional[Any]						= bstring.decode('''utf-8'''     ,       errors='''ignore'''			)
        return string


       def _lowerCAmelCase(						self     ,       __lowerCAmelCase     ,       __lowerCAmelCase = None			)  ->  Tuple[str]:
        return ()


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

__a:			Optional[Any]	       =       logging.get_logger(__name__)



@add_end_docstrings(a__ )
class       UpperCAmelCase  (						a__ ):





       '''simple docstring'''



       def __init__(						self     ,       **__lowerCAmelCase			)  ->  int:
        super().__init__(**__lowerCAmelCase			)

        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     ,       __lowerCAmelCase     ,       **__lowerCAmelCase			)  ->  List[str]:
        return super().__call__(__lowerCAmelCase     ,       **__lowerCAmelCase			)

       def _lowerCAmelCase(						self     ,       **__lowerCAmelCase			)  ->  Optional[Any]:
        lowercase__    :   str						= {}
        if "candidate_labels" in kwargs:
         lowercase__    :   str						= kwargs['''candidate_labels''']
        if "hypothesis_template" in kwargs:
         lowercase__    :   str						= kwargs['''hypothesis_template''']

        return preprocess_params, {}, {}

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

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase			)  ->  Optional[Any]:
        lowercase__    :   Any						= model_inputs.pop('''candidate_labels'''			)
        lowercase__    :   int						= model_inputs.pop('''text_inputs'''			)
        if isinstance(text_inputs[0]     ,       __lowerCAmelCase			):
         lowercase__    :   Union[str, Any]						= text_inputs[0]
        else:
         # Batching case.
         lowercase__    :   Optional[Any]						= text_inputs[0][0]

        lowercase__    :   Any						= self.model(**__lowerCAmelCase     ,       **__lowerCAmelCase			)

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


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

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

