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

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCamelCase  =logging.get_logger(__name__)

lowerCamelCase  ="▁"

lowerCamelCase  ={"vocab_file": "sentencepiece.bpe.model"}

lowerCamelCase  ={
    "vocab_file": {
        "facebook/xglm-564M": "https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model",
    }
}

lowerCamelCase  ={
    "facebook/xglm-564M": 2_0_4_8,
}




class 					_lowerCamelCase			(   UpperCamelCase_			):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	VOCAB_FILES_NAMES
     SCREAMING_SNAKE_CASE_		 =	PRETRAINED_VOCAB_FILES_MAP
     SCREAMING_SNAKE_CASE_		 =	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     SCREAMING_SNAKE_CASE_		 =	['''input_ids''', '''attention_mask''']






     def __init__(				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE="<s>"    ,    __SCREAMING_SNAKE_CASE="</s>"    ,    __SCREAMING_SNAKE_CASE="</s>"    ,    __SCREAMING_SNAKE_CASE="<s>"    ,    __SCREAMING_SNAKE_CASE="<unk>"    ,    __SCREAMING_SNAKE_CASE="<pad>"    ,    __SCREAMING_SNAKE_CASE = None    ,    **__SCREAMING_SNAKE_CASE    ,    )	-> None:




           """simple docstring"""







           UpperCamelCase__     : str											=					{} if sp_model_kwargs is None else sp_model_kwargs

           # Compatibility with the original tokenizer
           UpperCamelCase__     : Optional[Any]											=					7
           UpperCamelCase__     : Tuple											=					[F'''<madeupword{i}>''' for i in range(self.num_madeup_words	)]

           UpperCamelCase__     : Optional[Any]											=					kwargs.get('''additional_special_tokens'''    ,    []	)
           kwargs["additional_special_tokens"] += [
               word for word in madeup_words if word not in kwargs["additional_special_tokens"]
           ]

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

           UpperCamelCase__     : str											=					spm.SentencePieceProcessor(**self.sp_model_kwargs	)
           self.sp_model.Load(str(__SCREAMING_SNAKE_CASE	)	)
           UpperCamelCase__     : List[Any]											=					vocab_file

           # Original fairseq vocab and spm vocab must be "aligned":
           # Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
           # -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
           # fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
           # spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

           # The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
           UpperCamelCase__     : Any											=					1

           # Mimic fairseq token-to-id alignment for the first 4 token
           UpperCamelCase__     : Tuple											=					{'''<s>''': 0, '''<pad>''': 1, '''</s>''': 2, '''<unk>''': 3}

           UpperCamelCase__     : List[Any]											=					len(self.sp_model	)
           UpperCamelCase__     : str											=					{F'''<madeupword{i}>''': sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words	)}
           self.fairseq_tokens_to_ids.update(__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : Union[str, Any]											=					{v: k for k, v in self.fairseq_tokens_to_ids.items()}






     def __getstate__(				self	)	-> Any:




           """simple docstring"""







           UpperCamelCase__     : Optional[int]											=					self.__dict__.copy()
           UpperCamelCase__     : Union[str, Any]											=					None
           UpperCamelCase__     : str											=					self.sp_model.serialized_model_proto()
           return state






     def __setstate__(				self    ,    __SCREAMING_SNAKE_CASE	)	-> Tuple:




           """simple docstring"""







           UpperCamelCase__     : Optional[int]											=					d

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

           UpperCamelCase__     : List[str]											=					spm.SentencePieceProcessor(**self.sp_model_kwargs	)
           self.sp_model.LoadFromSerializedProto(self.sp_model_proto	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = None	)	-> List[int]:




           """simple docstring"""







           if token_ids_a is None:
                 return [self.sep_token_id] + token_ids_a
           UpperCamelCase__     : List[Any]											=					[self.sep_token_id]
           return sep + token_ids_a + sep + sep + token_ids_a






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = None    ,    __SCREAMING_SNAKE_CASE = False	)	-> List[int]:




           """simple docstring"""







           if already_has_special_tokens:
                 return super().get_special_tokens_mask(
                     token_ids_a=__SCREAMING_SNAKE_CASE    ,    token_ids_a=__SCREAMING_SNAKE_CASE    ,    already_has_special_tokens=__SCREAMING_SNAKE_CASE	)

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






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = None	)	-> List[int]:




           """simple docstring"""







           UpperCamelCase__     : Tuple											=					[self.sep_token_id]

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






     @property
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Dict:




           """simple docstring"""







           return len(self.sp_model	) + self.fairseq_offset + self.num_madeup_words






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : Tuple											=					{self.convert_ids_to_tokens(__SCREAMING_SNAKE_CASE	): i for i in range(self.vocab_size	)}
           vocab.update(self.added_tokens_encoder	)
           return vocab






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> List[str]:




           """simple docstring"""







           return self.sp_model.encode(__SCREAMING_SNAKE_CASE    ,    out_type=__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> Union[str, Any]:




           """simple docstring"""







           if token in self.fairseq_tokens_to_ids:
                 return self.fairseq_tokens_to_ids[token]
           UpperCamelCase__     : Dict											=					self.sp_model.PieceToId(__SCREAMING_SNAKE_CASE	)

           # Need to return unknown token if the SP model returned 0
           return spm_id + self.fairseq_offset if spm_id else self.unk_token_id






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> Union[str, Any]:




           """simple docstring"""







           if index in self.fairseq_ids_to_tokens:
                 return self.fairseq_ids_to_tokens[index]
           return self.sp_model.IdToPiece(index - self.fairseq_offset	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__     : str											=					''''''.join(__SCREAMING_SNAKE_CASE	).replace(__SCREAMING_SNAKE_CASE    ,    ''' '''	).strip()
           return out_string






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = None	)	-> Tuple[str]:




           """simple docstring"""







           if not os.path.isdir(__SCREAMING_SNAKE_CASE	):
                 logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''	)
                 return
           UpperCamelCase__     : Union[str, Any]											=					os.path.join(
               __SCREAMING_SNAKE_CASE    ,    (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']	)

           if os.path.abspath(self.vocab_file	) != os.path.abspath(__SCREAMING_SNAKE_CASE	) and os.path.isfile(self.vocab_file	):
                 copyfile(self.vocab_file    ,    __SCREAMING_SNAKE_CASE	)
           elif not os.path.isfile(self.vocab_file	):
                 with open(__SCREAMING_SNAKE_CASE    ,    '''wb'''	) as fi:
                       UpperCamelCase__     : Optional[Any]											=					self.sp_model.serialized_model_proto()
                       fi.write(__SCREAMING_SNAKE_CASE	)

           return (out_vocab_file,)




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

lowerCamelCase  =logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase_			)
class 					_lowerCamelCase			(   UpperCamelCase_			):





     """simple docstring"""







     def __init__(				self    ,    *__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)	-> Dict:




           """simple docstring"""







           super().__init__(*__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)
           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  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE=None    ,    __SCREAMING_SNAKE_CASE=None    ,    __SCREAMING_SNAKE_CASE=None	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : Tuple											=					{}
           UpperCamelCase__     : int											=					{}

           if prompt is not None:
                 UpperCamelCase__     : int											=					prompt

           if generate_kwargs is not None:
                 UpperCamelCase__     : int											=					generate_kwargs
           if max_new_tokens is not None:
                 if "generate_kwargs" not in forward_kwargs:
                       UpperCamelCase__     : Union[str, Any]											=					{}
                 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'''	)
                 UpperCamelCase__     : Dict											=					max_new_tokens
           return preprocess_params, forward_kwargs, {}






     def __call__(				self    ,    __SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)	-> List[str]:




           """simple docstring"""







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






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=None	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : List[Any]											=					load_image(__SCREAMING_SNAKE_CASE	)

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

                 UpperCamelCase__     : Optional[int]											=					self.model.config.model_type

                 if model_type == "git":
                       UpperCamelCase__     : Optional[int]											=					self.image_processor(images=__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)
                       UpperCamelCase__     : str											=					self.tokenizer(text=__SCREAMING_SNAKE_CASE    ,    add_special_tokens=__SCREAMING_SNAKE_CASE	).input_ids
                       UpperCamelCase__     : Dict											=					[self.tokenizer.cls_token_id] + input_ids
                       UpperCamelCase__     : int											=					torch.tensor(__SCREAMING_SNAKE_CASE	).unsqueeze(0	)
                       model_inputs.update({'''input_ids''': input_ids}	)

                 elif model_type == "pix2struct":
                       UpperCamelCase__     : Tuple											=					self.image_processor(images=__SCREAMING_SNAKE_CASE    ,    header_text=__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)

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

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

           else:
                 UpperCamelCase__     : str											=					self.image_processor(images=__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)

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

           return model_inputs






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=None	)	-> Optional[Any]:




           """simple docstring"""







           if (
               "input_ids" in model_inputs
               and isinstance(model_inputs['''input_ids''']    ,    __SCREAMING_SNAKE_CASE	)
               and all(x is None for x in model_inputs['''input_ids''']	)
           ):
                 UpperCamelCase__     : Union[str, Any]											=					None

           if generate_kwargs is None:
                 UpperCamelCase__     : Optional[Any]											=					{}
           # 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.
           UpperCamelCase__     : Any											=					model_inputs.pop(self.model.main_input_name	)
           UpperCamelCase__     : Any											=					self.model.generate(__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)
           return model_outputs






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : Union[str, Any]											=					[]
           for output_ids in model_outputs:
                 UpperCamelCase__     : str											=					{
                     '''generated_text''': self.tokenizer.decode(
                         __SCREAMING_SNAKE_CASE    ,    skip_special_tokens=__SCREAMING_SNAKE_CASE    ,    )
                 }
                 records.append(__SCREAMING_SNAKE_CASE	)
           return records


