





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


lowerCAmelCase							:       Any					 =	["""\nclass""", """\ndef""", """\n#""", """\n@""", """\nprint""", """\nif"""]





class 				a			(      __lowercase			):






				def __init__(  self			, _lowerCAmelCase			, _lowerCAmelCase			, _lowerCAmelCase=None			, _lowerCAmelCase=1					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				Any           =   tokenizer
										__SCREAMING_SNAKE_CASE:				str           =   dataset
										__SCREAMING_SNAKE_CASE:				str           =   len(_lowerCAmelCase					) if n_tasks is None else n_tasks
										__SCREAMING_SNAKE_CASE:				str           =   n_copies






				def __iter__(  self					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				List[str]           =   []
										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()					)
										__SCREAMING_SNAKE_CASE:				Any           =   self.tokenizer(_lowerCAmelCase			, padding=_lowerCAmelCase			, 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			(      __lowercase			):






				def __init__(  self			, _lowerCAmelCase			, _lowerCAmelCase			, _lowerCAmelCase					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				Dict           =   start_length
										__SCREAMING_SNAKE_CASE:				Any           =   eof_strings
										__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   tokenizer






				def __call__(  self			, _lowerCAmelCase			, _lowerCAmelCase			, **_lowerCAmelCase					):

										"""simple docstring"""




										__SCREAMING_SNAKE_CASE:				List[Any]           =   self.tokenizer.batch_decode(input_ids[:, self.start_length :]					)
										__SCREAMING_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(_lowerCAmelCase					)






def 		lowerCAmelCase  (	UpperCamelCase__   :   str					)      ->    List[str]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   re.split('''(%s)''' % '''|'''.join(UpperCamelCase__					)       ,			UpperCamelCase__					)
						# last string should be ""
						return "".join(string_list[:-2]					)






def 		lowerCAmelCase  (	UpperCamelCase__   :   int       ,			UpperCamelCase__   :   List[Any]       ,			UpperCamelCase__   :   List[str]       ,			UpperCamelCase__   :   Tuple       ,			UpperCamelCase__   :   Tuple       ,			UpperCamelCase__   :   int=20       ,			**UpperCamelCase__   :   Tuple					)      ->    str:

						"""simple docstring"""





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

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

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

						__SCREAMING_SNAKE_CASE:				Optional[Any]           =   [[] for _ in range(UpperCamelCase__					)]
						for task, generated_tokens in gen_token_dict.items():
												for s in generated_tokens:
																		__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   tokenizer.decode(UpperCamelCase__       ,			skip_special_tokens=UpperCamelCase__       ,			clean_up_tokenization_spaces=UpperCamelCase__					)
																		code_gens[task].append(remove_last_block(UpperCamelCase__					)					)
						return code_gens






def 		lowerCAmelCase  (	)      ->    str:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				str           =   HfArgumentParser(UpperCamelCase__					)
						__SCREAMING_SNAKE_CASE:				Any           =   parser.parse_args()

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

						if args.num_workers is None:
												__SCREAMING_SNAKE_CASE:				List[Any]           =   multiprocessing.cpu_count()

						# Use dataset load to feed to accelerate
						__SCREAMING_SNAKE_CASE:				List[str]           =   Accelerator()
						set_seed(args.seed       ,			device_specific=UpperCamelCase__					)

						# Load model and tokenizer
						__SCREAMING_SNAKE_CASE:				List[Any]           =   AutoTokenizer.from_pretrained(args.model_ckpt					)
						__SCREAMING_SNAKE_CASE:				Tuple           =   tokenizer.eos_token
						__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   AutoModelForCausalLM.from_pretrained(args.model_ckpt					)

						# Generation settings
						__SCREAMING_SNAKE_CASE:				Union[str, 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       ,			UpperCamelCase__       ,			UpperCamelCase__					)]					),
						}

						# Load evaluation dataset and metric
						__SCREAMING_SNAKE_CASE:				Any           =   load_dataset('''openai_humaneval'''					)
						__SCREAMING_SNAKE_CASE:				str           =   load_metric('''code_eval'''					)

						__SCREAMING_SNAKE_CASE:				Any           =   args.num_tasks if args.num_tasks is not None else len(human_eval['''test''']					)
						__SCREAMING_SNAKE_CASE:				Optional[int]           =   args.n_samples // args.batch_size

						__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   TokenizedDataset(UpperCamelCase__       ,			human_eval['''test''']       ,			n_copies=UpperCamelCase__       ,			n_tasks=UpperCamelCase__					)
						# do not confuse args.batch_size, which is actually the num_return_sequences
						__SCREAMING_SNAKE_CASE:				Optional[Any]           =   DataLoader(UpperCamelCase__       ,			batch_size=1					)

						# Run a quick test to see if code evaluation is enabled
						try:
												__SCREAMING_SNAKE_CASE:				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

						__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE:				Dict           =   accelerator.prepare(UpperCamelCase__       ,			UpperCamelCase__					)

						__SCREAMING_SNAKE_CASE:				Any           =   complete_code(
						    UpperCamelCase__       ,			UpperCamelCase__       ,			UpperCamelCase__       ,			UpperCamelCase__       ,			n_tasks=UpperCamelCase__       ,			batch_size=args.batch_size       ,			**UpperCamelCase__       ,			)

						if accelerator.is_main_process:
												__SCREAMING_SNAKE_CASE:				List[Any]           =   []

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

												# Evaluate completions with "code_eval" metric
												__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE:				str           =   code_eval_metric.compute(
												    references=UpperCamelCase__       ,			predictions=UpperCamelCase__       ,			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(UpperCamelCase__       ,			UpperCamelCase__					)


# 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()










from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


lowerCAmelCase							:       List[str]					 =	{
    """configuration_gpt_bigcode""": ["""GPT_BIGCODE_PRETRAINED_CONFIG_ARCHIVE_MAP""", """GPTBigCodeConfig"""],
}

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowerCAmelCase							:       str					 =	[
							    """GPT_BIGCODE_PRETRAINED_MODEL_ARCHIVE_LIST""",
							    """GPTBigCodeForSequenceClassification""",
							    """GPTBigCodeForTokenClassification""",
							    """GPTBigCodeForCausalLM""",
							    """GPTBigCodeModel""",
							    """GPTBigCodePreTrainedModel""",
							]

if TYPE_CHECKING:
							from .configuration_gpt_bigcode import GPT_BIGCODE_PRETRAINED_CONFIG_ARCHIVE_MAP, GPTBigCodeConfig

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_gpt_bigcode import (
														    GPT_BIGCODE_PRETRAINED_MODEL_ARCHIVE_LIST,
														    GPTBigCodeForCausalLM,
														    GPTBigCodeForSequenceClassification,
														    GPTBigCodeForTokenClassification,
														    GPTBigCodeModel,
														    GPTBigCodePreTrainedModel,
														)


else:
							import sys

							lowerCAmelCase							:       List[str]					 =	_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)



