






"""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


__A		:       Optional[int]  				=		['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']





class 					_UpperCAmelCase       ( _A							):




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






					def __iter__(				self     :						Optional[int]   )  ->  Union[str, Any]:
							lowercase_ : 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()   )
							lowercase_ : str							       =   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 					_UpperCAmelCase       ( _A							):




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






					def __call__(				self     :						Optional[Any]    ,						A     :						Tuple    ,						A     :						int    ,						**A     :						Optional[Any]   )  ->  Optional[int]:
							lowercase_ : Optional[Any]							       =   self.tokenizer.batch_decode(input_ids[:, self.start_length :]   )
							lowercase_ : Dict							       =   []
							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 						lowercase    (				__snake_case				: Optional[int]	):
		lowercase_ : Any							       =   re.split('''(%s)''' % '''|'''.join(__snake_case	)		,					__snake_case	)
		# last string should be ""
		return "".join(string_list[:-2]	)



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

						lowercase_       ,					lowercase_ : str							       =   accelerator.gather((generated_tokens, generated_tasks)	)
						lowercase_ : Tuple							       =   generated_tokens.cpu().numpy()
						lowercase_ : Any							       =   generated_tasks.cpu().numpy()

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

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



def 						lowercase    (				):
		# Setup configuration
		lowercase_ : List[str]							       =   HfArgumentParser(__snake_case	)
		lowercase_ : Tuple							       =   parser.parse_args()

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

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

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

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

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

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

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

		lowercase_ : Dict							       =   TokenizedDataset(__snake_case		,					human_eval['''test''']		,					n_copies=__snake_case		,					n_tasks=__snake_case	)
		# do not confuse args.batch_size, which is actually the num_return_sequences
		lowercase_ : Optional[int]							       =   DataLoader(__snake_case		,					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_ : Dict							       =   accelerator.prepare(__snake_case		,					__snake_case	)

		lowercase_ : Optional[int]							       =   complete_code(
		    __snake_case		,					__snake_case		,					__snake_case		,					__snake_case		,					n_tasks=__snake_case		,					batch_size=args.batch_size		,					**__snake_case		,					)

		if accelerator.is_main_process:
				lowercase_ : Union[str, Any]							       =   []

				for task in tqdm(range(__snake_case	)	):
						lowercase_ : str							       =   human_eval['''test'''][task]['''test''']
						lowercase_ : List[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_ : Optional[int]							       =   code_eval_metric.compute(
				    references=__snake_case		,					predictions=__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(__snake_case		,					__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"""



from ...configuration_utils import PretrainedConfig
from ...utils import logging


__A		:       Dict  				=		logging.get_logger(__name__)

__A		:       Union[str, Any]  				=		{
    '''facebook/vit-mae-base''': '''https://huggingface.co/facebook/vit-mae-base/resolve/main/config.json''',
    # See all ViT MAE models at https://huggingface.co/models?filter=vit-mae
}





class 					_UpperCAmelCase       ( _A							):
					SCREAMING_SNAKE_CASE_		:				Union[str, Any]							    =   "vit_mae"




					def __init__(				self     :						Dict    ,						A     :						List[str]=7_68    ,						A     :						Any=12    ,						A     :						Union[str, Any]=12    ,						A     :						Tuple=30_72    ,						A     :						Any="gelu"    ,						A     :						Tuple=0.0    ,						A     :						List[str]=0.0    ,						A     :						Tuple=0.02    ,						A     :						Tuple=1e-12    ,						A     :						int=2_24    ,						A     :						Dict=16    ,						A     :						int=3    ,						A     :						Tuple=True    ,						A     :						Tuple=16    ,						A     :						Optional[Any]=5_12    ,						A     :						Union[str, Any]=8    ,						A     :						List[Any]=20_48    ,						A     :						Dict=0.75    ,						A     :						Any=False    ,						**A     :						Optional[int]    ,						)  ->  Union[str, Any]:
							super().__init__(**A   )

							lowercase_ : List[Any]							       =   hidden_size
							lowercase_ : str							       =   num_hidden_layers
							lowercase_ : List[Any]							       =   num_attention_heads
							lowercase_ : Any							       =   intermediate_size
							lowercase_ : Optional[int]							       =   hidden_act
							lowercase_ : List[Any]							       =   hidden_dropout_prob
							lowercase_ : int							       =   attention_probs_dropout_prob
							lowercase_ : int							       =   initializer_range
							lowercase_ : Dict							       =   layer_norm_eps
							lowercase_ : Optional[Any]							       =   image_size
							lowercase_ : str							       =   patch_size
							lowercase_ : Dict							       =   num_channels
							lowercase_ : Any							       =   qkv_bias
							lowercase_ : Union[str, Any]							       =   decoder_num_attention_heads
							lowercase_ : Optional[Any]							       =   decoder_hidden_size
							lowercase_ : List[str]							       =   decoder_num_hidden_layers
							lowercase_ : List[Any]							       =   decoder_intermediate_size
							lowercase_ : Optional[Any]							       =   mask_ratio
							lowercase_ : Optional[Any]							       =   norm_pix_loss



