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







class   _lowercase						(       lowerCAmelCase_				):


     '''simple docstring'''





     def __init__(      self ,			__UpperCamelCase ,			__UpperCamelCase ,			__UpperCamelCase=None ,			__UpperCamelCase=1  )->					Union[str, Any]:
          UpperCAmelCase__							: Optional[Any] 							=						tokenizer
          UpperCAmelCase__							: Any 							=						dataset
          UpperCAmelCase__							: int 							=						len(__UpperCamelCase  ) if n_tasks is None else n_tasks
          UpperCAmelCase__							: Dict 							=						n_copies



     def __iter__(      self  )->					int:
          UpperCAmelCase__							: int 							=						[]
          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()  )
          UpperCAmelCase__							: Union[str, Any] 							=						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   _lowercase						(       lowerCAmelCase_				):


     '''simple docstring'''





     def __init__(      self ,			__UpperCamelCase ,			__UpperCamelCase ,			__UpperCamelCase  )->					Optional[Any]:
          UpperCAmelCase__							: Optional[int] 							=						start_length
          UpperCAmelCase__							: Union[str, Any] 							=						eof_strings
          UpperCAmelCase__							: int 							=						tokenizer



     def __call__(      self ,			__UpperCamelCase ,			__UpperCamelCase ,			**__UpperCamelCase  )->					str:
          UpperCAmelCase__							: Optional[Any] 							=						self.tokenizer.batch_decode(input_ids[:, self.start_length :]  )
          UpperCAmelCase__							: Optional[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 							a__ (				lowerCAmelCase	:       Dict		):






     '''simple docstring'''
     UpperCAmelCase__							: int 							=						re.split("(%s)" % "|".join(lowerCAmelCase		)       ,    lowerCAmelCase		)
     # last string should be ""
     return "".join(string_list[:-2]		)



def 							a__ (				lowerCAmelCase	:       Tuple       ,    lowerCAmelCase	:       Any       ,    lowerCAmelCase	:       str       ,    lowerCAmelCase	:       str       ,    lowerCAmelCase	:       Dict       ,    lowerCAmelCase	:       List[str]=20       ,    **lowerCAmelCase	:       Optional[Any]		):






     '''simple docstring'''
     UpperCAmelCase__							: List[str] 							=						defaultdict(lowerCAmelCase		)  # dict of list of generated tokens
     for step, batch in tqdm(enumerate(lowerCAmelCase		)		):
          with torch.no_grad():
               UpperCAmelCase__							: List[str] 							=						batch["ids"].shape[-1]
               UpperCAmelCase__							: Dict 							=						accelerator.unwrap_model(lowerCAmelCase		).generate(
                   input_ids=batch["ids"][:, : batch["input_len"]]       ,    num_return_sequences=lowerCAmelCase       ,    **lowerCAmelCase		)
               # each task is generated batch_size times
               UpperCAmelCase__							: Any 							=						batch["task_id"].repeat(lowerCAmelCase		)
               UpperCAmelCase__							: Optional[int] 							=						accelerator.pad_across_processes(
                   lowerCAmelCase       ,    dim=1       ,    pad_index=tokenizer.pad_token_id		)

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

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

     UpperCAmelCase__							: Tuple 							=						[[] for _ in range(lowerCAmelCase		)]
     for task, generated_tokens in gen_token_dict.items():
          for s in generated_tokens:
               UpperCAmelCase__							: Optional[int] 							=						tokenizer.decode(lowerCAmelCase       ,    skip_special_tokens=lowerCAmelCase       ,    clean_up_tokenization_spaces=lowerCAmelCase		)
               code_gens[task].append(remove_last_block(lowerCAmelCase		)		)
     return code_gens



def 							a__ (				):






     '''simple docstring'''
     # Setup configuration
     UpperCAmelCase__							: List[str] 							=						HfArgumentParser(lowerCAmelCase		)
     UpperCAmelCase__							: Dict 							=						parser.parse_args()

     transformers.logging.set_verbosity_error()
     # enables code execution in code_eval metric
     UpperCAmelCase__							: Dict 							=						args.HF_ALLOW_CODE_EVAL
     # make sure tokenizer plays nice with multiprocessing
     UpperCAmelCase__							: Optional[int] 							=						"false"

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

     # Use dataset load to feed to accelerate
     UpperCAmelCase__							: List[Any] 							=						Accelerator()
     set_seed(args.seed       ,    device_specific=lowerCAmelCase		)

     # Load model and tokenizer
     UpperCAmelCase__							: Tuple 							=						AutoTokenizer.from_pretrained(args.model_ckpt		)
     UpperCAmelCase__							: int 							=						tokenizer.eos_token
     UpperCAmelCase__							: Tuple 							=						AutoModelForCausalLM.from_pretrained(args.model_ckpt		)

     # Generation settings
     UpperCAmelCase__							: 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       ,    lowerCAmelCase       ,    lowerCAmelCase		)]		),
     }

     # Load evaluation dataset and metric
     UpperCAmelCase__							: Optional[Any] 							=						load_dataset("openai_humaneval"		)
     UpperCAmelCase__							: int 							=						load_metric("code_eval"		)

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

     UpperCAmelCase__							: Union[str, Any] 							=						TokenizedDataset(lowerCAmelCase       ,    human_eval["test"]       ,    n_copies=lowerCAmelCase       ,    n_tasks=lowerCAmelCase		)
     # do not confuse args.batch_size, which is actually the num_return_sequences
     UpperCAmelCase__							: List[Any] 							=						DataLoader(lowerCAmelCase       ,    batch_size=1		)

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

     UpperCAmelCase__							,	UpperCAmelCase__							: Optional[Any] 							=						accelerator.prepare(lowerCAmelCase       ,    lowerCAmelCase		)

     UpperCAmelCase__							: Union[str, Any] 							=						complete_code(
         lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase       ,    n_tasks=lowerCAmelCase       ,    batch_size=args.batch_size       ,    **lowerCAmelCase       ,    )

     if accelerator.is_main_process:
          UpperCAmelCase__							: Optional[int] 							=						[]

          for task in tqdm(range(lowerCAmelCase		)		):
               UpperCAmelCase__							: List[Any] 							=						human_eval["test"][task]["test"]
               UpperCAmelCase__							: Optional[int] 							=						F"check({human_eval['test'][task]['entry_point']})"
               references.append("\n" + test_func + "\n" + entry_point		)

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





"""simple docstring"""




from timeit import timeit



def 							a__ (				lowerCAmelCase	:       int		):






     '''simple docstring'''
     if number < 0:
          raise ValueError("the value of input must not be negative"		)
     UpperCAmelCase__							: Tuple 							=						0
     while number:
          number &= number - 1
          result += 1
     return result



def 							a__ (				lowerCAmelCase	:       int		):






     '''simple docstring'''
     if number < 0:
          raise ValueError("the value of input must not be negative"		)
     UpperCAmelCase__							: Union[str, Any] 							=						0
     while number:
          if number % 2 == 1:
               result += 1
          number >>= 1
     return result



def 							a__ (				):






     '''simple docstring'''

     def do_benchmark(lowerCAmelCase	:       int		) -> None:
          UpperCAmelCase__							: Dict 							=						"import __main__ as z"
          print(F"Benchmark when {number = }:"		)
          print(F"{get_set_bits_count_using_modulo_operator(lowerCAmelCase		) = }"		)
          UpperCAmelCase__							: Tuple 							=						timeit("z.get_set_bits_count_using_modulo_operator(25)"       ,    setup=lowerCAmelCase		)
          print(F"timeit() runs in {timing} seconds"		)
          print(F"{get_set_bits_count_using_brian_kernighans_algorithm(lowerCAmelCase		) = }"		)
          UpperCAmelCase__							: Any 							=						timeit(
              "z.get_set_bits_count_using_brian_kernighans_algorithm(25)"       ,    setup=lowerCAmelCase       ,    )
          print(F"timeit() runs in {timing} seconds"		)

     for number in (25, 37, 58, 0):
          do_benchmark(lowerCAmelCase		)
          print()


if __name__ == "__main__":
    import doctest

    doctest.testmod()
    benchmark()



