
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         =   ['\nclass', '\ndef', '\n#', '\n@', '\nprint', '\nif']
class 					lowercase__(							_a				):



	"""simple docstring"""




	def __init__(			self    :     str			,				SCREAMING_SNAKE_CASE_    :     Tuple			,				SCREAMING_SNAKE_CASE_    :     Any			,				SCREAMING_SNAKE_CASE_    :     List[str]=None			,				SCREAMING_SNAKE_CASE_    :     Dict=1			)			->	Dict:
					lowercase_	    =   tokenizer
					lowercase_	    =   dataset
					lowercase_	    =   len(_SCREAMING_SNAKE_CASE			) if n_tasks is None else n_tasks
					lowercase_	    =   n_copies




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



	"""simple docstring"""




	def __init__(			self    :     Optional[Any]			,				SCREAMING_SNAKE_CASE_    :     Optional[Any]			,				SCREAMING_SNAKE_CASE_    :     Dict			,				SCREAMING_SNAKE_CASE_    :     List[Any]			)			->	List[Any]:
					lowercase_	    =   start_length
					lowercase_	    =   eof_strings
					lowercase_	    =   tokenizer




	def __call__(			self    :     int			,				SCREAMING_SNAKE_CASE_    :     List[str]			,				SCREAMING_SNAKE_CASE_    :     Any			,				**SCREAMING_SNAKE_CASE_    :     Union[str, Any]			)			->	List[Any]:
					lowercase_	    =   self.tokenizer.batch_decode(input_ids[:, self.start_length :]			)
					lowercase_	    =   []
					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		(				snake_case__:   Dict   ):


				'''simple docstring'''
				lowercase_	    =   re.split('''(%s)''' % '''|'''.join(_lowercase   )		,  _lowercase   )
				# last string should be ""
				return "".join(string_list[:-2]   )




def 							a		(				snake_case__:   int		,  snake_case__:   str		,  snake_case__:   List[str]		,  snake_case__:   Optional[Any]		,  snake_case__:   List[str]		,  snake_case__:   str=20		,  **snake_case__:   int   ):


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

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

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

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




def 							a		(				):


				'''simple docstring'''
				lowercase_	    =   HfArgumentParser(_lowercase   )
				lowercase_	    =   parser.parse_args()

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

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

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

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

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

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

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

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

				lowercase_	    =   accelerator.prepare(_lowercase		,  _lowercase   )

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

				if accelerator.is_main_process:
								lowercase_	    =   []

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

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










import unittest

import numpy as np







def lowerCamelCase__  (     _lowercase					,		_lowercase					,		_lowercase					,		_lowercase = None					,		):

     '''simple docstring'''


     UpperCAmelCase_     :				Dict  =     np.shape(_lowercase )
     UpperCAmelCase_     :				Optional[Any]  =     np.shape(_lowercase )
     UpperCAmelCase_     :				Tuple  =     np.shape(_lowercase )

     if shape_a[0] != shape_b[0]:
          UpperCAmelCase_     :				Tuple  =     (
              '''Expected the same number of rows for A and B. '''
              f'''Instead found A of size {shape_a} and B of size {shape_b}'''
          )
          raise ValueError(_lowercase )

     if shape_b[1] != shape_c[1]:
          UpperCAmelCase_     :				List[Any]  =     (
              '''Expected the same number of columns for B and C. '''
              f'''Instead found B of size {shape_b} and C of size {shape_c}'''
          )
          raise ValueError(_lowercase )

     UpperCAmelCase_     :				Dict  =     pseudo_inv
     if a_inv is None:
          try:
               UpperCAmelCase_     :				Any  =     np.linalg.inv(_lowercase )
          except np.linalg.LinAlgError:
               raise ValueError(
                   '''Input matrix A is not invertible. Cannot compute Schur complement.''' )

     return mat_c - mat_b.T @ a_inv @ mat_b

class     __a(	unittest.TestCase  ):






    """simple docstring"""





    def 				a__ ( self    )					->						None:
         UpperCAmelCase_     :				str  =     np.array([[1, 2, 1], [2, 1, 2], [3, 2, 4]]    )
         UpperCAmelCase_     :				Any  =     np.array([[0, 3], [3, 0], [2, 3]]    )
         UpperCAmelCase_     :				List[str]  =     np.array([[2, 1], [6, 3]]    )

         UpperCAmelCase_     :				Tuple  =     schur_complement(_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_     :				List[Any]  =     np.block([[a, b], [b.T, c]]    )

         UpperCAmelCase_     :				List[Any]  =     np.linalg.det(_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_     :				List[Any]  =     np.linalg.det(_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_     :				Any  =     np.linalg.det(_SCREAMING_SNAKE_CASE    )

         self.assertAlmostEqual(_SCREAMING_SNAKE_CASE  ,det_a * det_s    )





    def 				a__ ( self    )					->						None:
         UpperCAmelCase_     :				str  =     np.array([[1, 2, 1], [2, 1, 2], [3, 2, 4]]    )
         UpperCAmelCase_     :				Optional[int]  =     np.array([[0, 3], [3, 0], [2, 3]]    )
         UpperCAmelCase_     :				Optional[int]  =     np.array([[2, 1], [6, 3]]    )

         with self.assertRaises(_SCREAMING_SNAKE_CASE    ):
              schur_complement(_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE    )





    def 				a__ ( self    )					->						None:
         UpperCAmelCase_     :				Optional[int]  =     np.array([[1, 2, 1], [2, 1, 2], [3, 2, 4]]    )
         UpperCAmelCase_     :				Optional[Any]  =     np.array([[0, 3], [3, 0], [2, 3]]    )
         UpperCAmelCase_     :				int  =     np.array([[2, 1, 3], [6, 3, 5]]    )

         with self.assertRaises(_SCREAMING_SNAKE_CASE    ):
              schur_complement(_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE    )


if __name__ == "__main__":
     import doctest

     doctest.testmod()
     unittest.main()