




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


UpperCAmelCase_							:	Any	     =		['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']





class      lowerCAmelCase (			__lowercase):


       def __init__( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None     ,	__SCREAMING_SNAKE_CASE=1	)   ->	Optional[int]:



            '''simple docstring'''




            __snake_case            =	tokenizer
            __snake_case            =	dataset
            __snake_case            =	len(_a	) if n_tasks is None else n_tasks
            __snake_case            =	n_copies




       def __iter__( self	)   ->	Optional[int]:



            '''simple docstring'''




            __snake_case            =	[]
            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            =	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      lowerCAmelCase (			__lowercase):


       def __init__( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE	)   ->	List[Any]:



            '''simple docstring'''




            __snake_case            =	start_length
            __snake_case            =	eof_strings
            __snake_case            =	tokenizer




       def __call__( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE	)   ->	Union[str, Any]:



            '''simple docstring'''




            __snake_case            =	self.tokenizer.batch_decode(input_ids[:, self.start_length :]	)
            __snake_case            =	[]
            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				:  Tuple						)-> Union[str, Any]:




     '''simple docstring'''





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







def       _UpperCamelCase			(_lowerCamelCase				:  Dict ,						_lowerCamelCase				:  List[Any] ,						_lowerCamelCase				:  int ,						_lowerCamelCase				:  List[Any] ,						_lowerCamelCase				:  List[Any] ,						_lowerCamelCase				:  Union[str, Any]=20 ,						**_lowerCamelCase				:  Union[str, Any]						)-> Union[str, Any]:




     '''simple docstring'''





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

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

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

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







def       _UpperCamelCase			()-> str:




     '''simple docstring'''





     __snake_case            =	HfArgumentParser(__a						)
     __snake_case            =	parser.parse_args()

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

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

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

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

     # Generation settings
     __snake_case            =	{
         '''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
     __snake_case            =	load_dataset('''openai_humaneval'''						)
     __snake_case            =	load_metric('''code_eval'''						)

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

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

     # Run a quick test to see if code evaluation is enabled
     try:
          __snake_case            =	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            =	accelerator.prepare(__a ,						__a						)

     __snake_case            =	complete_code(
         __a ,						__a ,						__a ,						__a ,						n_tasks=__a ,						batch_size=args.batch_size ,						**__a ,						)

     if accelerator.is_main_process:
          __snake_case            =	[]

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

          # Evaluate completions with "code_eval" metric
          __snake_case            =	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()









class 							UpperCAmelCase_						:
							"""simple docstring"""







							def __init__(     self			,				_a			,				_a			,				_a			)       ->       List[str]:
												_a      :       List[Any]   =						name
												_a      :       List[str]   =						value
												_a      :       List[str]   =						weight


							def __repr__(     self			)       ->       Optional[int]:
												return F"""{self.__class__.__name__}({self.name}, {self.value}, {self.weight})"""


							def 							__lowercase							(     self			)       ->       List[Any]:
												return self.value


							def 							__lowercase							(     self			)       ->       int:
												return self.name


							def 							__lowercase							(     self			)       ->       Optional[int]:
												return self.weight




							def 							__lowercase							(     self			)       ->       Optional[Any]:
												return self.value / self.weight



def   __UpperCAmelCase					(      __a :     Optional[int]						,__a :     Tuple						,__a :     List[str]      )     ->	List[str]:

					"""simple docstring"""

					_a      :       Optional[int]   =						[]
					for i in range(len(__a      )      ):
										menu.append(Things(name[i]						,value[i]						,weight[i]      )      )
					return menu



def   __UpperCAmelCase					(      __a :     int						,__a :     Union[str, Any]						,__a :     int      )     ->	Union[str, Any]:

					"""simple docstring"""

					_a      :       Union[str, Any]   =						sorted(__a						,key=__a						,reverse=__a      )
					_a      :       Any   =						[]
					_a ,   _a      :       Optional[int]   =						0.0, 0.0
					for i in range(len(__a      )      ):
										if (total_cost + items_copy[i].get_weight()) <= max_cost:
															result.append(items_copy[i]      )
															total_cost += items_copy[i].get_weight()
															total_value += items_copy[i].get_value()
					return (result, total_value)



def   __UpperCAmelCase					(      )     ->	int:

					"""simple docstring"""



if __name__ == "__main__":
			import doctest

			doctest.testmod()



