





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




class 							UpperCamelCase_							(       _a		):







 """simple docstring"""

 def __init__(   self  ,				UpperCAmelCase  ,				UpperCAmelCase  ,				UpperCAmelCase=None  ,				UpperCAmelCase=1							):
       __lowerCamelCase				   =       tokenizer
       __lowerCamelCase				   =       dataset
       __lowerCamelCase				   =       len(__lowerCamelCase							) if n_tasks is None else n_tasks
       __lowerCamelCase				   =       n_copies



 def __iter__(   self							):
       __lowerCamelCase				   =       []
       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()							)
       __lowerCamelCase				   =       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 							UpperCamelCase_							(       _a		):







 """simple docstring"""

 def __init__(   self  ,				UpperCAmelCase  ,				UpperCAmelCase  ,				UpperCAmelCase							):
       __lowerCamelCase				   =       start_length
       __lowerCamelCase				   =       eof_strings
       __lowerCamelCase				   =       tokenizer



 def __call__(   self  ,				UpperCAmelCase  ,				UpperCAmelCase  ,				**UpperCAmelCase							):
       __lowerCamelCase				   =       self.tokenizer.batch_decode(input_ids[:, self.start_length :]							)
       __lowerCamelCase				   =       []
       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  UpperCamelCase__     (   _A:    List[Any]				):




      '''simple docstring'''







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



def  UpperCamelCase__     (   _A:    int							,    _A:    Optional[int]							,    _A:    Tuple							,    _A:    int							,    _A:    Optional[Any]							,    _A:    Tuple=20							,    **_A:    Optional[Any]				):




      '''simple docstring'''







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

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

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

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



def  UpperCamelCase__     (   ):




      '''simple docstring'''







      __lowerCamelCase				   =       HfArgumentParser(__lowerCAmelCase				)
      __lowerCamelCase				   =       parser.parse_args()

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

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

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

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

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

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

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

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

      __lowerCamelCase   ,						__lowerCamelCase				   =       accelerator.prepare(__lowerCAmelCase							,    __lowerCAmelCase				)

      __lowerCamelCase				   =       complete_code(
          __lowerCAmelCase							,    __lowerCAmelCase							,    __lowerCAmelCase							,    __lowerCAmelCase							,    n_tasks=__lowerCAmelCase							,    batch_size=args.batch_size							,    **__lowerCAmelCase							,    )

      if accelerator.is_main_process:
            __lowerCamelCase				   =       []

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

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












from collections.abc import Iterator, MutableMapping
from dataclasses import dataclass
from typing import Generic, TypeVar

__magic_name__  :  Any					=       TypeVar("""KEY""")
__magic_name__  :  Dict					=       TypeVar("""VAL""")





@dataclass(frozen=_a    ,					slots=_a			)
class     SCREAMING_SNAKE_CASE__		(Generic[KEY, VAL]			):
     lowercase_ :      KEY
     lowercase_ :      VAL





class     SCREAMING_SNAKE_CASE__		(_Item			):
     def __init__(						self	:		Tuple						):



       """simple docstring"""
       super().__init__(__lowerCamelCase       ,						__lowerCamelCase						)



     def __bool__(						self	:		Tuple						):



       """simple docstring"""
       return False



__magic_name__  :  str					=       _DeletedItem()





class     SCREAMING_SNAKE_CASE__		(MutableMapping[KEY, VAL]			):
     def __init__(						self	:		Any       ,						__lowerCamelCase	:		int = 8       ,						__lowerCamelCase	:		float = 0.75						):



       """simple docstring"""
       lowerCAmelCase__										=      initial_block_size
       lowerCAmelCase__										=      [None] * initial_block_size
       assert 0.0 < capacity_factor < 1.0
       lowerCAmelCase__										=      capacity_factor
       lowerCAmelCase__										=      0
     def        A__      (						self	:		Union[str, Any]       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       return hash(__lowerCamelCase						) % len(self._buckets						)
     def        A__      (						self	:		Tuple       ,						__lowerCamelCase	:		int						):



       """simple docstring"""
       return (ind + 1) % len(self._buckets						)
     def        A__      (						self	:		Dict       ,						__lowerCamelCase	:		int       ,						__lowerCamelCase	:		KEY       ,						__lowerCamelCase	:		VAL						):



       """simple docstring"""
       lowerCAmelCase__										=      self._buckets[ind]
       if not stored:
         lowerCAmelCase__										=      _Item(__lowerCamelCase       ,						__lowerCamelCase						)
         self._len += 1
         return True
       elif stored.key == key:
         lowerCAmelCase__										=      _Item(__lowerCamelCase       ,						__lowerCamelCase						)
         return True
       else:
         return False
     def        A__      (						self	:		Any						):



       """simple docstring"""
       lowerCAmelCase__										=      len(self._buckets						) * self._capacity_factor
       return len(self						) >= int(__lowerCamelCase						)
     def        A__      (						self	:		Tuple						):



       """simple docstring"""
       if len(self._buckets						) <= self._initial_block_size:
         return False
       lowerCAmelCase__										=      len(self._buckets						) * self._capacity_factor / 2
       return len(self						) < limit
     def        A__      (						self	:		List[Any]       ,						__lowerCamelCase	:		int						):



       """simple docstring"""
       lowerCAmelCase__										=      self._buckets
       lowerCAmelCase__										=      [None] * new_size
       lowerCAmelCase__										=      0
       for item in old_buckets:
         if item:
           self._add_item(item.key       ,						item.val						)
     def        A__      (						self	:		Any						):



       """simple docstring"""
       self._resize(len(self._buckets						) * 2						)
     def        A__      (						self	:		Dict						):



       """simple docstring"""
       self._resize(len(self._buckets						) // 2						)
     def        A__      (						self	:		List[str]       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       lowerCAmelCase__										=      self._get_bucket_index(__lowerCamelCase						)
       for _ in range(len(self._buckets						)						):
         yield ind
         lowerCAmelCase__										=      self._get_next_ind(__lowerCamelCase						)
     def        A__      (						self	:		Dict       ,						__lowerCamelCase	:		KEY       ,						__lowerCamelCase	:		VAL						):



       """simple docstring"""
       for ind in self._iterate_buckets(__lowerCamelCase						):
         if self._try_set(__lowerCamelCase       ,						__lowerCamelCase       ,						__lowerCamelCase						):
           break
     def __setitem__(						self	:		str       ,						__lowerCamelCase	:		KEY       ,						__lowerCamelCase	:		VAL						):



       """simple docstring"""
       if self._is_full():
         self._size_up()

       self._add_item(__lowerCamelCase       ,						__lowerCamelCase						)
     def __delitem__(						self	:		Tuple       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       for ind in self._iterate_buckets(__lowerCamelCase						):
         lowerCAmelCase__										=      self._buckets[ind]
         if item is None:
           raise KeyError(__lowerCamelCase						)
         if item is _deleted:
           continue
         if item.key == key:
           lowerCAmelCase__										=      _deleted
           self._len -= 1
           break
       if self._is_sparse():
         self._size_down()
     def __getitem__(						self	:		Optional[int]       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       for ind in self._iterate_buckets(__lowerCamelCase						):
         lowerCAmelCase__										=      self._buckets[ind]
         if item is None:
           break
         if item is _deleted:
           continue
         if item.key == key:
           return item.val
       raise KeyError(__lowerCamelCase						)
     def __len__(						self	:		int						):



       """simple docstring"""
       return self._len
     def __iter__(						self	:		Union[str, Any]						):



       """simple docstring"""
       yield from (item.key for item in self._buckets if item)



     def __repr__(						self	:		Union[str, Any]						):



       """simple docstring"""
       lowerCAmelCase__										=      ''' ,'''.join(
           F"""{item.key}: {item.val}""" for item in self._buckets if item						)
       return F"""HashMap({val_string})"""

