


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





							def __init__(							self			,       _SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE=2			,       _SCREAMING_SNAKE_CASE=8			,       _SCREAMING_SNAKE_CASE=True			,       _SCREAMING_SNAKE_CASE=False			,       _SCREAMING_SNAKE_CASE=False			,       _SCREAMING_SNAKE_CASE=False			,       _SCREAMING_SNAKE_CASE=False			,       _SCREAMING_SNAKE_CASE=99			,       _SCREAMING_SNAKE_CASE=32			,       _SCREAMING_SNAKE_CASE=3			,       _SCREAMING_SNAKE_CASE=4			,       _SCREAMING_SNAKE_CASE=37			,       _SCREAMING_SNAKE_CASE=32			,       _SCREAMING_SNAKE_CASE=128			,       _SCREAMING_SNAKE_CASE=8			,       _SCREAMING_SNAKE_CASE=8			,       _SCREAMING_SNAKE_CASE=8			,       _SCREAMING_SNAKE_CASE=1.0			,       _SCREAMING_SNAKE_CASE=True			,       ):
								a_     			=   parent
								a_     			=   batch_size
								a_     			=   seq_length
								a_     			=   is_training
								a_     			=   use_token_type_ids
								a_     			=   use_input_mask
								a_     			=   use_labels
								a_     			=   use_mc_token_ids
								a_     			=   vocab_size
								a_     			=   hidden_size
								a_     			=   num_hidden_layers
								a_     			=   num_attention_heads
								a_     			=   intermediate_size
								a_     			=   num_buckets
								a_     			=   max_distance
								a_     			=   prompt_length
								a_     			=   prompt_types
								a_     			=   segment_types
								a_     			=   init_std
								a_     			=   return_dict





							def   __magic_name__      (							self    ):
								a_     			=   {}
								a_     			=   ids_tensor([self.batch_size, self.seq_length]			,       self.vocab_size    ).type(torch.intaa    )
								a_     			=   False

								a_     			=   self.get_config()

								return (config, input_ids)





							def   __magic_name__      (							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=_SCREAMING_SNAKE_CASE			,       init_std=self.init_std			,       return_dict=self.return_dict			,       )





							def   __magic_name__      (							self			,       _SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE			,       *_SCREAMING_SNAKE_CASE    ):
								a_     			=   CpmAntModel(config=_SCREAMING_SNAKE_CASE    )
								model.to(_SCREAMING_SNAKE_CASE    )
								model.eval()

								a_     			=   model(**_SCREAMING_SNAKE_CASE    ).last_hidden_state

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





							def   __magic_name__      (							self			,       _SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE			,       *_SCREAMING_SNAKE_CASE    ):
								a_     			=   CpmAntForCausalLM(_SCREAMING_SNAKE_CASE    )
								model.to(_SCREAMING_SNAKE_CASE    )
								a_     			=   input_ids["""input_ids"""].to(_SCREAMING_SNAKE_CASE    )
								model.eval()

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





							def   __magic_name__      (							self    ):
								a_			, a_     			=   self.prepare_config_and_inputs()
								return config, inputs_dict







@require_torch
class 			lowerCamelCase_					(    _SCREAMING_SNAKE_CASE   ,		_SCREAMING_SNAKE_CASE   ,		unittest.TestCase    ):
							_lowerCamelCase				: int       =					(CpmAntModel, CpmAntForCausalLM) if is_torch_available() else ()
							_lowerCamelCase				: Tuple       =					(
							    {"""feature-extraction""": CpmAntModel, """text-generation""": CpmAntForCausalLM} if is_torch_available() else {}
							)

							_lowerCamelCase				: Optional[Any]       =					False
							_lowerCamelCase				: Union[str, Any]       =					False
							_lowerCamelCase				: Any       =					False
							_lowerCamelCase				: List[Any]       =					False
							_lowerCamelCase				: Union[str, Any]       =					False





							def   __magic_name__      (							self    ):
								a_     			=   CpmAntModelTester(self    )
								a_     			=   ConfigTester(self			,       config_class=_SCREAMING_SNAKE_CASE    )





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





							def   __magic_name__      (							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   __magic_name__      (							self    ):
								a_			, a_     			=   self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_cpmant_model(_SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE    )





							def   __magic_name__      (							self    ):
								a_			, a_     			=   self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_lm_head_model(_SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE    )







@require_torch
class 			lowerCamelCase_					(    unittest.TestCase    ):





							@tooslow
							def   __magic_name__      (							self    ):
								a_     			=   """今天天气真好！"""
								a_     			=   """openbmb/cpm-ant-10b"""
								a_     			=   CpmAntModel.from_pretrained(_SCREAMING_SNAKE_CASE    )
								a_     			=   CpmAntTokenizer.from_pretrained(_SCREAMING_SNAKE_CASE    )
								a_     			=   tokenizer(_SCREAMING_SNAKE_CASE			,       return_tensors="""pt"""    )
								a_     			=   model(**_SCREAMING_SNAKE_CASE    ).last_hidden_state

								a_     			=   torch.tensor(
								    [[[6.1_7_0_8, 5.9_2_4_4, 1.0_8_3_5], [6.5_2_0_7, 6.2_8_9_3, -1_1.3_3_2_4], [-1.0_1_0_7, -0.0_5_7_6, -5.9_5_7_7]]]			,       )
								self.assertTrue(torch.allclose(hidden_states[:, :3, :3]			,       _SCREAMING_SNAKE_CASE			,       atol=1E-2    )    )







@require_torch
class 			lowerCamelCase_					(    unittest.TestCase    ):





							@tooslow
							def   __magic_name__      (							self    ):
								a_     			=   """今天天气真好！"""
								a_     			=   """openbmb/cpm-ant-10b"""
								a_     			=   CpmAntForCausalLM.from_pretrained(_SCREAMING_SNAKE_CASE    )
								a_     			=   CpmAntTokenizer.from_pretrained(_SCREAMING_SNAKE_CASE    )
								a_     			=   tokenizer(_SCREAMING_SNAKE_CASE			,       return_tensors="""pt"""    )
								a_     			=   model(**_SCREAMING_SNAKE_CASE    ).logits

								a_     			=   torch.tensor(
								    [[[-6.4_2_6_7, -6.4_0_8_3, -6.3_9_5_8], [-5.8_8_0_2, -5.9_4_4_7, -5.7_8_1_1], [-5.3_8_9_6, -5.4_8_2_0, -5.4_2_9_5]]]			,       )
								self.assertTrue(torch.allclose(hidden_states[:, :3, :3]			,       _SCREAMING_SNAKE_CASE			,       atol=1E-2    )    )





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





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



import random
import shutil
import tempfile
import unittest

import numpy as np

from transformers import TaTokenizer, TaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torch
from transformers.utils.import_utils import is_speech_available, is_torch_available


if is_torch_available():
			pass

if is_speech_available():
			from transformers import EncodecFeatureExtractor, MusicgenProcessor


_A							     =       random.Random()

def 			__SCREAMING_SNAKE_CASE		(		UpperCamelCase		:      List[Any]     ,					UpperCamelCase		:      Any=1.0     ,					UpperCamelCase		:      Union[str, Any]=None     ,					UpperCamelCase		:      Union[str, Any]=None	)     ->	List[Any]:



	"""simple docstring"""






	if rng is None:
		a_     			=   global_rng

	a_     			=   []
	for batch_idx in range(shape[0]	):
		values.append([]	)
		for _ in range(shape[1]	):
			values[-1].append(rng.random() * scale	)

	return values

@require_torch
@require_sentencepiece
class 			lowerCamelCase_					(    unittest.TestCase    ):





							def   __magic_name__      (							self    ):
								a_     			=   """facebook/musicgen-small"""
								a_     			=   tempfile.mkdtemp()





							def   __magic_name__      (							self			,       **_SCREAMING_SNAKE_CASE    ):
								return TaTokenizer.from_pretrained(self.checkpoint			,       **_SCREAMING_SNAKE_CASE    )





							def   __magic_name__      (							self			,       **_SCREAMING_SNAKE_CASE    ):
								return EncodecFeatureExtractor.from_pretrained(self.checkpoint			,       **_SCREAMING_SNAKE_CASE    )





							def   __magic_name__      (							self    ):
								shutil.rmtree(self.tmpdirname    )





							def   __magic_name__      (							self    ):
								a_     			=   self.get_tokenizer()
								a_     			=   self.get_feature_extractor()

								a_     			=   MusicgenProcessor(tokenizer=_SCREAMING_SNAKE_CASE			,       feature_extractor=_SCREAMING_SNAKE_CASE    )

								processor.save_pretrained(self.tmpdirname    )
								a_     			=   MusicgenProcessor.from_pretrained(self.tmpdirname    )

								self.assertEqual(processor.tokenizer.get_vocab()			,       tokenizer.get_vocab()    )
								self.assertIsInstance(processor.tokenizer			,       _SCREAMING_SNAKE_CASE    )

								self.assertEqual(processor.feature_extractor.to_json_string()			,       feature_extractor.to_json_string()    )
								self.assertIsInstance(processor.feature_extractor			,       _SCREAMING_SNAKE_CASE    )





							def   __magic_name__      (							self    ):
								a_     			=   MusicgenProcessor(tokenizer=self.get_tokenizer()			,       feature_extractor=self.get_feature_extractor()    )
								processor.save_pretrained(self.tmpdirname    )

								a_     			=   self.get_tokenizer(bos_token="""(BOS)"""			,       eos_token="""(EOS)"""    )
								a_     			=   self.get_feature_extractor(do_normalize=_SCREAMING_SNAKE_CASE			,       padding_value=1.0    )

								a_     			=   MusicgenProcessor.from_pretrained(
								    self.tmpdirname			,       bos_token="""(BOS)"""			,       eos_token="""(EOS)"""			,       do_normalize=_SCREAMING_SNAKE_CASE			,       padding_value=1.0    )

								self.assertEqual(processor.tokenizer.get_vocab()			,       tokenizer_add_kwargs.get_vocab()    )
								self.assertIsInstance(processor.tokenizer			,       _SCREAMING_SNAKE_CASE    )

								self.assertEqual(processor.feature_extractor.to_json_string()			,       feature_extractor_add_kwargs.to_json_string()    )
								self.assertIsInstance(processor.feature_extractor			,       _SCREAMING_SNAKE_CASE    )





							def   __magic_name__      (							self    ):
								a_     			=   self.get_feature_extractor()
								a_     			=   self.get_tokenizer()

								a_     			=   MusicgenProcessor(tokenizer=_SCREAMING_SNAKE_CASE			,       feature_extractor=_SCREAMING_SNAKE_CASE    )

								a_     			=   floats_list((3, 1000)    )

								a_     			=   feature_extractor(_SCREAMING_SNAKE_CASE			,       return_tensors="""np"""    )
								a_     			=   processor(_SCREAMING_SNAKE_CASE			,       return_tensors="""np"""    )

								for key in input_feat_extract.keys():
									self.assertAlmostEqual(input_feat_extract[key].sum()			,       input_processor[key].sum()			,       delta=1E-2    )





							def   __magic_name__      (							self    ):
								a_     			=   self.get_feature_extractor()
								a_     			=   self.get_tokenizer()

								a_     			=   MusicgenProcessor(tokenizer=_SCREAMING_SNAKE_CASE			,       feature_extractor=_SCREAMING_SNAKE_CASE    )

								a_     			=   """This is a test string"""

								a_     			=   processor(text=_SCREAMING_SNAKE_CASE    )

								a_     			=   tokenizer(_SCREAMING_SNAKE_CASE    )

								for key in encoded_tok.keys():
									self.assertListEqual(encoded_tok[key]			,       encoded_processor[key]    )





							def   __magic_name__      (							self    ):
								a_     			=   self.get_feature_extractor()
								a_     			=   self.get_tokenizer()

								a_     			=   MusicgenProcessor(tokenizer=_SCREAMING_SNAKE_CASE			,       feature_extractor=_SCREAMING_SNAKE_CASE    )

								a_     			=   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

								a_     			=   processor.batch_decode(sequences=_SCREAMING_SNAKE_CASE    )
								a_     			=   tokenizer.batch_decode(_SCREAMING_SNAKE_CASE    )

								self.assertListEqual(_SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE    )





							def   __magic_name__      (							self    ):
								a_     			=   self.get_feature_extractor()
								a_     			=   self.get_tokenizer()

								a_     			=   MusicgenProcessor(tokenizer=_SCREAMING_SNAKE_CASE			,       feature_extractor=_SCREAMING_SNAKE_CASE    )

								self.assertListEqual(
								    processor.model_input_names			,       feature_extractor.model_input_names			,       msg="""`processor` and `feature_extractor` model input names do not match"""			,       )





							def   __magic_name__      (							self    ):
								a_     			=   self.get_feature_extractor(padding_side="""left"""    )
								a_     			=   self.get_tokenizer()

								a_     			=   MusicgenProcessor(tokenizer=_SCREAMING_SNAKE_CASE			,       feature_extractor=_SCREAMING_SNAKE_CASE    )

								a_     			=   [floats_list((1, x)    )[0] for x in range(5			,       20			,       5    )]
								a_     			=   processor(_SCREAMING_SNAKE_CASE    ).padding_mask

								a_     			=   np.asarray(floats_list((3, 20)    )    )[:, None, :]
								a_     			=   processor.batch_decode(_SCREAMING_SNAKE_CASE			,       padding_mask=_SCREAMING_SNAKE_CASE    )

								self.assertIsInstance(_SCREAMING_SNAKE_CASE			,       _SCREAMING_SNAKE_CASE    )

								for audio in decoded_audios:
									self.assertIsInstance(_SCREAMING_SNAKE_CASE			,       np.ndarray    )

								self.assertTrue(decoded_audios[0].shape == (1, 10)    )
								self.assertTrue(decoded_audios[1].shape == (1, 15)    )
								self.assertTrue(decoded_audios[2].shape == (1, 20)    )