





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   lowercase__     :







  def __init__(						self     ,    _lowercase     ,    _lowercase=2     ,    _lowercase=8     ,    _lowercase=True     ,    _lowercase=False     ,    _lowercase=False     ,    _lowercase=False     ,    _lowercase=False     ,    _lowercase=99     ,    _lowercase=32     ,    _lowercase=3     ,    _lowercase=4     ,    _lowercase=37     ,    _lowercase=32     ,    _lowercase=128     ,    _lowercase=8     ,    _lowercase=8     ,    _lowercase=8     ,    _lowercase=1.0     ,    _lowercase=True     ,    ):
       lowerCAmelCase_       :							Optional[Any]			 =		parent
       lowerCAmelCase_       :							Optional[int]			 =		batch_size
       lowerCAmelCase_       :							Dict			 =		seq_length
       lowerCAmelCase_       :							str			 =		is_training
       lowerCAmelCase_       :							Dict			 =		use_token_type_ids
       lowerCAmelCase_       :							List[str]			 =		use_input_mask
       lowerCAmelCase_       :							Optional[Any]			 =		use_labels
       lowerCAmelCase_       :							List[Any]			 =		use_mc_token_ids
       lowerCAmelCase_       :							Union[str, Any]			 =		vocab_size
       lowerCAmelCase_       :							str			 =		hidden_size
       lowerCAmelCase_       :							Tuple			 =		num_hidden_layers
       lowerCAmelCase_       :							str			 =		num_attention_heads
       lowerCAmelCase_       :							str			 =		intermediate_size
       lowerCAmelCase_       :							Optional[int]			 =		num_buckets
       lowerCAmelCase_       :							Any			 =		max_distance
       lowerCAmelCase_       :							str			 =		prompt_length
       lowerCAmelCase_       :							Optional[Any]			 =		prompt_types
       lowerCAmelCase_       :							int			 =		segment_types
       lowerCAmelCase_       :							List[Any]			 =		init_std
       lowerCAmelCase_       :							List[Any]			 =		return_dict







  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							Tuple			 =		{}
       lowerCAmelCase_       :							List[Any]			 =		ids_tensor([self.batch_size, self.seq_length]     ,    self.vocab_size    ).type(torch.intaa    )
       lowerCAmelCase_       :							Optional[Any]			 =		False

       lowerCAmelCase_       :							Dict			 =		self.get_config()

       return (config, input_ids)







  def 						UpperCAmelCase__		(						self    ):
       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=_lowercase     ,    init_std=self.init_std     ,    return_dict=self.return_dict     ,    )







  def 						UpperCAmelCase__		(						self     ,    _lowercase     ,    _lowercase     ,    *_lowercase    ):
       lowerCAmelCase_       :							List[Any]			 =		CpmAntModel(config=_lowercase    )
       model.to(_lowercase    )
       model.eval()

       lowerCAmelCase_       :							Union[str, Any]			 =		model(**_lowercase    ).last_hidden_state

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







  def 						UpperCAmelCase__		(						self     ,    _lowercase     ,    _lowercase     ,    *_lowercase    ):
       lowerCAmelCase_       :							Dict			 =		CpmAntForCausalLM(_lowercase    )
       model.to(_lowercase    )
       lowerCAmelCase_       :							Optional[Any]			 =		input_ids['''input_ids'''].to(_lowercase    )
       model.eval()

       lowerCAmelCase_       :							Any			 =		model(**_lowercase    )
       self.parent.assertEqual(
           model_output.logits.shape     ,    (self.batch_size, self.seq_length, config.vocab_size + config.prompt_types * config.prompt_length)     ,    )







  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							Union[str, Any]			 =		self.prepare_config_and_inputs()
       return config, inputs_dict





@require_torch
class   lowercase__     (				__A  ,       __A  ,       unittest.TestCase	):
  __UpperCamelCase						  =	(CpmAntModel, CpmAntForCausalLM) if is_torch_available() else ()
  __UpperCamelCase						  =	(
      {'''feature-extraction''': CpmAntModel, '''text-generation''': CpmAntForCausalLM} if is_torch_available() else {}
  )

  __UpperCamelCase						  =	False
  __UpperCamelCase						  =	False
  __UpperCamelCase						  =	False
  __UpperCamelCase						  =	False
  __UpperCamelCase						  =	False







  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							Dict			 =		CpmAntModelTester(self    )
       lowerCAmelCase_       :							List[Any]			 =		ConfigTester(self     ,    config_class=_lowercase    )







  def 						UpperCAmelCase__		(						self    ):
       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 						UpperCAmelCase__		(						self    ):
       unittest.skip("""CPMAnt doesn\'t support input_embeds."""    )(self.test_inputs_embeds    )







  def 						UpperCAmelCase__		(						self    ):
       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 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							List[Any]			 =		self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_cpmant_model(_lowercase     ,    _lowercase    )







  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							List[Any]			 =		self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_lm_head_model(_lowercase     ,    _lowercase    )





@require_torch
class   lowercase__     (				unittest.TestCase	):







  @tooslow
  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							str			 =		'''今天天气真好！'''
       lowerCAmelCase_       :							Optional[Any]			 =		'''openbmb/cpm-ant-10b'''
       lowerCAmelCase_       :							List[str]			 =		CpmAntModel.from_pretrained(_lowercase    )
       lowerCAmelCase_       :							List[Any]			 =		CpmAntTokenizer.from_pretrained(_lowercase    )
       lowerCAmelCase_       :							Dict			 =		tokenizer(_lowercase     ,    return_tensors="""pt"""    )
       lowerCAmelCase_       :							Tuple			 =		model(**_lowercase    ).last_hidden_state

       lowerCAmelCase_       :							int			 =		torch.tensor(
           [[[6.1708, 5.9244, 1.0835], [6.5207, 6.2893, -11.3_324], [-1.0107, -0.0576, -5.9577]]]     ,    )
       self.assertTrue(torch.allclose(hidden_states[:, :3, :3]     ,    _lowercase     ,    atol=1e-2    )    )





@require_torch
class   lowercase__     (				unittest.TestCase	):







  @tooslow
  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							Tuple			 =		'''今天天气真好！'''
       lowerCAmelCase_       :							str			 =		'''openbmb/cpm-ant-10b'''
       lowerCAmelCase_       :							Tuple			 =		CpmAntForCausalLM.from_pretrained(_lowercase    )
       lowerCAmelCase_       :							str			 =		CpmAntTokenizer.from_pretrained(_lowercase    )
       lowerCAmelCase_       :							Tuple			 =		tokenizer(_lowercase     ,    return_tensors="""pt"""    )
       lowerCAmelCase_       :							Optional[int]			 =		model(**_lowercase    ).logits

       lowerCAmelCase_       :							Union[str, Any]			 =		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]     ,    _lowercase     ,    atol=1e-2    )    )







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







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








"""simple docstring"""





import numpy as np



def      __lowercase       (     a			:					Union[str, Any]							,		a			:					Optional[Any]							,		a			:					List[str]			)       ->       Dict:
      __snake_case   :     Dict	      =n_draws * class_counts / class_counts.sum()
      # floored means we don't overshoot n_samples, but probably undershoot
      __snake_case   :     int	      =np.floor(a			)
      # we add samples according to how much "left over" probability
      # they had, until we arrive at n_samples
      __snake_case   :     Dict	      =int(n_draws - floored.sum()			)
      if need_to_add > 0:
            __snake_case   :     List[str]	      =continuous - floored
            __snake_case   :     str	      =np.sort(np.unique(a			)			)[::-1]
            # add according to remainder, but break ties
            # randomly to avoid biases
            for value in values:
                  ((__snake_case)      ,					)   :     Tuple	      =np.where(remainder == value			)
                  # if we need_to_add less than what's in inds
                  # we draw randomly from them.
                  # if we need to add more, we add them all and
                  # go to the next value
                  __snake_case   :     int	      =min(len(a			)							,		a			)
                  __snake_case   :     Any	      =rng.choice(a							,		size=a							,		replace=a			)
                  floored[inds] += 1
                  need_to_add -= add_now
                  if need_to_add == 0:
                        break
      return floored.astype(np.intaa			)



def      __lowercase       (     a			:					Any							,		a			:					str							,		a			:					Dict							,		a			:					Union[str, Any]							,		a			:					Union[str, Any]=10			)       ->       Any:
      __snake_case      ,					__snake_case   :     int	      =np.unique(a							,		return_inverse=a			)
      __snake_case   :     Dict	      =classes.shape[0]
      __snake_case   :     int	      =np.bincount(a			)
      if np.min(a			) < 2:
            raise ValueError('''Minimum class count error'''			)
      if n_train < n_classes:
            raise ValueError(
                '''The train_size = %d should be greater or ''' '''equal to the number of classes = %d''' % (n_train, n_classes)			)
      if n_test < n_classes:
            raise ValueError(
                '''The test_size = %d should be greater or ''' '''equal to the number of classes = %d''' % (n_test, n_classes)			)
      __snake_case   :     str	      =np.split(np.argsort(a							,		kind='''mergesort'''			)							,		np.cumsum(a			)[:-1]			)
      for _ in range(a			):
            __snake_case   :     str	      =approximate_mode(a							,		a							,		a			)
            __snake_case   :     List[str]	      =class_counts - n_i
            __snake_case   :     Union[str, Any]	      =approximate_mode(a							,		a							,		a			)

            __snake_case   :     int	      =[]
            __snake_case   :     Union[str, Any]	      =[]

            for i in range(a			):
                  __snake_case   :     Union[str, Any]	      =rng.permutation(class_counts[i]			)
                  __snake_case   :     Dict	      =class_indices[i].take(a							,		mode='''clip'''			)
                  train.extend(perm_indices_class_i[: n_i[i]]			)
                  test.extend(perm_indices_class_i[n_i[i] : n_i[i] + t_i[i]]			)
            __snake_case   :     Any	      =rng.permutation(a			)
            __snake_case   :     int	      =rng.permutation(a			)

            yield train, test


