

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  __UpperCAmelCase					(	unittest.TestCase						):
   UpperCamelCase       					=					MODEL_FOR_CAUSAL_LM_MAPPING
   UpperCamelCase       					=					TF_MODEL_FOR_CAUSAL_LM_MAPPING



   @require_torch
   def 	__magic_name__   (							self							:					Optional[Any]						):
     UpperCAmelCase	:							Any						=			pipeline(task='''text-generation''',		model='''sshleifer/tiny-ctrl''',		framework='''pt'''						)
     # Using `do_sample=False` to force deterministic output
     UpperCAmelCase	:							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@@'''
                 )
             }
         ],		)

     UpperCAmelCase	:							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@@'''
                     )
                 }
             ],
         ],		)

     UpperCAmelCase	:							List[Any]						=			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						)},
         ],		)
     UpperCAmelCase	:							Tuple						=			text_generator.model.config.eos_token_id
     UpperCAmelCase	:							Dict						=			'''<pad>'''
     UpperCAmelCase	:							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 	__magic_name__   (							self							:					Union[str, Any]						):
     UpperCAmelCase	:							Optional[int]						=			pipeline(task='''text-generation''',		model='''sshleifer/tiny-ctrl''',		framework='''tf'''						)

     # Using `do_sample=False` to force deterministic output
     UpperCAmelCase	:							Union[str, Any]						=			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,'''
                 )
             }
         ],		)

     UpperCAmelCase	:							Union[str, Any]						=			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 	__magic_name__   (							self							:					Optional[int],		__A							:					int,		__A							:					List[Any],		__A							:					List[str]						):
     UpperCAmelCase	:							List[str]						=			TextGenerationPipeline(model=__A,		tokenizer=__A						)
     return text_generator, ["This is a test", "Another test"]



   def 	__magic_name__   (							self							:					List[Any]						):
     UpperCAmelCase	:							Dict						=			'''Hello I believe in'''
     UpperCAmelCase	:							str						=			pipeline('''text-generation''',		model='''hf-internal-testing/tiny-random-gpt2'''						)
     UpperCAmelCase	:							Dict						=			text_generator(__A						)
     self.assertEqual(
         __A,		[{'''generated_text''': '''Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe'''}],		)

     UpperCAmelCase	:							Optional[int]						=			text_generator(__A,		stop_sequence=''' fe'''						)
     self.assertEqual(__A,		[{'''generated_text''': '''Hello I believe in fe'''}]						)



   def 	__magic_name__   (							self							:					Union[str, Any],		__A							:					Any,		__A							:					List[str]						):
     UpperCAmelCase	:							Tuple						=			text_generator.model
     UpperCAmelCase	:							List[Any]						=			text_generator.tokenizer

     UpperCAmelCase	:							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'''						)						)

     UpperCAmelCase	:							Dict						=			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''']						)

     UpperCAmelCase	:							List[Any]						=			pipeline(task='''text-generation''',		model=__A,		tokenizer=__A,		return_full_text=__A						)
     UpperCAmelCase	:							Optional[int]						=			text_generator('''This is a test'''						)
     self.assertEqual(__A,		[{'''generated_text''': ANY(__A						)}]						)
     self.assertNotIn('''This is a test''',		outputs[0]['''generated_text''']						)

     UpperCAmelCase	:							str						=			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'''						)						)

     UpperCAmelCase	:							List[str]						=			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:
       UpperCAmelCase	:							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						):
       UpperCAmelCase	:							Optional[int]						=			text_generator('''test''',		return_full_text=__A,		return_text=__A						)
     with self.assertRaises(__A						):
       UpperCAmelCase	:							List[str]						=			text_generator('''test''',		return_full_text=__A,		return_tensors=__A						)
     with self.assertRaises(__A						):
       UpperCAmelCase	:							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__
     ):
       UpperCAmelCase	:							Dict						=			text_generator(''''''						)
       self.assertEqual(__A,		[{'''generated_text''': ANY(__A						)}]						)
     else:
       with self.assertRaises((ValueError, AssertionError)						):
         UpperCAmelCase	:							Optional[Any]						=			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	:							int						=			['''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						)

       UpperCAmelCase	:							Dict						=			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 	__magic_name__   (							self							:					Dict						):
     import torch

     # Classic `model_kwargs`
     UpperCAmelCase	:							str						=			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	:							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.)
     UpperCAmelCase	:							str						=			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	:							Tuple						=			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
     UpperCAmelCase	:							Dict						=			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	:							Union[str, Any]						=			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 	__magic_name__   (							self							:					Tuple						):
     import torch

     UpperCAmelCase	:							str						=			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 	__magic_name__   (							self							:					int						):
     import torch

     UpperCAmelCase	:							Optional[int]						=			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 	__magic_name__   (							self							:					List[Any]						):
     UpperCAmelCase	:							Union[str, Any]						=			'''Hello world'''
     UpperCAmelCase	:							Dict						=			pipeline('''text-generation''',		model='''hf-internal-testing/tiny-random-gpt2'''						)
     if text_generator.model.framework == "tf":
       UpperCAmelCase	:							Tuple						=			logging.get_logger('''transformers.generation.tf_utils'''						)
     else:
       UpperCAmelCase	:							Optional[int]						=			logging.get_logger('''transformers.generation.utils'''						)
     UpperCAmelCase	:							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:
       UpperCAmelCase	:							int						=			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:
       UpperCAmelCase	:							Union[str, Any]						=			text_generator(__A,		max_new_tokens=1						)
     self.assertNotIn(__A,		cl.out						)

     with CaptureLogger(__A						) as cl:
       UpperCAmelCase	:							Tuple						=			text_generator(__A,		max_length=1_0						)
     self.assertNotIn(__A,		cl.out						)








def       a__					(		UpperCAmelCase							:      int    ,			UpperCAmelCase							:      int      )  ->						str:
  if a < 0 or b < 0:
    raise ValueError('''the value of both inputs must be positive'''      )
  UpperCAmelCase	:							List[Any]						=			str(bin(UpperCAmelCase      )      )[2:]  # remove the leading "0b"
  UpperCAmelCase	:							List[str]						=			str(bin(UpperCAmelCase      )      )[2:]
  UpperCAmelCase	:							Optional[Any]						=			max(len(UpperCAmelCase      )    ,			len(UpperCAmelCase      )      )
  return "0b" + "".join(
      str(int('''1''' in (char_a, char_b)      )      )
      for char_a, char_b in zip(a_binary.zfill(UpperCAmelCase      )    ,			b_binary.zfill(UpperCAmelCase      )      )      )


if __name__ == "__main__":
     import doctest

     doctest.testmod()





