




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




class      UpperCAmelCase_  (						A_							):




    def __init__(   self						:	int     ,			snake_case_						:	Optional[Any]     ,			snake_case_						:	Any     ,			snake_case_						:	Tuple=None     ,			snake_case_						:	Optional[int]=1				)				->							Tuple:




     '''simple docstring'''

     A__	=    tokenizer
     A__	=    dataset
     A__	=    len(snake_case_				) if n_tasks is None else n_tasks
     A__	=    n_copies




    def __iter__(   self						:	Optional[int]				)				->							str:




     '''simple docstring'''

     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(snake_case_     ,			padding=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      UpperCAmelCase_  (						A_							):




    def __init__(   self						:	Tuple     ,			snake_case_						:	Optional[Any]     ,			snake_case_						:	List[Any]     ,			snake_case_						:	int				)				->							Union[str, Any]:




     '''simple docstring'''

     A__	=    start_length
     A__	=    eof_strings
     A__	=    tokenizer




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




     '''simple docstring'''

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







def   _SCREAMING_SNAKE_CASE       (							lowercase_						)						->		str:
 A__	=    re.split("(%s)" % "|".join(lowercase_						)      ,			lowercase_						)
 # last string should be ""
 return "".join(string_list[:-2]						)







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

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

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

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







def   _SCREAMING_SNAKE_CASE       (							)						->		str:
 # Setup configuration
 A__	=    HfArgumentParser(lowercase_						)
 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=lowercase_						)

 # 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      ,			lowercase_      ,			lowercase_						)]						),
 }

 # 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(lowercase_      ,			human_eval["test"]      ,			n_copies=lowercase_      ,			n_tasks=lowercase_						)
 # do not confuse args.batch_size, which is actually the num_return_sequences
 A__	=    DataLoader(lowercase_      ,			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__,				A__	=    accelerator.prepare(lowercase_      ,			lowercase_						)

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

 if accelerator.is_main_process:
  A__	=    []

  for task in tqdm(range(lowercase_						)						):
   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__,				A__	=    code_eval_metric.compute(
      references=lowercase_      ,			predictions=lowercase_      ,			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(lowercase_      ,			lowercase_						)


# 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 __future__ import annotations

SCREAMING_SNAKE_CASE    =	"#"




class      UpperCAmelCase_  :




    def __init__(   self						:	Dict				)				->							None:




     '''simple docstring'''

     A__	=    {}




    def 							__magic_name__							(   self						:	Optional[Any]     ,			snake_case_						:	str				)				->							None:




     '''simple docstring'''

     A__	=    self._trie
     for char in text:
      if char not in trie:
       A__	=    {}
      A__	=    trie[char]
     A__	=    True




    def 							__magic_name__							(   self						:	List[Any]     ,			snake_case_						:	str				)				->							tuple | list:




     '''simple docstring'''

     A__	=    self._trie
     for char in prefix:
      if char in trie:
       A__	=    trie[char]
      else:
       return []
     return self._elements(snake_case_				)




    def 							__magic_name__							(   self						:	Union[str, Any]     ,			snake_case_						:	dict				)				->							tuple:




     '''simple docstring'''

     A__	=    []
     for c, v in d.items():
      A__	=    [" "] if c == END else [(c + s) for s in self._elements(snake_case_				)]
      result.extend(snake_case_				)
     return tuple(snake_case_				)






SCREAMING_SNAKE_CASE    =	Trie()
SCREAMING_SNAKE_CASE    =	("depart", "detergent", "daring", "dog", "deer", "deal")
for word in words:
      trie.insert_word(word)







def   _SCREAMING_SNAKE_CASE       (							lowercase_						)						->		tuple:
 A__	=    trie.find_word(lowercase_						)
 return tuple(string + word for word in suffixes						)







def   _SCREAMING_SNAKE_CASE       (							)						->		None:
 print(autocomplete_using_trie("de"						)						)


if __name__ == "__main__":
      import doctest

      doctest.testmod()
      main()





