




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


__lowerCAmelCase	:       List[str]     =	logging.get_logger(__name__)




class 					snake_case__							(_UpperCamelCase     ):
				"""simple docstring"""



				def __init__(   self       :  Tuple      ,   __lowerCamelCase       :  int      ,   __lowerCamelCase       :  int      ,   __lowerCamelCase       :  float      ,   **__lowerCamelCase       :  Any							)      ->  int:
							a					=    feature_size
							a					=    sampling_rate
							a					=    padding_value

							a					=    kwargs.pop("padding_side"      ,   "right"							)
							a					=    kwargs.pop("return_attention_mask"      ,   __lowerCamelCase							)

							super().__init__(**__lowerCamelCase							)





				def 							__UpperCAmelCase							(   self       :  Optional[Any]      ,   __lowerCamelCase       :  Union[
				        BatchFeature,
				        List[BatchFeature],
				        Dict[str, BatchFeature],
				        Dict[str, List[BatchFeature]],
				        List[Dict[str, BatchFeature]],
				    ]      ,   __lowerCamelCase       :  Union[bool, str, PaddingStrategy] = True      ,   __lowerCamelCase       :  Optional[int] = None      ,   __lowerCamelCase       :  bool = False      ,   __lowerCamelCase       :  Optional[int] = None      ,   __lowerCamelCase       :  Optional[bool] = None      ,   __lowerCamelCase       :  Optional[Union[str, TensorType]] = None      ,   )      ->  BatchFeature:
							# 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(__lowerCamelCase      ,   (list, tuple)							) and isinstance(processed_features[0]      ,   (dict, BatchFeature)							):
										a					=    {
										    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()							)}"""							)

							a					=    processed_features[self.model_input_names[0]]
							a					=    (
							    return_attention_mask if return_attention_mask is not None else self.return_attention_mask
							)

							if len(__lowerCamelCase							) == 0:
										if return_attention_mask:
													a					=    []
										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

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

							if return_tensors is None:
										if is_tf_tensor(__lowerCamelCase							):
													a					=    "tf"
										elif is_torch_tensor(__lowerCamelCase							):
													a					=    "pt"
										elif isinstance(__lowerCamelCase      ,   (int, float, list, tuple, np.ndarray)							):
													a					=    "np"
										else:
													raise ValueError(
													    f"""type of {first_element} unknown: {type(__lowerCamelCase							)}. """
													    "Should be one of a python, numpy, pytorch or tensorflow object."							)

							for key, value in processed_features.items():
										if isinstance(value[0]      ,   (int, float)							):
													a					=    to_numpy(__lowerCamelCase							)
										else:
													a					=    [to_numpy(__lowerCamelCase							) for v in value]

        # Convert padding_strategy in PaddingStrategy
							a					=    self._get_padding_strategies(padding=__lowerCamelCase      ,   max_length=__lowerCamelCase							)

							a					=    processed_features[self.model_input_names[0]]

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

							a					=    []
							for i in range(__lowerCamelCase							):
										a					=    {k: v[i] for k, v in processed_features.items()}
										# truncation
										a					=    self._truncate(
										    __lowerCamelCase      ,   max_length=__lowerCamelCase      ,   pad_to_multiple_of=__lowerCamelCase      ,   truncation=__lowerCamelCase      ,   )
										truncated_inputs.append(__lowerCamelCase							)

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

							a					=    {}
							for i in range(__lowerCamelCase							):
										# padding
										a					=    self._pad(
										    truncated_inputs[i]      ,   max_length=__lowerCamelCase      ,   padding_strategy=__lowerCamelCase      ,   pad_to_multiple_of=__lowerCamelCase      ,   return_attention_mask=__lowerCamelCase      ,   )

										for key, value in outputs.items():
													if key not in batch_outputs:
																a					=    []
													if value.dtype is np.dtype(np.floataa							):
																a					=    value.astype(np.floataa							)
													batch_outputs[key].append(__lowerCamelCase							)

							return BatchFeature(__lowerCamelCase      ,   tensor_type=__lowerCamelCase							)





				def 							__UpperCAmelCase							(   self       :  List[Any]      ,   __lowerCamelCase       :  Union[Dict[str, np.ndarray], BatchFeature]      ,   __lowerCamelCase       :  Optional[int] = None      ,   __lowerCamelCase       :  PaddingStrategy = PaddingStrategy.DO_NOT_PAD      ,   __lowerCamelCase       :  Optional[int] = None      ,   __lowerCamelCase       :  Optional[bool] = None      ,   )      ->  dict:
							a					=    processed_features[self.model_input_names[0]]

							if padding_strategy == PaddingStrategy.LONGEST:
										a					=    len(__lowerCamelCase							)

							if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
										a					=    ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

							a					=    padding_strategy != PaddingStrategy.DO_NOT_PAD and len(__lowerCamelCase							) < max_length

							if return_attention_mask and "attention_mask" not in processed_features:
										a					=    np.ones(len(__lowerCamelCase							)      ,   dtype=np.intaa							)

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

							return processed_features





				def 							__UpperCAmelCase							(   self       :  Dict      ,   __lowerCamelCase       :  Union[Dict[str, np.ndarray], BatchFeature]      ,   __lowerCamelCase       :  Optional[int] = None      ,   __lowerCamelCase       :  Optional[int] = None      ,   __lowerCamelCase       :  Optional[bool] = None      ,   )      ->  List[str]:
							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."							)

							a					=    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):
										a					=    ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

							a					=    len(__lowerCamelCase							) > max_length

							if needs_to_be_truncated:
										a					=    processed_features[self.model_input_names[0]][:max_length]
										if "attention_mask" in processed_features:
													a					=    processed_features["attention_mask"][:max_length]

							return processed_features





				def 							__UpperCAmelCase							(   self       :  Union[str, Any]      ,   __lowerCamelCase       :  List[str]=False      ,   __lowerCamelCase       :  int=None							)      ->  int:

							# Get padding strategy
							if padding is not False:
										if padding is True:
													a					=    PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
										elif not isinstance(__lowerCamelCase      ,   __lowerCamelCase							):
													a					=    PaddingStrategy(__lowerCamelCase							)
										elif isinstance(__lowerCamelCase      ,   __lowerCamelCase							):
													a					=    padding
							else:
										a					=    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


__lowerCAmelCase	:       Union[str, Any]     =	['\nclass', '\ndef', '\n#', '\n@', '\nprint', '\nif']




class 					snake_case__							(_UpperCamelCase     ):
				"""simple docstring"""



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





				def __iter__(   self       :  Tuple							)      ->  str:
							a					=    []
							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()							)
							a					=    self.tokenizer(__lowerCamelCase      ,   padding=__lowerCamelCase      ,   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 					snake_case__							(_UpperCamelCase     ):
				"""simple docstring"""



				def __init__(   self       :  int      ,   __lowerCamelCase       :  Dict      ,   __lowerCamelCase       :  Dict      ,   __lowerCamelCase       :  Tuple							)      ->  Optional[Any]:
							a					=    start_length
							a					=    eof_strings
							a					=    tokenizer





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




def     __magic_name__       (				A	:      List[Any]     ):






			'''simple docstring'''







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




def     __magic_name__       (				A	:      Union[str, Any],						A	:      Optional[Any],						A	:      List[Any],						A	:      Optional[Any],						A	:      List[str],						A	:      List[Any]=20,						**A	:      Union[str, Any]     ):






			'''simple docstring'''







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

									a					,	a					=    accelerator.gather((generated_tokens, generated_tasks)     )
									a					=    generated_tokens.cpu().numpy()
									a					=    generated_tasks.cpu().numpy()

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

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




def     __magic_name__       (				):






			'''simple docstring'''







			a					=    HfArgumentParser(A     )
			a					=    parser.parse_args()

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

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

			# Use dataset load to feed to accelerate
			a					=    Accelerator()
			set_seed(args.seed,						device_specific=A     )

			# Load model and tokenizer
			a					=    AutoTokenizer.from_pretrained(args.model_ckpt     )
			a					=    tokenizer.eos_token
			a					=    AutoModelForCausalLM.from_pretrained(args.model_ckpt     )

			# Generation settings
			a					=    {
			    "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,						A,						A     )]     ),
			}

			# Load evaluation dataset and metric
			a					=    load_dataset("openai_humaneval"     )
			a					=    load_metric("code_eval"     )

			a					=    args.num_tasks if args.num_tasks is not None else len(human_eval["test"]     )
			a					=    args.n_samples // args.batch_size

			a					=    TokenizedDataset(A,						human_eval["test"],						n_copies=A,						n_tasks=A     )
			# do not confuse args.batch_size, which is actually the num_return_sequences
			a					=    DataLoader(A,						batch_size=1     )

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

			a					,	a					=    accelerator.prepare(A,						A     )

			a					=    complete_code(
			    A,						A,						A,						A,						n_tasks=A,						batch_size=args.batch_size,						**A,						)

			if accelerator.is_main_process:
						a					=    []

						for task in tqdm(range(A     )     ):
									a					=    human_eval["test"][task]["test"]
									a					=    F"""check({human_eval["test"][task]["entry_point"]})"""
									references.append("\n" + test_func + "\n" + entry_point     )

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


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






