
"""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


SCREAMING_SNAKE_CASE_            =    ['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']
class   _UpperCAmelCase				(		SCREAMING_SNAKE_CASE_				):




      def __init__(				self ,							lowercase_ ,							lowercase_ ,							lowercase_=None ,							lowercase_=1				)    ->       Tuple:
          UpperCAmelCase          =       tokenizer
          UpperCAmelCase          =       dataset
          UpperCAmelCase          =       len(lowercase_				) if n_tasks is None else n_tasks
          UpperCAmelCase          =       n_copies




      def __iter__(				self				)    ->       Tuple:
          UpperCAmelCase          =       []
          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          =       self.tokenizer(lowercase_ ,							padding=lowercase_ ,							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   _UpperCAmelCase				(		SCREAMING_SNAKE_CASE_				):




      def __init__(				self ,							lowercase_ ,							lowercase_ ,							lowercase_				)    ->       Optional[int]:
          UpperCAmelCase          =       start_length
          UpperCAmelCase          =       eof_strings
          UpperCAmelCase          =       tokenizer




      def __call__(				self ,							lowercase_ ,							lowercase_ ,							**lowercase_				)    ->       Dict:
          UpperCAmelCase          =       self.tokenizer.batch_decode(input_ids[:, self.start_length :]				)
          UpperCAmelCase          =       []
          for decoded_generation in decoded_generations:
              done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings				)				)
          return all(lowercase_				)




def      lowercase__    (       lowerCAmelCase					:       Dict			)			->   Optional[int]:






    """simple docstring"""




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




def      lowercase__    (       lowerCAmelCase					:       Tuple			,					lowerCAmelCase					:       Union[str, Any]			,					lowerCAmelCase					:       List[str]			,					lowerCAmelCase					:       Union[str, Any]			,					lowerCAmelCase					:       List[Any]			,					lowerCAmelCase					:       int=20			,					**lowerCAmelCase					:       List[Any]			)			->   Tuple:






    """simple docstring"""




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

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

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

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




def      lowercase__    (       )			->   Optional[int]:






    """simple docstring"""




    UpperCAmelCase          =       HfArgumentParser(lowerCAmelCase			)
    UpperCAmelCase          =       parser.parse_args()

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

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

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

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

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

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

    UpperCAmelCase          =       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          =       DataLoader(lowerCAmelCase			,					batch_size=1			)

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

    UpperCAmelCase          =       complete_code(
        lowerCAmelCase			,					lowerCAmelCase			,					lowerCAmelCase			,					lowerCAmelCase			,					n_tasks=lowerCAmelCase			,					batch_size=args.batch_size			,					**lowerCAmelCase			,					)

    if accelerator.is_main_process:
        UpperCAmelCase          =       []

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

        # Evaluate completions with "code_eval" metric
        UpperCAmelCase      ,				UpperCAmelCase          =       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 ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
class   _UpperCAmelCase				(		SCREAMING_SNAKE_CASE_				):
      __SCREAMING_SNAKE_CASE :				List[Any]				   =						["image_processor", "tokenizer"]
      __SCREAMING_SNAKE_CASE :				Dict				   =						"AutoImageProcessor"
      __SCREAMING_SNAKE_CASE :				Optional[int]				   =						"AutoTokenizer"




      def __init__(				self ,							lowercase_ ,							lowercase_				)    ->       List[str]:
          super().__init__(lowercase_ ,							lowercase_				)
          UpperCAmelCase          =       self.image_processor




      def __call__(				self ,							lowercase_=None ,							lowercase_=None ,							lowercase_=None ,							**lowercase_				)    ->       Any:

          if text is None and images is None:
              raise ValueError('You have to specify either text or images. Both cannot be none.'				)

          if text is not None:
              UpperCAmelCase          =       self.tokenizer(lowercase_ ,							return_tensors=lowercase_ ,							**lowercase_				)

          if images is not None:
              UpperCAmelCase          =       self.image_processor(lowercase_ ,							return_tensors=lowercase_ ,							**lowercase_				)

          if text is not None and images is not None:
              UpperCAmelCase          =       image_features.pixel_values
              return encoding
          elif text is not None:
              return encoding
          else:
              return BatchEncoding(data=dict(**lowercase_				) ,							tensor_type=lowercase_				)




      def      a_   (				self ,							*lowercase_ ,							**lowercase_				)    ->       Optional[Any]:
          return self.tokenizer.batch_decode(*lowercase_ ,							**lowercase_				)




      def      a_   (				self ,							*lowercase_ ,							**lowercase_				)    ->       Optional[int]:
          return self.tokenizer.decode(*lowercase_ ,							**lowercase_				)




      @property
      def      a_   (				self				)    ->       Optional[int]:
          return ["input_ids", "attention_mask", "pixel_values"]



