





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_        =							['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']



class    __lowerCAmelCase			(     _a	):






			def __init__(self   ,				__magic_name__   ,				__magic_name__   ,				__magic_name__=None   ,				__magic_name__=1       )	->  Optional[int]:


							'''simple docstring'''



							snake_case_		:       Union[str, Any]           =    tokenizer
							snake_case_		:       int           =    dataset
							snake_case_		:       List[str]           =    len(__magic_name__       ) if n_tasks is None else n_tasks
							snake_case_		:       Dict           =    n_copies






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


							'''simple docstring'''



							snake_case_		:       Optional[int]           =    []
							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[Any]           =    self.tokenizer(__magic_name__   ,				padding=__magic_name__   ,				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    __lowerCAmelCase			(     _a	):






			def __init__(self   ,				__magic_name__   ,				__magic_name__   ,				__magic_name__       )	->  Union[str, Any]:


							'''simple docstring'''



							snake_case_		:       Dict           =    start_length
							snake_case_		:       Any           =    eof_strings
							snake_case_		:       Tuple           =    tokenizer






			def __call__(self   ,				__magic_name__   ,				__magic_name__   ,				**__magic_name__       )	->  Dict:


							'''simple docstring'''



							snake_case_		:       List[Any]           =    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(__magic_name__       )






def 	lowerCamelCase_		(							_UpperCamelCase					)						->		Tuple:






				"""simple docstring"""


				snake_case_		:       Optional[int]           =    re.split('''(%s)''' % '''|'''.join(_UpperCamelCase					)					,							_UpperCamelCase					)
				# last string should be ""
				return "".join(string_list[:-2]					)






def 	lowerCamelCase_		(							_UpperCamelCase					,							_UpperCamelCase					,							_UpperCamelCase					,							_UpperCamelCase					,							_UpperCamelCase					,							_UpperCamelCase=20					,							**_UpperCamelCase					)						->		str:






				"""simple docstring"""


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

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

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

				snake_case_		:       Any           =    [[] for _ in range(_UpperCamelCase					)]
				for task, generated_tokens in gen_token_dict.items():
								for s in generated_tokens:
												snake_case_		:       Tuple           =    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_		(							)						->		Any:






				"""simple docstring"""


				snake_case_		:       Dict           =    HfArgumentParser(_UpperCamelCase					)
				snake_case_		:       Optional[Any]           =    parser.parse_args()

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

				if args.num_workers is None:
								snake_case_		:       Dict           =    multiprocessing.cpu_count()

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

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

				# Generation settings
				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
				snake_case_		:       List[Any]           =    load_dataset('''openai_humaneval'''					)
				snake_case_		:       Dict           =    load_metric('''code_eval'''					)

				snake_case_		:       Union[str, Any]           =    args.num_tasks if args.num_tasks is not None else len(human_eval['''test''']					)
				snake_case_		:       Optional[Any]           =    args.n_samples // args.batch_size

				snake_case_		:       List[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
				snake_case_		:       List[Any]           =    DataLoader(_UpperCamelCase					,							batch_size=1					)

				# Run a quick test to see if code evaluation is enabled
				try:
								snake_case_		:       int           =    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_		:       Any           =    accelerator.prepare(_UpperCamelCase					,							_UpperCamelCase					)

				snake_case_		:       List[Any]           =    complete_code(
				    _UpperCamelCase					,							_UpperCamelCase					,							_UpperCamelCase					,							_UpperCamelCase					,							n_tasks=_UpperCamelCase					,							batch_size=args.batch_size					,							**_UpperCamelCase					,							)

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

								for task in tqdm(range(_UpperCamelCase					)					):
												snake_case_		:       Optional[Any]           =    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_		:       Dict           =    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()




'''simple docstring'''
def 					lowerCamelCase_ ( __UpperCamelCase					:       list							,    __UpperCamelCase					:       int							,    __UpperCamelCase					:       int = 0							,    __UpperCamelCase					:       int = 0		)					->	int:







     """simple docstring"""
     _A				=							right or len(__UpperCamelCase		) - 1
     if left > right:
          return -1
     elif list_data[left] == key:
          return left
     elif list_data[right] == key:
          return right
     else:
          return search(__UpperCamelCase							,    __UpperCamelCase							,    left + 1							,    right - 1		)


if __name__ == "__main__":
    import doctest

    doctest.testmod()





