




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


class    lowerCamelCase    (		_lowerCamelCase						):






   '''simple docstring'''





   def __init__( self			:   Optional[int]       ,						lowerCamelCase_			:   Any       ,						lowerCamelCase_			:   int       ,						lowerCamelCase_			:   List[Any]=None       ,						lowerCamelCase_			:   Optional[Any]=1  )						->   List[str]:
        __magic_name__							:	Any    							=							tokenizer
        __magic_name__							:	Tuple    							=							dataset
        __magic_name__							:	Optional[Any]    							=							len(lowerCamelCase_  ) if n_tasks is None else n_tasks
        __magic_name__							:	Any    							=							n_copies


   def __iter__( self			:   Any  )						->   Union[str, Any]:
        __magic_name__							:	Optional[int]    							=							[]
        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__							:	int    							=							self.tokenizer(lowerCamelCase_       ,						padding=lowerCamelCase_       ,						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    lowerCamelCase    (		_lowerCamelCase						):






   '''simple docstring'''





   def __init__( self			:   int       ,						lowerCamelCase_			:   str       ,						lowerCamelCase_			:   Union[str, Any]       ,						lowerCamelCase_			:   Tuple  )						->   Optional[int]:
        __magic_name__							:	Optional[int]    							=							start_length
        __magic_name__							:	Optional[int]    							=							eof_strings
        __magic_name__							:	Any    							=							tokenizer


   def __call__( self			:   Union[str, Any]       ,						lowerCamelCase_			:   Union[str, Any]       ,						lowerCamelCase_			:   Optional[int]       ,						**lowerCamelCase_			:   List[Any]  )						->   Any:
        __magic_name__							:	List[Any]    							=							self.tokenizer.batch_decode(input_ids[:, self.start_length :]  )
        __magic_name__							:	List[Any]    							=							[]
        for decoded_generation in decoded_generations:
             done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings  )  )
        return all(lowerCamelCase_  )







def 		lowercase__   (		__A:	List[Any]				):





     '''simple docstring'''






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







def 		lowercase__   (		__A:	str							,__A:	Optional[Any]							,__A:	int							,__A:	str							,__A:	Union[str, Any]							,__A:	Optional[int]=2_0							,**__A:	Tuple				):





     '''simple docstring'''






     __magic_name__							:	Optional[Any]    							=							defaultdict(__A				)  # dict of list of generated tokens
     for step, batch in tqdm(enumerate(__A				)				):
          with torch.no_grad():
               __magic_name__							:	Any    							=							batch['''ids'''].shape[-1]
               __magic_name__							:	Union[str, Any]    							=							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__							:	Optional[Any]    							=							batch['''task_id'''].repeat(__A				)
               __magic_name__							:	Union[str, Any]    							=							accelerator.pad_across_processes(
                   __A							,dim=1							,pad_index=tokenizer.pad_token_id				)

               __magic_name__							:	Union[str, Any]    							=							accelerator.gather((generated_tokens, generated_tasks)				)
               __magic_name__							:	List[str]    							=							generated_tokens.cpu().numpy()
               __magic_name__							:	Any    							=							generated_tasks.cpu().numpy()

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

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







def 		lowercase__   (		):





     '''simple docstring'''






     __magic_name__							:	List[Any]    							=							HfArgumentParser(__A				)
     __magic_name__							:	Optional[Any]    							=							parser.parse_args()

     transformers.logging.set_verbosity_error()
     # enables code execution in code_eval metric
     __magic_name__							:	Union[str, Any]    							=							args.HF_ALLOW_CODE_EVAL
     # make sure tokenizer plays nice with multiprocessing
     __magic_name__							:	Any    							=							'''false'''

     if args.num_workers is None:
          __magic_name__							:	List[str]    							=							multiprocessing.cpu_count()

     # Use dataset load to feed to accelerate
     __magic_name__							:	Optional[Any]    							=							Accelerator()
     set_seed(args.seed							,device_specific=__A				)

     # Load model and tokenizer
     __magic_name__							:	int    							=							AutoTokenizer.from_pretrained(args.model_ckpt				)
     __magic_name__							:	Optional[Any]    							=							tokenizer.eos_token
     __magic_name__							:	List[str]    							=							AutoModelForCausalLM.from_pretrained(args.model_ckpt				)

     # Generation settings
     __magic_name__							:	Union[str, Any]    							=							{
         '''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__							:	Dict    							=							load_dataset('''openai_humaneval'''				)
     __magic_name__							:	List[Any]    							=							load_metric('''code_eval'''				)

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

     __magic_name__							:	Optional[Any]    							=							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__							:	Optional[Any]    							=							DataLoader(__A							,batch_size=1				)

     # Run a quick test to see if code evaluation is enabled
     try:
          __magic_name__							:	str    							=							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__							:	str    							=							accelerator.prepare(__A							,__A				)

     __magic_name__							:	List[str]    							=							complete_code(
         __A							,__A							,__A							,__A							,n_tasks=__A							,batch_size=args.batch_size							,**__A							,)

     if accelerator.is_main_process:
          __magic_name__							:	Tuple    							=							[]

          for task in tqdm(range(__A				)				):
               __magic_name__							:	str    							=							human_eval['''test'''][task]['''test''']
               __magic_name__							:	List[Any]    							=							F'''check({human_eval['test'][task]['entry_point']})'''
               references.append('''\n''' + test_func + '''\n''' + entry_point				)

          # Evaluate completions with "code_eval" metric
          __magic_name__							:	Optional[int]    							=							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()









import inspect
import os
import unittest
from pathlib import Path

import torch

import accelerate
from accelerate.test_utils import execute_subprocess_async
from accelerate.test_utils.testing import run_command


class    lowerCamelCase    (		unittest.TestCase						):






			'''simple docstring'''





			UpperCamelCase__								=inspect.getfile(accelerate.test_utils						)
			UpperCamelCase__								=os.path.sep.join(mod_file.split(os.path.sep						)[:-1] + ['''scripts''', '''test_cli.py''']						)

			UpperCamelCase__								=['''accelerate''', '''launch''']
			UpperCamelCase__								=Path.home() / '''.cache/huggingface/accelerate'''
			UpperCamelCase__								='''default_config.yaml'''
			UpperCamelCase__								=config_folder / config_file
			UpperCamelCase__								=config_folder / '''_default_config.yaml'''

			UpperCamelCase__								=Path('''tests/test_configs'''						)

			@classmethod
			def 							UpperCAmelCase__       ( cls			:   Optional[int]  )						->   int:
								if cls.config_path.is_file():
													cls.config_path.rename(cls.changed_path  )

			@classmethod
			def 							UpperCAmelCase__       ( cls			:   Dict  )						->   List[str]:
								if cls.changed_path.is_file():
													cls.changed_path.rename(cls.config_path  )

			def 							UpperCAmelCase__       ( self			:   str  )						->   Optional[Any]:
								__magic_name__							:	Dict    							=							self.base_cmd
								if torch.cuda.is_available() and (torch.cuda.device_count() > 1):
													cmd += ["--multi_gpu"]
								execute_subprocess_async(cmd + [self.test_file_path]       ,						env=os.environ.copy()  )

			def 							UpperCAmelCase__       ( self			:   int  )						->   Optional[int]:
								for config in sorted(self.test_config_path.glob('''**/*.yaml'''  )  ):
													with self.subTest(config_file=lowerCamelCase_  ):
																		execute_subprocess_async(
																		    self.base_cmd + ['''--config_file''', str(lowerCamelCase_  ), self.test_file_path]       ,						env=os.environ.copy()  )


			def 							UpperCAmelCase__       ( self			:   int  )						->   Union[str, Any]:
								execute_subprocess_async(['''accelerate''', '''test''']       ,						env=os.environ.copy()  )






class    lowerCamelCase    (		unittest.TestCase						):






			'''simple docstring'''





			UpperCamelCase__								='''test-tpu'''
			UpperCamelCase__								='''us-central1-a'''
			UpperCamelCase__								='''ls'''
			UpperCamelCase__								=['''accelerate''', '''tpu-config''']
			UpperCamelCase__								='''cd /usr/share'''
			UpperCamelCase__								='''tests/test_samples/test_command_file.sh'''
			UpperCamelCase__								='''Running gcloud compute tpus tpu-vm ssh'''

			def 							UpperCAmelCase__       ( self			:   Optional[Any]  )						->   Optional[Any]:
								__magic_name__							:	str    							=							run_command(
								    self.cmd
								    + ['''--command''', self.command, '''--tpu_zone''', self.tpu_zone, '''--tpu_name''', self.tpu_name, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Union[str, Any]  )						->   int:
								__magic_name__							:	int    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/0_12_0.yaml''',
								        '''--command''',
								        self.command,
								        '''--tpu_zone''',
								        self.tpu_zone,
								        '''--tpu_name''',
								        self.tpu_name,
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   List[Any]  )						->   Tuple:
								__magic_name__							:	Any    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--debug''']       ,						return_stdout=lowerCamelCase_  )
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Optional[int]  )						->   Tuple:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--command''', self.command, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   List[Any]  )						->   Dict:
								__magic_name__							:	str    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/latest.yaml''',
								        '''--command''',
								        self.command,
								        '''--command''',
								        '''echo "Hello World"''',
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls; echo "Hello World" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Optional[Any]  )						->   Optional[int]:
								__magic_name__							:	Union[str, Any]    							=							run_command(
								    self.cmd
								    + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--command_file''', self.command_file, '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   int  )						->   List[Any]:
								__magic_name__							:	List[Any]    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/0_12_0.yaml''',
								        '''--command_file''',
								        self.command_file,
								        '''--tpu_zone''',
								        self.tpu_zone,
								        '''--tpu_name''',
								        self.tpu_name,
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   str  )						->   Dict:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd + ['''--config_file''', '''tests/test_configs/latest.yaml''', '''--install_accelerate''', '''--debug''']       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate -U; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)


			def 							UpperCAmelCase__       ( self			:   Any  )						->   Tuple:
								__magic_name__							:	Tuple    							=							run_command(
								    self.cmd
								    + [
								        '''--config_file''',
								        '''tests/test_configs/latest.yaml''',
								        '''--install_accelerate''',
								        '''--accelerate_version''',
								        '''12.0.0''',
								        '''--debug''',
								    ]       ,						return_stdout=lowerCamelCase_       ,						)
								self.assertIn(
								    F'''{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate==12.0.0; echo "hello world"; echo "this is a second command" --worker all'''       ,						lowerCamelCase_       ,						)



