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






class   a_    (       a__				):



       """simple docstring"""






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






       def __iter__(						self					) ->Optional[Any]:
           SCREAMING_SNAKE_CASE							:							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()					)
           SCREAMING_SNAKE_CASE							:							List[str]								=      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   a_    (       a__				):



       """simple docstring"""






       def __init__(						self							,    _lowerCamelCase							,    _lowerCamelCase							,    _lowerCamelCase					) ->Optional[Any]:
           SCREAMING_SNAKE_CASE							:							Optional[Any]								=      start_length
           SCREAMING_SNAKE_CASE							:							List[str]								=      eof_strings
           SCREAMING_SNAKE_CASE							:							Dict								=      tokenizer






       def __call__(						self							,    _lowerCamelCase							,    _lowerCamelCase							,    **_lowerCamelCase					) ->Dict:
           SCREAMING_SNAKE_CASE							:							Any								=      self.tokenizer.batch_decode(input_ids[:, self.start_length :]					)
           SCREAMING_SNAKE_CASE							:							Optional[int]								=      []
           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__					):



    """simple docstring"""



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







def 							UpperCAmelCase_(  a__      ,		a__      ,		a__      ,		a__      ,		a__      ,		a__=20      ,		**a__					):



    """simple docstring"""



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

            SCREAMING_SNAKE_CASE	,				SCREAMING_SNAKE_CASE							:							Any								=      accelerator.gather((generated_tokens, generated_tasks)					)
            SCREAMING_SNAKE_CASE							:							Tuple								=      generated_tokens.cpu().numpy()
            SCREAMING_SNAKE_CASE							:							int								=      generated_tasks.cpu().numpy()

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

    SCREAMING_SNAKE_CASE							:							Union[str, Any]								=      [[] for _ in range(__snake_case					)]
    for task, generated_tokens in gen_token_dict.items():
        for s in generated_tokens:
            SCREAMING_SNAKE_CASE							:							List[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"""



    SCREAMING_SNAKE_CASE							:							List[Any]								=      HfArgumentParser(__snake_case					)
    SCREAMING_SNAKE_CASE							:							Dict								=      parser.parse_args()

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

    if args.num_workers is None:
        SCREAMING_SNAKE_CASE							:							Optional[int]								=      multiprocessing.cpu_count()

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

    # Load model and tokenizer
    SCREAMING_SNAKE_CASE							:							Any								=      AutoTokenizer.from_pretrained(args.model_ckpt					)
    SCREAMING_SNAKE_CASE							:							Dict								=      tokenizer.eos_token
    SCREAMING_SNAKE_CASE							:							Dict								=      AutoModelForCausalLM.from_pretrained(args.model_ckpt					)

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

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

    SCREAMING_SNAKE_CASE							:							Union[str, 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
    SCREAMING_SNAKE_CASE							:							Optional[int]								=      DataLoader(__snake_case      ,		batch_size=1					)

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

    SCREAMING_SNAKE_CASE	,				SCREAMING_SNAKE_CASE							:							List[str]								=      accelerator.prepare(__snake_case      ,		__snake_case					)

    SCREAMING_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:
        SCREAMING_SNAKE_CASE							:							Optional[Any]								=      []

        for task in tqdm(range(__snake_case					)					):
            SCREAMING_SNAKE_CASE							:							Any								=      human_eval['''test'''][task]['''test''']
            SCREAMING_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
        SCREAMING_SNAKE_CASE	,				SCREAMING_SNAKE_CASE							:							int								=      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     ,						__snake_case						)     ->			List[Any]:







 """simple docstring"""


 if b == 0:
  return 1
 if (b % 2) == 0:
  return actual_power(__snake_case     ,						int(b / 2						)						) * actual_power(__snake_case     ,						int(b / 2						)						)
 else:
  return a * actual_power(__snake_case     ,						int(b / 2						)						) * actual_power(__snake_case     ,						int(b / 2						)						)
def 				UpperCAmelCase_					(							__snake_case     ,						__snake_case						)     ->			float:







 """simple docstring"""


 if b < 0:
  return 1 / actual_power(__snake_case     ,						__snake_case						)
 return actual_power(__snake_case     ,						__snake_case						)


if __name__ == "__main__":
  print(power(-2, -3))

