





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


class 					__a      (lowerCamelCase     ):

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







													"""simple docstring"""
													UpperCAmelCase_			:      Union[str, Any]     						=  tokenizer
													UpperCAmelCase_			:      str     						=  dataset
													UpperCAmelCase_			:      str     						=  len(__magic_name__   ) if n_tasks is None else n_tasks
													UpperCAmelCase_			:      List[Any]     						=  n_copies



							def __iter__(							self							:     str   )					->       Dict:







													"""simple docstring"""
													UpperCAmelCase_			:      Optional[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()   )
													UpperCAmelCase_			:      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 					__a      (lowerCamelCase     ):

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







													"""simple docstring"""
													UpperCAmelCase_			:      List[Any]     						=  start_length
													UpperCAmelCase_			:      Dict     						=  eof_strings
													UpperCAmelCase_			:      List[str]     						=  tokenizer



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







													"""simple docstring"""
													UpperCAmelCase_			:      Any     						=  self.tokenizer.batch_decode(input_ids[:, self.start_length :]   )
													UpperCAmelCase_			:      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(__magic_name__   )





def        lowerCamelCase_      (  SCREAMING_SNAKE_CASE__      :			List[Any]   ) ->	Optional[Any]:
						UpperCAmelCase_			:      str     						=  re.split('''(%s)''' % '''|'''.join(SCREAMING_SNAKE_CASE__   ),			SCREAMING_SNAKE_CASE__   )
						# last string should be ""
						return "".join(string_list[:-2]   )





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

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

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

						UpperCAmelCase_			:      List[Any]     						=  [[] for _ in range(SCREAMING_SNAKE_CASE__   )]
						for task, generated_tokens in gen_token_dict.items():
												for s in generated_tokens:
																		UpperCAmelCase_			:      int     						=  tokenizer.decode(SCREAMING_SNAKE_CASE__,			skip_special_tokens=SCREAMING_SNAKE_CASE__,			clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE__   )
																		code_gens[task].append(remove_last_block(SCREAMING_SNAKE_CASE__   )   )
						return code_gens





def        lowerCamelCase_      (  ) ->	Tuple:
						# Setup configuration
						UpperCAmelCase_			:      Optional[Any]     						=  HfArgumentParser(SCREAMING_SNAKE_CASE__   )
						UpperCAmelCase_			:      List[str]     						=  parser.parse_args()

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

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

						# Use dataset load to feed to accelerate
						UpperCAmelCase_			:      Optional[Any]     						=  Accelerator()
						set_seed(args.seed,			device_specific=SCREAMING_SNAKE_CASE__   )

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

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

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

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

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

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

						UpperCAmelCase_			:      List[str]     						=  accelerator.prepare(SCREAMING_SNAKE_CASE__,			SCREAMING_SNAKE_CASE__   )

						UpperCAmelCase_			:      Optional[int]     						=  complete_code(
						    SCREAMING_SNAKE_CASE__,			SCREAMING_SNAKE_CASE__,			SCREAMING_SNAKE_CASE__,			SCREAMING_SNAKE_CASE__,			n_tasks=SCREAMING_SNAKE_CASE__,			batch_size=args.batch_size,			**SCREAMING_SNAKE_CASE__,			)

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

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

												# Evaluate completions with "code_eval" metric
												UpperCAmelCase_			:      Union[str, Any]     						=  code_eval_metric.compute(
												    references=SCREAMING_SNAKE_CASE__,			predictions=SCREAMING_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(SCREAMING_SNAKE_CASE__,			SCREAMING_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 unittest

import torch
from torch import nn

from diffusers.models.activations import get_activation


class 					__a      (unittest.TestCase     ):

							def     UpperCAmelCase__	(							self							:     Dict   )					->       Dict:







													"""simple docstring"""
													UpperCAmelCase_			:      Dict     						=  get_activation('''swish'''   )

													self.assertIsInstance(__magic_name__       ,     nn.SiLU   )

													self.assertEqual(act(torch.tensor(-1_00       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertNotEqual(act(torch.tensor(-1       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(0       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(20       ,     dtype=torch.floataa   )   ).item()       ,     20   )

							def     UpperCAmelCase__	(							self							:     Tuple   )					->       Tuple:







													"""simple docstring"""
													UpperCAmelCase_			:      Union[str, Any]     						=  get_activation('''silu'''   )

													self.assertIsInstance(__magic_name__       ,     nn.SiLU   )

													self.assertEqual(act(torch.tensor(-1_00       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertNotEqual(act(torch.tensor(-1       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(0       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(20       ,     dtype=torch.floataa   )   ).item()       ,     20   )

							def     UpperCAmelCase__	(							self							:     Tuple   )					->       Optional[int]:







													"""simple docstring"""
													UpperCAmelCase_			:      Optional[int]     						=  get_activation('''mish'''   )

													self.assertIsInstance(__magic_name__       ,     nn.Mish   )

													self.assertEqual(act(torch.tensor(-2_00       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertNotEqual(act(torch.tensor(-1       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(0       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(20       ,     dtype=torch.floataa   )   ).item()       ,     20   )



							def     UpperCAmelCase__	(							self							:     str   )					->       Optional[Any]:







													"""simple docstring"""
													UpperCAmelCase_			:      List[Any]     						=  get_activation('''gelu'''   )

													self.assertIsInstance(__magic_name__       ,     nn.GELU   )

													self.assertEqual(act(torch.tensor(-1_00       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertNotEqual(act(torch.tensor(-1       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(0       ,     dtype=torch.floataa   )   ).item()       ,     0   )
													self.assertEqual(act(torch.tensor(20       ,     dtype=torch.floataa   )   ).item()       ,     20   )





