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


_a														=		["""\nclass""", """\ndef""", """\n#""", """\n@""", """\nprint""", """\nif"""]






class     _UpperCAmelCase(    lowerCamelCase	):



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






							'''simple docstring'''
							_UpperCamelCase							 =  tokenizer
							_UpperCamelCase							 =  dataset
							_UpperCamelCase							 =  len(__a) if n_tasks is None else n_tasks
							_UpperCamelCase							 =  n_copies




		def __iter__(					self)       ->       Dict:






							'''simple docstring'''
							_UpperCamelCase							 =  []
							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())
							_UpperCamelCase							 =  self.tokenizer(__a				,		padding=__a				,		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(    lowerCamelCase	):



		def __init__(					self				,		__a				,		__a				,		__a)       ->       int:






							'''simple docstring'''
							_UpperCamelCase							 =  start_length
							_UpperCamelCase							 =  eof_strings
							_UpperCamelCase							 =  tokenizer




		def __call__(					self				,		__a				,		__a				,		**__a)       ->       List[str]:






							'''simple docstring'''
							_UpperCamelCase							 =  self.tokenizer.batch_decode(input_ids[:, self.start_length :])
							_UpperCamelCase							 =  []
							for decoded_generation in decoded_generations:
												done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings))
							return all(__a)





def       lowerCamelCase__		(			__snake_case	)							->		str:


					"""simple docstring"""





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



def       lowerCamelCase__		(			__snake_case,     __snake_case,     __snake_case,     __snake_case,     __snake_case,     __snake_case=20,     **__snake_case	)							->		Optional[int]:


					"""simple docstring"""





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

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

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

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



def       lowerCamelCase__		(			)							->		Any:


					"""simple docstring"""





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

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

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

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

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

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

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

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

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

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

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

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

					if accelerator.is_main_process:
										_UpperCamelCase							 =  []

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

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


# 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"""





def       lowerCamelCase__		(			__snake_case = 10_00	)							->		int:


					"""simple docstring"""





					_UpperCamelCase							 =  2**power
					_UpperCamelCase							 =  str(__snake_case	)
					_UpperCamelCase							 =  list(__snake_case	)
					_UpperCamelCase							 =  0

					for i in list_num:
										sum_of_num += int(__snake_case	)

					return sum_of_num


if __name__ == "__main__":
				_a														=		int(input("""Enter the power of 2: """).strip())
				print("""2 ^ """, power, """ = """, 2**power)
				_a														=		solution(power)
				print("""Sum of the digits is: """, result)






