



"""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    __lowercase     (					_UpperCamelCase):







						"""simple docstring"""







						def __init__(self   ,		lowercase__   ,		lowercase__   ,		lowercase__=None   ,		lowercase__=1       ):
											snake_case_					:		Union[str, Any] 					=					tokenizer
											snake_case_					:		Tuple 					=					dataset
											snake_case_					:		str 					=					len(__a       ) if n_tasks is None else n_tasks
											snake_case_					:		Union[str, Any] 					=					n_copies







						def __iter__(self       ):
											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_					:		Dict 					=					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    __lowercase     (					_UpperCamelCase):







						"""simple docstring"""







						def __init__(self   ,		lowercase__   ,		lowercase__   ,		lowercase__       ):
											snake_case_					:		Union[str, Any] 					=					start_length
											snake_case_					:		Union[str, Any] 					=					eof_strings
											snake_case_					:		Tuple 					=					tokenizer







						def __call__(self   ,		lowercase__   ,		lowercase__   ,		**lowercase__       ):
											snake_case_					:		List[str] 					=					self.tokenizer.batch_decode(input_ids[:, self.start_length :]       )
											snake_case_					:		Any 					=					[]
											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 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  List[str]			):



					"""simple docstring"""





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







def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  Optional[Any]				,    SCREAMING_SNAKE_CASE__ :  Any				,    SCREAMING_SNAKE_CASE__ :  List[str]				,    SCREAMING_SNAKE_CASE__ :  Any				,    SCREAMING_SNAKE_CASE__ :  List[str]				,    SCREAMING_SNAKE_CASE__ :  Optional[Any]=2_0				,    **SCREAMING_SNAKE_CASE__ :  Optional[Any]			):



					"""simple docstring"""





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

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

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

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







def 							SCREAMING_SNAKE_CASE__      (		):



					"""simple docstring"""





					snake_case_					:		Union[str, Any] 					=					HfArgumentParser(lowercase_			)
					snake_case_					:		str 					=					parser.parse_args()

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

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

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

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

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

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

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

					# Run a quick test to see if code evaluation is enabled
					try:
										snake_case_					:		Union[str, 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

					snake_case_					:		Tuple 					=					accelerator.prepare(lowercase_				,    lowercase_			)

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

					if accelerator.is_main_process:
										snake_case_					:		Union[str, Any] 					=					[]

										for task in tqdm(range(lowercase_			)			):
															snake_case_					:		List[Any] 					=					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_					:		List[str] 					=					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()





"""simple docstring"""

import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						_A		:							Union[str, Any]							       =		["""image_processor""", """tokenizer"""]
						_A		:							str							       =		"""ChineseCLIPImageProcessor"""
						_A		:							Tuple							       =		("""BertTokenizer""", """BertTokenizerFast""")







						def __init__(self   ,		lowercase__=None   ,		lowercase__=None   ,		**lowercase__       ):
											snake_case_					:		Any 					=					None
											if "feature_extractor" in kwargs:
																warnings.warn(
																    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
																    """ instead."""   ,		lowercase__   ,		)
																snake_case_					:		Optional[Any] 					=					kwargs.pop("""feature_extractor"""       )

											snake_case_					:		str 					=					image_processor if image_processor is not None else feature_extractor
											if image_processor is None:
																raise ValueError("""You need to specify an `image_processor`."""       )
											if tokenizer is None:
																raise ValueError("""You need to specify a `tokenizer`."""       )

											super().__init__(lowercase__   ,		lowercase__       )
											snake_case_					:		Union[str, Any] 					=					self.image_processor







						def __call__(self   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		**lowercase__       ):

											if text is None and images is None:
																raise ValueError("""You have to specify either text or images. Both cannot be none."""       )

											if text is not None:
																snake_case_					:		Any 					=					self.tokenizer(lowercase__   ,		return_tensors=lowercase__   ,		**lowercase__       )

											if images is not None:
																snake_case_					:		Tuple 					=					self.image_processor(lowercase__   ,		return_tensors=lowercase__   ,		**lowercase__       )

											if text is not None and images is not None:
																snake_case_					:		List[Any] 					=					image_features.pixel_values
																return encoding
											elif text is not None:
																return encoding
											else:
																return BatchEncoding(data=dict(**lowercase__       )   ,		tensor_type=lowercase__       )







						def 						__UpperCamelCase							(self   ,		*lowercase__   ,		**lowercase__       ):
											return self.tokenizer.batch_decode(*lowercase__   ,		**lowercase__       )







						def 						__UpperCamelCase							(self   ,		*lowercase__   ,		**lowercase__       ):
											return self.tokenizer.decode(*lowercase__   ,		**lowercase__       )







						@property
						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Optional[int] 					=					self.tokenizer.model_input_names
											snake_case_					:		Union[str, Any] 					=					self.image_processor.model_input_names
											return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names       )       )







						@property
						def 						__UpperCamelCase							(self       ):
											warnings.warn(
											    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""   ,		lowercase__   ,		)
											return self.image_processor_class




