






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






class 							__magic_name__     (	__UpperCAmelCase				):







	"""simple docstring"""



	def __init__(							self  ,      _lowercase  ,      _lowercase  ,      _lowercase=None  ,      _lowercase=1       )						->  List[Any]:
							lowercase_      :   List[Any]        =			tokenizer
							lowercase_      :   int        =			dataset
							lowercase_      :   str        =			len(__SCREAMING_SNAKE_CASE       ) if n_tasks is None else n_tasks
							lowercase_      :   int        =			n_copies







	def __iter__(							self       )						->  str:
							lowercase_      :   List[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()       )
							lowercase_      :   str        =			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 							__magic_name__     (	__UpperCAmelCase				):







	"""simple docstring"""



	def __init__(							self  ,      _lowercase  ,      _lowercase  ,      _lowercase       )						->  Tuple:
							lowercase_      :   List[str]        =			start_length
							lowercase_      :   List[Any]        =			eof_strings
							lowercase_      :   str        =			tokenizer







	def __call__(							self  ,      _lowercase  ,      _lowercase  ,      **_lowercase       )						->  str:
							lowercase_      :   int        =			self.tokenizer.batch_decode(input_ids[:, self.start_length :]       )
							lowercase_      :   List[str]        =			[]
							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 	_UpperCAmelCase    (  a		:	List[Any]							)			->			Optional[Any]:





						"""simple docstring"""
						lowercase_      :   Optional[Any]        =			re.split('(%s)' % '|'.join(_UpperCAmelCase							)						,			_UpperCAmelCase							)
						# last string should be ""
						return "".join(string_list[:-2]							)






def 	_UpperCAmelCase    (  a		:	List[str]						,			a		:	Optional[int]						,			a		:	int						,			a		:	Union[str, Any]						,			a		:	Tuple						,			a		:	List[Any]=2_0						,			**a		:	str							)			->			List[Any]:





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

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

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

						lowercase_      :   Optional[Any]        =			[[] for _ in range(_UpperCAmelCase							)]
						for task, generated_tokens in gen_token_dict.items():
												for s in generated_tokens:
																		lowercase_      :   Dict        =			tokenizer.decode(_UpperCAmelCase						,			skip_special_tokens=_UpperCAmelCase						,			clean_up_tokenization_spaces=_UpperCAmelCase							)
																		code_gens[task].append(remove_last_block(_UpperCAmelCase							)							)
						return code_gens






def 	_UpperCAmelCase    (  )			->			Tuple:





						"""simple docstring"""
						# Setup configuration
						lowercase_      :   Tuple        =			HfArgumentParser(_UpperCAmelCase							)
						lowercase_      :   Tuple        =			parser.parse_args()

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

						if args.num_workers is None:
												lowercase_      :   Optional[int]        =			multiprocessing.cpu_count()

						# Use dataset load to feed to accelerate
						lowercase_      :   Tuple        =			Accelerator()
						set_seed(args.seed						,			device_specific=_UpperCAmelCase							)

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

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

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

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

						lowercase_      :   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
						lowercase_      :   str        =			DataLoader(_UpperCAmelCase						,			batch_size=1							)

						# Run a quick test to see if code evaluation is enabled
						try:
												lowercase_      :   Union[str, 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

						lowercase_  ,      lowercase_      :   str        =			accelerator.prepare(_UpperCAmelCase						,			_UpperCAmelCase							)

						lowercase_      :   List[Any]        =			complete_code(
						    _UpperCAmelCase						,			_UpperCAmelCase						,			_UpperCAmelCase						,			_UpperCAmelCase						,			n_tasks=_UpperCAmelCase						,			batch_size=args.batch_size						,			**_UpperCAmelCase						,			)

						if accelerator.is_main_process:
												lowercase_      :   Dict        =			[]

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

												# Evaluate completions with "code_eval" metric
												lowercase_  ,      lowercase_      :   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'''

import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format="%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s",
    datefmt="%m/%d/%Y %H:%M:%S",
    level=logging.INFO,
)
A:      Tuple  				=				logging.getLogger(__name__)






def 	_UpperCAmelCase    (  a		:	str							)			->			List[Any]:





						"""simple docstring"""
						lowercase_      :   List[str]        =			git.Repo(search_parent_directories=a							)
						lowercase_      :   Union[str, Any]        =			{
						    'repo_id': str(a							),
						    'repo_sha': str(repo.head.object.hexsha							),
						    'repo_branch': str(repo.active_branch							),
						}

						with open(os.path.join(a						,			'git_log.json'							)						,			'w'							) as f:
												json.dump(a						,			a						,			indent=4							)






def 	_UpperCAmelCase    (  a		:	str							)			->			Union[str, Any]:





						"""simple docstring"""
						if params.n_gpu <= 0:
												lowercase_      :   int        =			0
												lowercase_      :   Union[str, Any]        =			-1
												lowercase_      :   List[str]        =			True
												lowercase_      :   Optional[Any]        =			False
												return

						assert torch.cuda.is_available()

						logger.info('Initializing GPUs'							)
						if params.n_gpu > 1:
												assert params.local_rank != -1

												lowercase_      :   Dict        =			int(os.environ['WORLD_SIZE']							)
												lowercase_      :   Union[str, Any]        =			int(os.environ['N_GPU_NODE']							)
												lowercase_      :   Optional[int]        =			int(os.environ['RANK']							)

												# number of nodes / node ID
												lowercase_      :   int        =			params.world_size // params.n_gpu_per_node
												lowercase_      :   str        =			params.global_rank // params.n_gpu_per_node
												lowercase_      :   Dict        =			True

												assert params.n_nodes == int(os.environ['N_NODES']							)
												assert params.node_id == int(os.environ['NODE_RANK']							)

						# local job (single GPU)
						else:
												assert params.local_rank == -1

												lowercase_      :   str        =			1
												lowercase_      :   Dict        =			0
												lowercase_      :   Tuple        =			0
												lowercase_      :   List[Any]        =			0
												lowercase_      :   int        =			1
												lowercase_      :   Tuple        =			1
												lowercase_      :   str        =			False

						# sanity checks
						assert params.n_nodes >= 1
						assert 0 <= params.node_id < params.n_nodes
						assert 0 <= params.local_rank <= params.global_rank < params.world_size
						assert params.world_size == params.n_nodes * params.n_gpu_per_node

						# define whether this is the master process / if we are in multi-node distributed mode
						lowercase_      :   List[str]        =			params.node_id == 0 and params.local_rank == 0
						lowercase_      :   Optional[Any]        =			params.n_nodes > 1

						# summary
						lowercase_      :   int        =			f"--- Global rank: {params.global_rank} - "
						logger.info(PREFIX + 'Number of nodes: %i' % params.n_nodes							)
						logger.info(PREFIX + 'Node ID        : %i' % params.node_id							)
						logger.info(PREFIX + 'Local rank     : %i' % params.local_rank							)
						logger.info(PREFIX + 'World size     : %i' % params.world_size							)
						logger.info(PREFIX + 'GPUs per node  : %i' % params.n_gpu_per_node							)
						logger.info(PREFIX + 'Master         : %s' % str(params.is_master							)							)
						logger.info(PREFIX + 'Multi-node     : %s' % str(params.multi_node							)							)
						logger.info(PREFIX + 'Multi-GPU      : %s' % str(params.multi_gpu							)							)
						logger.info(PREFIX + 'Hostname       : %s' % socket.gethostname()							)

						# set GPU device
						torch.cuda.set_device(params.local_rank							)

						# initialize multi-GPU
						if params.multi_gpu:
												logger.info('Initializing PyTorch distributed'							)
												torch.distributed.init_process_group(
												    init_method='env://'						,			backend='nccl'						,			)






def 	_UpperCAmelCase    (  a		:	Dict							)			->			Optional[int]:





						"""simple docstring"""
						np.random.seed(args.seed							)
						torch.manual_seed(args.seed							)
						if args.n_gpu > 0:
												torch.cuda.manual_seed_all(args.seed							)
