






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__       		=  ["""\nclass""", """\ndef""", """\n#""", """\n@""", """\nprint""", """\nif"""]




class       snake_case__ (				lowercase_):

       '''simple docstring'''



       def __init__(				self      ,    a__      ,    a__      ,    a__=None      ,    a__=1     )			->   Tuple:






            '''simple docstring'''





            __snake_case      :Dict 							=    tokenizer
            __snake_case      :Any 							=    dataset
            __snake_case      :Dict 							=    len(a__     ) if n_tasks is None else n_tasks
            __snake_case      :Union[str, Any] 							=    n_copies



       def __iter__(				self     )			->   Dict:






            '''simple docstring'''





            __snake_case      :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()     )
            __snake_case      :Union[str, Any] 							=    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       snake_case__ (				lowercase_):

       '''simple docstring'''



       def __init__(				self      ,    a__      ,    a__      ,    a__     )			->   Union[str, Any]:






            '''simple docstring'''





            __snake_case      :List[Any] 							=    start_length
            __snake_case      :List[Any] 							=    eof_strings
            __snake_case      :Optional[Any] 							=    tokenizer



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






            '''simple docstring'''





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




     '''simple docstring'''






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






def 	UpperCamelCase	(							snake_case__  : str						,snake_case__  : str						,snake_case__  : List[Any]						,snake_case__  : List[str]						,snake_case__  : Union[str, Any]						,snake_case__  : Optional[Any]=20						,**snake_case__  : Tuple	):




     '''simple docstring'''






     __snake_case      :Any 							=    defaultdict(snake_case__	)  # dict of list of generated tokens
     for step, batch in tqdm(enumerate(snake_case__	)	):
          with torch.no_grad():
               __snake_case      :Optional[Any] 							=    batch["""ids"""].shape[-1]
               __snake_case      :Union[str, Any] 							=    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
               __snake_case      :Union[str, Any] 							=    batch["""task_id"""].repeat(snake_case__	)
               __snake_case      :Dict 							=    accelerator.pad_across_processes(
                   snake_case__						,dim=1						,pad_index=tokenizer.pad_token_id	)

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

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

     __snake_case      :Any 							=    [[] for _ in range(snake_case__	)]
     for task, generated_tokens in gen_token_dict.items():
          for s in generated_tokens:
               __snake_case      :Optional[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 	UpperCamelCase	(							):




     '''simple docstring'''






     __snake_case      :Union[str, Any] 							=    HfArgumentParser(snake_case__	)
     __snake_case      :int 							=    parser.parse_args()

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

     if args.num_workers is None:
          __snake_case      :List[Any] 							=    multiprocessing.cpu_count()

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

     # Load model and tokenizer
     __snake_case      :Dict 							=    AutoTokenizer.from_pretrained(args.model_ckpt	)
     __snake_case      :int 							=    tokenizer.eos_token
     __snake_case      :Optional[int] 							=    AutoModelForCausalLM.from_pretrained(args.model_ckpt	)

     # Generation settings
     __snake_case      :Dict 							=    {
         """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
     __snake_case      :Optional[int] 							=    load_dataset("""openai_humaneval"""	)
     __snake_case      :List[Any] 							=    load_metric("""code_eval"""	)

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

     __snake_case      :Any 							=    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
     __snake_case      :Optional[Any] 							=    DataLoader(snake_case__						,batch_size=1	)

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

     __snake_case      :str 							=    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:
          __snake_case      :Union[str, Any] 							=    []

          for task in tqdm(range(snake_case__	)	):
               __snake_case      :int 							=    human_eval["""test"""][task]["""test"""]
               __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
          __snake_case							,			__snake_case      :Tuple 							=    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()










def 	UpperCamelCase	(							snake_case__  : float						,snake_case__  : int	):




     '''simple docstring'''






     if digit_amount > 0:
          return round(number - int(snake_case__	)						,snake_case__	)
     return number - int(snake_case__	)


if __name__ == "__main__":
     print(decimal_isolate(1.5_3, 0))
     print(decimal_isolate(3_5.3_4_5, 1))
     print(decimal_isolate(3_5.3_4_5, 2))
     print(decimal_isolate(3_5.3_4_5, 3))
     print(decimal_isolate(-1_4.7_8_9, 3))
     print(decimal_isolate(0, 2))
     print(decimal_isolate(-1_4.1_2_3, 1))
     print(decimal_isolate(-1_4.1_2_3, 2))
     print(decimal_isolate(-1_4.1_2_3, 3))


