


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


snake_case__						:   List[str]			 = ['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']



class 						snake_case_(				a__	):






     def __init__(		self     :    Dict    , UpperCamelCase_     :    int    , UpperCamelCase_     :    int    , UpperCamelCase_     :    Tuple=None    , UpperCamelCase_     :    List[Any]=1 ):
           lowerCAmelCase    :		Dict				=      tokenizer
           lowerCAmelCase    :		str				=      dataset
           lowerCAmelCase    :		Dict				=      len(UpperCamelCase_ ) if n_tasks is None else n_tasks
           lowerCAmelCase    :		int				=      n_copies






     def __iter__(		self     :    Any ):
           lowerCAmelCase    :		List[str]				=      []
           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    :		List[str]				=      self.tokenizer(UpperCamelCase_    , padding=UpperCamelCase_    , 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_(				a__	):






     def __init__(		self     :    str    , UpperCamelCase_     :    Any    , UpperCamelCase_     :    Tuple    , UpperCamelCase_     :    Union[str, Any] ):
           lowerCAmelCase    :		Optional[int]				=      start_length
           lowerCAmelCase    :		str				=      eof_strings
           lowerCAmelCase    :		Tuple				=      tokenizer






     def __call__(		self     :    Tuple    , UpperCamelCase_     :    str    , UpperCamelCase_     :    List[str]    , **UpperCamelCase_     :    Optional[Any] ):
           lowerCAmelCase    :		str				=      self.tokenizer.batch_decode(input_ids[:, self.start_length :] )
           lowerCAmelCase    :		Any				=      []
           for decoded_generation in decoded_generations:
                 done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings ) )
           return all(UpperCamelCase_ )



def    _snake_case  (  _snake_case :	str	):
      lowerCAmelCase    :		Tuple				=      re.split('''(%s)''' % '''|'''.join(_snake_case	)       ,      _snake_case	)
      # last string should be ""
      return "".join(string_list[:-2]	)


def    _snake_case  (  _snake_case :	int       ,      _snake_case :	int       ,      _snake_case :	List[str]       ,      _snake_case :	str       ,      _snake_case :	List[Any]       ,      _snake_case :	Any=20       ,      **_snake_case :	Optional[int]	):
      lowerCAmelCase    :		int				=      defaultdict(_snake_case	)  # dict of list of generated tokens
      for step, batch in tqdm(enumerate(_snake_case	)	):
            with torch.no_grad():
                  lowerCAmelCase    :		Dict				=      batch['''ids'''].shape[-1]
                  lowerCAmelCase    :		Tuple				=      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
                  lowerCAmelCase    :		Optional[Any]				=      batch['''task_id'''].repeat(_snake_case	)
                  lowerCAmelCase    :		Dict				=      accelerator.pad_across_processes(
                      _snake_case       ,      dim=1       ,      pad_index=tokenizer.pad_token_id	)

                  lowerCAmelCase,					lowerCAmelCase    :		Dict				=      accelerator.gather((generated_tokens, generated_tasks)	)
                  lowerCAmelCase    :		Any				=      generated_tokens.cpu().numpy()
                  lowerCAmelCase    :		List[Any]				=      generated_tasks.cpu().numpy()

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

      lowerCAmelCase    :		Any				=      [[] for _ in range(_snake_case	)]
      for task, generated_tokens in gen_token_dict.items():
            for s in generated_tokens:
                  lowerCAmelCase    :		Dict				=      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    _snake_case  (  ):
      # Setup configuration
      lowerCAmelCase    :		List[Any]				=      HfArgumentParser(_snake_case	)
      lowerCAmelCase    :		Tuple				=      parser.parse_args()

      transformers.logging.set_verbosity_error()
      # enables code execution in code_eval metric
      lowerCAmelCase    :		Union[str, Any]				=      args.HF_ALLOW_CODE_EVAL
      # make sure tokenizer plays nice with multiprocessing
      lowerCAmelCase    :		List[Any]				=      '''false'''

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

      # Use dataset load to feed to accelerate
      lowerCAmelCase    :		int				=      Accelerator()
      set_seed(args.seed       ,      device_specific=_snake_case	)

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

      # Generation settings
      lowerCAmelCase    :		Optional[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
      lowerCAmelCase    :		int				=      load_dataset('''openai_humaneval'''	)
      lowerCAmelCase    :		List[str]				=      load_metric('''code_eval'''	)

      lowerCAmelCase    :		Dict				=      args.num_tasks if args.num_tasks is not None else len(human_eval['''test''']	)
      lowerCAmelCase    :		List[Any]				=      args.n_samples // args.batch_size

      lowerCAmelCase    :		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
      lowerCAmelCase    :		str				=      DataLoader(_snake_case       ,      batch_size=1	)

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

      lowerCAmelCase,					lowerCAmelCase    :		Tuple				=      accelerator.prepare(_snake_case       ,      _snake_case	)

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

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

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








import random

def      A      (      _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    )	-> tuple:
       lowerCamelCase ,      lowerCamelCase ,      lowerCamelCase   :      Any			 =					[], [], []
       for element in data:
              if element < pivot:
                     less.append(_SCREAMING_SNAKE_CASE    )
              elif element > pivot:
                     greater.append(_SCREAMING_SNAKE_CASE    )
              else:
                     equal.append(_SCREAMING_SNAKE_CASE    )
       return less, equal, greater


def      A      (      _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    )	-> str:
       # index = len(items) // 2 when trying to find the median
       #   (value of index when items is sorted)

       # invalid input
       if index >= len(_SCREAMING_SNAKE_CASE    ) or index < 0:
              return None

       lowerCamelCase   :      List[Any]			 =					items[random.randint(0 ,len(_SCREAMING_SNAKE_CASE    ) - 1    )]
       lowerCamelCase   :      Dict			 =					0
       lowerCamelCase ,      lowerCamelCase ,      lowerCamelCase   :      Tuple			 =					_partition(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    )
       lowerCamelCase   :      Union[str, Any]			 =					len(_SCREAMING_SNAKE_CASE    )
       lowerCamelCase   :      str			 =					len(_SCREAMING_SNAKE_CASE    )

       # index is the pivot
       if m <= index < m + count:
              return pivot
       # must be in smaller
       elif m > index:
              return quick_select(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    )
       # must be in larger
       else:
              return quick_select(_SCREAMING_SNAKE_CASE ,index - (m + count)    )






