

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




class 					_lowerCAmelCase  (	lowercase   ):







 """simple docstring"""







 def __init__(			self							:			Dict,					UpperCAmelCase__							:			Optional[int],					UpperCAmelCase__							:			Tuple,					UpperCAmelCase__							:			Tuple=None,					UpperCAmelCase__							:			Tuple=1	):
    __lowercase								=						tokenizer
    __lowercase								=						dataset
    __lowercase								=						len(UpperCAmelCase__	) if n_tasks is None else n_tasks
    __lowercase								=						n_copies



 def __iter__(			self							:			int	):
    __lowercase								=						[]
    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								=						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 					_lowerCAmelCase  (	lowercase   ):







 """simple docstring"""







 def __init__(			self							:			int,					UpperCAmelCase__							:			Any,					UpperCAmelCase__							:			int,					UpperCAmelCase__							:			str	):
    __lowercase								=						start_length
    __lowercase								=						eof_strings
    __lowercase								=						tokenizer



 def __call__(			self							:			Union[str, Any],					UpperCAmelCase__							:			List[str],					UpperCAmelCase__							:			Optional[Any],					**UpperCAmelCase__							:			Dict	):
    __lowercase								=						self.tokenizer.batch_decode(input_ids[:, self.start_length :]	)
    __lowercase								=						[]
    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   _A  (  UpperCamelCase_     :       List[Any])	->						int:
   '''simple docstring'''


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




def   _A  (  UpperCamelCase_     :       Dict,			UpperCamelCase_     :       Dict,			UpperCamelCase_     :       Tuple,			UpperCamelCase_     :       Optional[Any],			UpperCamelCase_     :       Union[str, Any],			UpperCamelCase_     :       List[Any]=20,			**UpperCamelCase_     :       int)	->						int:
   '''simple docstring'''


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

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

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

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




def   _A  (  )	->						Dict:
   '''simple docstring'''


   __lowercase								=						HfArgumentParser(UpperCamelCase_)
   __lowercase								=						parser.parse_args()

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

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

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

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

   # Generation settings
   __lowercase								=						{
       "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,			UpperCamelCase_,			UpperCamelCase_)]),
   }

   # Load evaluation dataset and metric
   __lowercase								=						load_dataset("openai_humaneval")
   __lowercase								=						load_metric("code_eval")

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

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

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

   __lowercase								=						complete_code(
       UpperCamelCase_,			UpperCamelCase_,			UpperCamelCase_,			UpperCamelCase_,			n_tasks=UpperCamelCase_,			batch_size=args.batch_size,			**UpperCamelCase_,			)

   if accelerator.is_main_process:
      __lowercase								=						[]

      for task in tqdm(range(UpperCamelCase_)):
         __lowercase								=						human_eval["test"][task]["test"]
         __lowercase								=						F"""check({human_eval["test"][task]["entry_point"]})"""
         references.append("\n" + test_func + "\n" + entry_point)

      # Evaluate completions with "code_eval" metric
      __lowercase		,__lowercase								=						code_eval_metric.compute(
          references=UpperCamelCase_,			predictions=UpperCamelCase_,			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(UpperCamelCase_,			UpperCamelCase_)


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






import unicodedata
from dataclasses import dataclass
from typing import Optional, Union

import numpy as np

from transformers.data.data_collator import DataCollatorMixin
from transformers.file_utils import PaddingStrategy
from transformers.tokenization_utils_base import PreTrainedTokenizerBase


def 		_A     (    A__     ,				A__     ,				A__     ,				A__      ):






 """simple docstring"""



 if isinstance(A__     ,				A__      ):
  __lowercase							  = np.full((len(A__      ), sequence_length, 2)     ,				A__      )
 else:
  __lowercase							  = np.full((len(A__      ), sequence_length)     ,				A__      )

 for i, tensor in enumerate(A__      ):
  if padding_side == "right":
   if isinstance(A__     ,				A__      ):
    __lowercase							  = tensor[:sequence_length]
   else:
    __lowercase							  = tensor[:sequence_length]
  else:
   if isinstance(A__     ,				A__      ):
    __lowercase							  = tensor[:sequence_length]
   else:
    __lowercase							  = tensor[:sequence_length]

 return out_tensor.tolist()


def 		_A     (    A__      ):






 """simple docstring"""



 __lowercase							  = ord(A__      )
 if (cp >= 33 and cp <= 47) or (cp >= 58 and cp <= 64) or (cp >= 91 and cp <= 96) or (cp >= 123 and cp <= 126):
  return True
 __lowercase							  = unicodedata.category(A__      )
 if cat.startswith('''P'''      ):
  return True
 return False






@dataclass
class 			lowercase_	(lowerCamelCase__      ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE    :  PreTrainedTokenizerBase
  SCREAMING_SNAKE_CASE    :  Union[bool, str, PaddingStrategy]							=			True
  SCREAMING_SNAKE_CASE    :  Optional[int]							=			None
  SCREAMING_SNAKE_CASE    :  Optional[int]							=			None
  SCREAMING_SNAKE_CASE    :  int							=			-1_0_0
  SCREAMING_SNAKE_CASE    :  str							=			"pt"






  def 					SCREAMING_SNAKE_CASE      (      self   :     List[str]					,lowercase__   :     List[str]			):
   import torch

   __lowercase							  = '''label''' if '''label''' in features[0].keys() else '''labels'''
   __lowercase							  = [feature[label_name] for feature in features] if label_name in features[0].keys() else None
   __lowercase							  = self.tokenizer.pad(
       lowercase__					,padding=self.padding					,max_length=self.max_length					,pad_to_multiple_of=self.pad_to_multiple_of					,return_tensors='''pt''' if labels is None else None					,)

   if labels is None:
    return batch

   __lowercase							  = torch.tensor(batch['''entity_ids''']			).shape[1]
   __lowercase							  = self.tokenizer.padding_side
   if padding_side == "right":
    __lowercase							  = [
        list(lowercase__			) + [self.label_pad_token_id] * (sequence_length - len(lowercase__			)) for label in labels
    ]
   else:
    __lowercase							  = [
        [self.label_pad_token_id] * (sequence_length - len(lowercase__			)) + list(lowercase__			) for label in labels
    ]

   __lowercase							  = [feature['''ner_tags'''] for feature in features]
   __lowercase							  = padding_tensor(lowercase__					,-1					,lowercase__					,lowercase__			)
   __lowercase							  = [feature['''original_entity_spans'''] for feature in features]
   __lowercase							  = padding_tensor(lowercase__					,(-1, -1)					,lowercase__					,lowercase__			)
   __lowercase							  = {k: torch.tensor(lowercase__					,dtype=torch.intaa			) for k, v in batch.items()}

   return batch

