





"""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


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




class 			__lowercase     (     __lowercase							):







					def __init__(  self		,     _UpperCAmelCase		,     _UpperCAmelCase		,     _UpperCAmelCase=None		,     _UpperCAmelCase=1       ):
									__a	:  Union[str, Any]         =					tokenizer
									__a	:  Dict         =					dataset
									__a	:  str         =					len(UpperCAmelCase__       ) if n_tasks is None else n_tasks
									__a	:  List[Any]         =					n_copies







					def __iter__(  self       ):
									__a	:  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()       )
									__a	:  Optional[Any]         =					self.tokenizer(UpperCAmelCase__		,     padding=UpperCAmelCase__		,     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 			__lowercase     (     __lowercase							):







					def __init__(  self		,     _UpperCAmelCase		,     _UpperCAmelCase		,     _UpperCAmelCase       ):
									__a	:  Optional[int]         =					start_length
									__a	:  Dict         =					eof_strings
									__a	:  Any         =					tokenizer







					def __call__(  self		,     _UpperCAmelCase		,     _UpperCAmelCase		,     **_UpperCAmelCase       ):
									__a	:  str         =					self.tokenizer.batch_decode(input_ids[:, self.start_length :]       )
									__a	:  int         =					[]
									for decoded_generation in decoded_generations:
													done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings       )       )
									return all(UpperCAmelCase__       )






def       __A (     a_  :Optional[int])						->						Any:
				__a	:  Tuple         =					re.split('''(%s)''' % '''|'''.join(a_)				,    a_)
				# last string should be ""
				return "".join(string_list[:-2])

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

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

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

				__a	:  List[str]         =					[[] for _ in range(a_)]
				for task, generated_tokens in gen_token_dict.items():
								for s in generated_tokens:
												__a	:  Dict         =					tokenizer.decode(a_				,    skip_special_tokens=a_				,    clean_up_tokenization_spaces=a_)
												code_gens[task].append(remove_last_block(a_))
				return code_gens

def       __A (     )						->						Any:
				# Setup configuration
				__a	:  Tuple         =					HfArgumentParser(a_)
				__a	:  Dict         =					parser.parse_args()

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

				if args.num_workers is None:
								__a	:  List[str]         =					multiprocessing.cpu_count()

				# Use dataset load to feed to accelerate
				__a	:  List[Any]         =					Accelerator()
				set_seed(args.seed				,    device_specific=a_)

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

				# Generation settings
				__a	:  Optional[int]         =					{
				    '''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				,    a_				,    a_)]),
				}

				# Load evaluation dataset and metric
				__a	:  List[Any]         =					load_dataset('''openai_humaneval''')
				__a	:  Optional[int]         =					load_metric('''code_eval''')

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

				__a	:  Any         =					TokenizedDataset(a_				,    human_eval['''test''']				,    n_copies=a_				,    n_tasks=a_)
				# do not confuse args.batch_size, which is actually the num_return_sequences
				__a	:  int         =					DataLoader(a_				,    batch_size=1)

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

				__a    ,  __a	:  List[str]         =					accelerator.prepare(a_				,    a_)

				__a	:  Union[str, Any]         =					complete_code(
				    a_				,    a_				,    a_				,    a_				,    n_tasks=a_				,    batch_size=args.batch_size				,    **a_				,    )

				if accelerator.is_main_process:
								__a	:  Optional[int]         =					[]

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

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


# 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"""






def       __A (     a_  :float)						->						float:

				if edge <= 0 or not isinstance(a_				,    a_):
								raise ValueError('''Length must be a positive.''')
				return 3 * ((25 + 10 * (5 ** (1 / 2))) ** (1 / 2)) * (edge**2)

def       __A (     a_  :float)						->						float:

				if edge <= 0 or not isinstance(a_				,    a_):
								raise ValueError('''Length must be a positive.''')
				return ((15 + (7 * (5 ** (1 / 2)))) / 4) * (edge**3)


if __name__ == "__main__":
		import doctest

		doctest.testmod()