
'''simple docstring'''
import argparse
import logging
import math
import time

import torch

from transformers import TransfoXLCorpus, TransfoXLLMHeadModel


logging.basicConfig(
    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s', datefmt='%m/%d/%Y %H:%M:%S', level=logging.INFO
)
lowerCAmelCase :  Optional[Any]      =      logging.getLogger(__name__)







def 						A_(			):
							UpperCamelCase					       =			argparse.ArgumentParser(description='PyTorch Transformer Language Model')
							parser.add_argument('--model_name'						,			type=A						,			default='transfo-xl-wt103'						,			help='pretrained model name')
							parser.add_argument(
							    '--split'						,			type=A						,			default='test'						,			choices=['all', 'valid', 'test']						,			help='which split to evaluate')
							parser.add_argument('--batch_size'						,			type=A						,			default=10						,			help='batch size')
							parser.add_argument('--tgt_len'						,			type=A						,			default=128						,			help='number of tokens to predict')
							parser.add_argument('--ext_len'						,			type=A						,			default=0						,			help='length of the extended context')
							parser.add_argument('--mem_len'						,			type=A						,			default=1600						,			help='length of the retained previous heads')
							parser.add_argument('--clamp_len'						,			type=A						,			default=1000						,			help='max positional embedding index')
							parser.add_argument('--no_cuda'						,			action='store_true'						,			help='Do not use CUDA even though CUA is available')
							parser.add_argument('--work_dir'						,			type=A						,			required=A						,			help='path to the work_dir')
							parser.add_argument('--no_log'						,			action='store_true'						,			help='do not log the eval result')
							parser.add_argument('--same_length'						,			action='store_true'						,			help='set same length attention with masking')
							parser.add_argument('--server_ip'						,			type=A						,			default=''						,			help='Can be used for distant debugging.')
							parser.add_argument('--server_port'						,			type=A						,			default=''						,			help='Can be used for distant debugging.')
							UpperCamelCase					       =			parser.parse_args()
							assert args.ext_len >= 0, "extended context length must be non-negative"

							if args.server_ip and args.server_port:
														# Distant debugging - see https://code.visualstudio.com/docs/python/debugging#_attach-to-a-local-script
														import ptvsd

														print('Waiting for debugger attach')
														ptvsd.enable_attach(address=(args.server_ip, args.server_port)						,			redirect_output=A)
														ptvsd.wait_for_attach()

							UpperCamelCase					       =			torch.device('cuda' if torch.cuda.is_available() and not args.no_cuda else 'cpu')
							logger.info('device: {}'.format(A))

							# Load a pre-processed dataset
							# You can also build the corpus yourself using TransfoXLCorpus methods
							# The pre-processing involve computing word frequencies to prepare the Adaptive input and SoftMax
							# and tokenizing the dataset
							# The pre-processed corpus is a convertion (using the conversion script )
							UpperCamelCase					       =			TransfoXLCorpus.from_pretrained(args.model_name)

							UpperCamelCase					       =			corpus.get_iterator('valid'						,			args.batch_size						,			args.tgt_len						,			device=A						,			ext_len=args.ext_len)
							UpperCamelCase					       =			corpus.get_iterator('test'						,			args.batch_size						,			args.tgt_len						,			device=A						,			ext_len=args.ext_len)

							# Load a pre-trained model
							UpperCamelCase					       =			TransfoXLLMHeadModel.from_pretrained(args.model_name)
							model.to(A)

							logger.info(
							    'Evaluating with bsz {} tgt_len {} ext_len {} mem_len {} clamp_len {}'.format(
							        args.batch_size						,			args.tgt_len						,			args.ext_len						,			args.mem_len						,			args.clamp_len))

							model.reset_memory_length(args.mem_len)
							if args.clamp_len > 0:
														UpperCamelCase					       =			args.clamp_len
							if args.same_length:
														UpperCamelCase					       =			True

							###############################################################################
							# Evaluation code
							###############################################################################
							def evaluate(A		:						Dict):
														# Turn on evaluation mode which disables dropout.
														model.eval()
														UpperCamelCase       ,     UpperCamelCase					       =			0, 0.0
														UpperCamelCase					       =			time.time()
														with torch.no_grad():
																					UpperCamelCase					       =			None
																					for idx, (data, target, seq_len) in enumerate(A):
																												UpperCamelCase					       =			model(A						,			lm_labels=A						,			mems=A)
																												UpperCamelCase       ,     UpperCamelCase       ,     UpperCamelCase					       =			ret
																												UpperCamelCase					       =			loss.mean()
																												total_loss += seq_len * loss.item()
																												total_len += seq_len
																					UpperCamelCase					       =			time.time() - start_time
														logger.info('Time : {:.2f}s, {:.2f}ms/segment'.format(A						,			1000 * total_time / (idx + 1)))
														return total_loss / total_len

							# Run on test data.
							if args.split == "all":
														UpperCamelCase					       =			evaluate(A)
														UpperCamelCase					       =			evaluate(A)
							elif args.split == "valid":
														UpperCamelCase					       =			evaluate(A)
														UpperCamelCase					       =			None
							elif args.split == "test":
														UpperCamelCase					       =			evaluate(A)
														UpperCamelCase					       =			None

							def format_log(A		:						Any						,			A		:						Optional[Any]):
														UpperCamelCase					       =			'| {0} loss {1:5.2f} | {0} ppl {2:9.3f} '.format(A						,			A						,			math.exp(A))
														return log_str

							UpperCamelCase					       =			''
							if valid_loss is not None:
														log_str += format_log(A						,			'valid')
							if test_loss is not None:
														log_str += format_log(A						,			'test')

							logger.info('=' * 100)
							logger.info(A)
							logger.info('=' * 100)


if __name__ == "__main__":
						main()






"""simple docstring"""




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		,	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		,	)    ->    str:
									"""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			)    ->    Tuple:
									"""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			)    ->    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=lowercase__		,	init_std=self.init_std		,	return_dict=self.return_dict		,	)


				def 							A					(				self		,	lowercase__		,	lowercase__		,	*lowercase__			)    ->    Union[str, Any]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	CpmAntModel(config=lowercase__			)
									model.to(lowercase__			)
									model.eval()

									SCREAMING_SNAKE_CASE				=	model(**lowercase__			).last_hidden_state

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


				def 							A					(				self		,	lowercase__		,	lowercase__		,	*lowercase__			)    ->    int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	CpmAntForCausalLM(lowercase__			)
									model.to(lowercase__			)
									SCREAMING_SNAKE_CASE				=	input_ids['input_ids'].to(lowercase__			)
									model.eval()

									SCREAMING_SNAKE_CASE				=	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 							A					(				self			)    ->    Optional[Any]:
									"""simple docstring"""







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






@require_torch
class      UpperCamelCase	(       __magic_name__  ,      __magic_name__  ,      unittest.TestCase      ):

				"""simple docstring"""



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

				UpperCAmelCase_				:      List[Any]						  =				False
				UpperCAmelCase_				:      Optional[int]						  =				False
				UpperCAmelCase_				:      Optional[Any]						  =				False
				UpperCAmelCase_				:      List[str]						  =				False
				UpperCAmelCase_				:      Dict						  =				False


				def 							A					(				self			)    ->    Union[str, Any]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	CpmAntModelTester(self			)
									SCREAMING_SNAKE_CASE				=	ConfigTester(self		,	config_class=lowercase__			)


				def 							A					(				self			)    ->    Tuple:
									"""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			)    ->    List[str]:
									"""simple docstring"""







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


				def 							A					(				self			)    ->    Optional[int]:
									"""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			)    ->    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(lowercase__		,	lowercase__			)




				def 							A					(				self			)    ->    str:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE				=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_lm_head_model(lowercase__		,	lowercase__			)






@require_torch
class      UpperCamelCase	(       unittest.TestCase      ):

				"""simple docstring"""



				@tooslow
				def 							A					(				self			)    ->    Any:
									"""simple docstring"""







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

									SCREAMING_SNAKE_CASE				=	torch.tensor(
									    [[[6.1_708, 5.9_244, 1.0_835], [6.5_207, 6.2_893, -11.3_324], [-1.0_107, -0.0_576, -5.9_577]]]		,	)
									self.assertTrue(torch.allclose(hidden_states[:, :3, :3]		,	lowercase__		,	atol=1E-2			)			)






@require_torch
class      UpperCamelCase	(       unittest.TestCase      ):

				"""simple docstring"""



				@tooslow
				def 							A					(				self			)    ->    str:
									"""simple docstring"""







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

									SCREAMING_SNAKE_CASE				=	torch.tensor(
									    [[[-6.4_267, -6.4_083, -6.3_958], [-5.8_802, -5.9_447, -5.7_811], [-5.3_896, -5.4_820, -5.4_295]]]		,	)
									self.assertTrue(torch.allclose(hidden_states[:, :3, :3]		,	lowercase__		,	atol=1E-2			)			)


				@tooslow
				def 							A					(				self			)    ->    Optional[Any]:
									"""simple docstring"""







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




				@tooslow
				def 							A					(				self			)    ->    str:
									"""simple docstring"""







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