

'''simple docstring'''






import os
import re
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


lowerCAmelCase__   		=      logging.get_logger(__name__)

lowerCAmelCase__   		=      {'''vocab_file''': '''spiece.model'''}

lowerCAmelCase__   		=      {
    '''vocab_file''': {
        '''google/bigbird-roberta-base''': '''https://huggingface.co/google/bigbird-roberta-base/resolve/main/spiece.model''',
        '''google/bigbird-roberta-large''': (
            '''https://huggingface.co/google/bigbird-roberta-large/resolve/main/spiece.model'''
        ),
        '''google/bigbird-base-trivia-itc''': (
            '''https://huggingface.co/google/bigbird-base-trivia-itc/resolve/main/spiece.model'''
        ),
    }
}

lowerCAmelCase__   		=      {
    '''google/bigbird-roberta-base''': 4096,
    '''google/bigbird-roberta-large''': 4096,
    '''google/bigbird-base-trivia-itc''': 4096,
}






class      lowercase_      (lowerCamelCase__ ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE       :    Tuple							     =     VOCAB_FILES_NAMES
  SCREAMING_SNAKE_CASE       :    Union[str, Any]							     =     PRETRAINED_VOCAB_FILES_MAP
  SCREAMING_SNAKE_CASE       :    Tuple							     =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  SCREAMING_SNAKE_CASE       :    str							     =     ['input_ids', 'attention_mask']
  SCREAMING_SNAKE_CASE       :    List[int]							     =     []






  def __init__( self  :	str   ,lowercase__  :	int   ,lowercase__  :	Optional[Any]="<unk>"   ,lowercase__  :	str="<s>"   ,lowercase__  :	List[Any]="</s>"   ,lowercase__  :	Dict="<pad>"   ,lowercase__  :	int="[SEP]"   ,lowercase__  :	str="[MASK]"   ,lowercase__  :	Tuple="[CLS]"   ,lowercase__  :	Optional[Dict[str, Any]] = None   ,**lowercase__  :	str   ,):
   __lowercase				    =					AddedToken(lowercase__   ,lstrip=lowercase__   ,rstrip=lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else bos_token
   __lowercase				    =					AddedToken(lowercase__   ,lstrip=lowercase__   ,rstrip=lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else eos_token
   __lowercase				    =					AddedToken(lowercase__   ,lstrip=lowercase__   ,rstrip=lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else unk_token
   __lowercase				    =					AddedToken(lowercase__   ,lstrip=lowercase__   ,rstrip=lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else pad_token
   __lowercase				    =					AddedToken(lowercase__   ,lstrip=lowercase__   ,rstrip=lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else cls_token
   __lowercase				    =					AddedToken(lowercase__   ,lstrip=lowercase__   ,rstrip=lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else sep_token

   # Mask token behave like a normal word, i.e. include the space before it
   __lowercase				    =					AddedToken(lowercase__   ,lstrip=lowercase__   ,rstrip=lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else mask_token

   __lowercase				    =					{} if sp_model_kwargs is None else sp_model_kwargs

   super().__init__(
       bos_token=lowercase__   ,eos_token=lowercase__   ,unk_token=lowercase__   ,pad_token=lowercase__   ,sep_token=lowercase__   ,mask_token=lowercase__   ,cls_token=lowercase__   ,sp_model_kwargs=self.sp_model_kwargs   ,**lowercase__   ,)

   __lowercase				    =					vocab_file

   __lowercase				    =					spm.SentencePieceProcessor(**self.sp_model_kwargs     )
   self.sp_model.Load(lowercase__     )






  @property
  def    SCREAMING_SNAKE_CASE ( self  :	List[str]     ):
   return self.sp_model.get_piece_size()






  def    SCREAMING_SNAKE_CASE ( self  :	Tuple     ):
   __lowercase				    =					{self.convert_ids_to_tokens(lowercase__     ): i for i in range(self.vocab_size     )}
   vocab.update(self.added_tokens_encoder     )
   return vocab






  def __getstate__( self  :	Dict     ):
   __lowercase				    =					self.__dict__.copy()
   __lowercase				    =					None
   return state






  def __setstate__( self  :	str   ,lowercase__  :	Optional[int]     ):
   __lowercase				    =					d

   # for backward compatibility
   if not hasattr(self   ,'''sp_model_kwargs'''     ):
    __lowercase				    =					{}

   __lowercase				    =					spm.SentencePieceProcessor(**self.sp_model_kwargs     )
   self.sp_model.Load(self.vocab_file     )






  def    SCREAMING_SNAKE_CASE ( self  :	Tuple   ,lowercase__  :	str     ):
   return self.sp_model.encode(lowercase__   ,out_type=lowercase__     )






  def    SCREAMING_SNAKE_CASE ( self  :	Union[str, Any]   ,lowercase__  :	Dict     ):
   return self.sp_model.piece_to_id(lowercase__     )






  def    SCREAMING_SNAKE_CASE ( self  :	int   ,lowercase__  :	List[str]     ):
   __lowercase				    =					self.sp_model.IdToPiece(lowercase__     )
   return token






  def    SCREAMING_SNAKE_CASE ( self  :	str   ,lowercase__  :	Tuple     ):
   __lowercase				    =					[]
   __lowercase				    =					''''''
   __lowercase				    =					False
   for token in tokens:
    # make sure that special tokens are not decoded using sentencepiece model
    if token in self.all_special_tokens:
     if not prev_is_special:
      out_string += " "
     out_string += self.sp_model.decode(lowercase__     ) + token
     __lowercase				    =					True
     __lowercase				    =					[]
    else:
     current_sub_tokens.append(lowercase__     )
     __lowercase				    =					False
   out_string += self.sp_model.decode(lowercase__     )
   return out_string.strip()






  def    SCREAMING_SNAKE_CASE ( self  :	int   ,lowercase__  :	List[int]   ,lowercase__  :	bool = False   ,lowercase__  :	bool = None   ,lowercase__  :	bool = True   ,**lowercase__  :	Optional[Any]   ,):
   __lowercase				    =					kwargs.pop('''use_source_tokenizer'''   ,lowercase__     )

   __lowercase				    =					self.convert_ids_to_tokens(lowercase__   ,skip_special_tokens=lowercase__     )

   # To avoid mixing byte-level and unicode for byte-level BPT
   # we need to build string separately for added tokens and byte-level tokens
   # cf. https://github.com/huggingface/transformers/issues/1133
   __lowercase				    =					[]
   __lowercase				    =					[]
   for token in filtered_tokens:
    if skip_special_tokens and token in self.all_special_ids:
     continue
    if token in self.added_tokens_encoder:
     if current_sub_text:
      sub_texts.append(self.convert_tokens_to_string(lowercase__     )     )
      __lowercase				    =					[]
     sub_texts.append(lowercase__     )
    else:
     current_sub_text.append(lowercase__     )
   if current_sub_text:
    sub_texts.append(self.convert_tokens_to_string(lowercase__     )     )

   # Mimic the behavior of the Rust tokenizer:
   # No space before [MASK] and [SEP]
   if spaces_between_special_tokens:
    __lowercase				    =					re.sub(r''' (\[(MASK|SEP)\])'''   ,r'''\1'''   ,''' '''.join(lowercase__     )     )
   else:
    __lowercase				    =					''''''.join(lowercase__     )

   __lowercase				    =					(
       clean_up_tokenization_spaces
       if clean_up_tokenization_spaces is not None
       else self.clean_up_tokenization_spaces
   )
   if clean_up_tokenization_spaces:
    __lowercase				    =					self.clean_up_tokenization(lowercase__     )
    return clean_text
   else:
    return text






  def    SCREAMING_SNAKE_CASE ( self  :	Any   ,lowercase__  :	str   ,lowercase__  :	Optional[str] = None     ):
   if not os.path.isdir(lowercase__     ):
    logger.error(F"Vocabulary path ({save_directory}) should be a directory"     )
    return
   __lowercase				    =					os.path.join(
       lowercase__   ,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']     )

   if os.path.abspath(self.vocab_file     ) != os.path.abspath(lowercase__     ) and os.path.isfile(self.vocab_file     ):
    copyfile(self.vocab_file   ,lowercase__     )
   elif not os.path.isfile(self.vocab_file     ):
    with open(lowercase__   ,'''wb'''     ) as fi:
     __lowercase				    =					self.sp_model.serialized_model_proto()
     fi.write(lowercase__     )

   return (out_vocab_file,)






  def    SCREAMING_SNAKE_CASE ( self  :	str   ,lowercase__  :	List[int]   ,lowercase__  :	Optional[List[int]] = None     ):
   if token_ids_a is None:
    return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
   __lowercase				    =					[self.cls_token_id]
   __lowercase				    =					[self.sep_token_id]
   return cls + token_ids_a + sep + token_ids_a + sep






  def    SCREAMING_SNAKE_CASE ( self  :	str   ,lowercase__  :	List[int]   ,lowercase__  :	Optional[List[int]] = None   ,lowercase__  :	bool = False     ):
   if already_has_special_tokens:
    return super().get_special_tokens_mask(
        token_ids_a=lowercase__   ,token_ids_a=lowercase__   ,already_has_special_tokens=lowercase__     )

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






  def    SCREAMING_SNAKE_CASE ( self  :	Union[str, Any]   ,lowercase__  :	List[int]   ,lowercase__  :	Optional[List[int]] = None     ):
   __lowercase				    =					[self.sep_token_id]
   __lowercase				    =					[self.cls_token_id]
   if token_ids_a is None:
    return len(cls + token_ids_a + sep     ) * [0]
   return len(cls + token_ids_a + sep     ) * [0] + len(token_ids_a + sep     ) * [1]







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:     Optional[int]									= logging.get_logger(__name__)



@add_end_docstrings(SCREAMING_SNAKE_CASE__ )
class    lowercase_		(SCREAMING_SNAKE_CASE__ ):



						def __init__(			self	:   Any					,							**snake_case__	:   str  ):

								"""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	:   str					,							snake_case__	:   Union[str, List[str], "Image", List["Image"]]					,							**snake_case__	:   Optional[Any]  ):

								"""simple docstring"""

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



						def 	__a					(			self	:   str					,							**snake_case__	:   str  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							{}
								if "candidate_labels" in kwargs:
										SCREAMING_SNAKE_CASE_					      =							kwargs['candidate_labels']
								if "hypothesis_template" in kwargs:
										SCREAMING_SNAKE_CASE_					      =							kwargs['hypothesis_template']

								return preprocess_params, {}, {}



						def 	__a					(			self	:   List[str]					,							snake_case__	:   Optional[int]					,							snake_case__	:   Optional[Any]=None					,							snake_case__	:   List[str]="This is a photo of {}."  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							load_image(snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							self.image_processor(images=[image]					,							return_tensors=self.framework  )
								SCREAMING_SNAKE_CASE_					      =							candidate_labels
								SCREAMING_SNAKE_CASE_					      =							[hypothesis_template.format(snake_case__  ) for x in candidate_labels]
								SCREAMING_SNAKE_CASE_					      =							self.tokenizer(snake_case__					,							return_tensors=self.framework					,							padding=snake_case__  )
								SCREAMING_SNAKE_CASE_					      =							[text_inputs]
								return inputs



						def 	__a					(			self	:   List[Any]					,							snake_case__	:   Any  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							model_inputs.pop('candidate_labels'  )
								SCREAMING_SNAKE_CASE_					      =							model_inputs.pop('text_inputs'  )
								if isinstance(text_inputs[0]					,							snake_case__  ):
										SCREAMING_SNAKE_CASE_					      =							text_inputs[0]
								else:
										# Batching case.
										SCREAMING_SNAKE_CASE_					      =							text_inputs[0][0]

								SCREAMING_SNAKE_CASE_					      =							self.model(**snake_case__					,							**snake_case__  )

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






						def 	__a					(			self	:   List[Any]					,							snake_case__	:   Optional[Any]  ):

								"""simple docstring"""

								SCREAMING_SNAKE_CASE_					      =							model_outputs.pop('candidate_labels'  )
								SCREAMING_SNAKE_CASE_					      =							model_outputs['logits'][0]
								if self.framework == "pt":
										SCREAMING_SNAKE_CASE_					      =							logits.softmax(dim=-1  ).squeeze(-1  )
										SCREAMING_SNAKE_CASE_					      =							probs.tolist()
										if not isinstance(snake_case__					,							snake_case__  ):
												SCREAMING_SNAKE_CASE_					      =							[scores]
								elif self.framework == "tf":
										SCREAMING_SNAKE_CASE_					      =							stable_softmax(snake_case__					,							axis=-1  )
										SCREAMING_SNAKE_CASE_					      =							probs.numpy().tolist()
								else:
										raise ValueError(f'''Unsupported framework: {self.framework}'''  )

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