


'''simple docstring'''




import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
       import tensorflow as tf

if is_torch_available():
       import torch


__snake_case    						=logging.get_logger(__name__)

class 		UpperCAmelCase_      :





     def __init__( self       :      Optional[int]    ,      UpperCAmelCase__       :      str = None    ,      UpperCAmelCase__       :      uuid.UUID = None    ,      UpperCAmelCase__       :      Any=None    ,      UpperCAmelCase__       :      List[str]=None    )				->  List[str]:






        if not conversation_id:
           lowerCAmelCase     =			uuid.uuida()
        if past_user_inputs is None:
           lowerCAmelCase     =			[]
        if generated_responses is None:
           lowerCAmelCase     =			[]

        lowerCAmelCase     =			conversation_id
        lowerCAmelCase     =			past_user_inputs
        lowerCAmelCase     =			generated_responses
        lowerCAmelCase     =			text





     def __eq__( self       :      Union[str, Any]    ,      UpperCAmelCase__       :      List[Any]    )				->  List[str]:






        if not isinstance(UpperCamelCase__    ,      UpperCamelCase__    ):
           return False
        if self.uuid == other.uuid:
           return True
        return (
            self.new_user_input == other.new_user_input
            and self.past_user_inputs == other.past_user_inputs
            and self.generated_responses == other.generated_responses
        )





     def 							__UpperCAmelCase      ( self       :      Optional[int]    ,      UpperCAmelCase__       :      str    ,      UpperCAmelCase__       :      bool = False    )				->  Tuple:






        if self.new_user_input:
           if overwrite:
              logger.warning(
                  F'''User input added while unprocessed input was existing: "{self.new_user_input}" was overwritten '''
                  F'''with: "{text}".'''    )
              lowerCAmelCase     =			text
           else:
              logger.warning(
                  F'''User input added while unprocessed input was existing: "{self.new_user_input}" new input '''
                  F'''ignored: "{text}". Set `overwrite` to True to overwrite unprocessed user input'''    )
        else:
           lowerCAmelCase     =			text





     def 							__UpperCAmelCase      ( self       :      Union[str, Any]    )				->  str:






        if self.new_user_input:
           self.past_user_inputs.append(self.new_user_input    )
        lowerCAmelCase     =			None





     def 							__UpperCAmelCase      ( self       :      Optional[Any]    ,      UpperCAmelCase__       :      str    )				->  Optional[int]:
        self.generated_responses.append(UpperCamelCase__    )





     def 							__UpperCAmelCase      ( self       :      Tuple    )				->  List[Any]:






        for user_input, generated_response in zip(self.past_user_inputs    ,      self.generated_responses    ):
           yield True, user_input
           yield False, generated_response
        if self.new_user_input:
           yield True, self.new_user_input





     def __repr__( self       :      Union[str, Any]    )				->  Tuple:
        lowerCAmelCase     =			F'''Conversation id: {self.uuid} \n'''
        for is_user, text in self.iter_texts():
           lowerCAmelCase     =			'user' if is_user else 'bot'
           output += F'''{name} >> {text} \n'''
        return output




@add_end_docstrings(
    _A       ,	r'''
        min_length_for_response (`int`, *optional*, defaults to 32):
            The minimum length (in number of tokens) for a response.
        minimum_tokens (`int`, *optional*, defaults to 10):
            The minimum length of tokens to leave for a response.
    '''       ,	)
class 		UpperCAmelCase_      (  _A		):





     def __init__( self       :      List[str]    ,      *UpperCAmelCase__       :      List[Any]    ,      **UpperCAmelCase__       :      List[Any]    )				->  str:
        super().__init__(*UpperCamelCase__    ,      **UpperCamelCase__    )
        if self.tokenizer.pad_token_id is None:
           lowerCAmelCase     =			self.tokenizer.eos_token





     def 							__UpperCAmelCase      ( self       :      str    ,      UpperCAmelCase__       :      Optional[Any]=None    ,      UpperCAmelCase__       :      Tuple=None    ,      UpperCAmelCase__       :      Tuple=None    ,      **UpperCAmelCase__       :      Tuple    )				->  Any:
        lowerCAmelCase     =			{}
        lowerCAmelCase     =			{}
        lowerCAmelCase     =			{}

        if min_length_for_response is not None:
           lowerCAmelCase     =			min_length_for_response
        if minimum_tokens is not None:
           lowerCAmelCase     =			minimum_tokens

        if "max_length" in generate_kwargs:
           lowerCAmelCase     =			generate_kwargs['max_length']
           # self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
        if clean_up_tokenization_spaces is not None:
           lowerCAmelCase     =			clean_up_tokenization_spaces

        if generate_kwargs:
           forward_params.update(UpperCamelCase__    )
        return preprocess_params, forward_params, postprocess_params





     def __call__( self       :      int    ,      UpperCAmelCase__       :      Union[Conversation, List[Conversation]]    ,      UpperCAmelCase__       :      Any=0    ,      **UpperCAmelCase__       :      List[Any]    )				->  str:
        lowerCAmelCase     =			super().__call__(UpperCamelCase__    ,      num_workers=UpperCamelCase__    ,      **UpperCamelCase__    )
        if isinstance(UpperCamelCase__    ,      UpperCamelCase__    ) and len(UpperCamelCase__    ) == 1:
           return outputs[0]
        return outputs





     def 							__UpperCAmelCase      ( self       :      List[Any]    ,      UpperCAmelCase__       :      Conversation    ,      UpperCAmelCase__       :      Union[str, Any]=3_2    )				->  Dict[str, Any]:






        if not isinstance(UpperCamelCase__    ,      UpperCamelCase__    ):
           raise ValueError('ConversationalPipeline, expects Conversation as inputs'    )
        if conversation.new_user_input is None:
           raise ValueError(
               F'''Conversation with UUID {type(conversation.uuid    )} does not contain new user input to process. '''
               'Add user inputs with the conversation\'s `add_user_input` method'    )
        if hasattr(self.tokenizer    ,      '_build_conversation_input_ids'    ):
           lowerCAmelCase     =			self.tokenizer._build_conversation_input_ids(UpperCamelCase__    )
        else:
           # If the tokenizer cannot handle conversations, we default to only the old version
           lowerCAmelCase     =			self._legacy_parse_and_tokenize(UpperCamelCase__    )

        if self.framework == "pt":
           lowerCAmelCase     =			torch.LongTensor([input_ids]    )
        elif self.framework == "tf":
           lowerCAmelCase     =			tf.constant([input_ids]    )
        return {"input_ids": input_ids, "conversation": conversation}





     def 							__UpperCAmelCase      ( self       :      Dict    ,      UpperCAmelCase__       :      Dict    ,      UpperCAmelCase__       :      List[str]=1_0    ,      **UpperCAmelCase__       :      int    )				->  Dict:
        lowerCAmelCase     =			generate_kwargs.get('max_length'    ,      self.model.config.max_length    )

        lowerCAmelCase     =			model_inputs['input_ids'].shape[1]
        if max_length - minimum_tokens < n:
           logger.warning(F'''Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})'''    )
           lowerCAmelCase     =			max_length - minimum_tokens
           lowerCAmelCase     =			model_inputs['input_ids'][:, -trim:]
           if "attention_mask" in model_inputs:
              lowerCAmelCase     =			model_inputs['attention_mask'][:, -trim:]
        lowerCAmelCase     =			model_inputs.pop('conversation'    )
        lowerCAmelCase     =			max_length
        lowerCAmelCase     =			self.model.generate(**UpperCamelCase__    ,      **UpperCamelCase__    )
        if self.model.config.is_encoder_decoder:
           lowerCAmelCase     =			1
        else:
           lowerCAmelCase     =			n
        return {"output_ids": output_ids[:, start_position:], "conversation": conversation}





     def 							__UpperCAmelCase      ( self       :      Optional[int]    ,      UpperCAmelCase__       :      Optional[int]    ,      UpperCAmelCase__       :      List[str]=True    )				->  str:
        lowerCAmelCase     =			model_outputs['output_ids']
        lowerCAmelCase     =			self.tokenizer.decode(
            output_ids[0]    ,      skip_special_tokens=UpperCamelCase__    ,      clean_up_tokenization_spaces=UpperCamelCase__    ,      )
        lowerCAmelCase     =			model_outputs['conversation']
        conversation.mark_processed()
        conversation.append_response(UpperCamelCase__    )
        return conversation





     def 							__UpperCAmelCase      ( self       :      List[Any]    ,      UpperCAmelCase__       :      Conversation    )				->  Dict:
        lowerCAmelCase     =			self.tokenizer.eos_token_id
        lowerCAmelCase     =			[]
        for is_user, text in conversation.iter_texts():
           if eos_token_id is not None:
              input_ids.extend(self.tokenizer.encode(UpperCamelCase__    ,      add_special_tokens=UpperCamelCase__    ) + [eos_token_id]    )
           else:
              input_ids.extend(self.tokenizer.encode(UpperCamelCase__    ,      add_special_tokens=UpperCamelCase__    )    )

        if len(UpperCamelCase__    ) > self.tokenizer.model_max_length:
           lowerCAmelCase     =			input_ids[-self.tokenizer.model_max_length :]
        return input_ids







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


__lowerCAmelCase			:     Optional[int]           =   ['\nclass', '\ndef', '\n#', '\n@', '\nprint', '\nif']




class        UpperCAmelCase_			( _A     ):


  '''simple docstring'''







  def __init__( self			: Union[str, Any]			,							UpperCamelCase__			: List[Any]			,							UpperCamelCase__			: Optional[int]			,							UpperCamelCase__			: Any=None			,							UpperCamelCase__			: Union[str, Any]=1    )			->	str:







    """simple docstring"""






    __magic_name__				     =       tokenizer
    __magic_name__				     =       dataset
    __magic_name__				     =       len(UpperCamelCase__    ) if n_tasks is None else n_tasks
    __magic_name__				     =       n_copies





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







    """simple docstring"""






    __magic_name__				     =       []
    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()    )
    __magic_name__				     =       self.tokenizer(UpperCamelCase__			,							padding=UpperCamelCase__			,							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_			( _A     ):


  '''simple docstring'''







  def __init__( self			: Union[str, Any]			,							UpperCamelCase__			: Dict			,							UpperCamelCase__			: str			,							UpperCamelCase__			: str    )			->	List[str]:







    """simple docstring"""






    __magic_name__				     =       start_length
    __magic_name__				     =       eof_strings
    __magic_name__				     =       tokenizer





  def __call__( self			: List[Any]			,							UpperCamelCase__			: Tuple			,							UpperCamelCase__			: Optional[Any]			,							**UpperCamelCase__			: Optional[int]    )			->	List[str]:







    """simple docstring"""






    __magic_name__				     =       self.tokenizer.batch_decode(input_ids[:, self.start_length :]    )
    __magic_name__				     =       []
    for decoded_generation in decoded_generations:
      done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings    )    )
    return all(UpperCamelCase__    )







def a__    (       A_					):






  '''simple docstring'''





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

def a__    (       A_,     A_,     A_,     A_,     A_,     A_=20,     **A_					):






  '''simple docstring'''





  __magic_name__				     =       defaultdict(A_					)  # dict of list of generated tokens
  for step, batch in tqdm(enumerate(A_					)					):
    with torch.no_grad():
      __magic_name__				     =       batch["""ids"""].shape[-1]
      __magic_name__				     =       accelerator.unwrap_model(A_					).generate(
          input_ids=batch["""ids"""][:, : batch["""input_len"""]],     num_return_sequences=A_,     **A_					)
      # each task is generated batch_size times
      __magic_name__				     =       batch["""task_id"""].repeat(A_					)
      __magic_name__				     =       accelerator.pad_across_processes(
          A_,     dim=1,     pad_index=tokenizer.pad_token_id					)

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

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

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

def a__    (       ):






  '''simple docstring'''





  __magic_name__				     =       HfArgumentParser(A_					)
  __magic_name__				     =       parser.parse_args()

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

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

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

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

  # Generation settings
  __magic_name__				     =       {
      """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,     A_,     A_					)]					),
  }

  # Load evaluation dataset and metric
  __magic_name__				     =       load_dataset("""openai_humaneval"""					)
  __magic_name__				     =       load_metric("""code_eval"""					)

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

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

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

  __magic_name__     ,    __magic_name__				     =       accelerator.prepare(A_,     A_					)

  __magic_name__				     =       complete_code(
      A_,     A_,     A_,     A_,     n_tasks=A_,     batch_size=args.batch_size,     **A_,     )

  if accelerator.is_main_process:
    __magic_name__				     =       []

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

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


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

