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







class   _lowercase						(       lowerCAmelCase_				):


					'''simple docstring'''





					def __init__(      self ,			__UpperCamelCase ,			__UpperCamelCase ,			__UpperCamelCase=None ,			__UpperCamelCase=1  )->					Union[str, Any]:
										UpperCAmelCase__							: Optional[Any] 							=						tokenizer
										UpperCAmelCase__							: Any 							=						dataset
										UpperCAmelCase__							: int 							=						len(__UpperCamelCase  ) if n_tasks is None else n_tasks
										UpperCAmelCase__							: Dict 							=						n_copies



					def __iter__(      self  )->					int:
										UpperCAmelCase__							: 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()  )
										UpperCAmelCase__							: Union[str, Any] 							=						self.tokenizer(__UpperCamelCase ,			padding=__UpperCamelCase ,			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						(       lowerCAmelCase_				):


					'''simple docstring'''





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



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





def 							a__ (				lowerCAmelCase	:       Dict		):






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



def 							a__ (				lowerCAmelCase	:       Tuple       ,    lowerCAmelCase	:       Any       ,    lowerCAmelCase	:       str       ,    lowerCAmelCase	:       str       ,    lowerCAmelCase	:       Dict       ,    lowerCAmelCase	:       List[str]=20       ,    **lowerCAmelCase	:       Optional[Any]		):






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

															UpperCAmelCase__							: str 							=						accelerator.gather((generated_tokens, generated_tasks)		)
															UpperCAmelCase__							: Optional[Any] 							=						generated_tokens.cpu().numpy()
															UpperCAmelCase__							: Any 							=						generated_tasks.cpu().numpy()

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

					UpperCAmelCase__							: Tuple 							=						[[] for _ in range(lowerCAmelCase		)]
					for task, generated_tokens in gen_token_dict.items():
										for s in generated_tokens:
															UpperCAmelCase__							: Optional[int] 							=						tokenizer.decode(lowerCAmelCase       ,    skip_special_tokens=lowerCAmelCase       ,    clean_up_tokenization_spaces=lowerCAmelCase		)
															code_gens[task].append(remove_last_block(lowerCAmelCase		)		)
					return code_gens



def 							a__ (				):






					'''simple docstring'''
					UpperCAmelCase__							: List[str] 							=						HfArgumentParser(lowerCAmelCase		)
					UpperCAmelCase__							: Dict 							=						parser.parse_args()

					transformers.logging.set_verbosity_error()
					# enables code execution in code_eval metric
					UpperCAmelCase__							: Dict 							=						args.HF_ALLOW_CODE_EVAL
					# make sure tokenizer plays nice with multiprocessing
					UpperCAmelCase__							: Optional[int] 							=						"false"

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

					# Use dataset load to feed to accelerate
					UpperCAmelCase__							: List[Any] 							=						Accelerator()
					set_seed(args.seed       ,    device_specific=lowerCAmelCase		)

					# Load model and tokenizer
					UpperCAmelCase__							: Tuple 							=						AutoTokenizer.from_pretrained(args.model_ckpt		)
					UpperCAmelCase__							: int 							=						tokenizer.eos_token
					UpperCAmelCase__							: Tuple 							=						AutoModelForCausalLM.from_pretrained(args.model_ckpt		)

					# Generation settings
					UpperCAmelCase__							: 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       ,    lowerCAmelCase       ,    lowerCAmelCase		)]		),
					}

					# Load evaluation dataset and metric
					UpperCAmelCase__							: Optional[Any] 							=						load_dataset("openai_humaneval"		)
					UpperCAmelCase__							: int 							=						load_metric("code_eval"		)

					UpperCAmelCase__							: Union[str, Any] 							=						args.num_tasks if args.num_tasks is not None else len(human_eval["test"]		)
					UpperCAmelCase__							: Union[str, Any] 							=						args.n_samples // args.batch_size

					UpperCAmelCase__							: Union[str, Any] 							=						TokenizedDataset(lowerCAmelCase       ,    human_eval["test"]       ,    n_copies=lowerCAmelCase       ,    n_tasks=lowerCAmelCase		)
					# do not confuse args.batch_size, which is actually the num_return_sequences
					UpperCAmelCase__							: List[Any] 							=						DataLoader(lowerCAmelCase       ,    batch_size=1		)

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

					UpperCAmelCase__							: Union[str, Any] 							=						complete_code(
					    lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase       ,    n_tasks=lowerCAmelCase       ,    batch_size=args.batch_size       ,    **lowerCAmelCase       ,    )

					if accelerator.is_main_process:
										UpperCAmelCase__							: Optional[int] 							=						[]

										for task in tqdm(range(lowerCAmelCase		)		):
															UpperCAmelCase__							: List[Any] 							=						human_eval["test"][task]["test"]
															UpperCAmelCase__							: Optional[int] 							=						F"check({human_eval['test'][task]['entry_point']})"
															references.append("\n" + test_func + "\n" + entry_point		)

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


# 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 math import isqrt



def 							a__ (				lowerCAmelCase	:       int		):






     '''simple docstring'''
     UpperCAmelCase__							: Tuple 							=						[True] * max_number
     for i in range(2       ,    isqrt(max_number - 1		) + 1		):
          if is_prime[i]:
               for j in range(i**2       ,    lowerCAmelCase       ,    lowerCAmelCase		):
                    UpperCAmelCase__							: List[Any] 							=						False

     return [i for i in range(2       ,    lowerCAmelCase		) if is_prime[i]]



def 							a__ (				lowerCAmelCase	:       int = 10**8		):






     '''simple docstring'''
     UpperCAmelCase__							: Dict 							=						calculate_prime_numbers(max_number // 2		)

     UpperCAmelCase__							: Optional[int] 							=						0
     UpperCAmelCase__							: Optional[int] 							=						0
     UpperCAmelCase__							: Tuple 							=						len(lowerCAmelCase		) - 1
     while left <= right:
          while prime_numbers[left] * prime_numbers[right] >= max_number:
               right -= 1
          semiprimes_count += right - left + 1
          left += 1

     return semiprimes_count


if __name__ == "__main__":
    print(f"""{solution() = }""")



