






import unittest

from transformers import (
    MODEL_FOR_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
    TextGenerationPipeline,
    logging,
    pipeline,
)
from transformers.testing_utils import (
    CaptureLogger,
    is_pipeline_test,
    require_accelerate,
    require_tf,
    require_torch,
    require_torch_gpu,
    require_torch_or_tf,
)

from .test_pipelines_common import ANY







@is_pipeline_test
@require_torch_or_tf
class  _lowercase   (   unittest.TestCase						):
      _a							:						List[str]										=							MODEL_FOR_CAUSAL_LM_MAPPING
      _a							:						Any										=							TF_MODEL_FOR_CAUSAL_LM_MAPPING
      @require_torch
      def 						lowercase__					(     self   ):
       snake_case__  :    int							       =pipeline(task="""text-generation"""				,			model="""sshleifer/tiny-ctrl"""				,			framework="""pt"""   )
       # Using `do_sample=False` to force deterministic output
       snake_case__  :    Optional[Any]							       =text_generator("""This is a test"""				,			do_sample=a   )
       self.assertEqual(
           a				,			[
               {
                   """generated_text""": (
                       """This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope."""
                       """ oscope. FiliFili@@"""
                   )
               }
           ]				,			)

       snake_case__  :    List[str]							       =text_generator(["""This is a test""", """This is a second test"""]   )
       self.assertEqual(
           a				,			[
               [
                   {
                       """generated_text""": (
                           """This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope."""
                           """ oscope. FiliFili@@"""
                       )
                   }
               ],
               [
                   {
                       """generated_text""": (
                           """This is a second test ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy"""
                           """ oscope. oscope. FiliFili@@"""
                       )
                   }
               ],
           ]				,			)

       snake_case__  :    int							       =text_generator("""This is a test"""				,			do_sample=a				,			num_return_sequences=2				,			return_tensors=a   )
       self.assertEqual(
           a				,			[
               {"""generated_token_ids""": ANY(a   )},
               {"""generated_token_ids""": ANY(a   )},
           ]				,			)
       snake_case__  :    Optional[int]							       =text_generator.model.config.eos_token_id
       snake_case__  :    List[str]							       ="""<pad>"""
       snake_case__  :    str							       =text_generator(
           ["""This is a test""", """This is a second test"""]				,			do_sample=a				,			num_return_sequences=2				,			batch_size=2				,			return_tensors=a				,			)
       self.assertEqual(
           a				,			[
               [
                   {"""generated_token_ids""": ANY(a   )},
                   {"""generated_token_ids""": ANY(a   )},
               ],
               [
                   {"""generated_token_ids""": ANY(a   )},
                   {"""generated_token_ids""": ANY(a   )},
               ],
           ]				,			)
      @require_tf
      def 						lowercase__					(     self   ):
       snake_case__  :    List[Any]							       =pipeline(task="""text-generation"""				,			model="""sshleifer/tiny-ctrl"""				,			framework="""tf"""   )

       # Using `do_sample=False` to force deterministic output
       snake_case__  :    str							       =text_generator("""This is a test"""				,			do_sample=a   )
       self.assertEqual(
           a				,			[
               {
                   """generated_text""": (
                       """This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵"""
                       """ please,"""
                   )
               }
           ]				,			)

       snake_case__  :    Optional[int]							       =text_generator(["""This is a test""", """This is a second test"""]				,			do_sample=a   )
       self.assertEqual(
           a				,			[
               [
                   {
                       """generated_text""": (
                           """This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵"""
                           """ please,"""
                       )
                   }
               ],
               [
                   {
                       """generated_text""": (
                           """This is a second test Chieftain Chieftain prefecture prefecture prefecture Cannes Cannes"""
                           """ Cannes 閲閲Cannes Cannes Cannes 攵 please,"""
                       )
                   }
               ],
           ]				,			)
      def 						lowercase__					(     self				,			a				,			a				,			a   ):
       snake_case__  :    str							       =TextGenerationPipeline(model=a				,			tokenizer=a   )
       return text_generator, ["This is a test", "Another test"]
      def 						lowercase__					(     self   ):
       snake_case__  :    Tuple							       ="""Hello I believe in"""
       snake_case__  :    Any							       =pipeline("""text-generation"""				,			model="""hf-internal-testing/tiny-random-gpt2"""   )
       snake_case__  :    Union[str, Any]							       =text_generator(a   )
       self.assertEqual(
           a				,			[{"""generated_text""": """Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe"""}]				,			)

       snake_case__  :    Optional[Any]							       =text_generator(a				,			stop_sequence=""" fe"""   )
       self.assertEqual(a				,			[{"""generated_text""": """Hello I believe in fe"""}]   )
      def 						lowercase__					(     self				,			a				,			a   ):
       snake_case__  :    Optional[Any]							       =text_generator.model
       snake_case__  :    Any							       =text_generator.tokenizer

       snake_case__  :    List[str]							       =text_generator("""This is a test"""   )
       self.assertEqual(a				,			[{"""generated_text""": ANY(a   )}]   )
       self.assertTrue(outputs[0]["""generated_text"""].startswith("""This is a test"""   )   )

       snake_case__  :    Union[str, Any]							       =text_generator("""This is a test"""				,			return_full_text=a   )
       self.assertEqual(a				,			[{"""generated_text""": ANY(a   )}]   )
       self.assertNotIn("""This is a test"""				,			outputs[0]["""generated_text"""]   )

       snake_case__  :    List[str]							       =pipeline(task="""text-generation"""				,			model=a				,			tokenizer=a				,			return_full_text=a   )
       snake_case__  :    Union[str, Any]							       =text_generator("""This is a test"""   )
       self.assertEqual(a				,			[{"""generated_text""": ANY(a   )}]   )
       self.assertNotIn("""This is a test"""				,			outputs[0]["""generated_text"""]   )

       snake_case__  :    int							       =text_generator("""This is a test"""				,			return_full_text=a   )
       self.assertEqual(a				,			[{"""generated_text""": ANY(a   )}]   )
       self.assertTrue(outputs[0]["""generated_text"""].startswith("""This is a test"""   )   )

       snake_case__  :    Any							       =text_generator(["""This is great !""", """Something else"""]				,			num_return_sequences=2				,			do_sample=a   )
       self.assertEqual(
           a				,			[
               [{"""generated_text""": ANY(a   )}, {"""generated_text""": ANY(a   )}],
               [{"""generated_text""": ANY(a   )}, {"""generated_text""": ANY(a   )}],
           ]				,			)

       if text_generator.tokenizer.pad_token is not None:
        snake_case__  :    int							       =text_generator(
            ["""This is great !""", """Something else"""]				,			num_return_sequences=2				,			batch_size=2				,			do_sample=a   )
        self.assertEqual(
            a				,			[
                [{"""generated_text""": ANY(a   )}, {"""generated_text""": ANY(a   )}],
                [{"""generated_text""": ANY(a   )}, {"""generated_text""": ANY(a   )}],
            ]				,			)

       with self.assertRaises(a   ):
        snake_case__  :    int							       =text_generator("""test"""				,			return_full_text=a				,			return_text=a   )
       with self.assertRaises(a   ):
        snake_case__  :    Optional[Any]							       =text_generator("""test"""				,			return_full_text=a				,			return_tensors=a   )
       with self.assertRaises(a   ):
        snake_case__  :    Optional[int]							       =text_generator("""test"""				,			return_text=a				,			return_tensors=a   )

       # Empty prompt is slighly special
       # it requires BOS token to exist.
       # Special case for Pegasus which will always append EOS so will
       # work even without BOS.
       if (
           text_generator.tokenizer.bos_token_id is not None
           or "Pegasus" in tokenizer.__class__.__name__
           or "Git" in model.__class__.__name__
       ):
        snake_case__  :    List[Any]							       =text_generator(""""""   )
        self.assertEqual(a				,			[{"""generated_text""": ANY(a   )}]   )
       else:
        with self.assertRaises((ValueError, AssertionError)   ):
         snake_case__  :    List[str]							       =text_generator(""""""   )

       if text_generator.framework == "tf":
        # TF generation does not support max_new_tokens, and it's impossible
        # to control long generation with only max_length without
        # fancy calculation, dismissing tests for now.
        return
       # We don't care about infinite range models.
       # They already work.
       # Skip this test for XGLM, since it uses sinusoidal positional embeddings which are resized on-the-fly.
       snake_case__  :    Optional[Any]							       =["""RwkvForCausalLM""", """XGLMForCausalLM""", """GPTNeoXForCausalLM"""]
       if (
           tokenizer.model_max_length < 1_0_0_0_0
           and text_generator.model.__class__.__name__ not in EXTRA_MODELS_CAN_HANDLE_LONG_INPUTS
       ):
        # Handling of large generations
        with self.assertRaises((RuntimeError, IndexError, ValueError, AssertionError)   ):
         text_generator("""This is a test""" * 5_0_0				,			max_new_tokens=2_0   )

        snake_case__  :    int							       =text_generator("""This is a test""" * 5_0_0				,			handle_long_generation="""hole"""				,			max_new_tokens=2_0   )
        # Hole strategy cannot work
        with self.assertRaises(a   ):
         text_generator(
             """This is a test""" * 5_0_0				,			handle_long_generation="""hole"""				,			max_new_tokens=tokenizer.model_max_length + 1_0				,			)
      @require_torch
      @require_accelerate
      @require_torch_gpu
      def 						lowercase__					(     self   ):
       import torch

       # Classic `model_kwargs`
       snake_case__  :    List[Any]							       =pipeline(
           model="""hf-internal-testing/tiny-random-bloom"""				,			model_kwargs={"""device_map""": """auto""", """torch_dtype""": torch.bfloataa}				,			)
       self.assertEqual(pipe.model.device				,			torch.device(0   )   )
       self.assertEqual(pipe.model.lm_head.weight.dtype				,			torch.bfloataa   )
       snake_case__  :    List[str]							       =pipe("""This is a test"""   )
       self.assertEqual(
           a				,			[
               {
                   """generated_text""": (
                       """This is a test test test test test test test test test test test test test test test test"""
                       """ test"""
                   )
               }
           ]				,			)

       # Upgraded those two to real pipeline arguments (they just get sent for the model as they're unlikely to mean anything else.)
       snake_case__  :    Union[str, Any]							       =pipeline(model="""hf-internal-testing/tiny-random-bloom"""				,			device_map="""auto"""				,			torch_dtype=torch.bfloataa   )
       self.assertEqual(pipe.model.device				,			torch.device(0   )   )
       self.assertEqual(pipe.model.lm_head.weight.dtype				,			torch.bfloataa   )
       snake_case__  :    Optional[int]							       =pipe("""This is a test"""   )
       self.assertEqual(
           a				,			[
               {
                   """generated_text""": (
                       """This is a test test test test test test test test test test test test test test test test"""
                       """ test"""
                   )
               }
           ]				,			)

       # torch_dtype will be automatically set to float32 if not provided - check: https://github.com/huggingface/transformers/pull/20602
       snake_case__  :    Union[str, Any]							       =pipeline(model="""hf-internal-testing/tiny-random-bloom"""				,			device_map="""auto"""   )
       self.assertEqual(pipe.model.device				,			torch.device(0   )   )
       self.assertEqual(pipe.model.lm_head.weight.dtype				,			torch.floataa   )
       snake_case__  :    int							       =pipe("""This is a test"""   )
       self.assertEqual(
           a				,			[
               {
                   """generated_text""": (
                       """This is a test test test test test test test test test test test test test test test test"""
                       """ test"""
                   )
               }
           ]				,			)
      @require_torch
      @require_torch_gpu
      def 						lowercase__					(     self   ):
       import torch

       snake_case__  :    Tuple							       =pipeline(model="""hf-internal-testing/tiny-random-bloom"""				,			device=0				,			torch_dtype=torch.floataa   )
       pipe("""This is a test"""   )
      @require_torch
      @require_accelerate
      @require_torch_gpu
      def 						lowercase__					(     self   ):
       import torch

       snake_case__  :    Dict							       =pipeline(model="""hf-internal-testing/tiny-random-bloom"""				,			device_map="""auto"""				,			torch_dtype=torch.floataa   )
       pipe("""This is a test"""				,			do_sample=a				,			top_p=0.5   )



      def 						lowercase__					(     self   ):
       snake_case__  :    List[Any]							       ="""Hello world"""
       snake_case__  :    Dict							       =pipeline("""text-generation"""				,			model="""hf-internal-testing/tiny-random-gpt2"""   )
       if text_generator.model.framework == "tf":
        snake_case__  :    Union[str, Any]							       =logging.get_logger("""transformers.generation.tf_utils"""   )
       else:
        snake_case__  :    Tuple							       =logging.get_logger("""transformers.generation.utils"""   )
       snake_case__  :    Union[str, Any]							       ="""Both `max_new_tokens`"""  # The beggining of the message to be checked in this test

       # Both are set by the user -> log warning
       with CaptureLogger(a   ) as cl:
        snake_case__  :    Tuple							       =text_generator(a				,			max_length=1_0				,			max_new_tokens=1   )
       self.assertIn(a				,			cl.out   )

       # The user only sets one -> no warning
       with CaptureLogger(a   ) as cl:
        snake_case__  :    Optional[Any]							       =text_generator(a				,			max_new_tokens=1   )
       self.assertNotIn(a				,			cl.out   )

       with CaptureLogger(a   ) as cl:
        snake_case__  :    Optional[int]							       =text_generator(a				,			max_length=1_0   )
       self.assertNotIn(a				,			cl.out   )













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	:		Dict										=   logging.get_logger(__name__)

__lowerCamelCase	:		Any										=   {"""vocab_file""": """spiece.model"""}

__lowerCamelCase	:		Optional[Any]										=   {
    """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	:		List[str]										=   {
    """google/bigbird-roberta-base""": 40_96,
    """google/bigbird-roberta-large""": 40_96,
    """google/bigbird-base-trivia-itc""": 40_96,
}







class  _lowercase   (   _A						):
      _a							:						int										=							VOCAB_FILES_NAMES
      _a							:						List[str]										=							PRETRAINED_VOCAB_FILES_MAP
      _a							:						Any										=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _a							:						Tuple										=							['input_ids', 'attention_mask']
      _a							:						List[int]										=							[]
      def __init__(     self				,			a				,			a="<unk>"				,			a="<s>"				,			a="</s>"				,			a="<pad>"				,			a="[SEP]"				,			a="[MASK]"				,			a="[CLS]"				,			a = None				,			**a				,			):
       snake_case__  :    str							       =AddedToken(a				,			lstrip=a				,			rstrip=a   ) if isinstance(a				,			a   ) else bos_token
       snake_case__  :    Optional[int]							       =AddedToken(a				,			lstrip=a				,			rstrip=a   ) if isinstance(a				,			a   ) else eos_token
       snake_case__  :    Any							       =AddedToken(a				,			lstrip=a				,			rstrip=a   ) if isinstance(a				,			a   ) else unk_token
       snake_case__  :    List[str]							       =AddedToken(a				,			lstrip=a				,			rstrip=a   ) if isinstance(a				,			a   ) else pad_token
       snake_case__  :    int							       =AddedToken(a				,			lstrip=a				,			rstrip=a   ) if isinstance(a				,			a   ) else cls_token
       snake_case__  :    Tuple							       =AddedToken(a				,			lstrip=a				,			rstrip=a   ) if isinstance(a				,			a   ) else sep_token

       # Mask token behave like a normal word, i.e. include the space before it
       snake_case__  :    Union[str, Any]							       =AddedToken(a				,			lstrip=a				,			rstrip=a   ) if isinstance(a				,			a   ) else mask_token

       snake_case__  :    Union[str, Any]							       ={} if sp_model_kwargs is None else sp_model_kwargs

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

       snake_case__  :    Optional[int]							       =vocab_file

       snake_case__  :    List[str]							       =spm.SentencePieceProcessor(**self.sp_model_kwargs   )
       self.sp_model.Load(a   )
      @property
      def 						lowercase__					(     self   ):
       return self.sp_model.get_piece_size()
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       ={self.convert_ids_to_tokens(a   ): i for i in range(self.vocab_size   )}
       vocab.update(self.added_tokens_encoder   )
       return vocab
      def __getstate__(     self   ):
       snake_case__  :    List[Any]							       =self.__dict__.copy()
       snake_case__  :    Optional[int]							       =None
       return state
      def __setstate__(     self				,			a   ):
       snake_case__  :    Optional[Any]							       =d

       # for backward compatibility
       if not hasattr(self				,			"""sp_model_kwargs"""   ):
        snake_case__  :    Optional[Any]							       ={}

       snake_case__  :    Tuple							       =spm.SentencePieceProcessor(**self.sp_model_kwargs   )
       self.sp_model.Load(self.vocab_file   )
      def 						lowercase__					(     self				,			a   ):
       return self.sp_model.encode(a				,			out_type=a   )
      def 						lowercase__					(     self				,			a   ):
       return self.sp_model.piece_to_id(a   )
      def 						lowercase__					(     self				,			a   ):
       snake_case__  :    Dict							       =self.sp_model.IdToPiece(a   )
       return token
      def 						lowercase__					(     self				,			a   ):
       snake_case__  :    Optional[int]							       =[]
       snake_case__  :    Dict							       =""""""
       snake_case__  :    Dict							       =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(a   ) + token
         snake_case__  :    Optional[Any]							       =True
         snake_case__  :    Dict							       =[]
        else:
         current_sub_tokens.append(a   )
         snake_case__  :    List[Any]							       =False
       out_string += self.sp_model.decode(a   )
       return out_string.strip()
      def 						lowercase__					(     self				,			a				,			a = False				,			a = None				,			a = True				,			**a				,			):
       snake_case__  :    List[str]							       =kwargs.pop("""use_source_tokenizer"""				,			a   )

       snake_case__  :    int							       =self.convert_ids_to_tokens(a				,			skip_special_tokens=a   )

       # 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
       snake_case__  :    List[Any]							       =[]
       snake_case__  :    int							       =[]
       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(a   )   )
          snake_case__  :    List[Any]							       =[]
         sub_texts.append(a   )
        else:
         current_sub_text.append(a   )
       if current_sub_text:
        sub_texts.append(self.convert_tokens_to_string(a   )   )

       # Mimic the behavior of the Rust tokenizer:
       # No space before [MASK] and [SEP]
       if spaces_between_special_tokens:
        snake_case__  :    Dict							       =re.sub(R""" (\[(MASK|SEP)\])"""				,			R"""\1"""				,			""" """.join(a   )   )
       else:
        snake_case__  :    Dict							       ="""""".join(a   )

       snake_case__  :    str							       =(
           clean_up_tokenization_spaces
           if clean_up_tokenization_spaces is not None
           else self.clean_up_tokenization_spaces
       )
       if clean_up_tokenization_spaces:
        snake_case__  :    Any							       =self.clean_up_tokenization(a   )
        return clean_text
       else:
        return text
      def 						lowercase__					(     self				,			a				,			a = None   ):
       if not os.path.isdir(a   ):
        logger.error(F"Vocabulary path ({save_directory}) should be a directory"   )
        return
       snake_case__  :    Any							       =os.path.join(
           a				,			(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]   )

       if os.path.abspath(self.vocab_file   ) != os.path.abspath(a   ) and os.path.isfile(self.vocab_file   ):
        copyfile(self.vocab_file				,			a   )
       elif not os.path.isfile(self.vocab_file   ):
        with open(a				,			"""wb"""   ) as fi:
         snake_case__  :    Any							       =self.sp_model.serialized_model_proto()
         fi.write(a   )

       return (out_vocab_file,)
      def 						lowercase__					(     self				,			a				,			a = None   ):
       if token_ids_a is None:
        return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
       snake_case__  :    Union[str, Any]							       =[self.cls_token_id]
       snake_case__  :    Dict							       =[self.sep_token_id]
       return cls + token_ids_a + sep + token_ids_a + sep
      def 						lowercase__					(     self				,			a				,			a = None				,			a = False   ):
       if already_has_special_tokens:
        return super().get_special_tokens_mask(
            token_ids_a=a				,			token_ids_a=a				,			already_has_special_tokens=a   )

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



      def 						lowercase__					(     self				,			a				,			a = None   ):
       snake_case__  :    List[str]							       =[self.sep_token_id]
       snake_case__  :    List[Any]							       =[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]





