





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


UpperCAmelCase_  							=  ['\nclass', '\ndef', '\n#', '\n@', '\nprint', '\nif']






class 					lowercase__ (			_UpperCamelCase       ):

   '''simple docstring'''


   def __init__(					self,	__magic_name__,	__magic_name__,	__magic_name__=None,	__magic_name__=1							)				->   Tuple:






       """simple docstring"""
       UpperCamelCase__							:       Union[str, Any]      =   tokenizer
       UpperCamelCase__							:       Tuple      =   dataset
       UpperCamelCase__							:       str      =   len(__a							) if n_tasks is None else n_tasks
       UpperCamelCase__							:       Union[str, Any]      =   n_copies




   def __iter__(					self							)				->   Optional[int]:






       """simple docstring"""
       UpperCamelCase__							:       Optional[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__							:       Dict      =   self.tokenizer(__a,	padding=__a,	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__ (			_UpperCamelCase       ):

   '''simple docstring'''


   def __init__(					self,	__magic_name__,	__magic_name__,	__magic_name__							)				->   int:






       """simple docstring"""
       UpperCamelCase__							:       Union[str, Any]      =   start_length
       UpperCamelCase__							:       Union[str, Any]      =   eof_strings
       UpperCamelCase__							:       Tuple      =   tokenizer




   def __call__(					self,	__magic_name__,	__magic_name__,	**__magic_name__							)				->   Union[str, Any]:






       """simple docstring"""
       UpperCamelCase__							:       List[str]      =   self.tokenizer.batch_decode(input_ids[:, self.start_length :]							)
       UpperCamelCase__							:       Any      =   []
       for decoded_generation in decoded_generations:
           done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings							)							)
       return all(__a							)




def 	lowerCAmelCase_					(						__UpperCAmelCase:    Optional[Any]   )   ->			int:
    UpperCamelCase__							:       List[Any]      =   re.split('''(%s)''' % '''|'''.join(lowercase_   )	,     lowercase_   )
    # last string should be ""
    return "".join(string_list[:-2]   )




def 	lowerCAmelCase_					(						__UpperCAmelCase:    List[Any]	,     __UpperCAmelCase:    Optional[Any]	,     __UpperCAmelCase:    str	,     __UpperCAmelCase:    Union[str, Any]	,     __UpperCAmelCase:    str	,     __UpperCAmelCase:    Union[str, Any]=20	,     **__UpperCAmelCase:    Tuple   )   ->			Dict:
    UpperCamelCase__							:       List[Any]      =   defaultdict(lowercase_   )  # dict of list of generated tokens
    for step, batch in tqdm(enumerate(lowercase_   )   ):
        with torch.no_grad():
            UpperCamelCase__							:       Optional[int]      =   batch["ids"].shape[-1]
            UpperCamelCase__							:       int      =   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__							:       Optional[Any]      =   batch["task_id"].repeat(lowercase_   )
            UpperCamelCase__							:       Any      =   accelerator.pad_across_processes(
                lowercase_	,     dim=1	,     pad_index=tokenizer.pad_token_id   )

            UpperCamelCase__							:       Dict      =   accelerator.gather((generated_tokens, generated_tasks)   )
            UpperCamelCase__							:       Any      =   generated_tokens.cpu().numpy()
            UpperCamelCase__							:       Optional[int]      =   generated_tasks.cpu().numpy()

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

    UpperCamelCase__							:       Tuple      =   [[] for _ in range(lowercase_   )]
    for task, generated_tokens in gen_token_dict.items():
        for s in generated_tokens:
            UpperCamelCase__							:       Optional[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_					(						)   ->			List[str]:
    UpperCamelCase__							:       Union[str, Any]      =   HfArgumentParser(lowercase_   )
    UpperCamelCase__							:       str      =   parser.parse_args()

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

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

    # Use dataset load to feed to accelerate
    UpperCamelCase__							:       Optional[Any]      =   Accelerator()
    set_seed(args.seed	,     device_specific=lowercase_   )

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

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

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

    UpperCamelCase__							:       Tuple      =   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__							:       List[Any]      =   DataLoader(lowercase_	,     batch_size=1   )

    # Run a quick test to see if code evaluation is enabled
    try:
        UpperCamelCase__							:       Union[str, 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__							:       Tuple      =   accelerator.prepare(lowercase_	,     lowercase_   )

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

    if accelerator.is_main_process:
        UpperCamelCase__							:       Union[str, Any]      =   []

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

        # Evaluate completions with "code_eval" metric
        UpperCamelCase__							:       List[str]      =   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()








import unittest

from transformers import BigBirdConfig, is_flax_available
from transformers.testing_utils import require_flax, slow

from ...test_modeling_flax_common import FlaxModelTesterMixin, ids_tensor, random_attention_mask


if is_flax_available():
  import jax

  from transformers.models.big_bird.modeling_flax_big_bird import (
      FlaxBigBirdForCausalLM,
      FlaxBigBirdForMaskedLM,
      FlaxBigBirdForMultipleChoice,
      FlaxBigBirdForPreTraining,
      FlaxBigBirdForQuestionAnswering,
      FlaxBigBirdForSequenceClassification,
      FlaxBigBirdForTokenClassification,
      FlaxBigBirdModel,
  )






class 					lowercase__ (			unittest.TestCase       ):

   '''simple docstring'''


   def __init__(					self,	__magic_name__,	__magic_name__=2,	__magic_name__=56,	__magic_name__=True,	__magic_name__=True,	__magic_name__=True,	__magic_name__=True,	__magic_name__=99,	__magic_name__=32,	__magic_name__=2,	__magic_name__=2,	__magic_name__=7,	__magic_name__="gelu_new",	__magic_name__=0.1,	__magic_name__=0.1,	__magic_name__=512,	__magic_name__=16,	__magic_name__=2,	__magic_name__=0.02,	__magic_name__=4,	__magic_name__="block_sparse",	__magic_name__=True,	__magic_name__=False,	__magic_name__=2,	__magic_name__=3,	)				->   Optional[int]:






       """simple docstring"""
       UpperCamelCase__							:       List[str]      =   parent
       UpperCamelCase__							:       List[Any]      =   batch_size
       UpperCamelCase__							:       str      =   seq_length
       UpperCamelCase__							:       List[str]      =   is_training
       UpperCamelCase__							:       List[Any]      =   use_attention_mask
       UpperCamelCase__							:       Any      =   use_token_type_ids
       UpperCamelCase__							:       Tuple      =   use_labels
       UpperCamelCase__							:       Any      =   vocab_size
       UpperCamelCase__							:       Optional[int]      =   hidden_size
       UpperCamelCase__							:       Union[str, Any]      =   num_hidden_layers
       UpperCamelCase__							:       Optional[Any]      =   num_attention_heads
       UpperCamelCase__							:       List[str]      =   intermediate_size
       UpperCamelCase__							:       List[str]      =   hidden_act
       UpperCamelCase__							:       Union[str, Any]      =   hidden_dropout_prob
       UpperCamelCase__							:       Union[str, Any]      =   attention_probs_dropout_prob
       UpperCamelCase__							:       int      =   max_position_embeddings
       UpperCamelCase__							:       List[Any]      =   type_vocab_size
       UpperCamelCase__							:       Tuple      =   type_sequence_label_size
       UpperCamelCase__							:       Optional[Any]      =   initializer_range
       UpperCamelCase__							:       Dict      =   num_choices

       UpperCamelCase__							:       Dict      =   rescale_embeddings
       UpperCamelCase__							:       Tuple      =   attention_type
       UpperCamelCase__							:       Union[str, Any]      =   use_bias
       UpperCamelCase__							:       Union[str, Any]      =   block_size
       UpperCamelCase__							:       Optional[int]      =   num_random_blocks




   def    UpperCamelCase__						(					self							)				->   int:






       """simple docstring"""
       UpperCamelCase__							:       int      =   ids_tensor([self.batch_size, self.seq_length],	self.vocab_size							)

       UpperCamelCase__							:       int      =   None
       if self.use_attention_mask:
           UpperCamelCase__							:       List[str]      =   random_attention_mask([self.batch_size, self.seq_length]							)

       UpperCamelCase__							:       Optional[int]      =   None
       if self.use_token_type_ids:
           UpperCamelCase__							:       List[str]      =   ids_tensor([self.batch_size, self.seq_length],	self.type_vocab_size							)

       UpperCamelCase__							:       str      =   BigBirdConfig(
           vocab_size=self.vocab_size,	hidden_size=self.hidden_size,	num_hidden_layers=self.num_hidden_layers,	num_attention_heads=self.num_attention_heads,	intermediate_size=self.intermediate_size,	hidden_act=self.hidden_act,	hidden_dropout_prob=self.hidden_dropout_prob,	attention_probs_dropout_prob=self.attention_probs_dropout_prob,	max_position_embeddings=self.max_position_embeddings,	type_vocab_size=self.type_vocab_size,	is_decoder=__magic_name__,	initializer_range=self.initializer_range,	attention_type=self.attention_type,	block_size=self.block_size,	num_random_blocks=self.num_random_blocks,	use_bias=self.use_bias,	rescale_embeddings=self.rescale_embeddings,	)

       return config, input_ids, token_type_ids, attention_mask




   def    UpperCamelCase__						(					self							)				->   str:






       """simple docstring"""
       UpperCamelCase__							:       List[str]      =   self.prepare_config_and_inputs()
       UpperCamelCase__     ,UpperCamelCase__     ,UpperCamelCase__     ,UpperCamelCase__							:       Tuple      =   config_and_inputs
       UpperCamelCase__							:       Union[str, Any]      =   {
           '''input_ids''': input_ids,
           '''token_type_ids''': token_type_ids,
           '''attention_mask''': attention_mask,
       }
       return config, inputs_dict






@require_flax
class 					lowercase__ (			__lowerCamelCase    ,       unittest.TestCase       ):

   '''simple docstring'''


   a		:							Any     	=	(
       (
           FlaxBigBirdForCausalLM,
           FlaxBigBirdModel,
           FlaxBigBirdForPreTraining,
           FlaxBigBirdForMaskedLM,
           FlaxBigBirdForMultipleChoice,
           FlaxBigBirdForQuestionAnswering,
           FlaxBigBirdForSequenceClassification,
           FlaxBigBirdForTokenClassification,
       )
       if is_flax_available()
       else ()
   )

   a		:							List[str]     	=	False
   a		:							str     	=	False




   def    UpperCamelCase__						(					self							)				->   Any:






       """simple docstring"""
       UpperCamelCase__							:       Union[str, Any]      =   FlaxBigBirdModelTester(self							)




   @slow
   # copied from `test_modeling_flax_common` because it takes much longer than other models
   def    UpperCamelCase__						(					self							)				->   Dict:






       """simple docstring"""
       super().test_from_pretrained_save_pretrained()




   @slow
   # copied from `test_modeling_flax_common` because it takes much longer than other models
   def    UpperCamelCase__						(					self							)				->   Optional[Any]:






       """simple docstring"""
       super().test_from_pretrained_with_no_automatic_init()




   @slow
   # copied from `test_modeling_flax_common` because it takes much longer than other models
   def    UpperCamelCase__						(					self							)				->   Dict:






       """simple docstring"""
       super().test_no_automatic_init()




   @slow
   # copied from `test_modeling_flax_common` because it takes much longer than other models
   def    UpperCamelCase__						(					self							)				->   Optional[Any]:






       """simple docstring"""
       super().test_hidden_states_output()




   @slow
   def    UpperCamelCase__						(					self							)				->   List[str]:






       """simple docstring"""
       for model_class_name in self.all_model_classes:
           UpperCamelCase__							:       List[str]      =   model_class_name.from_pretrained('''google/bigbird-roberta-base'''							)
           self.assertIsNotNone(__magic_name__							)




   def    UpperCamelCase__						(					self							)				->   str:






       """simple docstring"""
       if self.test_attn_probs:
           super().test_attention_outputs()




   @slow
   # copied from `test_modeling_flax_common` because it takes much longer than other models
   def    UpperCamelCase__						(					self							)				->   Any:






       """simple docstring"""
       UpperCamelCase__     ,UpperCamelCase__							:       Dict      =   self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
           with self.subTest(model_class.__name__							):
               UpperCamelCase__							:       Union[str, Any]      =   self._prepare_for_class(__magic_name__,	__magic_name__							)
               UpperCamelCase__							:       Tuple      =   model_class(__magic_name__							)

               @jax.jit
               def model_jitted(__magic_name__,	__magic_name__=None,	**__magic_name__							):
                   return model(input_ids=__magic_name__,	attention_mask=__magic_name__,	**__magic_name__							)

               with self.subTest('''JIT Enabled'''							):
                   UpperCamelCase__							:       str      =   model_jitted(**__magic_name__							).to_tuple()

               with self.subTest('''JIT Disabled'''							):
                   with jax.disable_jit():
                       UpperCamelCase__							:       Dict      =   model_jitted(**__magic_name__							).to_tuple()

               self.assertEqual(len(__magic_name__							),	len(__magic_name__							)							)
               for jitted_output, output in zip(__magic_name__,	__magic_name__							):
                   self.assertEqual(jitted_output.shape,	output.shape							)




   def    UpperCamelCase__						(					self,	__magic_name__,	__magic_name__,	__magic_name__,	__magic_name__=1E-5,	__magic_name__="outputs",	__magic_name__=None							)				->   List[Any]:






       """simple docstring"""
       # `bigbird_block_sparse_attention` in `FlaxBigBird` returns `attention_probs = None`, while in PyTorch version,
       # an effort was done to return `attention_probs` (yet to be verified).
       if name.startswith('''outputs.attentions'''							):
           return
       else:
           super().check_pt_flax_outputs(__magic_name__,	__magic_name__,	__magic_name__,	__magic_name__,	__magic_name__,	__magic_name__							)

