


import os
import sys
import unittest


_SCREAMING_SNAKE_CASE     :					int      =       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
_SCREAMING_SNAKE_CASE     :					int      =       os.path.join(git_repo_path, "src", "transformers")

_SCREAMING_SNAKE_CASE     :					Optional[int]      =       "\n{0} = None\n"

_SCREAMING_SNAKE_CASE     :					Optional[Any]      =       "\nclass {0}(metaclass=DummyObject):\n    _backends = {1}\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, {1})\n"


_SCREAMING_SNAKE_CASE     :					List[str]      =       "\ndef {0}(*args, **kwargs):\n    requires_backends({0}, {1})\n"

class     A__   (       unittest.TestCase      ):







    """simple docstring"""

    def 						a_     (			self       ):
         snake_case											=							find_backend('''    _import_structure["models.albert"].append("AlbertTokenizerFast")'''       )
         self.assertIsNone(_SCREAMING_SNAKE_CASE       )

         snake_case											=							find_backend('''    if not is_tokenizers_available():'''       )
         self.assertEqual(_SCREAMING_SNAKE_CASE     ,   '''tokenizers'''       )

         snake_case											=							find_backend('''    if not is_tensorflow_text_available():'''       )
         self.assertEqual(_SCREAMING_SNAKE_CASE     ,   '''tensorflow_text'''       )

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

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

         snake_case											=							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 						a_     (			self       ):
         snake_case											=							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 						a_     (			self       ):
         snake_case											=							create_dummy_object('''CONSTANT'''     ,   '''\'torch\''''       )
         self.assertEqual(_SCREAMING_SNAKE_CASE     ,   '''\nCONSTANT = None\n'''       )

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

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

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




    def 						a_     (			self       ):
         snake_case											=							'''# 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"])
'''
         snake_case											=							create_dummy_files({'''torch''': ['''CONSTANT''', '''function''', '''FakeClass''']}       )
         self.assertEqual(dummy_files['''torch''']     ,   _SCREAMING_SNAKE_CASE       )










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     __a(	_a  ):






    """simple docstring"""





    def __init__( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE=None  ,_SCREAMING_SNAKE_CASE=1    )					->						Dict:
         UpperCAmelCase_     :				List[Any]  =     tokenizer
         UpperCAmelCase_     :				int  =     dataset
         UpperCAmelCase_     :				Dict  =     len(_SCREAMING_SNAKE_CASE    ) if n_tasks is None else n_tasks
         UpperCAmelCase_     :				Optional[int]  =     n_copies





    def __iter__( self    )					->						Any:
         UpperCAmelCase_     :				List[Any]  =     []
         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(_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(	_a  ):






    """simple docstring"""





    def __init__( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE    )					->						List[Any]:
         UpperCAmelCase_     :				str  =     start_length
         UpperCAmelCase_     :				Optional[int]  =     eof_strings
         UpperCAmelCase_     :				str  =     tokenizer





    def __call__( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,**_SCREAMING_SNAKE_CASE    )					->						List[Any]:
         UpperCAmelCase_     :				Optional[Any]  =     self.tokenizer.batch_decode(input_ids[:, self.start_length :]    )
         UpperCAmelCase_     :				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(_SCREAMING_SNAKE_CASE    )







def lowerCamelCase__  (     _lowercase ):

     '''simple docstring'''


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







def lowerCamelCase__  (     _lowercase					,		_lowercase					,		_lowercase					,		_lowercase					,		_lowercase					,		_lowercase=20					,		**_lowercase ):

     '''simple docstring'''


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

               UpperCAmelCase_,       UpperCAmelCase_     :				List[str]  =     accelerator.gather((generated_tokens, generated_tasks) )
               UpperCAmelCase_     :				Union[str, Any]  =     generated_tokens.cpu().numpy()
               UpperCAmelCase_     :				Union[str, Any]  =     generated_tasks.cpu().numpy()

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

     UpperCAmelCase_     :				Union[str, Any]  =     [[] for _ in range(_lowercase )]
     for task, generated_tokens in gen_token_dict.items():
          for s in generated_tokens:
               UpperCAmelCase_     :				int  =     tokenizer.decode(_lowercase					,		skip_special_tokens=_lowercase					,		clean_up_tokenization_spaces=_lowercase )
               code_gens[task].append(remove_last_block(_lowercase ) )
     return code_gens







def lowerCamelCase__  (     ):

     '''simple docstring'''


     UpperCAmelCase_     :				Optional[Any]  =     HfArgumentParser(_lowercase )
     UpperCAmelCase_     :				int  =     parser.parse_args()

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

     if args.num_workers is None:
          UpperCAmelCase_     :				Optional[Any]  =     multiprocessing.cpu_count()

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

     # Load model and tokenizer
     UpperCAmelCase_     :				Tuple  =     AutoTokenizer.from_pretrained(args.model_ckpt )
     UpperCAmelCase_     :				Any  =     tokenizer.eos_token
     UpperCAmelCase_     :				List[str]  =     AutoModelForCausalLM.from_pretrained(args.model_ckpt )

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

     # Load evaluation dataset and metric
     UpperCAmelCase_     :				Tuple  =     load_dataset('''openai_humaneval''' )
     UpperCAmelCase_     :				Dict  =     load_metric('''code_eval''' )

     UpperCAmelCase_     :				Optional[int]  =     args.num_tasks if args.num_tasks is not None else len(human_eval['''test'''] )
     UpperCAmelCase_     :				str  =     args.n_samples // args.batch_size

     UpperCAmelCase_     :				str  =     TokenizedDataset(_lowercase					,		human_eval['''test''']					,		n_copies=_lowercase					,		n_tasks=_lowercase )
     # do not confuse args.batch_size, which is actually the num_return_sequences
     UpperCAmelCase_     :				Optional[Any]  =     DataLoader(_lowercase					,		batch_size=1 )

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

     UpperCAmelCase_     :				int  =     complete_code(
         _lowercase					,		_lowercase					,		_lowercase					,		_lowercase					,		n_tasks=_lowercase					,		batch_size=args.batch_size					,		**_lowercase					,		)

     if accelerator.is_main_process:
          UpperCAmelCase_     :				Any  =     []

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

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


# 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()