



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

class 					_A (			__UpperCAmelCase					):



       def __init__(	self							:  Optional[Any]  ,	__SCREAMING_SNAKE_CASE							:  Union[str, Any]  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  List[Any]=None  ,	__SCREAMING_SNAKE_CASE							:  List[Any]=1):


           '''simple docstring'''

           __a		 =       tokenizer
           __a		 =       dataset
           __a		 =       len(__SCREAMING_SNAKE_CASE) if n_tasks is None else n_tasks
           __a		 =       n_copies





       def __iter__(	self							:  int):


           '''simple docstring'''

           __a		 =       []
           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())
           __a		 =       self.tokenizer(__SCREAMING_SNAKE_CASE  ,	padding=__SCREAMING_SNAKE_CASE  ,	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 (			__UpperCAmelCase					):



       def __init__(	self							:  Dict  ,	__SCREAMING_SNAKE_CASE							:  List[Any]  ,	__SCREAMING_SNAKE_CASE							:  Union[str, Any]  ,	__SCREAMING_SNAKE_CASE							:  Dict):


           '''simple docstring'''

           __a		 =       start_length
           __a		 =       eof_strings
           __a		 =       tokenizer





       def __call__(	self							:  Union[str, Any]  ,	__SCREAMING_SNAKE_CASE							:  List[Any]  ,	__SCREAMING_SNAKE_CASE							:  str  ,	**__SCREAMING_SNAKE_CASE							:  Tuple):


           '''simple docstring'''

           __a		 =       self.tokenizer.batch_decode(input_ids[:, self.start_length :])
           __a		 =       []
           for decoded_generation in decoded_generations:
               done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings))
           return all(__SCREAMING_SNAKE_CASE)





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





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

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

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

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





def   __snake_case			(    ):
    # Setup configuration
    __a		 =       HfArgumentParser(_UpperCAmelCase       )
    __a		 =       parser.parse_args()

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

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

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

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

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

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

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

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

    __a ,     __a		 =       accelerator.prepare(_UpperCAmelCase     ,				_UpperCAmelCase       )

    __a		 =       complete_code(
        _UpperCAmelCase     ,				_UpperCAmelCase     ,				_UpperCAmelCase     ,				_UpperCAmelCase     ,				n_tasks=_UpperCAmelCase     ,				batch_size=args.batch_size     ,				**_UpperCAmelCase     ,				)

    if accelerator.is_main_process:
        __a		 =       []

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

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










import os
import sys
import unittest


__snake_case		:Union[str, Any]   	=  os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, '''utils'''))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
__snake_case		:List[str]   	=  os.path.join(git_repo_path, '''src''', '''transformers''')

__snake_case		:Any   	=  '''
{0} = None
'''

__snake_case		:Dict   	=  '''
class {0}(metaclass=DummyObject):
    _backends = {1}

    def __init__(self, *args, **kwargs):
        requires_backends(self, {1})
'''


__snake_case		:str   	=  '''
def {0}(*args, **kwargs):
    requires_backends({0}, {1})
'''

class 					_A (			unittest.TestCase					):



       def 	_lowerCamelCase    (	self							:  List[Any]):


           '''simple docstring'''

           __a		 =       find_backend('''    _import_structure["models.albert"].append("AlbertTokenizerFast")''')
           self.assertIsNone(__SCREAMING_SNAKE_CASE)

           __a		 =       find_backend('''    if not is_tokenizers_available():''')
           self.assertEqual(__SCREAMING_SNAKE_CASE  ,	'''tokenizers''')

           __a		 =       find_backend('''    if not is_tensorflow_text_available():''')
           self.assertEqual(__SCREAMING_SNAKE_CASE  ,	'''tensorflow_text''')

           __a		 =       find_backend('''    if not (is_sentencepiece_available() and is_tokenizers_available()):''')
           self.assertEqual(__SCREAMING_SNAKE_CASE  ,	'''sentencepiece_and_tokenizers''')

           __a		 =       find_backend(
               '''    if not (is_sentencepiece_available() and is_tensorflow_text_available()):''')
           self.assertEqual(__SCREAMING_SNAKE_CASE  ,	'''sentencepiece_and_tensorflow_text''')

           __a		 =       find_backend(
               '''    if not (is_sentencepiece_available() and is_tokenizers_available() and is_vision_available()):''')
           self.assertEqual(__SCREAMING_SNAKE_CASE  ,	'''sentencepiece_and_tokenizers_and_vision''')



       def 	_lowerCamelCase    (	self							:  Optional[Any]):


           '''simple docstring'''

           __a		 =       read_init()
           # We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
           self.assertIn('''torch'''  ,	__SCREAMING_SNAKE_CASE)
           self.assertIn('''tensorflow_text'''  ,	__SCREAMING_SNAKE_CASE)
           self.assertIn('''sentencepiece_and_tokenizers'''  ,	__SCREAMING_SNAKE_CASE)

           # Likewise, we can't assert on the exact content of a key
           self.assertIn('''BertModel'''  ,	objects['''torch'''])
           self.assertIn('''TFBertModel'''  ,	objects['''tf'''])
           self.assertIn('''FlaxBertModel'''  ,	objects['''flax'''])
           self.assertIn('''BertModel'''  ,	objects['''torch'''])
           self.assertIn('''TFBertTokenizer'''  ,	objects['''tensorflow_text'''])
           self.assertIn('''convert_slow_tokenizer'''  ,	objects['''sentencepiece_and_tokenizers'''])



       def 	_lowerCamelCase    (	self							:  Any):


           '''simple docstring'''

           __a		 =       create_dummy_object('''CONSTANT'''  ,	'''\'torch\'''')
           self.assertEqual(__SCREAMING_SNAKE_CASE  ,	'''\nCONSTANT = None\n''')

           __a		 =       create_dummy_object('''function'''  ,	'''\'torch\'''')
           self.assertEqual(
               __SCREAMING_SNAKE_CASE  ,	'''\ndef function(*args, **kwargs):\n    requires_backends(function, \'torch\')\n''')

           __a		 =       '''
class FakeClass(metaclass=DummyObject):
    _backends = \'torch\'

    def __init__(self, *args, **kwargs):
        requires_backends(self, \'torch\')
'''
           __a		 =       create_dummy_object('''FakeClass'''  ,	'''\'torch\'''')
           self.assertEqual(__SCREAMING_SNAKE_CASE  ,	__SCREAMING_SNAKE_CASE)





       def 	_lowerCamelCase    (	self							:  Optional[Any]):


           '''simple docstring'''

           __a		 =       '''# This file is autogenerated by the command `make fix-copies`, do not edit.
from ..utils import DummyObject, requires_backends


CONSTANT = None


def function(*args, **kwargs):
    requires_backends(function, ["torch"])


class FakeClass(metaclass=DummyObject):
    _backends = ["torch"]

    def __init__(self, *args, **kwargs):
        requires_backends(self, ["torch"])
'''
           __a		 =       create_dummy_files({'''torch''': ['''CONSTANT''', '''function''', '''FakeClass''']})
           self.assertEqual(dummy_files['''torch''']  ,	__SCREAMING_SNAKE_CASE)





