






import functools
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

import huggingface_hub.utils as hf_hub_utils
from tqdm import auto as tqdm_lib


a_							=			threading.Lock()
a_							=			None

a_							=			{
    '''debug''': logging.DEBUG,
    '''info''': logging.INFO,
    '''warning''': logging.WARNING,
    '''error''': logging.ERROR,
    '''critical''': logging.CRITICAL,
}

a_							=			logging.WARNING

a_							=			True


def 				_a     (   )       ->      Tuple:
						"""simple docstring"""







						lowerCAmelCase__       					=  os.getenv("TRANSFORMERS_VERBOSITY"			,      _lowerCamelCase      )
						if env_level_str:
												if env_level_str in log_levels:
																		return log_levels[env_level_str]
												else:
																		logging.getLogger().warning(
																		    F"Unknown option TRANSFORMERS_VERBOSITY={env_level_str}, "
																		    F"has to be one of: { ', '.join(log_levels.keys()      ) }"      )
						return _default_log_level


def 				_a     (   )       ->      str:
						"""simple docstring"""







						return __name__.split("."      )[0]


def 				_a     (   )       ->      logging.Logger:
						"""simple docstring"""







						return logging.getLogger(_get_library_name()      )


def 				_a     (   )       ->      None:
						"""simple docstring"""







						global _default_handler

						with _lock:
												if _default_handler:
																		# This library has already configured the library root logger.
																		return
												lowerCAmelCase__       					=  logging.StreamHandler()  # Set sys.stderr as stream.
												lowerCAmelCase__       					=  sys.stderr.flush

												# Apply our default configuration to the library root logger.
												lowerCAmelCase__       					=  _get_library_root_logger()
												library_root_logger.addHandler(_default_handler      )
												library_root_logger.setLevel(_get_default_logging_level()      )
												lowerCAmelCase__       					=  False


def 				_a     (   )       ->      None:
						"""simple docstring"""







						global _default_handler

						with _lock:
												if not _default_handler:
																		return

												lowerCAmelCase__       					=  _get_library_root_logger()
												library_root_logger.removeHandler(_default_handler      )
												library_root_logger.setLevel(logging.NOTSET      )
												lowerCAmelCase__       					=  None


def 				_a     (   )       ->      Optional[Any]:
						"""simple docstring"""







						return log_levels


def 				_a     (   UpperCamelCase_    : Optional[str] = None      )       ->      logging.Logger:
						"""simple docstring"""







						if name is None:
												lowerCAmelCase__       					=  _get_library_name()

						_configure_library_root_logger()
						return logging.getLogger(_lowerCamelCase      )


def 				_a     (   )       ->      int:
						"""simple docstring"""







						_configure_library_root_logger()
						return _get_library_root_logger().getEffectiveLevel()


def 				_a     (   UpperCamelCase_    : int      )       ->      None:
						"""simple docstring"""







						_configure_library_root_logger()
						_get_library_root_logger().setLevel(_lowerCamelCase      )


def 				_a     (   )       ->      int:
						"""simple docstring"""







						return set_verbosity(_lowerCamelCase      )


def 				_a     (   )       ->      Tuple:
						"""simple docstring"""







						return set_verbosity(_lowerCamelCase      )


def 				_a     (   )       ->      str:
						"""simple docstring"""







						return set_verbosity(_lowerCamelCase      )


def 				_a     (   )       ->      Tuple:
						"""simple docstring"""







						return set_verbosity(_lowerCamelCase      )


def 				_a     (   )       ->      None:
						"""simple docstring"""







						_configure_library_root_logger()

						assert _default_handler is not None
						_get_library_root_logger().removeHandler(_default_handler      )


def 				_a     (   )       ->      None:
						"""simple docstring"""







						_configure_library_root_logger()

						assert _default_handler is not None
						_get_library_root_logger().addHandler(_default_handler      )


def 				_a     (   UpperCamelCase_    : logging.Handler      )       ->      None:
						"""simple docstring"""







						_configure_library_root_logger()

						assert handler is not None
						_get_library_root_logger().addHandler(_lowerCamelCase      )


def 				_a     (   UpperCamelCase_    : logging.Handler      )       ->      None:
						"""simple docstring"""







						_configure_library_root_logger()

						assert handler is not None and handler not in _get_library_root_logger().handlers
						_get_library_root_logger().removeHandler(_lowerCamelCase      )


def 				_a     (   )       ->      None:
						"""simple docstring"""







						_configure_library_root_logger()
						lowerCAmelCase__       					=  False


def 				_a     (   )       ->      None:
						"""simple docstring"""







						_configure_library_root_logger()
						lowerCAmelCase__       					=  True


def 				_a     (   )       ->      None:
						"""simple docstring"""







						lowerCAmelCase__       					=  _get_library_root_logger().handlers

						for handler in handlers:
												lowerCAmelCase__       					=  logging.Formatter("[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s"      )
												handler.setFormatter(_lowerCamelCase      )


def 				_a     (   )       ->      None:
						"""simple docstring"""







						lowerCAmelCase__       					=  _get_library_root_logger().handlers

						for handler in handlers:
												handler.setFormatter(_lowerCamelCase      )


def 				_a     (   self    : Any			,      *UpperCamelCase_    : Optional[Any]			,      **UpperCamelCase_    : str      )       ->      Optional[int]:
						"""simple docstring"""







						lowerCAmelCase__       					=  os.getenv("TRANSFORMERS_NO_ADVISORY_WARNINGS"			,      _lowerCamelCase      )
						if no_advisory_warnings:
												return
						self.warning(*_lowerCamelCase			,      **_lowerCamelCase      )


a_							=			warning_advice


@functools.lru_cache(_lowerCamelCase      )
def 				_a     (   self    : Union[str, Any]			,      *UpperCamelCase_    : List[str]			,      **UpperCamelCase_    : Dict      )       ->      int:
						"""simple docstring"""







						self.warning(*_lowerCamelCase			,      **_lowerCamelCase      )


a_							=			warning_once


class    lowercase__   :

	def __init__(							self						,			*__UpperCAmelCase						,			**__UpperCAmelCase				)->    Any:  # pylint: disable=unused-argument







							'''simple docstring'''





							lowerCAmelCase__       					=  args[0] if args else None

	def __iter__(							self				)->    List[Any]:







							'''simple docstring'''





							return iter(self._iterator				)

	def __getattr__(							self						,			__UpperCAmelCase				)->    List[str]:







							'''simple docstring'''





							def empty_fn(*__UpperCAmelCase						,			**__UpperCAmelCase				):  # pylint: disable=unused-argument
													return

							return empty_fn

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







							'''simple docstring'''





							return self


	def __exit__(							self						,			__UpperCAmelCase						,			__UpperCAmelCase						,			__UpperCAmelCase				)->    Tuple:







							'''simple docstring'''





							return





class    lowercase__   :

	def __call__(							self						,			*__UpperCAmelCase						,			**__UpperCAmelCase				)->    List[Any]:







							'''simple docstring'''





							if _tqdm_active:
													return tqdm_lib.tqdm(*__SCREAMING_SNAKE_CASE						,			**__SCREAMING_SNAKE_CASE				)
							else:
													return EmptyTqdm(*__SCREAMING_SNAKE_CASE						,			**__SCREAMING_SNAKE_CASE				)

	def 						UpperCAmelCase     (							self						,			*__UpperCAmelCase						,			**__UpperCAmelCase				)->    str:







							'''simple docstring'''





							lowerCAmelCase__       					=  None
							if _tqdm_active:
													return tqdm_lib.tqdm.set_lock(*__SCREAMING_SNAKE_CASE						,			**__SCREAMING_SNAKE_CASE				)


	def 						UpperCAmelCase     (							self				)->    Optional[Any]:







							'''simple docstring'''





							if _tqdm_active:
													return tqdm_lib.tqdm.get_lock()





a_							=			_tqdm_cls()


def 				_a     (   )       ->      bool:
						"""simple docstring"""







						global _tqdm_active
						return bool(_tqdm_active      )


def 				_a     (   )       ->      List[str]:
						"""simple docstring"""







						global _tqdm_active
						lowerCAmelCase__       					=  True
						hf_hub_utils.enable_progress_bars()





def 				_a     (   )       ->      Optional[Any]:
						"""simple docstring"""







						global _tqdm_active
						lowerCAmelCase__       					=  False
						hf_hub_utils.disable_progress_bars()














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






class      a						(					__snake_case		):






    def __init__(       self     :							Dict       ,						__SCREAMING_SNAKE_CASE     :							Any       ,						__SCREAMING_SNAKE_CASE     :							Optional[Any]       ,						__SCREAMING_SNAKE_CASE     :							Any=None       ,						__SCREAMING_SNAKE_CASE     :							Tuple=1					)     ->		str:
      lowerCamelCase_         =						tokenizer
      lowerCamelCase_         =						dataset
      lowerCamelCase_         =						len(__SCREAMING_SNAKE_CASE					) if n_tasks is None else n_tasks
      lowerCamelCase_         =						n_copies






    def __iter__(       self     :							Dict					)     ->		Any:
      lowerCamelCase_         =						[]
      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()					)
      lowerCamelCase_         =						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						(					__snake_case		):






    def __init__(       self     :							Union[str, Any]       ,						__SCREAMING_SNAKE_CASE     :							str       ,						__SCREAMING_SNAKE_CASE     :							int       ,						__SCREAMING_SNAKE_CASE     :							Any					)     ->		List[Any]:
      lowerCamelCase_         =						start_length
      lowerCamelCase_         =						eof_strings
      lowerCamelCase_         =						tokenizer






    def __call__(       self     :							Union[str, Any]       ,						__SCREAMING_SNAKE_CASE     :							Optional[int]       ,						__SCREAMING_SNAKE_CASE     :							Dict       ,						**__SCREAMING_SNAKE_CASE     :							Union[str, Any]					)     ->		Optional[Any]:
      lowerCamelCase_         =						self.tokenizer.batch_decode(input_ids[:, self.start_length :]					)
      lowerCamelCase_         =						[]
      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__       (							_lowerCamelCase       :		List[Any] )      ->				Tuple:
  lowerCamelCase_         =						re.split('(%s)' % '|'.join(_lowerCamelCase )		,				_lowerCamelCase )
  # last string should be ""
  return "".join(string_list[:-2] )







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

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

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

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







def      lowerCamelCase__       (							)      ->				Tuple:
  # Setup configuration
  lowerCamelCase_         =						HfArgumentParser(_lowerCamelCase )
  lowerCamelCase_         =						parser.parse_args()

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

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

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

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

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

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

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

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

  lowerCamelCase_     ,	lowerCamelCase_         =						accelerator.prepare(_lowerCamelCase		,				_lowerCamelCase )

  lowerCamelCase_         =						complete_code(
      _lowerCamelCase		,				_lowerCamelCase		,				_lowerCamelCase		,				_lowerCamelCase		,				n_tasks=_lowerCamelCase		,				batch_size=args.batch_size		,				**_lowerCamelCase		,				)

  if accelerator.is_main_process:
    lowerCamelCase_         =						[]

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

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




