



"""simple docstring"""






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


__snake_case							:     str   =					['\nclass', '\ndef', '\n#', '\n@', '\nprint', '\nif']



class     A__     (						__SCREAMING_SNAKE_CASE							):




	'''simple docstring'''





	def __init__(  self:	int ,    _SCREAMING_SNAKE_CASE:	List[Any] ,    _SCREAMING_SNAKE_CASE:	str ,    _SCREAMING_SNAKE_CASE:	Optional[Any]=None ,    _SCREAMING_SNAKE_CASE:	List[Any]=1)       -> List[Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						Tuple				= tokenizer
				__lowerCAmelCase   :						List[str]				= dataset
				__lowerCAmelCase   :						List[str]				= len(_SCREAMING_SNAKE_CASE) if n_tasks is None else n_tasks
				__lowerCAmelCase   :						str				= n_copies







	def __iter__(  self:	List[str])       -> Optional[int]:

				"""simple docstring"""

				__lowerCAmelCase   :						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())
				__lowerCAmelCase   :						Dict				= self.tokenizer(_SCREAMING_SNAKE_CASE ,    padding=_SCREAMING_SNAKE_CASE ,    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__     (						__SCREAMING_SNAKE_CASE							):




	'''simple docstring'''





	def __init__(  self:	Tuple ,    _SCREAMING_SNAKE_CASE:	Union[str, Any] ,    _SCREAMING_SNAKE_CASE:	Optional[int] ,    _SCREAMING_SNAKE_CASE:	str)       -> int:

				"""simple docstring"""

				__lowerCAmelCase   :						Union[str, Any]				= start_length
				__lowerCAmelCase   :						Any				= eof_strings
				__lowerCAmelCase   :						Tuple				= tokenizer







	def __call__(  self:	Optional[int] ,    _SCREAMING_SNAKE_CASE:	Union[str, Any] ,    _SCREAMING_SNAKE_CASE:	str ,    **_SCREAMING_SNAKE_CASE:	Union[str, Any])       -> Any:

				"""simple docstring"""

				__lowerCAmelCase   :						List[str]				= self.tokenizer.batch_decode(input_ids[:, self.start_length :])
				__lowerCAmelCase   :						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(_SCREAMING_SNAKE_CASE)






def     _lowercase       (						__snake_case			)		->     int:
			__lowerCAmelCase   :						List[Any]				= re.split("(%s)" % "|".join(__snake_case			)      ,__snake_case			)
			# last string should be ""
			return "".join(string_list[:-2]			)


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

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

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

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


def     _lowercase       (						)		->     Dict:
			# Setup configuration
			__lowerCAmelCase   :						Any				= HfArgumentParser(__snake_case			)
			__lowerCAmelCase   :						Dict				= parser.parse_args()

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

			if args.num_workers is None:
						__lowerCAmelCase   :						Optional[Any]				= multiprocessing.cpu_count()

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

			# Load model and tokenizer
			__lowerCAmelCase   :						Union[str, Any]				= AutoTokenizer.from_pretrained(args.model_ckpt			)
			__lowerCAmelCase   :						Any				= tokenizer.eos_token
			__lowerCAmelCase   :						Optional[int]				= AutoModelForCausalLM.from_pretrained(args.model_ckpt			)

			# Generation settings
			__lowerCAmelCase   :						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      ,__snake_case      ,__snake_case			)]			),
			}

			# Load evaluation dataset and metric
			__lowerCAmelCase   :						int				= load_dataset("openai_humaneval"			)
			__lowerCAmelCase   :						List[str]				= load_metric("code_eval"			)

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

			__lowerCAmelCase   :						Any				= TokenizedDataset(__snake_case      ,human_eval["test"]      ,n_copies=__snake_case      ,n_tasks=__snake_case			)
			# do not confuse args.batch_size, which is actually the num_return_sequences
			__lowerCAmelCase   :						str				= DataLoader(__snake_case      ,batch_size=1			)

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

			__lowerCAmelCase		,			__lowerCAmelCase   :						Dict				= accelerator.prepare(__snake_case      ,__snake_case			)

			__lowerCAmelCase   :						Optional[Any]				= complete_code(
			    __snake_case      ,__snake_case      ,__snake_case      ,__snake_case      ,n_tasks=__snake_case      ,batch_size=args.batch_size      ,**__snake_case      ,)

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

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

						# Evaluate completions with "code_eval" metric
						__lowerCAmelCase		,			__lowerCAmelCase   :						List[Any]				= code_eval_metric.compute(
						    references=__snake_case      ,predictions=__snake_case      ,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(__snake_case      ,__snake_case			)


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




"""simple docstring"""






import re

from filelock import FileLock


try:
	import nltk

	__snake_case							:     Any   =					True
except (ImportError, ModuleNotFoundError):
	__snake_case							:     Optional[int]   =					False

if NLTK_AVAILABLE:
	with FileLock('.lock') as lock:
		nltk.download('punkt', quiet=True)


def     _lowercase       (						__snake_case			)		->     str:
			re.sub("<n>"      ,""      ,__snake_case			)  # remove pegasus newline char
			assert NLTK_AVAILABLE, "nltk must be installed to separate newlines between sentences. (pip install nltk)"
			return "\n".join(nltk.sent_tokenize(__snake_case			)			)