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






 """simple docstring"""




 def __init__(    self ,       __magic_name__ ,       __magic_name__=2 ,       __magic_name__=8 ,       __magic_name__=True ,       __magic_name__=False ,       __magic_name__=False ,       __magic_name__=False ,       __magic_name__=False ,       __magic_name__=99 ,       __magic_name__=32 ,       __magic_name__=3 ,       __magic_name__=4 ,       __magic_name__=37 ,       __magic_name__=32 ,       __magic_name__=128 ,       __magic_name__=8 ,       __magic_name__=8 ,       __magic_name__=8 ,       __magic_name__=1.0 ,       __magic_name__=True ,       ):

        """simple docstring"""

        A_							:       int	      =				parent
        A_							:       Any	      =				batch_size
        A_							:       Dict	      =				seq_length
        A_							:       List[str]	      =				is_training
        A_							:       List[Any]	      =				use_token_type_ids
        A_							:       Dict	      =				use_input_mask
        A_							:       Union[str, Any]	      =				use_labels
        A_							:       Union[str, Any]	      =				use_mc_token_ids
        A_							:       Optional[int]	      =				vocab_size
        A_							:       Tuple	      =				hidden_size
        A_							:       Optional[Any]	      =				num_hidden_layers
        A_							:       int	      =				num_attention_heads
        A_							:       List[Any]	      =				intermediate_size
        A_							:       Optional[int]	      =				num_buckets
        A_							:       Union[str, Any]	      =				max_distance
        A_							:       Tuple	      =				prompt_length
        A_							:       Any	      =				prompt_types
        A_							:       Tuple	      =				segment_types
        A_							:       int	      =				init_std
        A_							:       List[str]	      =				return_dict
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       Tuple	      =				{}
        A_							:       str	      =				ids_tensor([self.batch_size, self.seq_length] ,       self.vocab_size      ).type(torch.intaa      )
        A_							:       List[str]	      =				False

        A_							:       Optional[Any]	      =				self.get_config()

        return (config, input_ids)
 def  UpperCAmelCase   (    self      ):

        """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=lowerCAmelCase_ ,       init_std=self.init_std ,       return_dict=self.return_dict ,       )
 def  UpperCAmelCase   (    self ,       __magic_name__ ,       __magic_name__ ,       *__magic_name__      ):

        """simple docstring"""

        A_							:       Any	      =				CpmAntModel(config=lowerCAmelCase_      )
        model.to(lowerCAmelCase_      )
        model.eval()

        A_							:       str	      =				model(**lowerCAmelCase_      ).last_hidden_state

        self.parent.assertEqual(hidden_states.shape ,       (self.batch_size, self.seq_length, config.hidden_size)      )
 def  UpperCAmelCase   (    self ,       __magic_name__ ,       __magic_name__ ,       *__magic_name__      ):

        """simple docstring"""

        A_							:       List[str]	      =				CpmAntForCausalLM(lowerCAmelCase_      )
        model.to(lowerCAmelCase_      )
        A_							:       Optional[int]	      =				input_ids['''input_ids'''].to(lowerCAmelCase_      )
        model.eval()

        A_							:       Tuple	      =				model(**lowerCAmelCase_      )
        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      ):

        """simple docstring"""

        A_      ,							A_							:       List[str]	      =				self.prepare_config_and_inputs()
        return config, inputs_dict






@require_torch
class     __UpperCAmelCase(    A__      ,	A__      ,	unittest.TestCase				):






 """simple docstring"""




 __magic_name__          =   (CpmAntModel, CpmAntForCausalLM) if is_torch_available() else ()
 __magic_name__          =   (
     {"""feature-extraction""": CpmAntModel, """text-generation""": CpmAntForCausalLM} if is_torch_available() else {}
 )

 __magic_name__          =   False
 __magic_name__          =   False
 __magic_name__          =   False
 __magic_name__          =   False
 __magic_name__          =   False
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       str	      =				CpmAntModelTester(self      )
        A_							:       Optional[int]	      =				ConfigTester(self ,       config_class=lowerCAmelCase_      )
 def  UpperCAmelCase   (    self      ):

        """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  UpperCAmelCase   (    self      ):

        """simple docstring"""

        unittest.skip('''CPMAnt doesn\'t support input_embeds.'''      )(self.test_inputs_embeds      )
 def  UpperCAmelCase   (    self      ):

        """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  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_      ,							A_							:       Optional[int]	      =				self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_cpmant_model(lowerCAmelCase_ ,       lowerCAmelCase_      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_      ,							A_							:       str	      =				self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_lm_head_model(lowerCAmelCase_ ,       lowerCAmelCase_      )






@require_torch
class     __UpperCAmelCase(    unittest.TestCase				):






 """simple docstring"""




 @tooslow
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       str	      =				'''今天天气真好！'''
        A_							:       Union[str, Any]	      =				'''openbmb/cpm-ant-10b'''
        A_							:       str	      =				CpmAntModel.from_pretrained(lowerCAmelCase_      )
        A_							:       str	      =				CpmAntTokenizer.from_pretrained(lowerCAmelCase_      )
        A_							:       int	      =				tokenizer(lowerCAmelCase_ ,       return_tensors='''pt'''      )
        A_							:       Union[str, Any]	      =				model(**lowerCAmelCase_      ).last_hidden_state

        A_							:       Union[str, Any]	      =				torch.tensor(
            [[[6.17_08, 5.92_44, 1.08_35], [6.52_07, 6.28_93, -11.33_24], [-1.01_07, -0.05_76, -5.95_77]]] ,       )
        self.assertTrue(torch.allclose(hidden_states[:, :3, :3] ,       lowerCAmelCase_ ,       atol=1e-2      )      )






@require_torch
class     __UpperCAmelCase(    unittest.TestCase				):






 """simple docstring"""




 @tooslow
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       str	      =				'''今天天气真好！'''
        A_							:       List[Any]	      =				'''openbmb/cpm-ant-10b'''
        A_							:       Tuple	      =				CpmAntForCausalLM.from_pretrained(lowerCAmelCase_      )
        A_							:       int	      =				CpmAntTokenizer.from_pretrained(lowerCAmelCase_      )
        A_							:       Optional[int]	      =				tokenizer(lowerCAmelCase_ ,       return_tensors='''pt'''      )
        A_							:       Union[str, Any]	      =				model(**lowerCAmelCase_      ).logits

        A_							:       str	      =				torch.tensor(
            [[[-6.42_67, -6.40_83, -6.39_58], [-5.88_02, -5.94_47, -5.78_11], [-5.38_96, -5.48_20, -5.42_95]]] ,       )
        self.assertTrue(torch.allclose(hidden_states[:, :3, :3] ,       lowerCAmelCase_ ,       atol=1e-2      )      )
 @tooslow
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

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

        """simple docstring"""

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




'''simple docstring'''





import os
import sys
from pathlib import Path

import pytest

from datasets import Dataset, IterableDataset
from datasets.distributed import split_dataset_by_node

from .utils import execute_subprocess_async, get_torch_dist_unique_port, require_torch
def 	snake_case_							(				):
      snake_case_    							=				Dataset.from_dict({"i": range(17      )}      )
      snake_case_    							=				len(_a      )
      snake_case_    							=				3
      snake_case_    							=				[
          split_dataset_by_node(_a			,       rank=_a			,       world_size=_a      ) for rank in range(_a      )
      ]
      assert sum(len(_a      ) for ds in datasets_per_rank      ) == full_size
      assert len({tuple(x.values()      ) for ds in datasets_per_rank for x in ds}      ) == full_size
def 	snake_case_							(				):
      def gen():
            return ({"i": i} for i in range(17      ))

      snake_case_    							=				3
      snake_case_    							=				IterableDataset.from_generator(_a      )
      snake_case_    							=				len(list(_a      )      )
      snake_case_    							=				[
          split_dataset_by_node(_a			,       rank=_a			,       world_size=_a      ) for rank in range(_a      )
      ]
      assert sum(len(list(_a      )      ) for ds in datasets_per_rank      ) == full_size
      assert len({tuple(x.values()      ) for ds in datasets_per_rank for x in ds}      ) == full_size
@pytest.mark.parametrize("shards_per_node"			,       [1, 2, 3]      )
def 	snake_case_							(				_a  :				List[Any]      ):
      def gen(_a  :				Dict      ):
            for shard in shards:
                  yield from ({"i": i, "shard": shard} for i in range(17      ))

      snake_case_    							=				3
      snake_case_    							=				shards_per_node * world_size
      snake_case_    							=				{"shards": [F'''shard_{shard_idx}.txt''' for shard_idx in range(_a      )]}
      snake_case_    							=				IterableDataset.from_generator(_a			,       gen_kwargs=_a      )
      snake_case_    							=				len(list(_a      )      )
      assert full_ds.n_shards == world_size * shards_per_node
      snake_case_    							=				[
          split_dataset_by_node(_a			,       rank=_a			,       world_size=_a      ) for rank in range(_a      )
      ]
      assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
      assert sum(len(list(_a      )      ) for ds in datasets_per_rank      ) == full_size
      assert len({tuple(x.values()      ) for ds in datasets_per_rank for x in ds}      ) == full_size
def 	snake_case_							(				):
      def gen():
            return ({"i": i} for i in range(17      ))

      snake_case_    							=				2
      snake_case_    							=				IterableDataset.from_generator(_a      )
      snake_case_    							=				len(list(_a      )      )

      snake_case_    							=				split_dataset_by_node(_a			,       rank=0			,       world_size=_a      ).shuffle(seed=42      )
      assert len(list(_a      )      ) == 1 + full_size // world_size
      with pytest.raises(_a      ):
            split_dataset_by_node(_a			,       rank=0			,       world_size=_a      ).shuffle()

      snake_case_    							=				split_dataset_by_node(full_ds.shuffle(seed=42      )			,       rank=0			,       world_size=_a      )
      assert len(list(_a      )      ) == 1 + full_size // world_size
      with pytest.raises(_a      ):
            split_dataset_by_node(full_ds.shuffle()			,       rank=0			,       world_size=_a      )
@pytest.mark.parametrize("streaming"			,       [False, True]      )
@require_torch
@pytest.mark.skipif(os.name == "nt"			,       reason="execute_subprocess_async doesn't support windows"      )
@pytest.mark.integration
def 	snake_case_							(				_a  :				Any      ):
      snake_case_    							=				2
      snake_case_    							=				get_torch_dist_unique_port()
      snake_case_    							=				Path(_a      ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
      snake_case_    							=				F'''
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    '''.split()
      snake_case_    							=				F'''
        --streaming={streaming}
    '''.split()
      snake_case_    							=				[sys.executable] + distributed_args + args
      execute_subprocess_async(_a			,       env=os.environ.copy()      )




@pytest.mark.parametrize(
    "nproc_per_node, num_workers"			,       [
        (2, 2),  # each node has 2 shards and each worker has 1 shards
        (3, 2),  # each node uses all the shards but skips examples, and each worker has 2 shards
    ]			,       )
@require_torch
@pytest.mark.skipif(os.name == "nt"			,       reason="execute_subprocess_async doesn't support windows"      )
@pytest.mark.integration
def 	snake_case_							(				_a  :				int			,       _a  :				List[Any]      ):
      snake_case_    							=				True
      snake_case_    							=				get_torch_dist_unique_port()
      snake_case_    							=				Path(_a      ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
      snake_case_    							=				F'''
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    '''.split()
      snake_case_    							=				F'''
        --streaming={streaming}
        --num_workers={num_workers}
    '''.split()
      snake_case_    							=				[sys.executable] + distributed_args + args
      execute_subprocess_async(_a			,       env=os.environ.copy()      )

