



from __future__ import annotations

import unittest

from transformers import DebertaVaConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
				import tensorflow as tf

				from transformers import (
				    TFDebertaVaForMaskedLM,
				    TFDebertaVaForQuestionAnswering,
				    TFDebertaVaForSequenceClassification,
				    TFDebertaVaForTokenClassification,
				    TFDebertaVaModel,
				)


class       a							:



		def __init__(     self     :List[str]   ,__lowercase     :Union[str, Any]   ,__lowercase     :str=1_3   ,__lowercase     :List[Any]=7   ,__lowercase     :Any=True   ,__lowercase     :List[Any]=True   ,__lowercase     :List[Any]=True   ,__lowercase     :List[str]=True   ,__lowercase     :str=9_9   ,__lowercase     :Union[str, Any]=3_2   ,__lowercase     :List[Any]=2   ,__lowercase     :int=4   ,__lowercase     :List[str]=3_7   ,__lowercase     :Optional[int]="gelu"   ,__lowercase     :Tuple=0.1   ,__lowercase     :List[str]=0.1   ,__lowercase     :Optional[int]=5_1_2   ,__lowercase     :Any=1_6   ,__lowercase     :int=2   ,__lowercase     :List[str]=0.02   ,__lowercase     :Any=False   ,__lowercase     :Optional[int]=True   ,__lowercase     :Tuple="None"   ,__lowercase     :Any=3   ,__lowercase     :Tuple=4   ,__lowercase     :Union[str, Any]=None   ,):
						snake_case__	:      Optional[int]     		=	parent
						snake_case__	:      Dict     		=	batch_size
						snake_case__	:      Dict     		=	seq_length
						snake_case__	:      int     		=	is_training
						snake_case__	:      int     		=	use_input_mask
						snake_case__	:      Tuple     		=	use_token_type_ids
						snake_case__	:      Dict     		=	use_labels
						snake_case__	:      Tuple     		=	vocab_size
						snake_case__	:      Optional[int]     		=	hidden_size
						snake_case__	:      Optional[Any]     		=	num_hidden_layers
						snake_case__	:      List[str]     		=	num_attention_heads
						snake_case__	:      List[Any]     		=	intermediate_size
						snake_case__	:      List[Any]     		=	hidden_act
						snake_case__	:      Optional[Any]     		=	hidden_dropout_prob
						snake_case__	:      str     		=	attention_probs_dropout_prob
						snake_case__	:      Dict     		=	max_position_embeddings
						snake_case__	:      Optional[int]     		=	type_vocab_size
						snake_case__	:      str     		=	type_sequence_label_size
						snake_case__	:      Tuple     		=	initializer_range
						snake_case__	:      Optional[Any]     		=	num_labels
						snake_case__	:      int     		=	num_choices
						snake_case__	:      int     		=	relative_attention
						snake_case__	:      List[str]     		=	position_biased_input
						snake_case__	:      Dict     		=	pos_att_type
						snake_case__	:      Optional[Any]     		=	scope



		def 	__lowerCamelCase			(     self     :int  ):
						snake_case__	:      Tuple     		=	ids_tensor([self.batch_size, self.seq_length]   ,self.vocab_size  )

						snake_case__	:      List[Any]     		=	None
						if self.use_input_mask:
										snake_case__	:      str     		=	random_attention_mask([self.batch_size, self.seq_length]  )

						snake_case__	:      List[Any]     		=	None
						if self.use_token_type_ids:
										snake_case__	:      Optional[int]     		=	ids_tensor([self.batch_size, self.seq_length]   ,self.type_vocab_size  )

						snake_case__	:      Union[str, Any]     		=	None
						snake_case__	:      Dict     		=	None
						snake_case__	:      str     		=	None
						if self.use_labels:
										snake_case__	:      Any     		=	ids_tensor([self.batch_size]   ,self.type_sequence_label_size  )
										snake_case__	:      str     		=	ids_tensor([self.batch_size, self.seq_length]   ,self.num_labels  )

						snake_case__	:      Dict     		=	DebertaVaConfig(
						    vocab_size=self.vocab_size   ,hidden_size=self.hidden_size   ,num_hidden_layers=self.num_hidden_layers   ,num_attention_heads=self.num_attention_heads   ,intermediate_size=self.intermediate_size   ,hidden_act=self.hidden_act   ,hidden_dropout_prob=self.hidden_dropout_prob   ,attention_probs_dropout_prob=self.attention_probs_dropout_prob   ,max_position_embeddings=self.max_position_embeddings   ,type_vocab_size=self.type_vocab_size   ,relative_attention=self.relative_attention   ,position_biased_input=self.position_biased_input   ,initializer_range=self.initializer_range   ,return_dict=__lowercase   ,)

						return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels



		def 	__lowerCamelCase			(     self     :Optional[Any]   ,__lowercase     :List[Any]   ,__lowercase     :int   ,__lowercase     :Tuple   ,__lowercase     :Tuple   ,__lowercase     :Union[str, Any]   ,__lowercase     :Tuple   ,__lowercase     :List[Any]  ):
						snake_case__	:      Union[str, Any]     		=	TFDebertaVaModel(config=__lowercase  )
						snake_case__	:      List[Any]     		=	{'''input_ids''': input_ids, '''attention_mask''': input_mask, '''token_type_ids''': token_type_ids}

						snake_case__	:      Optional[int]     		=	[input_ids, input_mask]
						snake_case__	:      Any     		=	model(__lowercase  )

						snake_case__	:      List[Any]     		=	model(__lowercase  )

						self.parent.assertEqual(result.last_hidden_state.shape   ,(self.batch_size, self.seq_length, self.hidden_size)  )



		def 	__lowerCamelCase			(     self     :Any   ,__lowercase     :Optional[Any]   ,__lowercase     :Optional[Any]   ,__lowercase     :str   ,__lowercase     :str   ,__lowercase     :Union[str, Any]   ,__lowercase     :str   ,__lowercase     :str  ):
						snake_case__	:      str     		=	TFDebertaVaForMaskedLM(config=__lowercase  )
						snake_case__	:      str     		=	{
						    '''input_ids''': input_ids,
						    '''attention_mask''': input_mask,
						    '''token_type_ids''': token_type_ids,
						}
						snake_case__	:      Dict     		=	model(__lowercase  )
						self.parent.assertEqual(result.logits.shape   ,(self.batch_size, self.seq_length, self.vocab_size)  )



		def 	__lowerCamelCase			(     self     :Dict   ,__lowercase     :Dict   ,__lowercase     :Any   ,__lowercase     :int   ,__lowercase     :Optional[int]   ,__lowercase     :int   ,__lowercase     :Tuple   ,__lowercase     :Dict  ):
						snake_case__	:      List[str]     		=	self.num_labels
						snake_case__	:      Any     		=	TFDebertaVaForSequenceClassification(config=__lowercase  )
						snake_case__	:      Dict     		=	{
						    '''input_ids''': input_ids,
						    '''attention_mask''': input_mask,
						    '''token_type_ids''': token_type_ids,
						}

						snake_case__	:      Any     		=	model(__lowercase  )
						self.parent.assertEqual(result.logits.shape   ,(self.batch_size, self.num_labels)  )



		def 	__lowerCamelCase			(     self     :Optional[int]   ,__lowercase     :int   ,__lowercase     :str   ,__lowercase     :List[str]   ,__lowercase     :Optional[int]   ,__lowercase     :List[str]   ,__lowercase     :List[str]   ,__lowercase     :Optional[int]  ):
						snake_case__	:      Union[str, Any]     		=	self.num_labels
						snake_case__	:      Optional[Any]     		=	TFDebertaVaForTokenClassification(config=__lowercase  )
						snake_case__	:      int     		=	{
						    '''input_ids''': input_ids,
						    '''attention_mask''': input_mask,
						    '''token_type_ids''': token_type_ids,
						}
						snake_case__	:      Tuple     		=	model(__lowercase  )
						self.parent.assertEqual(result.logits.shape   ,(self.batch_size, self.seq_length, self.num_labels)  )



		def 	__lowerCamelCase			(     self     :Dict   ,__lowercase     :Union[str, Any]   ,__lowercase     :Tuple   ,__lowercase     :str   ,__lowercase     :Optional[int]   ,__lowercase     :List[Any]   ,__lowercase     :List[str]   ,__lowercase     :Any  ):
						snake_case__	:      Any     		=	TFDebertaVaForQuestionAnswering(config=__lowercase  )
						snake_case__	:      List[str]     		=	{
						    '''input_ids''': input_ids,
						    '''attention_mask''': input_mask,
						    '''token_type_ids''': token_type_ids,
						}

						snake_case__	:      List[str]     		=	model(__lowercase  )
						self.parent.assertEqual(result.start_logits.shape   ,(self.batch_size, self.seq_length)  )
						self.parent.assertEqual(result.end_logits.shape   ,(self.batch_size, self.seq_length)  )






		def 	__lowerCamelCase			(     self     :str  ):
						snake_case__	:      Optional[Any]     		=	self.prepare_config_and_inputs()
						(
						    (
						    snake_case__
						)		, (
						    snake_case__
						)		, (
						    snake_case__
						)		, (
						    snake_case__
						)		, (
						    snake_case__
						)		, (
						    snake_case__
						)		, (
						    snake_case__
						)		, 
						)	:      str     		=	config_and_inputs
						snake_case__	:      Optional[int]     		=	{'''input_ids''': input_ids, '''token_type_ids''': token_type_ids, '''attention_mask''': input_mask}
						return config, inputs_dict


@require_tf
class       a							(			__lowerCamelCase  ,							__lowerCamelCase  ,							unittest.TestCase				):
		__lowerCAmelCase	:    str											=					(
		    (
		        TFDebertaVaModel,
		        TFDebertaVaForMaskedLM,
		        TFDebertaVaForQuestionAnswering,
		        TFDebertaVaForSequenceClassification,
		        TFDebertaVaForTokenClassification,
		    )
		    if is_tf_available()
		    else ()
		)
		__lowerCAmelCase	:    Tuple											=					(
		    {
		        """feature-extraction""": TFDebertaVaModel,
		        """fill-mask""": TFDebertaVaForMaskedLM,
		        """question-answering""": TFDebertaVaForQuestionAnswering,
		        """text-classification""": TFDebertaVaForSequenceClassification,
		        """token-classification""": TFDebertaVaForTokenClassification,
		        """zero-shot""": TFDebertaVaForSequenceClassification,
		    }
		    if is_tf_available()
		    else {}
		)

		__lowerCAmelCase	:    Tuple											=					False
		__lowerCAmelCase	:    Any											=					False



		def 	__lowerCamelCase			(     self     :Union[str, Any]  ):
						snake_case__	:      int     		=	TFDebertaVaModelTester(self  )
						snake_case__	:      Optional[int]     		=	ConfigTester(self   ,config_class=__lowercase   ,hidden_size=3_7  )



		def 	__lowerCamelCase			(     self     :Dict  ):
						self.config_tester.run_common_tests()



		def 	__lowerCamelCase			(     self     :List[str]  ):
						snake_case__	:      int     		=	self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*__lowercase  )



		def 	__lowerCamelCase			(     self     :int  ):
						snake_case__	:      List[str]     		=	self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_masked_lm(*__lowercase  )



		def 	__lowerCamelCase			(     self     :Dict  ):
						snake_case__	:      Any     		=	self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_question_answering(*__lowercase  )



		def 	__lowerCamelCase			(     self     :List[Any]  ):
						snake_case__	:      List[Any]     		=	self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_sequence_classification(*__lowercase  )



		def 	__lowerCamelCase			(     self     :Tuple  ):
						snake_case__	:      Any     		=	self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_token_classification(*__lowercase  )






		@slow
		def 	__lowerCamelCase			(     self     :List[Any]  ):
						snake_case__	:      Union[str, Any]     		=	TFDebertaVaModel.from_pretrained('''kamalkraj/deberta-v2-xlarge'''  )
						self.assertIsNotNone(__lowercase  )


@require_tf
class       a							(			unittest.TestCase				):



		@unittest.skip(reason='''Model not available yet'''  )
		def 	__lowerCamelCase			(     self     :Dict  ):
						pass






		@slow
		def 	__lowerCamelCase			(     self     :Optional[Any]  ):
						snake_case__	:      Union[str, Any]     		=	TFDebertaVaModel.from_pretrained('''kamalkraj/deberta-v2-xlarge'''  )
						snake_case__	:      Optional[int]     		=	tf.constant([[0, 3_1_4_1_4, 2_3_2, 3_2_8, 7_4_0, 1_1_4_0, 1_2_6_9_5, 6_9, 4_6_0_7_8, 1_5_8_8, 2]]  )
						snake_case__	:      Optional[Any]     		=	tf.constant([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]  )
						snake_case__	:      List[str]     		=	model(__lowercase   ,attention_mask=__lowercase  )[0]

						snake_case__	:      Any     		=	tf.constant(
						    [[[0.2356, 0.1948, 0.0369], [-0.1063, 0.3586, -0.5152], [-0.6399, -0.0259, -0.2525]]]  )
						tf.debugging.assert_near(output[:, 1:4, 1:4]   ,__lowercase   ,atol=1e-4  )






import json
import multiprocessing
import os
import re
from collections import defaultdict

import torch
from accelerate import Accelerator
from accelerate.utils import set_seed
from arguments import HumanEvalArguments
from datasets import load_dataset, load_metric
from torch.utils.data import IterableDataset
from torch.utils.data.dataloader import DataLoader
from tqdm import tqdm

import transformers
from transformers import AutoModelForCausalLM, AutoTokenizer, HfArgumentParser, StoppingCriteria, StoppingCriteriaList


A__								=						['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']


class       a							(			__lowerCamelCase				):



		def __init__(     self     :str   ,__lowercase     :Dict   ,__lowercase     :str   ,__lowercase     :Any=None   ,__lowercase     :Union[str, Any]=1  ):
						snake_case__	:      Optional[int]     		=	tokenizer
						snake_case__	:      List[Any]     		=	dataset
						snake_case__	:      Any     		=	len(__lowercase  ) if n_tasks is None else n_tasks
						snake_case__	:      List[Any]     		=	n_copies






		def __iter__(     self     :Tuple  ):
						snake_case__	:      Union[str, Any]     		=	[]
						for task in range(self.n_tasks  ):
										# without strip, the model generate commented codes ...
										prompts.append(self.tokenizer.eos_token + self.dataset[task]['''prompt'''].strip()  )
						snake_case__	:      Optional[int]     		=	self.tokenizer(__lowercase   ,padding=__lowercase   ,return_tensors='''pt'''  )
						for task in range(self.n_tasks  ):
										for _ in range(self.n_copies  ):
														yield {
														    "ids": outputs.input_ids[task],
														    "task_id": task,
														    "input_len": outputs.attention_mask[task].sum(),
														}


class       a							(			__lowerCamelCase				):



		def __init__(     self     :Union[str, Any]   ,__lowercase     :List[Any]   ,__lowercase     :int   ,__lowercase     :List[str]  ):
						snake_case__	:      List[str]     		=	start_length
						snake_case__	:      Any     		=	eof_strings
						snake_case__	:      Optional[Any]     		=	tokenizer






		def __call__(     self     :str   ,__lowercase     :str   ,__lowercase     :Any   ,**__lowercase     :Optional[int]  ):
						snake_case__	:      Tuple     		=	self.tokenizer.batch_decode(input_ids[:, self.start_length :]  )
						snake_case__	:      Union[str, Any]     		=	[]
						for decoded_generation in decoded_generations:
										done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings  )  )
						return all(__lowercase  )



def 			_lowerCAmelCase       (					__lowerCAmelCase      )   ->					List[Any]:


				"""simple docstring"""

				snake_case__	:      List[Any]     		=	re.split('''(%s)''' % '''|'''.join(__lowerCAmelCase      )  ,  __lowerCAmelCase      )
				# last string should be ""
				return "".join(string_list[:-2]      )



def 			_lowerCAmelCase       (					__lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase=20  ,  **__lowerCAmelCase      )   ->					List[Any]:


				"""simple docstring"""

				snake_case__	:      Optional[int]     		=	defaultdict(__lowerCAmelCase      )  # dict of list of generated tokens
				for step, batch in tqdm(enumerate(__lowerCAmelCase      )      ):
								with torch.no_grad():
												snake_case__	:      List[Any]     		=	batch['''ids'''].shape[-1]
												snake_case__	:      str     		=	accelerator.unwrap_model(__lowerCAmelCase      ).generate(
												    input_ids=batch['''ids'''][:, : batch['''input_len''']]  ,  num_return_sequences=__lowerCAmelCase  ,  **__lowerCAmelCase      )
												# each task is generated batch_size times
												snake_case__	:      str     		=	batch['''task_id'''].repeat(__lowerCAmelCase      )
												snake_case__	:      Optional[int]     		=	accelerator.pad_across_processes(
												    __lowerCAmelCase  ,  dim=1  ,  pad_index=tokenizer.pad_token_id      )

												snake_case__		, snake_case__	:      Any     		=	accelerator.gather((generated_tokens, generated_tasks)      )
												snake_case__	:      Union[str, Any]     		=	generated_tokens.cpu().numpy()
												snake_case__	:      Optional[Any]     		=	generated_tasks.cpu().numpy()

												for task, generated_tokens in zip(__lowerCAmelCase  ,  __lowerCAmelCase      ):
																gen_token_dict[task].append(__lowerCAmelCase      )

				snake_case__	:      List[Any]     		=	[[] for _ in range(__lowerCAmelCase      )]
				for task, generated_tokens in gen_token_dict.items():
								for s in generated_tokens:
												snake_case__	:      int     		=	tokenizer.decode(__lowerCAmelCase  ,  skip_special_tokens=__lowerCAmelCase  ,  clean_up_tokenization_spaces=__lowerCAmelCase      )
												code_gens[task].append(remove_last_block(__lowerCAmelCase      )      )
				return code_gens



def 			_lowerCAmelCase       (					)   ->					Optional[int]:


				"""simple docstring"""

				snake_case__	:      List[str]     		=	HfArgumentParser(__lowerCAmelCase      )
				snake_case__	:      Optional[Any]     		=	parser.parse_args()

				transformers.logging.set_verbosity_error()
				# enables code execution in code_eval metric
				snake_case__	:      Union[str, Any]     		=	args.HF_ALLOW_CODE_EVAL
				# make sure tokenizer plays nice with multiprocessing
				snake_case__	:      Union[str, Any]     		=	'''false'''

				if args.num_workers is None:
								snake_case__	:      Union[str, Any]     		=	multiprocessing.cpu_count()

				# Use dataset load to feed to accelerate
				snake_case__	:      Dict     		=	Accelerator()
				set_seed(args.seed  ,  device_specific=__lowerCAmelCase      )

				# Load model and tokenizer
				snake_case__	:      str     		=	AutoTokenizer.from_pretrained(args.model_ckpt      )
				snake_case__	:      List[str]     		=	tokenizer.eos_token
				snake_case__	:      Optional[Any]     		=	AutoModelForCausalLM.from_pretrained(args.model_ckpt      )

				# Generation settings
				snake_case__	:      List[Any]     		=	{
				    '''do_sample''': args.do_sample,
				    '''temperature''': args.temperature,
				    '''max_new_tokens''': args.max_new_tokens,
				    '''top_p''': args.top_p,
				    '''top_k''': args.top_k,
				    '''stopping_criteria''': StoppingCriteriaList([EndOfFunctionCriteria(0  ,  __lowerCAmelCase  ,  __lowerCAmelCase      )]      ),
				}

				# Load evaluation dataset and metric
				snake_case__	:      List[str]     		=	load_dataset('''openai_humaneval'''      )
				snake_case__	:      str     		=	load_metric('''code_eval'''      )

				snake_case__	:      List[Any]     		=	args.num_tasks if args.num_tasks is not None else len(human_eval['''test''']      )
				snake_case__	:      str     		=	args.n_samples // args.batch_size

				snake_case__	:      Tuple     		=	TokenizedDataset(__lowerCAmelCase  ,  human_eval['''test''']  ,  n_copies=__lowerCAmelCase  ,  n_tasks=__lowerCAmelCase      )
				# do not confuse args.batch_size, which is actually the num_return_sequences
				snake_case__	:      Tuple     		=	DataLoader(__lowerCAmelCase  ,  batch_size=1      )

				# Run a quick test to see if code evaluation is enabled
				try:
								snake_case__	:      Optional[Any]     		=	code_eval_metric.compute(references=['''''']  ,  predictions=[['''''']]      )
				except ValueError as exception:
								print(
								    '''Code evaluation not enabled. Read the warning below carefully and then use `--HF_ALLOW_CODE_EVAL="1"`'''
								    ''' flag to enable code evaluation.'''      )
								raise exception

				snake_case__		, snake_case__	:      Optional[Any]     		=	accelerator.prepare(__lowerCAmelCase  ,  __lowerCAmelCase      )

				snake_case__	:      Dict     		=	complete_code(
				    __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  n_tasks=__lowerCAmelCase  ,  batch_size=args.batch_size  ,  **__lowerCAmelCase  ,  )

				if accelerator.is_main_process:
								snake_case__	:      Optional[Any]     		=	[]

								for task in tqdm(range(__lowerCAmelCase      )      ):
												snake_case__	:      str     		=	human_eval['''test'''][task]['''test''']
												snake_case__	:      int     		=	f"""check({human_eval['test'][task]['entry_point']})"""
												references.append('''\n''' + test_func + '''\n''' + entry_point      )

								# Evaluate completions with "code_eval" metric
								snake_case__		, snake_case__	:      Any     		=	code_eval_metric.compute(
								    references=__lowerCAmelCase  ,  predictions=__lowerCAmelCase  ,  num_workers=args.num_workers      )
								print(f"""Results: {pass_at_k}"""      )

								# Save results to json file
								with open(args.output_file  ,  '''w'''      ) as fp:
												json.dump(__lowerCAmelCase  ,  __lowerCAmelCase      )


# For some reason the folliwng seems to be necessary sometimes for code_eval to work nice with multiprocessing
# https://stackoverflow.com/questions/60804599/python-multiprocessing-keeps-spawning-the-whole-script
if __name__ == "__main__":
				main()

