'''simple docstring'''



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       __SCREAMING_SNAKE_CASE	(						unittest.TestCase			):



     '''simple docstring'''






     UpperCAmelCase__       :						Optional[int]      =   MODEL_FOR_CAUSAL_LM_MAPPING
     UpperCAmelCase__       :						Tuple      =   TF_MODEL_FOR_CAUSAL_LM_MAPPING

     @require_torch
     def 	UpperCamelCase(					self	):
       _snake_case     		=				pipeline(task="text-generation"						,   model="sshleifer/tiny-ctrl"						,   framework="pt"	)
       # Using `do_sample=False` to force deterministic output
       _snake_case     		=				text_generator("This is a test"						,   do_sample=lowerCamelCase	)
       self.assertEqual(
           lowerCamelCase						,   [
               {
                   "generated_text": (
                       "This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope."
                       " oscope. FiliFili@@"
                   )
               }
           ]						,   )

       _snake_case     		=				text_generator(["This is a test", "This is a second test"]	)
       self.assertEqual(
           lowerCamelCase						,   [
               [
                   {
                       "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     		=				text_generator("This is a test"						,   do_sample=lowerCamelCase						,   num_return_sequences=2						,   return_tensors=lowerCamelCase	)
       self.assertEqual(
           lowerCamelCase						,   [
               {"generated_token_ids": ANY(lowerCamelCase	)},
               {"generated_token_ids": ANY(lowerCamelCase	)},
           ]						,   )
       _snake_case     		=				text_generator.model.config.eos_token_id
       _snake_case     		=				"<pad>"
       _snake_case     		=				text_generator(
           ["This is a test", "This is a second test"]						,   do_sample=lowerCamelCase						,   num_return_sequences=2						,   batch_size=2						,   return_tensors=lowerCamelCase						,   )
       self.assertEqual(
           lowerCamelCase						,   [
               [
                   {"generated_token_ids": ANY(lowerCamelCase	)},
                   {"generated_token_ids": ANY(lowerCamelCase	)},
               ],
               [
                   {"generated_token_ids": ANY(lowerCamelCase	)},
                   {"generated_token_ids": ANY(lowerCamelCase	)},
               ],
           ]						,   )

     @require_tf
     def 	UpperCamelCase(					self	):
       _snake_case     		=				pipeline(task="text-generation"						,   model="sshleifer/tiny-ctrl"						,   framework="tf"	)

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

       _snake_case     		=				text_generator(["This is a test", "This is a second test"]						,   do_sample=lowerCamelCase	)
       self.assertEqual(
           lowerCamelCase						,   [
               [
                   {
                       "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 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase						,   lowerCamelCase	):
       _snake_case     		=				TextGenerationPipeline(model=lowerCamelCase						,   tokenizer=lowerCamelCase	)
       return text_generator, ["This is a test", "Another test"]

     def 	UpperCamelCase(					self	):
       _snake_case     		=				"Hello I believe in"
       _snake_case     		=				pipeline("text-generation"						,   model="hf-internal-testing/tiny-random-gpt2"	)
       _snake_case     		=				text_generator(lowerCamelCase	)
       self.assertEqual(
           lowerCamelCase						,   [{"generated_text": "Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe"}]						,   )

       _snake_case     		=				text_generator(lowerCamelCase						,   stop_sequence=" fe"	)
       self.assertEqual(lowerCamelCase						,   [{"generated_text": "Hello I believe in fe"}]	)

     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase	):
       _snake_case     		=				text_generator.model
       _snake_case     		=				text_generator.tokenizer

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

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

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

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

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

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

       with self.assertRaises(lowerCamelCase	):
         _snake_case     		=				text_generator("test"						,   return_full_text=lowerCamelCase						,   return_text=lowerCamelCase	)
       with self.assertRaises(lowerCamelCase	):
         _snake_case     		=				text_generator("test"						,   return_full_text=lowerCamelCase						,   return_tensors=lowerCamelCase	)
       with self.assertRaises(lowerCamelCase	):
         _snake_case     		=				text_generator("test"						,   return_text=lowerCamelCase						,   return_tensors=lowerCamelCase	)

       # 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     		=				text_generator(""	)
         self.assertEqual(lowerCamelCase						,   [{"generated_text": ANY(lowerCamelCase	)}]	)
       else:
         with self.assertRaises((ValueError, AssertionError)	):
           _snake_case     		=				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     		=				["RwkvForCausalLM", "XGLMForCausalLM", "GPTNeoXForCausalLM"]
       if (
           tokenizer.model_max_length < 10_000
           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" * 500						,   max_new_tokens=20	)

         _snake_case     		=				text_generator("This is a test" * 500						,   handle_long_generation="hole"						,   max_new_tokens=20	)
         # Hole strategy cannot work
         with self.assertRaises(lowerCamelCase	):
           text_generator(
               "This is a test" * 500						,   handle_long_generation="hole"						,   max_new_tokens=tokenizer.model_max_length + 10						,   )

     @require_torch
     @require_accelerate
     @require_torch_gpu
     def 	UpperCamelCase(					self	):
       import torch

       # Classic `model_kwargs`
       _snake_case     		=				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     		=				pipe("This is a test"	)
       self.assertEqual(
           lowerCamelCase						,   [
               {
                   "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     		=				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     		=				pipe("This is a test"	)
       self.assertEqual(
           lowerCamelCase						,   [
               {
                   "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     		=				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     		=				pipe("This is a test"	)
       self.assertEqual(
           lowerCamelCase						,   [
               {
                   "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 	UpperCamelCase(					self	):
       import torch

       _snake_case     		=				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 	UpperCamelCase(					self	):
       import torch

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






     def 	UpperCamelCase(					self	):
       _snake_case     		=				"Hello world"
       _snake_case     		=				pipeline("text-generation"						,   model="hf-internal-testing/tiny-random-gpt2"	)
       if text_generator.model.framework == "tf":
         _snake_case     		=				logging.get_logger("transformers.generation.tf_utils"	)
       else:
         _snake_case     		=				logging.get_logger("transformers.generation.utils"	)
       _snake_case     		=				"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(lowerCamelCase	) as cl:
         _snake_case     		=				text_generator(lowerCamelCase						,   max_length=10						,   max_new_tokens=1	)
       self.assertIn(lowerCamelCase						,   cl.out	)

       # The user only sets one -> no warning
       with CaptureLogger(lowerCamelCase	) as cl:
         _snake_case     		=				text_generator(lowerCamelCase						,   max_new_tokens=1	)
       self.assertNotIn(lowerCamelCase						,   cl.out	)

       with CaptureLogger(lowerCamelCase	) as cl:
         _snake_case     		=				text_generator(lowerCamelCase						,   max_length=10	)
       self.assertNotIn(lowerCamelCase						,   cl.out	)









'''simple docstring'''



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


__magic_name__   :						Optional[int]    =			logging.get_logger(__name__)

__magic_name__   :						str    =			"""▁"""

__magic_name__   :						int    =			{"""vocab_file""": """sentencepiece.bpe.model"""}

__magic_name__   :						List[str]    =			{
    """vocab_file""": {
        """facebook/xglm-564M""": """https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model""",
    }
}

__magic_name__   :						Union[str, Any]    =			{
    """facebook/xglm-564M""": 2_048,
}


class       __SCREAMING_SNAKE_CASE	(						__UpperCamelCase			):



     '''simple docstring'''






     UpperCAmelCase__       :						Union[str, Any]      =   VOCAB_FILES_NAMES
     UpperCAmelCase__       :						str      =   PRETRAINED_VOCAB_FILES_MAP
     UpperCAmelCase__       :						Any      =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     UpperCAmelCase__       :						Any      =   ['''input_ids''', '''attention_mask''']

     def __init__(					self						,   lowerCamelCase						,   lowerCamelCase="<s>"						,   lowerCamelCase="</s>"						,   lowerCamelCase="</s>"						,   lowerCamelCase="<s>"						,   lowerCamelCase="<unk>"						,   lowerCamelCase="<pad>"						,   lowerCamelCase = None						,   **lowerCamelCase						,   ):
       _snake_case     		=				{} if sp_model_kwargs is None else sp_model_kwargs

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

       _snake_case     		=				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=lowerCamelCase						,   eos_token=lowerCamelCase						,   unk_token=lowerCamelCase						,   sep_token=lowerCamelCase						,   cls_token=lowerCamelCase						,   pad_token=lowerCamelCase						,   sp_model_kwargs=self.sp_model_kwargs						,   **lowerCamelCase						,   )

       _snake_case     		=				spm.SentencePieceProcessor(**self.sp_model_kwargs	)
       self.sp_model.Load(str(lowerCamelCase	)	)
       _snake_case     		=				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
       _snake_case     		=				1

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

       _snake_case     		=				len(self.sp_model	)
       _snake_case     		=				{F'''<madeupword{i}>''': sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words	)}
       self.fairseq_tokens_to_ids.update(lowerCamelCase	)

       _snake_case     		=				{v: k for k, v in self.fairseq_tokens_to_ids.items()}

     def __getstate__(					self	):
       _snake_case     		=				self.__dict__.copy()
       _snake_case     		=				None
       _snake_case     		=				self.sp_model.serialized_model_proto()
       return state

     def __setstate__(					self						,   lowerCamelCase	):
       _snake_case     		=				d

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

       _snake_case     		=				spm.SentencePieceProcessor(**self.sp_model_kwargs	)
       self.sp_model.LoadFromSerializedProto(self.sp_model_proto	)

     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase = None	):

       if token_ids_a is None:
         return [self.sep_token_id] + token_ids_a
       _snake_case     		=				[self.sep_token_id]
       return sep + token_ids_a + sep + sep + token_ids_a

     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase = None						,   lowerCamelCase = False	):

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

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

     def 	UpperCamelCase(					self						,   lowerCamelCase						,   lowerCamelCase = None	):
       _snake_case     		=				[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 	UpperCamelCase(					self	):
       return len(self.sp_model	) + self.fairseq_offset + self.num_madeup_words

     def 	UpperCamelCase(					self	):
       _snake_case     		=				{self.convert_ids_to_tokens(lowerCamelCase	): i for i in range(self.vocab_size	)}
       vocab.update(self.added_tokens_encoder	)
       return vocab

     def 	UpperCamelCase(					self						,   lowerCamelCase	):
       return self.sp_model.encode(lowerCamelCase						,   out_type=lowerCamelCase	)

     def 	UpperCamelCase(					self						,   lowerCamelCase	):
       if token in self.fairseq_tokens_to_ids:
         return self.fairseq_tokens_to_ids[token]
       _snake_case     		=				self.sp_model.PieceToId(lowerCamelCase	)

       # 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 	UpperCamelCase(					self						,   lowerCamelCase	):
       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 	UpperCamelCase(					self						,   lowerCamelCase	):
       _snake_case     		=				"".join(lowerCamelCase	).replace(lowerCamelCase						,   " "	).strip()
       return out_string






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

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

       return (out_vocab_file,)


