




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

					'''simple docstring'''





					lowerCAmelCase    :	str       		=  MODEL_FOR_CAUSAL_LM_MAPPING
					lowerCAmelCase    :	Optional[int]       		=  TF_MODEL_FOR_CAUSAL_LM_MAPPING
					@require_torch
					def   lowerCamelCase_       (     self       :      Optional[Any]    ):
									UpperCAmelCase_										=					pipeline(task='''text-generation''' ,   model='''sshleifer/tiny-ctrl''' ,   framework='''pt'''    )
									# Using `do_sample=False` to force deterministic output
									UpperCAmelCase_										=					text_generator('''This is a test''' ,   do_sample=__snake_case    )
									self.assertEqual(
									    __snake_case ,   [
									        {
									            '''generated_text''': (
									                '''This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'''
									                ''' oscope. FiliFili@@'''
									            )
									        }
									    ] ,   )

									UpperCAmelCase_										=					text_generator(['''This is a test''', '''This is a second test''']    )
									self.assertEqual(
									    __snake_case ,   [
									        [
									            {
									                '''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@@'''
									                )
									            }
									        ],
									    ] ,   )

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

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

									UpperCAmelCase_										=					text_generator(['''This is a test''', '''This is a second test'''] ,   do_sample=__snake_case    )
									self.assertEqual(
									    __snake_case ,   [
									        [
									            {
									                '''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   lowerCamelCase_       (     self       :      List[Any] ,   __snake_case       :      Optional[Any] ,   __snake_case       :      int ,   __snake_case       :      List[str]    ):
									UpperCAmelCase_										=					TextGenerationPipeline(model=__snake_case ,   tokenizer=__snake_case    )
									return text_generator, ["This is a test", "Another test"]
					def   lowerCamelCase_       (     self       :      Optional[int]    ):
									UpperCAmelCase_										=					'''Hello I believe in'''
									UpperCAmelCase_										=					pipeline('''text-generation''' ,   model='''hf-internal-testing/tiny-random-gpt2'''    )
									UpperCAmelCase_										=					text_generator(__snake_case    )
									self.assertEqual(
									    __snake_case ,   [{'''generated_text''': '''Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe'''}] ,   )

									UpperCAmelCase_										=					text_generator(__snake_case ,   stop_sequence=''' fe'''    )
									self.assertEqual(__snake_case ,   [{'''generated_text''': '''Hello I believe in fe'''}]    )
					def   lowerCamelCase_       (     self       :      Any ,   __snake_case       :      Optional[Any] ,   __snake_case       :      Union[str, Any]    ):
									UpperCAmelCase_										=					text_generator.model
									UpperCAmelCase_										=					text_generator.tokenizer

									UpperCAmelCase_										=					text_generator('''This is a test'''    )
									self.assertEqual(__snake_case ,   [{'''generated_text''': ANY(__snake_case    )}]    )
									self.assertTrue(outputs[0]['''generated_text'''].startswith('''This is a test'''    )    )

									UpperCAmelCase_										=					text_generator('''This is a test''' ,   return_full_text=__snake_case    )
									self.assertEqual(__snake_case ,   [{'''generated_text''': ANY(__snake_case    )}]    )
									self.assertNotIn('''This is a test''' ,   outputs[0]['''generated_text''']    )

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

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

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

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

									with self.assertRaises(__snake_case    ):
													UpperCAmelCase_										=					text_generator('''test''' ,   return_full_text=__snake_case ,   return_text=__snake_case    )
									with self.assertRaises(__snake_case    ):
													UpperCAmelCase_										=					text_generator('''test''' ,   return_full_text=__snake_case ,   return_tensors=__snake_case    )
									with self.assertRaises(__snake_case    ):
													UpperCAmelCase_										=					text_generator('''test''' ,   return_text=__snake_case ,   return_tensors=__snake_case    )

									# 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__
									):
													UpperCAmelCase_										=					text_generator(''''''    )
													self.assertEqual(__snake_case ,   [{'''generated_text''': ANY(__snake_case    )}]    )
									else:
													with self.assertRaises((ValueError, AssertionError)    ):
																	UpperCAmelCase_										=					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.
									UpperCAmelCase_										=					['''RwkvForCausalLM''', '''XGLMForCausalLM''', '''GPTNeoXForCausalLM''']
									if (
									    tokenizer.model_max_length < 1_00_00
									    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_00 ,   max_new_tokens=20    )

													UpperCAmelCase_										=					text_generator('''This is a test''' * 5_00 ,   handle_long_generation='''hole''' ,   max_new_tokens=20    )
													# Hole strategy cannot work
													with self.assertRaises(__snake_case    ):
																	text_generator(
																	    '''This is a test''' * 5_00 ,   handle_long_generation='''hole''' ,   max_new_tokens=tokenizer.model_max_length + 10 ,   )
					@require_torch
					@require_accelerate
					@require_torch_gpu
					def   lowerCamelCase_       (     self       :      List[Any]    ):
									import torch

									# Classic `model_kwargs`
									UpperCAmelCase_										=					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    )
									UpperCAmelCase_										=					pipe('''This is a test'''    )
									self.assertEqual(
									    __snake_case ,   [
									        {
									            '''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.)
									UpperCAmelCase_										=					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    )
									UpperCAmelCase_										=					pipe('''This is a test'''    )
									self.assertEqual(
									    __snake_case ,   [
									        {
									            '''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
									UpperCAmelCase_										=					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    )
									UpperCAmelCase_										=					pipe('''This is a test'''    )
									self.assertEqual(
									    __snake_case ,   [
									        {
									            '''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   lowerCamelCase_       (     self       :      str    ):
									import torch

									UpperCAmelCase_										=					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   lowerCamelCase_       (     self       :      Any    ):
									import torch

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




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

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

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











import argparse
import logging
import pickle
from collections import Counter


logging.basicConfig(
    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s', datefmt='%m/%d/%Y %H:%M:%S', level=logging.INFO
)
_lowerCamelCase      			=						logging.getLogger(__name__)

if __name__ == "__main__":
 _lowerCamelCase      			=						argparse.ArgumentParser(
     description='Token Counts for smoothing the masking probabilities in MLM (cf XLM/word2vec)'
 )
 parser.add_argument(
     '--data_file', type=str, default='data/dump.bert-base-uncased.pickle', help='The binarized dataset.'
 )
 parser.add_argument(
     '--token_counts_dump', type=str, default='data/token_counts.bert-base-uncased.pickle', help='The dump file.'
 )
 parser.add_argument('--vocab_size', default=3_05_22, type=int)
 _lowerCamelCase      			=						parser.parse_args()

 logger.info(F"Loading data from {args.data_file}")
 with open(args.data_file, 'rb') as fp:
  _lowerCamelCase      			=						pickle.load(fp)

 logger.info('Counting occurrences for MLM.')
 _lowerCamelCase      			=						Counter()
 for tk_ids in data:
  counter.update(tk_ids)
 _lowerCamelCase      			=						[0] * args.vocab_size
 for k, v in counter.items():
  _lowerCamelCase      			=						v

 logger.info(F"Dump to {args.token_counts_dump}")
 with open(args.token_counts_dump, 'wb') as handle:
  pickle.dump(counts, handle, protocol=pickle.HIGHEST_PROTOCOL)





