





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


a__:      Any									=		['\nclass', '\ndef', '\n#', '\n@', '\nprint', '\nif']




class     SCREAMING_SNAKE_CASE__ (						UpperCAmelCase_      ):



 def __init__(							self,__lowerCamelCase,__lowerCamelCase,__lowerCamelCase=None,__lowerCamelCase=1 ):
        A__											=						tokenizer
        A__											=						dataset
        A__											=						len(__lowercase ) if n_tasks is None else n_tasks
        A__											=						n_copies




 def __iter__(							self ):
        A__											=						[]
        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() )
        A__											=						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     SCREAMING_SNAKE_CASE__ (						UpperCAmelCase_      ):



 def __init__(							self,__lowerCamelCase,__lowerCamelCase,__lowerCamelCase ):
        A__											=						start_length
        A__											=						eof_strings
        A__											=						tokenizer




 def __call__(							self,__lowerCamelCase,__lowerCamelCase,**__lowerCamelCase ):
        A__											=						self.tokenizer.batch_decode(input_ids[:, self.start_length :] )
        A__											=						[]
        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 	UpperCamelCase__(       UpperCamelCase__			:       Optional[Any]  )->str:
       A__											=						re.split('''(%s)''' % '''|'''.join(UpperCamelCase__  ) ,     UpperCamelCase__  )
       # last string should be ""
       return "".join(string_list[:-2]  )


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

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

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

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


def 	UpperCamelCase__(       )->int:
       A__											=						HfArgumentParser(UpperCamelCase__  )
       A__											=						parser.parse_args()

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

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

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

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

       # Generation settings
       A__											=						{
           '''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 ,     UpperCamelCase__ ,     UpperCamelCase__  )]  ),
       }

       # Load evaluation dataset and metric
       A__											=						load_dataset('''openai_humaneval'''  )
       A__											=						load_metric('''code_eval'''  )

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

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

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

       A__											=						accelerator.prepare(UpperCamelCase__ ,     UpperCamelCase__  )

       A__											=						complete_code(
           UpperCamelCase__ ,     UpperCamelCase__ ,     UpperCamelCase__ ,     UpperCamelCase__ ,     n_tasks=UpperCamelCase__ ,     batch_size=args.batch_size ,     **UpperCamelCase__ ,     )

       if accelerator.is_main_process:
              A__											=						[]

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

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


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






# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch

from ..models.auto import AutoModelForSequenceClassification, AutoTokenizer
from .base import PipelineTool


class 	lowerCamelCase_     (   UpperCAmelCase_ ):






    '''simple docstring'''






    a__	:						Optional[Any]						=      """facebook/bart-large-mnli"""
    a__	:						int						=      (
        """This is a tool that classifies an English text using provided labels. It takes two inputs: `text`, which """
        """should be the text to classify, and `labels`, which should be the list of labels to use for classification. """
        """It returns the most likely label in the list of provided `labels` for the input text."""
    )
    a__	:						Optional[Any]						=      """text_classifier"""
    a__	:						Any						=      AutoTokenizer
    a__	:						str						=      AutoModelForSequenceClassification

    a__	:						str						=      ["""text""", ["""text"""]]
    a__	:						Optional[int]						=      ["""text"""]







    def 			UpperCamelCase__   (	self)       ->	Union[str, Any]:
        super().setup()
        __UpperCamelCase     :int			  =		self.model.config
        __UpperCamelCase     :Optional[Any]			  =		-1
        for idx, label in config.idalabel.items():
            if label.lower().startswith('''entail'''):
                __UpperCamelCase     :List[Any]			  =		int(__lowercase)
        if self.entailment_id == -1:
            raise ValueError('''Could not determine the entailment ID from the model config, please pass it at init.''')







    def 			UpperCamelCase__   (	self , __lowercase , __lowercase)       ->	Union[str, Any]:
        __UpperCamelCase     :Any			  =		labels
        return self.pre_processor(
            [text] * len(__lowercase) , [f"""This example is {label}""" for label in labels] , return_tensors='''pt''' , padding='''max_length''' , )







    def 			UpperCamelCase__   (	self , __lowercase)       ->	Optional[Any]:
        __UpperCamelCase     :List[Any]			  =		outputs.logits
        __UpperCamelCase     :Any			  =		torch.argmax(logits[:, 2]).item()
        return self._labels[label_id]





