






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


lowercase  :							Tuple			  =    ['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']




class 		UpperCAmelCase_						(	SCREAMING_SNAKE_CASE__					):


						'''simple docstring'''




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






						def __iter__(	self				) ->				Union[str, Any]:
								snake_case_				:							Optional[Any]						 =       []
								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_				:							Optional[Any]						 =       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_						(	SCREAMING_SNAKE_CASE__					):


						'''simple docstring'''




						def __init__(	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				) ->				Optional[Any]:
								snake_case_				:							int						 =       start_length
								snake_case_				:							Union[str, Any]						 =       eof_strings
								snake_case_				:							List[str]						 =       tokenizer






						def __call__(	self		,				_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE		,				**_SCREAMING_SNAKE_CASE				) ->				List[Any]:
								snake_case_				:							Optional[Any]						 =       self.tokenizer.batch_decode(input_ids[:, self.start_length :]				)
								snake_case_				:							Optional[int]						 =       []
								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__	(    _a       :			Dict    ):
		snake_case_				:							int						 =       re.split("(%s)" % "|".join(__snake_case    )  ,      __snake_case    )
		# last string should be ""
		return "".join(string_list[:-2]    )

def      lowerCAmelCase__	(    _a       :			Optional[int]  ,      _a       :			Dict  ,      _a       :			Any  ,      _a       :			List[str]  ,      _a       :			int  ,      _a       :			Tuple=20  ,      **_a       :			List[str]    ):
		snake_case_				:							Any						 =       defaultdict(__snake_case    )  # dict of list of generated tokens
		for step, batch in tqdm(enumerate(__snake_case    )    ):
				with torch.no_grad():
						snake_case_				:							int						 =       batch["ids"].shape[-1]
						snake_case_				:							Tuple						 =       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
						snake_case_				:							Tuple						 =       batch["task_id"].repeat(__snake_case    )
						snake_case_				:							Optional[int]						 =       accelerator.pad_across_processes(
						    __snake_case  ,      dim=1  ,      pad_index=tokenizer.pad_token_id    )

						snake_case_ ,       snake_case_				:							Optional[int]						 =       accelerator.gather((generated_tokens, generated_tasks)    )
						snake_case_				:							Tuple						 =       generated_tokens.cpu().numpy()
						snake_case_				:							List[Any]						 =       generated_tasks.cpu().numpy()

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

		snake_case_				:							Optional[int]						 =       [[] for _ in range(__snake_case    )]
		for task, generated_tokens in gen_token_dict.items():
				for s in generated_tokens:
						snake_case_				:							str						 =       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__	(    ):
		snake_case_				:							Optional[Any]						 =       HfArgumentParser(__snake_case    )
		snake_case_				:							Tuple						 =       parser.parse_args()

		transformers.logging.set_verbosity_error()
		# enables code execution in code_eval metric
		snake_case_				:							Tuple						 =       args.HF_ALLOW_CODE_EVAL
		# make sure tokenizer plays nice with multiprocessing
		snake_case_				:							List[str]						 =       "false"

		if args.num_workers is None:
				snake_case_				:							Union[str, Any]						 =       multiprocessing.cpu_count()

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

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

		# Generation settings
		snake_case_				:							List[str]						 =       {
		    "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
		snake_case_				:							Optional[Any]						 =       load_dataset("openai_humaneval"    )
		snake_case_				:							Optional[int]						 =       load_metric("code_eval"    )

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

		snake_case_				:							str						 =       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
		snake_case_				:							Optional[Any]						 =       DataLoader(__snake_case  ,      batch_size=1    )

		# Run a quick test to see if code evaluation is enabled
		try:
				snake_case_				:							int						 =       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_				:							List[Any]						 =       accelerator.prepare(__snake_case  ,      __snake_case    )

		snake_case_				:							Optional[int]						 =       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:
				snake_case_				:							Optional[Any]						 =       []

				for task in tqdm(range(__snake_case    )    ):
						snake_case_				:							Optional[int]						 =       human_eval["test"][task]["test"]
						snake_case_				:							List[str]						 =       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_				:							List[Any]						 =       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()








def      lowerCAmelCase__	(    _a       :			int    ):
		snake_case_				:							str						 =       1
		for i in range(1  ,      num + 1    ):
				fact *= i
		return fact

def      lowerCAmelCase__	(    _a       :			int    ):
		snake_case_				:							List[str]						 =       0
		while number > 0:
				snake_case_				:							Dict						 =       number % 10
				sum_of_digits += last_digit
				snake_case_				:							List[Any]						 =       number // 10  # Removing the last_digit from the given number
		return sum_of_digits

def      lowerCAmelCase__	(    _a       :			int = 1_00    ):
		snake_case_				:							Optional[Any]						 =       factorial(_a    )
		snake_case_				:							Optional[int]						 =       split_and_add(_a    )
		return result


if __name__ == "__main__":
							print(solution(int(input('''Enter the Number: ''').strip())))
