



import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor

class        __snake_case					(      unittest.TestCase):







							'''simple docstring'''






							def _a					(							self     ,     a_ = False						):
											a__ 					=						10

											if only_cross_attention:
															a__ 					=						12
											else:
															# when only cross attention is not set, the cross attention dim must be the same as the query dim
															a__ 					=						query_dim

											return {
											    "query_dim": query_dim,
											    "cross_attention_dim": cross_attention_dim,
											    "heads": 2,
											    "dim_head": 4,
											    "added_kv_proj_dim": 6,
											    "norm_num_groups": 1,
											    "only_cross_attention": only_cross_attention,
											    "processor": AttnAddedKVProcessor(),
											}


							def _a					(							self     ,     a_     ,     a_						):
											a__ 					=						2

											a__ 					=						torch.rand(_UpperCamelCase     ,     _UpperCamelCase     ,     3     ,     2						)
											a__ 					=						torch.rand(_UpperCamelCase     ,     4     ,     _UpperCamelCase						)
											a__ 					=						None

											return {
											    "hidden_states": hidden_states,
											    "encoder_hidden_states": encoder_hidden_states,
											    "attention_mask": attention_mask,
											}







							def _a					(							self						):
											torch.manual_seed(0						)

											a__ 					=						self.get_constructor_arguments(only_cross_attention=_UpperCamelCase						)
											a__ 					=						Attention(**_UpperCamelCase						)

											self.assertTrue(attn.to_k is not None						)
											self.assertTrue(attn.to_v is not None						)

											a__ 					=						self.get_forward_arguments(
											    query_dim=constructor_args["""query_dim"""]     ,     added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]						)

											a__ 					=						attn(**_UpperCamelCase						)

											# only self attention

											torch.manual_seed(0						)

											a__ 					=						self.get_constructor_arguments(only_cross_attention=_UpperCamelCase						)
											a__ 					=						Attention(**_UpperCamelCase						)

											self.assertTrue(attn.to_k is None						)
											self.assertTrue(attn.to_v is None						)

											a__ 					=						self.get_forward_arguments(
											    query_dim=constructor_args["""query_dim"""]     ,     added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]						)

											a__ 					=						attn(**_UpperCamelCase						)

											self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all()						)




class        __snake_case					(      unittest.TestCase):







							'''simple docstring'''






							def _a					(							self						):
											a__ 					=						DiffusionPipeline.from_pretrained("""hf-internal-testing/tiny-stable-diffusion-pipe"""     ,     safety_checker=_UpperCamelCase						)

											a__ 					=						pipe(
											    """foo"""     ,     num_inference_steps=2     ,     generator=torch.Generator("""cpu"""						).manual_seed(0						)     ,     output_type="""np"""     ,     ).images

											# the initial conversion succeeds
											a__ 					=						DiffusionPipeline.from_pretrained(
											    """hf-internal-testing/tiny-stable-diffusion-pipe"""     ,     device_map="""sequential"""     ,     safety_checker=_UpperCamelCase						)

											a__ 					=						pipe(
											    """foo"""     ,     num_inference_steps=2     ,     generator=torch.Generator("""cpu"""						).manual_seed(0						)     ,     output_type="""np"""     ,     ).images

											with tempfile.TemporaryDirectory() as tmpdir:
															# save the converted model
															pipe.save_pretrained(_UpperCamelCase						)

															# can also load the converted weights
															a__ 					=						DiffusionPipeline.from_pretrained(_UpperCamelCase     ,     device_map="""sequential"""     ,     safety_checker=_UpperCamelCase						)

											a__ 					=						pipe(
											    """foo"""     ,     num_inference_steps=2     ,     generator=torch.Generator("""cpu"""						).manual_seed(0						)     ,     output_type="""np"""     ,     ).images

											self.assertTrue(np.allclose(_UpperCamelCase     ,     _UpperCamelCase     ,     atol=1E-5						)						)
											self.assertTrue(np.allclose(_UpperCamelCase     ,     _UpperCamelCase     ,     atol=1E-5						)						)







import unittest

from transformers.testing_utils import is_torch_available, require_torch, tooslow

from ...generation.test_utils import torch_device
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
    import torch

    from transformers import (
        CpmAntConfig,
        CpmAntForCausalLM,
        CpmAntModel,
        CpmAntTokenizer,
    )
@require_torch
class       __a					:


     """simple docstring"""


     def __init__(	self			: int				,_UpperCamelCase			: str				,_UpperCamelCase			: Optional[int]=2				,_UpperCamelCase			: Tuple=8				,_UpperCamelCase			: Union[str, Any]=True				,_UpperCamelCase			: Tuple=False				,_UpperCamelCase			: Tuple=False				,_UpperCamelCase			: Optional[Any]=False				,_UpperCamelCase			: Union[str, Any]=False				,_UpperCamelCase			: Dict=9_9				,_UpperCamelCase			: int=3_2				,_UpperCamelCase			: Tuple=3				,_UpperCamelCase			: Dict=4				,_UpperCamelCase			: Optional[int]=3_7				,_UpperCamelCase			: int=3_2				,_UpperCamelCase			: Optional[int]=1_2_8				,_UpperCamelCase			: Optional[Any]=8				,_UpperCamelCase			: Optional[int]=8				,_UpperCamelCase			: Tuple=8				,_UpperCamelCase			: Optional[int]=1.0				,_UpperCamelCase			: List[Any]=True				,)		->   Optional[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=parent
         SCREAMING_SNAKE_CASE__								=batch_size
         SCREAMING_SNAKE_CASE__								=seq_length
         SCREAMING_SNAKE_CASE__								=is_training
         SCREAMING_SNAKE_CASE__								=use_token_type_ids
         SCREAMING_SNAKE_CASE__								=use_input_mask
         SCREAMING_SNAKE_CASE__								=use_labels
         SCREAMING_SNAKE_CASE__								=use_mc_token_ids
         SCREAMING_SNAKE_CASE__								=vocab_size
         SCREAMING_SNAKE_CASE__								=hidden_size
         SCREAMING_SNAKE_CASE__								=num_hidden_layers
         SCREAMING_SNAKE_CASE__								=num_attention_heads
         SCREAMING_SNAKE_CASE__								=intermediate_size
         SCREAMING_SNAKE_CASE__								=num_buckets
         SCREAMING_SNAKE_CASE__								=max_distance
         SCREAMING_SNAKE_CASE__								=prompt_length
         SCREAMING_SNAKE_CASE__								=prompt_types
         SCREAMING_SNAKE_CASE__								=segment_types
         SCREAMING_SNAKE_CASE__								=init_std
         SCREAMING_SNAKE_CASE__								=return_dict




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


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								={}
         SCREAMING_SNAKE_CASE__								=ids_tensor([self.batch_size, self.seq_length]				,self.vocab_size						).type(torch.intaa						)
         SCREAMING_SNAKE_CASE__								=False

         SCREAMING_SNAKE_CASE__								=self.get_config()

         return (config, input_ids)




     def   __A				(	self			: Tuple						)		->   List[Any]:


         '''simple docstring'''





         return CpmAntConfig(
             vocab_size=self.vocab_size				,hidden_size=self.hidden_size				,num_hidden_layers=self.num_hidden_layers				,num_attention_heads=self.num_attention_heads				,dim_ff=self.intermediate_size				,position_bias_num_buckets=self.num_buckets				,position_bias_max_distance=self.max_distance				,prompt_types=self.prompt_types				,prompt_length=self.prompt_length				,segment_types=self.segment_types				,use_cache=_UpperCamelCase				,init_std=self.init_std				,return_dict=self.return_dict				,)




     def   __A				(	self			: Any				,_UpperCamelCase			: Dict				,_UpperCamelCase			: List[Any]				,*_UpperCamelCase			: str						)		->   List[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=CpmAntModel(config=_UpperCamelCase						)
         model.to(_UpperCamelCase						)
         model.eval()

         SCREAMING_SNAKE_CASE__								=model(**_UpperCamelCase						).last_hidden_state

         self.parent.assertEqual(hidden_states.shape				,(self.batch_size, self.seq_length, config.hidden_size)						)




     def   __A				(	self			: int				,_UpperCamelCase			: str				,_UpperCamelCase			: Tuple				,*_UpperCamelCase			: Dict						)		->   str:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=CpmAntForCausalLM(_UpperCamelCase						)
         model.to(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=input_ids["""input_ids"""].to(_UpperCamelCase						)
         model.eval()

         SCREAMING_SNAKE_CASE__								=model(**_UpperCamelCase						)
         self.parent.assertEqual(
             model_output.logits.shape				,(self.batch_size, self.seq_length, config.vocab_size + config.prompt_types * config.prompt_length)				,)




     def   __A				(	self			: str						)		->   List[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__			,		SCREAMING_SNAKE_CASE__								=self.prepare_config_and_inputs()
         return config, inputs_dict






@require_torch
class       __a					( __lowerCamelCase							,     __lowerCamelCase							,     unittest.TestCase     ):


     """simple docstring"""


     _A				:    List[Any]									=  (CpmAntModel, CpmAntForCausalLM) if is_torch_available() else ()
     _A				:    Any									=  (
         {"feature-extraction": CpmAntModel, "text-generation": CpmAntForCausalLM} if is_torch_available() else {}
     )

     _A				:    Optional[int]									=  False
     _A				:    Optional[int]									=  False
     _A				:    Optional[Any]									=  False
     _A				:    Any									=  False
     _A				:    int									=  False




     def   __A				(	self			: int						)		->   List[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								=CpmAntModelTester(self						)
         SCREAMING_SNAKE_CASE__								=ConfigTester(self				,config_class=_UpperCamelCase						)




     def   __A				(	self			: Optional[int]						)		->   Dict:


         '''simple docstring'''





         self.config_tester.create_and_test_config_common_properties()
         self.config_tester.create_and_test_config_to_json_string()
         self.config_tester.create_and_test_config_to_json_file()
         self.config_tester.create_and_test_config_from_and_save_pretrained()
         self.config_tester.check_config_can_be_init_without_params()
         self.config_tester.check_config_arguments_init()




     def   __A				(	self			: Tuple						)		->   List[Any]:


         '''simple docstring'''





         unittest.skip("""CPMAnt doesn't support input_embeds."""						)(self.test_inputs_embeds						)




     def   __A				(	self			: str						)		->   Dict:


         '''simple docstring'''





         unittest.skip(
             """CPMAnt doesn't support retain grad in hidden_states or attentions, because prompt management will peel off the output.hidden_states from graph.\
                 So is attentions. We strongly recommand you use loss to tune model."""						)(self.test_retain_grad_hidden_states_attentions						)




     def   __A				(	self			: Any						)		->   List[str]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__			,		SCREAMING_SNAKE_CASE__								=self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_cpmant_model(_UpperCamelCase				,_UpperCamelCase						)




     def   __A				(	self			: str						)		->   Tuple:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__			,		SCREAMING_SNAKE_CASE__								=self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_lm_head_model(_UpperCamelCase				,_UpperCamelCase						)






@require_torch
class       __a					( unittest.TestCase     ):


     """simple docstring"""


     @tooslow
     def   __A				(	self			: Optional[Any]						)		->   Tuple:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								="""今天天气真好！"""
         SCREAMING_SNAKE_CASE__								="""openbmb/cpm-ant-10b"""
         SCREAMING_SNAKE_CASE__								=CpmAntModel.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=CpmAntTokenizer.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=tokenizer(_UpperCamelCase				,return_tensors="""pt"""						)
         SCREAMING_SNAKE_CASE__								=model(**_UpperCamelCase						).last_hidden_state

         SCREAMING_SNAKE_CASE__								=torch.tensor(
             [[[6.1708, 5.9244, 1.0835], [6.5207, 6.2893, -11.3324], [-1.0107, -0.0576, -5.9577]]]				,)
         self.assertTrue(torch.allclose(hidden_states[:, :3, :3]				,_UpperCamelCase				,atol=1e-2						)						)






@require_torch
class       __a					( unittest.TestCase     ):


     """simple docstring"""


     @tooslow
     def   __A				(	self			: List[Any]						)		->   Optional[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								="""今天天气真好！"""
         SCREAMING_SNAKE_CASE__								="""openbmb/cpm-ant-10b"""
         SCREAMING_SNAKE_CASE__								=CpmAntForCausalLM.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=CpmAntTokenizer.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=tokenizer(_UpperCamelCase				,return_tensors="""pt"""						)
         SCREAMING_SNAKE_CASE__								=model(**_UpperCamelCase						).logits

         SCREAMING_SNAKE_CASE__								=torch.tensor(
             [[[-6.4267, -6.4083, -6.3958], [-5.8802, -5.9447, -5.7811], [-5.3896, -5.4820, -5.4295]]]				,)
         self.assertTrue(torch.allclose(hidden_states[:, :3, :3]				,_UpperCamelCase				,atol=1e-2						)						)




     @tooslow
     def   __A				(	self			: Any						)		->   Any:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								="""openbmb/cpm-ant-10b"""
         SCREAMING_SNAKE_CASE__								=CpmAntForCausalLM.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=CpmAntTokenizer.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								="""今天天气不错，"""
         SCREAMING_SNAKE_CASE__								="""今天天气不错，阳光明媚，我和妈妈一起去超市买东西。\n在超市里，我看到了一个很好玩的玩具，它的名字叫“机器人”。它有一个圆圆的脑袋，两只圆圆的眼睛，还有一个圆圆的"""
         SCREAMING_SNAKE_CASE__								=tokenizer(_UpperCamelCase				,return_tensors="""pt"""						)
         SCREAMING_SNAKE_CASE__								=model.generate(**_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=tokenizer.batch_decode(_UpperCamelCase						)
         self.assertEqual(_UpperCamelCase				,_UpperCamelCase						)




     @tooslow
     def   __A				(	self			: Optional[Any]						)		->   Union[str, Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__								="""openbmb/cpm-ant-10b"""
         SCREAMING_SNAKE_CASE__								=CpmAntForCausalLM.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=CpmAntTokenizer.from_pretrained(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=["""今天天气不错，""", """新年快乐，万事如意！"""]
         SCREAMING_SNAKE_CASE__								=[
             """今天天气不错，阳光明媚，我和妈妈一起去超市买东西。\n在超市里，我看到了一个很好玩的玩具，它的名字叫“机器人”。它有一个圆圆的脑袋，两只圆圆的眼睛，还有一个圆圆的""",
             """新年快乐，万事如意！在这辞旧迎新的美好时刻，我谨代表《农村新技术》杂志社全体同仁，向一直以来关心、支持《农村新技术》杂志发展的各级领导、各界朋友和广大读者致以最诚挚的""",
         ]
         SCREAMING_SNAKE_CASE__								=tokenizer(_UpperCamelCase				,return_tensors="""pt"""				,padding=_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=model.generate(**_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__								=tokenizer.batch_decode(_UpperCamelCase						)
         self.assertEqual(_UpperCamelCase				,_UpperCamelCase						)






