



from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


A__								=						logging.get_logger(__name__)


class       a							(			__lowerCamelCase				):



		def __init__(     self     :Any   ,__lowercase     :int   ,__lowercase     :int   ,__lowercase     :float   ,**__lowercase     :List[str]  ):
						snake_case__	:      Optional[int]     		=	feature_size
						snake_case__	:      Any     		=	sampling_rate
						snake_case__	:      Dict     		=	padding_value

						snake_case__	:      Union[str, Any]     		=	kwargs.pop('''padding_side'''   ,'''right'''  )
						snake_case__	:      str     		=	kwargs.pop('''return_attention_mask'''   ,__lowercase  )

						super().__init__(**__lowercase  )



		def 	__lowerCamelCase			(     self     :Any   ,__lowercase     :Union[
		        BatchFeature,
		        List[BatchFeature],
		        Dict[str, BatchFeature],
		        Dict[str, List[BatchFeature]],
		        List[Dict[str, BatchFeature]],
		    ]   ,__lowercase     :Union[bool, str, PaddingStrategy] = True   ,__lowercase     :Optional[int] = None   ,__lowercase     :bool = False   ,__lowercase     :Optional[int] = None   ,__lowercase     :Optional[bool] = None   ,__lowercase     :Optional[Union[str, TensorType]] = None   ,):
						# If we have a list of dicts, let's convert it in a dict of lists
						# We do this to allow using this method as a collate_fn function in PyTorch Dataloader
						if isinstance(__lowercase   ,(list, tuple)  ) and isinstance(processed_features[0]   ,(dict, BatchFeature)  ):
										snake_case__	:      int     		=	{
										    key: [example[key] for example in processed_features] for key in processed_features[0].keys()
										}

						# The model's main input name, usually `input_values`, has be passed for padding
						if self.model_input_names[0] not in processed_features:
										raise ValueError(
										    '''You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`'''
										    F""" to this method that includes {self.model_input_names[0]}, but you provided"""
										    F""" {list(processed_features.keys()  )}"""  )

						snake_case__	:      Optional[int]     		=	processed_features[self.model_input_names[0]]
						snake_case__	:      str     		=	(
						    return_attention_mask if return_attention_mask is not None else self.return_attention_mask
						)

						if len(__lowercase  ) == 0:
										if return_attention_mask:
														snake_case__	:      Tuple     		=	[]
										return processed_features

						# If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
						# and rebuild them afterwards if no return_tensors is specified
						# Note that we lose the specific device the tensor may be on for PyTorch

						snake_case__	:      Tuple     		=	required_input[0]
						if isinstance(__lowercase   ,(list, tuple)  ):
										# first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
										snake_case__	:      Any     		=	0
										while len(required_input[index]  ) == 0:
														index += 1
										if index < len(__lowercase  ):
														snake_case__	:      Tuple     		=	required_input[index][0]

						if return_tensors is None:
										if is_tf_tensor(__lowercase  ):
														snake_case__	:      Tuple     		=	'''tf'''
										elif is_torch_tensor(__lowercase  ):
														snake_case__	:      Tuple     		=	'''pt'''
										elif isinstance(__lowercase   ,(int, float, list, tuple, np.ndarray)  ):
														snake_case__	:      Union[str, Any]     		=	'''np'''
										else:
														raise ValueError(
														    F"""type of {first_element} unknown: {type(__lowercase  )}. """
														    '''Should be one of a python, numpy, pytorch or tensorflow object.'''  )

						for key, value in processed_features.items():
										if isinstance(value[0]   ,(int, float)  ):
														snake_case__	:      Dict     		=	to_numpy(__lowercase  )
										else:
														snake_case__	:      str     		=	[to_numpy(__lowercase  ) for v in value]

        # Convert padding_strategy in PaddingStrategy
						snake_case__	:      str     		=	self._get_padding_strategies(padding=__lowercase   ,max_length=__lowercase  )

						snake_case__	:      Tuple     		=	processed_features[self.model_input_names[0]]

						snake_case__	:      int     		=	len(__lowercase  )
						if not all(len(__lowercase  ) == batch_size for v in processed_features.values()  ):
										raise ValueError('''Some items in the output dictionary have a different batch size than others.'''  )

						snake_case__	:      Optional[Any]     		=	[]
						for i in range(__lowercase  ):
										snake_case__	:      str     		=	{k: v[i] for k, v in processed_features.items()}
										# truncation
										snake_case__	:      Optional[int]     		=	self._truncate(
										    __lowercase   ,max_length=__lowercase   ,pad_to_multiple_of=__lowercase   ,truncation=__lowercase   ,)
										truncated_inputs.append(__lowercase  )

						if padding_strategy == PaddingStrategy.LONGEST:
										# make sure that `max_length` cannot be longer than the longest truncated length
										snake_case__	:      List[Any]     		=	max(len(input_slice[self.model_input_names[0]]  ) for input_slice in truncated_inputs  )
										snake_case__	:      Optional[Any]     		=	PaddingStrategy.MAX_LENGTH

						snake_case__	:      List[str]     		=	{}
						for i in range(__lowercase  ):
										# padding
										snake_case__	:      Tuple     		=	self._pad(
										    truncated_inputs[i]   ,max_length=__lowercase   ,padding_strategy=__lowercase   ,pad_to_multiple_of=__lowercase   ,return_attention_mask=__lowercase   ,)

										for key, value in outputs.items():
														if key not in batch_outputs:
																		snake_case__	:      Tuple     		=	[]
														if value.dtype is np.dtype(np.floataa  ):
																		snake_case__	:      Union[str, Any]     		=	value.astype(np.floataa  )
														batch_outputs[key].append(__lowercase  )

						return BatchFeature(__lowercase   ,tensor_type=__lowercase  )



		def 	__lowerCamelCase			(     self     :str   ,__lowercase     :Union[Dict[str, np.ndarray], BatchFeature]   ,__lowercase     :Optional[int] = None   ,__lowercase     :PaddingStrategy = PaddingStrategy.DO_NOT_PAD   ,__lowercase     :Optional[int] = None   ,__lowercase     :Optional[bool] = None   ,):
						snake_case__	:      Optional[int]     		=	processed_features[self.model_input_names[0]]

						if padding_strategy == PaddingStrategy.LONGEST:
										snake_case__	:      List[str]     		=	len(__lowercase  )

						if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
										snake_case__	:      Union[str, Any]     		=	((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

						snake_case__	:      str     		=	padding_strategy != PaddingStrategy.DO_NOT_PAD and len(__lowercase  ) < max_length

						if return_attention_mask and "attention_mask" not in processed_features:
										snake_case__	:      Dict     		=	np.ones(len(__lowercase  )   ,dtype=np.intaa  )

						if needs_to_be_padded:
										snake_case__	:      int     		=	max_length - len(__lowercase  )
										if self.padding_side == "right":
														if return_attention_mask:
																		snake_case__	:      Dict     		=	np.pad(
																		    processed_features['''attention_mask''']   ,(0, difference)  )
														snake_case__	:      Optional[int]     		=	((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
														snake_case__	:      Optional[int]     		=	np.pad(
														    __lowercase   ,__lowercase   ,'''constant'''   ,constant_values=self.padding_value  )
										elif self.padding_side == "left":
														if return_attention_mask:
																		snake_case__	:      Dict     		=	np.pad(
																		    processed_features['''attention_mask''']   ,(difference, 0)  )
														snake_case__	:      Any     		=	((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
														snake_case__	:      Optional[int]     		=	np.pad(
														    __lowercase   ,__lowercase   ,'''constant'''   ,constant_values=self.padding_value  )
										else:
														raise ValueError('''Invalid padding strategy:''' + str(self.padding_side  )  )

						return processed_features



		def 	__lowerCamelCase			(     self     :Optional[int]   ,__lowercase     :Union[Dict[str, np.ndarray], BatchFeature]   ,__lowercase     :Optional[int] = None   ,__lowercase     :Optional[int] = None   ,__lowercase     :Optional[bool] = None   ,):
						if not truncation:
										return processed_features
						elif truncation and max_length is None:
										raise ValueError('''When setting ``truncation=True``, make sure that ``max_length`` is defined.'''  )

						snake_case__	:      Tuple     		=	processed_features[self.model_input_names[0]]

						# find `max_length` that fits `pad_to_multiple_of`
						if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
										snake_case__	:      Optional[int]     		=	((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

						snake_case__	:      Tuple     		=	len(__lowercase  ) > max_length

						if needs_to_be_truncated:
										snake_case__	:      Dict     		=	processed_features[self.model_input_names[0]][:max_length]
										if "attention_mask" in processed_features:
														snake_case__	:      List[str]     		=	processed_features['''attention_mask'''][:max_length]

						return processed_features






		def 	__lowerCamelCase			(     self     :Tuple   ,__lowercase     :int=False   ,__lowercase     :int=None  ):

						# Get padding strategy
						if padding is not False:
										if padding is True:
														snake_case__	:      Optional[Any]     		=	PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
										elif not isinstance(__lowercase   ,__lowercase  ):
														snake_case__	:      Optional[Any]     		=	PaddingStrategy(__lowercase  )
										elif isinstance(__lowercase   ,__lowercase  ):
														snake_case__	:      List[Any]     		=	padding
						else:
										snake_case__	:      Optional[int]     		=	PaddingStrategy.DO_NOT_PAD

						# Set max length if needed
						if max_length is None:
										if padding_strategy == PaddingStrategy.MAX_LENGTH:
														raise ValueError(
														    F"""When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined"""  )

        # Test if we have a padding value
						if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
										raise ValueError(
										    '''Asking to pad but the feature_extractor does not have a padding value. Please select a value to use'''
										    ''' as `padding_value`. For example: `feature_extractor.padding_value = 0.0`.'''  )

						return padding_strategy






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       a							(			__lowerCamelCase				):



		def __init__(     self     :str   ,__lowercase     :Dict   ,__lowercase     :str   ,__lowercase     :Any=None   ,__lowercase     :Union[str, Any]=1  ):
						snake_case__	:      Optional[int]     		=	tokenizer
						snake_case__	:      List[Any]     		=	dataset
						snake_case__	:      Any     		=	len(__lowercase  ) if n_tasks is None else n_tasks
						snake_case__	:      List[Any]     		=	n_copies






		def __iter__(     self     :Tuple  ):
						snake_case__	:      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()  )
						snake_case__	:      Optional[int]     		=	self.tokenizer(__lowercase   ,padding=__lowercase   ,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     :Union[str, Any]   ,__lowercase     :List[Any]   ,__lowercase     :int   ,__lowercase     :List[str]  ):
						snake_case__	:      List[str]     		=	start_length
						snake_case__	:      Any     		=	eof_strings
						snake_case__	:      Optional[Any]     		=	tokenizer






		def __call__(     self     :str   ,__lowercase     :str   ,__lowercase     :Any   ,**__lowercase     :Optional[int]  ):
						snake_case__	:      Tuple     		=	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(__lowercase  )



def 			_lowerCAmelCase       (					__lowerCAmelCase      )   ->					List[Any]:


				"""simple docstring"""

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



def 			_lowerCAmelCase       (					__lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase=20  ,  **__lowerCAmelCase      )   ->					List[Any]:


				"""simple docstring"""

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

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

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

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



def 			_lowerCAmelCase       (					)   ->					Optional[int]:


				"""simple docstring"""

				snake_case__	:      List[str]     		=	HfArgumentParser(__lowerCAmelCase      )
				snake_case__	:      Optional[Any]     		=	parser.parse_args()

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

				if args.num_workers is None:
								snake_case__	:      Union[str, Any]     		=	multiprocessing.cpu_count()

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

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

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

				snake_case__	:      List[Any]     		=	args.num_tasks if args.num_tasks is not None else len(human_eval['''test''']      )
				snake_case__	:      str     		=	args.n_samples // args.batch_size

				snake_case__	:      Tuple     		=	TokenizedDataset(__lowerCAmelCase  ,  human_eval['''test''']  ,  n_copies=__lowerCAmelCase  ,  n_tasks=__lowerCAmelCase      )
				# do not confuse args.batch_size, which is actually the num_return_sequences
				snake_case__	:      Tuple     		=	DataLoader(__lowerCAmelCase  ,  batch_size=1      )

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

				snake_case__		, snake_case__	:      Optional[Any]     		=	accelerator.prepare(__lowerCAmelCase  ,  __lowerCAmelCase      )

				snake_case__	:      Dict     		=	complete_code(
				    __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  n_tasks=__lowerCAmelCase  ,  batch_size=args.batch_size  ,  **__lowerCAmelCase  ,  )

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

								for task in tqdm(range(__lowerCAmelCase      )      ):
												snake_case__	:      str     		=	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__	:      Any     		=	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()

