





import inspect
import warnings
from typing import Any, Dict, Optional, Union

from packaging import version







def       UpperCAmelCase							(     *_lowerCamelCase      :  Optional[Any]			,   _lowerCamelCase      :  Optional[Union[Dict, Any]] = None			,   _lowerCamelCase      :  Dict=True			,   _lowerCamelCase      :  Dict=2     ):





						'''simple docstring'''


						from .. import __version__

						SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					take_from
						SCREAMING_SNAKE_CASE__     :				str   =					()
						if not isinstance(args[0]			,   _lowerCamelCase     ):
												SCREAMING_SNAKE_CASE__     :				Dict   =					(args,)

						for attribute, version_name, message in args:
												if version.parse(version.parse(_lowerCamelCase     ).base_version     ) >= version.parse(_lowerCamelCase     ):
																		raise ValueError(
																		    f"""The deprecation tuple {(attribute, version_name, message)} should be removed since diffusers'"""
																		    f""" version {__version__} is >= {version_name}"""     )

												SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					None
												if isinstance(_lowerCamelCase			,   _lowerCamelCase     ) and attribute in deprecated_kwargs:
																		values += (deprecated_kwargs.pop(_lowerCamelCase     ),)
																		SCREAMING_SNAKE_CASE__     :				Optional[int]   =					f"""The `{attribute}` argument is deprecated and will be removed in version {version_name}."""
												elif hasattr(_lowerCamelCase			,   _lowerCamelCase     ):
																		values += (getattr(_lowerCamelCase			,   _lowerCamelCase     ),)
																		SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					f"""The `{attribute}` attribute is deprecated and will be removed in version {version_name}."""
												elif deprecated_kwargs is None:
																		SCREAMING_SNAKE_CASE__     :				List[Any]   =					f"""`{attribute}` is deprecated and will be removed in version {version_name}."""

												if warning is not None:
																		SCREAMING_SNAKE_CASE__     :				List[str]   =					warning + " " if standard_warn else ""
																		warnings.warn(warning + message			,   _lowerCamelCase			,   stacklevel=_lowerCamelCase     )

						if isinstance(_lowerCamelCase			,   _lowerCamelCase     ) and len(_lowerCamelCase     ) > 0:
												SCREAMING_SNAKE_CASE__     :				List[str]   =					inspect.getouterframes(inspect.currentframe()     )[1]
												SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					call_frame.filename
												SCREAMING_SNAKE_CASE__     :				Tuple   =					call_frame.lineno
												SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					call_frame.function
												SCREAMING_SNAKE_CASE__     :				List[str]   =					next(iter(deprecated_kwargs.items()     )     )
												raise TypeError(f"""{function} in {filename} line {line_number-1} got an unexpected keyword argument `{key}`"""     )

						if len(_lowerCamelCase     ) == 0:
												return
						elif len(_lowerCamelCase     ) == 1:
												return values[0]
						return values






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


__lowercase       :str       =  ["\nclass", "\ndef", "\n#", "\n@", "\nprint", "\nif"]



class    _a     (				lowercase__							):






							"""simple docstring"""

							def __init__(      self				:     List[str]							,	a				:     Optional[int]							,	a				:     str							,	a				:     int=None							,	a				:     Optional[Any]=1    )	->Optional[Any]:
													SCREAMING_SNAKE_CASE__     :				Dict   =					tokenizer
													SCREAMING_SNAKE_CASE__     :				Optional[int]   =					dataset
													SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					len(a    ) if n_tasks is None else n_tasks
													SCREAMING_SNAKE_CASE__     :				Dict   =					n_copies






							def __iter__(      self				:     str    )	->Tuple:
													SCREAMING_SNAKE_CASE__     :				str   =					[]
													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()    )
													SCREAMING_SNAKE_CASE__     :				int   =					self.tokenizer(a							,	padding=a							,	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     (				lowercase__							):






							"""simple docstring"""

							def __init__(      self				:     Dict							,	a				:     int							,	a				:     int							,	a				:     Tuple    )	->Dict:
													SCREAMING_SNAKE_CASE__     :				Dict   =					start_length
													SCREAMING_SNAKE_CASE__     :				Any   =					eof_strings
													SCREAMING_SNAKE_CASE__     :				Any   =					tokenizer






							def __call__(      self				:     Any							,	a				:     Optional[int]							,	a				:     int							,	**a				:     Union[str, Any]    )	->List[str]:
													SCREAMING_SNAKE_CASE__     :				Dict   =					self.tokenizer.batch_decode(input_ids[:, self.start_length :]    )
													SCREAMING_SNAKE_CASE__     :				int   =					[]
													for decoded_generation in decoded_generations:
																			done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings    )    )
													return all(a    )







def       UpperCAmelCase							(     _lowerCamelCase      :  Dict     ):





						'''simple docstring'''


						SCREAMING_SNAKE_CASE__     :				List[Any]   =					re.split("(%s)" % "|".join(_lowerCamelCase     )			,   _lowerCamelCase     )
						# last string should be ""
						return "".join(string_list[:-2]     )







def       UpperCAmelCase							(     _lowerCamelCase      :  Dict			,   _lowerCamelCase      :  Optional[Any]			,   _lowerCamelCase      :  Tuple			,   _lowerCamelCase      :  str			,   _lowerCamelCase      :  int			,   _lowerCamelCase      :  str=20			,   **_lowerCamelCase      :  Dict     ):





						'''simple docstring'''


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

																		SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Any   =					accelerator.gather((generated_tokens, generated_tasks)     )
																		SCREAMING_SNAKE_CASE__     :				Dict   =					generated_tokens.cpu().numpy()
																		SCREAMING_SNAKE_CASE__     :				Any   =					generated_tasks.cpu().numpy()

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

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







def       UpperCAmelCase							(     ):





						'''simple docstring'''


						SCREAMING_SNAKE_CASE__     :				Optional[Any]   =					HfArgumentParser(_lowerCamelCase     )
						SCREAMING_SNAKE_CASE__     :				Union[str, Any]   =					parser.parse_args()

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

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

						# Use dataset load to feed to accelerate
						SCREAMING_SNAKE_CASE__     :				Dict   =					Accelerator()
						set_seed(args.seed			,   device_specific=_lowerCamelCase     )

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

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

						# Load evaluation dataset and metric
						SCREAMING_SNAKE_CASE__     :				str   =					load_dataset("openai_humaneval"     )
						SCREAMING_SNAKE_CASE__     :				Any   =					load_metric("code_eval"     )

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

						SCREAMING_SNAKE_CASE__     :				Dict   =					TokenizedDataset(_lowerCamelCase			,   human_eval["test"]			,   n_copies=_lowerCamelCase			,   n_tasks=_lowerCamelCase     )
						# do not confuse args.batch_size, which is actually the num_return_sequences
						SCREAMING_SNAKE_CASE__     :				Optional[int]   =					DataLoader(_lowerCamelCase			,   batch_size=1     )

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

						SCREAMING_SNAKE_CASE__,							SCREAMING_SNAKE_CASE__     :				Optional[int]   =					accelerator.prepare(_lowerCamelCase			,   _lowerCamelCase     )

						SCREAMING_SNAKE_CASE__     :				Tuple   =					complete_code(
						    _lowerCamelCase			,   _lowerCamelCase			,   _lowerCamelCase			,   _lowerCamelCase			,   n_tasks=_lowerCamelCase			,   batch_size=args.batch_size			,   **_lowerCamelCase			,   )

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

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

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


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