




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


UpperCamelCase_		:     Union[str, Any]          =		['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']

class   _a							(		__lowerCAmelCase				):






     def __init__(					self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE=None   ,_SCREAMING_SNAKE_CASE=1       )				->     Dict:
          _snake_case				= tokenizer
          _snake_case				= dataset
          _snake_case				= len(_SCREAMING_SNAKE_CASE       ) if n_tasks is None else n_tasks
          _snake_case				= n_copies






     def __iter__(					self       )				->     Union[str, Any]:
          _snake_case				= []
          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()       )
          _snake_case				= 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							(		__lowerCAmelCase				):






     def __init__(					self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE       )				->     List[Any]:
          _snake_case				= start_length
          _snake_case				= eof_strings
          _snake_case				= tokenizer






     def __call__(					self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE   ,**_SCREAMING_SNAKE_CASE       )				->     List[Any]:
          _snake_case				= self.tokenizer.batch_decode(input_ids[:, self.start_length :]       )
          _snake_case				= []
          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       __a      (  _UpperCamelCase:       List[Any]       )		->					List[str]:

     """simple docstring"""




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





def       __a      (  _UpperCamelCase:       List[Any]   ,		_UpperCamelCase:       int   ,		_UpperCamelCase:       str   ,		_UpperCamelCase:       List[str]   ,		_UpperCamelCase:       Optional[int]   ,		_UpperCamelCase:       Union[str, Any]=20   ,		**_UpperCamelCase:       Union[str, Any]       )		->					Optional[Any]:

     """simple docstring"""




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

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

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

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





def       __a      (  )		->					int:

     """simple docstring"""




     _snake_case				= HfArgumentParser(_UpperCamelCase       )
     _snake_case				= parser.parse_args()

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

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

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

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

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

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

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

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

     _snake_case				,   _snake_case				= accelerator.prepare(_UpperCamelCase   ,		_UpperCamelCase       )

     _snake_case				= complete_code(
         _UpperCamelCase   ,		_UpperCamelCase   ,		_UpperCamelCase   ,		_UpperCamelCase   ,		n_tasks=_UpperCamelCase   ,		batch_size=args.batch_size   ,		**_UpperCamelCase   ,		)

     if accelerator.is_main_process:
          _snake_case				= []

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

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








'''simple docstring'''





from typing import Any

class   _a							:






     def __init__(					self   ,_SCREAMING_SNAKE_CASE       )				->     List[str]:
          _snake_case				= data
          _snake_case				= None







class   _a							:






     def __init__(					self       )				->     List[Any]:
          _snake_case				= None






     def 							_lowercase      (					self       )				->     str:
          _snake_case				= self.head
          while temp is not None:
               print(temp.data   ,end=" "       )
               _snake_case				= temp.next
          print()






     def 							_lowercase      (					self   ,_SCREAMING_SNAKE_CASE       )				->     Dict:
          _snake_case				= Node(_SCREAMING_SNAKE_CASE       )
          _snake_case				= self.head
          _snake_case				= new_node






     def 							_lowercase      (					self   ,_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE       )				->     Tuple:
          if node_data_a == node_data_a:
               return
          else:
               _snake_case				= self.head
               while node_a is not None and node_a.data != node_data_a:
                    _snake_case				= node_a.next

               _snake_case				= self.head
               while node_a is not None and node_a.data != node_data_a:
                    _snake_case				= node_a.next

               if node_a is None or node_a is None:
                    return

               _snake_case				,   _snake_case				= node_a.data, node_a.data


if __name__ == "__main__":
   UpperCamelCase_		:     str          =		LinkedList()
   for i in range(5, 0, -1):
      ll.push(i)

   ll.print_list()

   ll.swap_nodes(1, 4)
   print('''After swapping''')
   ll.print_list()


