






"""simple docstring"""





import os
import sys
import warnings
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
  import PIL.Image

  from .features import FeatureType


_SCREAMING_SNAKE_CASE					:				Optional[List[str]]					    =						None
_SCREAMING_SNAKE_CASE					:				Optional[int]					    =						'''<''' if sys.byteorder == '''little''' else '''>'''
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
_SCREAMING_SNAKE_CASE					:				List[str]					    =						[
    np.dtype('''|b1'''),
    np.dtype('''|u1'''),
    np.dtype('''<u2'''),
    np.dtype('''>u2'''),
    np.dtype('''<i2'''),
    np.dtype('''>i2'''),
    np.dtype('''<u4'''),
    np.dtype('''>u4'''),
    np.dtype('''<i4'''),
    np.dtype('''>i4'''),
    np.dtype('''<f4'''),
    np.dtype('''>f4'''),
    np.dtype('''<f8'''),
    np.dtype('''>f8'''),
]






@dataclass
class      a						:
    SCREAMING_SNAKE_CASE						:     bool						= True
    SCREAMING_SNAKE_CASE						:     Optional[str]						= None
    # Automatically constructed
    SCREAMING_SNAKE_CASE						:     ClassVar[str]						= "PIL.Image.Image"
    SCREAMING_SNAKE_CASE						:     ClassVar[Any]						= pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}		)
    SCREAMING_SNAKE_CASE						:     str						= field(default="""Image"""  ,      init=__snake_case  ,      repr=__snake_case		)






    def __call__(       self     :							Union[str, Any]					)     ->		Tuple:
      return self.pa_type






    def    UpperCamelCase       (       self     :							str       ,						__SCREAMING_SNAKE_CASE     :							Union[str, bytes, dict, np.ndarray, "PIL.Image.Image"]					)     ->		dict:
      if config.PIL_AVAILABLE:
        import PIL.Image
      else:
        raise ImportError('To support encoding images, please install \'Pillow\'.'					)

      if isinstance(__SCREAMING_SNAKE_CASE       ,						__SCREAMING_SNAKE_CASE					):
        lowerCamelCase_         =						np.array(__SCREAMING_SNAKE_CASE					)

      if isinstance(__SCREAMING_SNAKE_CASE       ,						__SCREAMING_SNAKE_CASE					):
        return {"path": value, "bytes": None}
      elif isinstance(__SCREAMING_SNAKE_CASE       ,						__SCREAMING_SNAKE_CASE					):
        return {"path": None, "bytes": value}
      elif isinstance(__SCREAMING_SNAKE_CASE       ,						np.ndarray					):
        # convert the image array to PNG/TIFF bytes
        return encode_np_array(__SCREAMING_SNAKE_CASE					)
      elif isinstance(__SCREAMING_SNAKE_CASE       ,						PIL.Image.Image					):
        # convert the PIL image to bytes (default format is PNG/TIFF)
        return encode_pil_image(__SCREAMING_SNAKE_CASE					)
      elif value.get('path'					) is not None and os.path.isfile(value['path']					):
        # we set "bytes": None to not duplicate the data if they're already available locally
        return {"bytes": None, "path": value.get('path'					)}
      elif value.get('bytes'					) is not None or value.get('path'					) is not None:
        # store the image bytes, and path is used to infer the image format using the file extension
        return {"bytes": value.get('bytes'					), "path": value.get('path'					)}
      else:
        raise ValueError(
            F'''An image sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''					)






    def    UpperCamelCase       (       self     :							int       ,						__SCREAMING_SNAKE_CASE     :							dict       ,						__SCREAMING_SNAKE_CASE     :							Optional[Any]=None					)     ->		"PIL.Image.Image":
      if not self.decode:
        raise RuntimeError('Decoding is disabled for this feature. Please use Image(decode=True) instead.'					)

      if config.PIL_AVAILABLE:
        import PIL.Image
      else:
        raise ImportError('To support decoding images, please install \'Pillow\'.'					)

      if token_per_repo_id is None:
        lowerCamelCase_         =						{}

      lowerCamelCase_     ,	lowerCamelCase_         =						value['path'], value['bytes']
      if bytes_ is None:
        if path is None:
          raise ValueError(F'''An image should have one of \'path\' or \'bytes\' but both are None in {value}.'''					)
        else:
          if is_local_path(__SCREAMING_SNAKE_CASE					):
            lowerCamelCase_         =						PIL.Image.open(__SCREAMING_SNAKE_CASE					)
          else:
            lowerCamelCase_         =						path.split('::'					)[-1]
            try:
              lowerCamelCase_         =						string_to_dict(__SCREAMING_SNAKE_CASE       ,						config.HUB_DATASETS_URL					)['repo_id']
              lowerCamelCase_         =						token_per_repo_id.get(__SCREAMING_SNAKE_CASE					)
            except ValueError:
              lowerCamelCase_         =						None
            with xopen(__SCREAMING_SNAKE_CASE       ,						'rb'       ,						use_auth_token=__SCREAMING_SNAKE_CASE					) as f:
              lowerCamelCase_         =						BytesIO(f.read()					)
            lowerCamelCase_         =						PIL.Image.open(bytes_					)
      else:
        lowerCamelCase_         =						PIL.Image.open(BytesIO(bytes_					)					)
      image.load()  # to avoid "Too many open files" errors
      return image






    def    UpperCamelCase       (       self     :							Optional[Any]					)     ->		Union["FeatureType", Dict[str, "FeatureType"]]:
      from .features import Value

      return (
          self
          if self.decode
          else {
              "bytes": Value('binary'					),
              "path": Value('string'					),
          }
      )






    def    UpperCamelCase       (       self     :							List[Any]       ,						__SCREAMING_SNAKE_CASE     :							Union[pa.StringArray, pa.StructArray, pa.ListArray]					)     ->		pa.StructArray:
      if pa.types.is_string(storage.type					):
        lowerCamelCase_         =						pa.array([None] * len(__SCREAMING_SNAKE_CASE					)       ,						type=pa.binary()					)
        lowerCamelCase_         =						pa.StructArray.from_arrays([bytes_array, storage]       ,						['bytes', 'path']       ,						mask=storage.is_null()					)
      elif pa.types.is_binary(storage.type					):
        lowerCamelCase_         =						pa.array([None] * len(__SCREAMING_SNAKE_CASE					)       ,						type=pa.string()					)
        lowerCamelCase_         =						pa.StructArray.from_arrays([storage, path_array]       ,						['bytes', 'path']       ,						mask=storage.is_null()					)
      elif pa.types.is_struct(storage.type					):
        if storage.type.get_field_index('bytes'					) >= 0:
          lowerCamelCase_         =						storage.field('bytes'					)
        else:
          lowerCamelCase_         =						pa.array([None] * len(__SCREAMING_SNAKE_CASE					)       ,						type=pa.binary()					)
        if storage.type.get_field_index('path'					) >= 0:
          lowerCamelCase_         =						storage.field('path'					)
        else:
          lowerCamelCase_         =						pa.array([None] * len(__SCREAMING_SNAKE_CASE					)       ,						type=pa.string()					)
        lowerCamelCase_         =						pa.StructArray.from_arrays([bytes_array, path_array]       ,						['bytes', 'path']       ,						mask=storage.is_null()					)
      elif pa.types.is_list(storage.type					):
        lowerCamelCase_         =						pa.array(
            [encode_np_array(np.array(__SCREAMING_SNAKE_CASE					)					)['bytes'] if arr is not None else None for arr in storage.to_pylist()]       ,						type=pa.binary()       ,						)
        lowerCamelCase_         =						pa.array([None] * len(__SCREAMING_SNAKE_CASE					)       ,						type=pa.string()					)
        lowerCamelCase_         =						pa.StructArray.from_arrays(
            [bytes_array, path_array]       ,						['bytes', 'path']       ,						mask=bytes_array.is_null()					)
      return array_cast(__SCREAMING_SNAKE_CASE       ,						self.pa_type					)






    def    UpperCamelCase       (       self     :							str       ,						__SCREAMING_SNAKE_CASE     :							pa.StructArray					)     ->		pa.StructArray:

      @no_op_if_value_is_null
      def path_to_bytes(__SCREAMING_SNAKE_CASE     :							int					):
        with xopen(__SCREAMING_SNAKE_CASE       ,						'rb'					) as f:
          lowerCamelCase_         =						f.read()
        return bytes_

      lowerCamelCase_         =						pa.array(
          [
              (path_to_bytes(x['path']					) if x['bytes'] is None else x['bytes']) if x is not None else None
              for x in storage.to_pylist()
          ]       ,						type=pa.binary()       ,						)
      lowerCamelCase_         =						pa.array(
          [os.path.basename(__SCREAMING_SNAKE_CASE					) if path is not None else None for path in storage.field('path'					).to_pylist()]       ,						type=pa.string()       ,						)
      lowerCamelCase_         =						pa.StructArray.from_arrays([bytes_array, path_array]       ,						['bytes', 'path']       ,						mask=bytes_array.is_null()					)
      return array_cast(__SCREAMING_SNAKE_CASE       ,						self.pa_type					)







def      lowerCamelCase__       (							)      ->				List[str]:
  if config.PIL_AVAILABLE:
    import PIL.Image
  else:
    raise ImportError('To support encoding images, please install \'Pillow\'.' )

  global _IMAGE_COMPRESSION_FORMATS
  if _IMAGE_COMPRESSION_FORMATS is None:
    PIL.Image.init()
    lowerCamelCase_         =						list(set(PIL.Image.OPEN.keys() ) & set(PIL.Image.SAVE.keys() ) )
  return _IMAGE_COMPRESSION_FORMATS







def      lowerCamelCase__       (							_lowerCamelCase       :		"PIL.Image.Image" )      ->				bytes:
  lowerCamelCase_         =						BytesIO()
  if image.format in list_image_compression_formats():
    lowerCamelCase_         =						image.format
  else:
    lowerCamelCase_         =						'PNG' if image.mode in ['1', 'L', 'LA', 'RGB', 'RGBA'] else 'TIFF'
  image.save(_lowerCamelCase		,				format=_lowerCamelCase )
  return buffer.getvalue()







def      lowerCamelCase__       (							_lowerCamelCase       :		"PIL.Image.Image" )      ->				dict:
  if hasattr(_lowerCamelCase		,				'filename' ) and image.filename != "":
    return {"path": image.filename, "bytes": None}
  else:
    return {"path": None, "bytes": image_to_bytes(_lowerCamelCase )}







def      lowerCamelCase__       (							_lowerCamelCase       :		np.ndarray )      ->				dict:
  if config.PIL_AVAILABLE:
    import PIL.Image
  else:
    raise ImportError('To support encoding images, please install \'Pillow\'.' )

  lowerCamelCase_         =						array.dtype
  lowerCamelCase_         =						dtype.byteorder if dtype.byteorder != '=' else _NATIVE_BYTEORDER
  lowerCamelCase_         =						dtype.kind
  lowerCamelCase_         =						dtype.itemsize

  lowerCamelCase_         =						None

  # Multi-channel array case (only np.dtype("|u1") is allowed)
  if array.shape[2:]:
    lowerCamelCase_         =						np.dtype('|u1' )
    if dtype_kind not in ["u", "i"]:
      raise TypeError(
          F'''Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays.''' )
    if dtype is not dest_dtype:
      warnings.warn(F'''Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\'''' )
    # Exact match
  elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
    lowerCamelCase_         =						dtype
  else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
    while dtype_itemsize >= 1:
      lowerCamelCase_         =						dtype_byteorder + dtype_kind + str(_lowerCamelCase )
      lowerCamelCase_         =						np.dtype(_lowerCamelCase )
      if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
        warnings.warn(F'''Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\'''' )
        break
      else:
        dtype_itemsize //= 2

  if dest_dtype is None:
    raise TypeError(
        F'''Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}''' )
  lowerCamelCase_         =						PIL.Image.fromarray(array.astype(_lowerCamelCase ) )
  return {"path": None, "bytes": image_to_bytes(_lowerCamelCase )}







def      lowerCamelCase__       (							_lowerCamelCase       :		Union[List[str], List[dict], List[np.ndarray], List["PIL.Image.Image"]] )      ->				List[dict]:
  if config.PIL_AVAILABLE:
    import PIL.Image
  else:
    raise ImportError('To support encoding images, please install \'Pillow\'.' )

  if objs:
    lowerCamelCase_     ,	lowerCamelCase_         =						first_non_null_value(_lowerCamelCase )
    if isinstance(_lowerCamelCase		,				_lowerCamelCase ):
      return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
    if isinstance(_lowerCamelCase		,				np.ndarray ):
      lowerCamelCase_         =						no_op_if_value_is_null(_lowerCamelCase )
      return [obj_to_image_dict_func(_lowerCamelCase ) for obj in objs]
    elif isinstance(_lowerCamelCase		,				PIL.Image.Image ):
      lowerCamelCase_         =						no_op_if_value_is_null(_lowerCamelCase )
      return [obj_to_image_dict_func(_lowerCamelCase ) for obj in objs]
    else:
      return objs
  else:
    return objs












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




