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


_lowercase		:				str	       =		["\nclass", "\ndef", "\n#", "\n@", "\nprint", "\nif"]





class 					__magic_name__		(					_UpperCAmelCase):







			def __init__(			self       :						Optional[int]       ,    lowercase_       :						Any       ,    lowercase_       :						Optional[Any]       ,    lowercase_       :						Any=None       ,    lowercase_       :						Union[str, Any]=1					):
										lowercase_      :    List[str]       		= tokenizer
										lowercase_      :    Optional[Any]       		= dataset
										lowercase_      :    int       		= len(lowercase_					) if n_tasks is None else n_tasks
										lowercase_      :    Optional[int]       		= n_copies







			def __iter__(			self       :						List[str]					):
										lowercase_      :    Optional[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()					)
										lowercase_      :    str       		= self.tokenizer(lowercase_       ,    padding=lowercase_       ,    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 					__magic_name__		(					_UpperCAmelCase):







			def __init__(			self       :						Any       ,    lowercase_       :						List[str]       ,    lowercase_       :						str       ,    lowercase_       :						List[Any]					):
										lowercase_      :    List[str]       		= start_length
										lowercase_      :    Optional[Any]       		= eof_strings
										lowercase_      :    str       		= tokenizer







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




def 							lowerCamelCase			(   UpperCAmelCase__ :      str				)    ->   List[str]:
							lowercase_      :    Optional[Any]       		= re.split("""(%s)""" % """|""".join(UpperCAmelCase__				)   ,   UpperCAmelCase__				)
							# last string should be ""
							return "".join(string_list[:-2]				)




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

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

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

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




def 							lowerCamelCase			(   )    ->   List[str]:
							# Setup configuration
							lowercase_      :    str       		= HfArgumentParser(UpperCAmelCase__				)
							lowercase_      :    int       		= parser.parse_args()

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

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

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

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

							# Generation settings
							lowercase_      :    Union[str, 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   ,   UpperCAmelCase__   ,   UpperCAmelCase__				)]				),
							}

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

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

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

							# Run a quick test to see if code evaluation is enabled
							try:
														lowercase_      :    Dict       		= 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_		,		lowercase_      :    Union[str, Any]       		= accelerator.prepare(UpperCAmelCase__   ,   UpperCAmelCase__				)

							lowercase_      :    Optional[int]       		= complete_code(
							    UpperCAmelCase__   ,   UpperCAmelCase__   ,   UpperCAmelCase__   ,   UpperCAmelCase__   ,   n_tasks=UpperCAmelCase__   ,   batch_size=args.batch_size   ,   **UpperCAmelCase__   ,   )

							if accelerator.is_main_process:
														lowercase_      :    List[Any]       		= []

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

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


# 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 inspect
import unittest

from transformers import BitConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch
						from torch import nn

						from transformers import BitBackbone, BitForImageClassification, BitImageProcessor, BitModel
						from transformers.models.bit.modeling_bit import BIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
						from PIL import Image





class 					__magic_name__		:







			def __init__(			self       :						Tuple       ,    lowercase_       :						Tuple       ,    lowercase_       :						Any=3       ,    lowercase_       :						int=32       ,    lowercase_       :						str=3       ,    lowercase_       :						int=10       ,    lowercase_       :						Any=[8, 16, 32, 64]       ,    lowercase_       :						Tuple=[1, 1, 2, 1]       ,    lowercase_       :						Any=True       ,    lowercase_       :						int=True       ,    lowercase_       :						Any="relu"       ,    lowercase_       :						List[Any]=3       ,    lowercase_       :						Tuple=None       ,    lowercase_       :						Union[str, Any]=["stage2", "stage3", "stage4"]       ,    lowercase_       :						Optional[int]=[2, 3, 4]       ,    lowercase_       :						List[str]=1       ,    ):
										lowercase_      :    Any       		= parent
										lowercase_      :    str       		= batch_size
										lowercase_      :    Any       		= image_size
										lowercase_      :    Optional[Any]       		= num_channels
										lowercase_      :    Any       		= embeddings_size
										lowercase_      :    Union[str, Any]       		= hidden_sizes
										lowercase_      :    Any       		= depths
										lowercase_      :    Dict       		= is_training
										lowercase_      :    Tuple       		= use_labels
										lowercase_      :    str       		= hidden_act
										lowercase_      :    Optional[Any]       		= num_labels
										lowercase_      :    Tuple       		= scope
										lowercase_      :    Any       		= len(lowercase_					)
										lowercase_      :    Optional[Any]       		= out_features
										lowercase_      :    Tuple       		= out_indices
										lowercase_      :    str       		= num_groups







			def 			SCREAMING_SNAKE_CASE_  (			self       :						List[Any]					):
										lowercase_      :    int       		= floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]					)

										lowercase_      :    List[Any]       		= None
										if self.use_labels:
																	lowercase_      :    List[str]       		= ids_tensor([self.batch_size]       ,    self.num_labels					)

										lowercase_      :    int       		= self.get_config()

										return config, pixel_values, labels







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Union[str, Any]					):
										return BitConfig(
										    num_channels=self.num_channels       ,    embeddings_size=self.embeddings_size       ,    hidden_sizes=self.hidden_sizes       ,    depths=self.depths       ,    hidden_act=self.hidden_act       ,    num_labels=self.num_labels       ,    out_features=self.out_features       ,    out_indices=self.out_indices       ,    num_groups=self.num_groups       ,    )







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Union[str, Any]       ,    lowercase_       :						int       ,    lowercase_       :						List[str]       ,    lowercase_       :						List[str]					):
										lowercase_      :    Optional[int]       		= BitModel(config=lowercase_					)
										model.to(lowercase_					)
										model.eval()
										lowercase_      :    List[Any]       		= model(lowercase_					)
										self.parent.assertEqual(
										    result.last_hidden_state.shape       ,    (self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32)       ,    )







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[Any]       ,    lowercase_       :						Tuple       ,    lowercase_       :						Optional[int]       ,    lowercase_       :						List[Any]					):
										lowercase_      :    Union[str, Any]       		= self.num_labels
										lowercase_      :    Tuple       		= BitForImageClassification(lowercase_					)
										model.to(lowercase_					)
										model.eval()
										lowercase_      :    Any       		= model(lowercase_       ,    labels=lowercase_					)
										self.parent.assertEqual(result.logits.shape       ,    (self.batch_size, self.num_labels)					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[Any]       ,    lowercase_       :						Union[str, Any]       ,    lowercase_       :						List[Any]       ,    lowercase_       :						Optional[int]					):
										lowercase_      :    Any       		= BitBackbone(config=lowercase_					)
										model.to(lowercase_					)
										model.eval()
										lowercase_      :    Dict       		= model(lowercase_					)

										# verify feature maps
										self.parent.assertEqual(len(result.feature_maps					)       ,    len(config.out_features					)					)
										self.parent.assertListEqual(list(result.feature_maps[0].shape					)       ,    [self.batch_size, self.hidden_sizes[1], 4, 4]					)

										# verify channels
										self.parent.assertEqual(len(model.channels					)       ,    len(config.out_features					)					)
										self.parent.assertListEqual(model.channels       ,    config.hidden_sizes[1:]					)

										# verify backbone works with out_features=None
										lowercase_      :    List[str]       		= None
										lowercase_      :    Dict       		= BitBackbone(config=lowercase_					)
										model.to(lowercase_					)
										model.eval()
										lowercase_      :    Tuple       		= model(lowercase_					)

										# verify feature maps
										self.parent.assertEqual(len(result.feature_maps					)       ,    1					)
										self.parent.assertListEqual(list(result.feature_maps[0].shape					)       ,    [self.batch_size, self.hidden_sizes[-1], 1, 1]					)

										# verify channels
										self.parent.assertEqual(len(model.channels					)       ,    1					)
										self.parent.assertListEqual(model.channels       ,    [config.hidden_sizes[-1]]					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Any					):
										lowercase_      :    Optional[int]       		= self.prepare_config_and_inputs()
										lowercase_		,		lowercase_		,		lowercase_      :    Optional[Any]       		= config_and_inputs
										lowercase_      :    Any       		= {"""pixel_values""": pixel_values}
										return config, inputs_dict





@require_torch
class 					__magic_name__		(					_UpperCAmelCase,  _UpperCAmelCase,  unittest.TestCase):
			UpperCamelCase__       = (BitModel, BitForImageClassification, BitBackbone) if is_torch_available() else ()
			UpperCamelCase__       = (
			    {'''feature-extraction''': BitModel, '''image-classification''': BitForImageClassification}
			    if is_torch_available()
			    else {}
			)

			UpperCamelCase__       = False
			UpperCamelCase__       = False
			UpperCamelCase__       = False
			UpperCamelCase__       = False
			UpperCamelCase__       = False







			def 			SCREAMING_SNAKE_CASE_  (			self       :						str					):
										lowercase_      :    int       		= BitModelTester(self					)
										lowercase_      :    Optional[int]       		= ConfigTester(self       ,    config_class=lowercase_       ,    has_text_modality=lowercase_					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Any					):
										self.create_and_test_config_common_properties()
										self.config_tester.create_and_test_config_to_json_string()
										self.config_tester.create_and_test_config_to_json_file()
										self.config_tester.create_and_test_config_from_and_save_pretrained()
										self.config_tester.create_and_test_config_with_num_labels()
										self.config_tester.check_config_can_be_init_without_params()
										self.config_tester.check_config_arguments_init()







			def 			SCREAMING_SNAKE_CASE_  (			self       :						List[str]					):
										return







			@unittest.skip(reason="""Bit does not output attentions"""					)
			def 			SCREAMING_SNAKE_CASE_  (			self       :						Any					):
										pass







			@unittest.skip(reason="""Bit does not use inputs_embeds"""					)
			def 			SCREAMING_SNAKE_CASE_  (			self       :						Any					):
										pass







			@unittest.skip(reason="""Bit does not support input and output embeddings"""					)
			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[int]					):
										pass







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Dict					):
										lowercase_		,		lowercase_      :    Optional[int]       		= self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
																	lowercase_      :    Optional[Any]       		= model_class(lowercase_					)
																	lowercase_      :    Union[str, Any]       		= inspect.signature(model.forward					)
																	# signature.parameters is an OrderedDict => so arg_names order is deterministic
																	lowercase_      :    Union[str, Any]       		= [*signature.parameters.keys()]

																	lowercase_      :    Optional[int]       		= ["""pixel_values"""]
																	self.assertListEqual(arg_names[:1]       ,    lowercase_					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Any					):
										lowercase_      :    int       		= self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_model(*lowercase_					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Tuple					):
										lowercase_      :    Any       		= self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_backbone(*lowercase_					)







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Any					):
										lowercase_		,		lowercase_      :    int       		= self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
																	lowercase_      :    List[Any]       		= model_class(config=lowercase_					)
																	for name, module in model.named_modules():
																								if isinstance(lowercase_       ,    (nn.BatchNormad, nn.GroupNorm)					):
																															self.assertTrue(
																															    torch.all(module.weight == 1					)       ,    msg=f'''Parameter {name} of model {model_class} seems not properly initialized'''       ,    )
																															self.assertTrue(
																															    torch.all(module.bias == 0					)       ,    msg=f'''Parameter {name} of model {model_class} seems not properly initialized'''       ,    )







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[Any]					):
										def check_hidden_states_output(lowercase_       :						Optional[int]       ,    lowercase_       :						int       ,    lowercase_       :						int					):
																	lowercase_      :    Optional[Any]       		= model_class(lowercase_					)
																	model.to(lowercase_					)
																	model.eval()

																	with torch.no_grad():
																								lowercase_      :    List[Any]       		= model(**self._prepare_for_class(lowercase_       ,    lowercase_					)					)

																	lowercase_      :    str       		= outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

																	lowercase_      :    Optional[int]       		= self.model_tester.num_stages
																	self.assertEqual(len(lowercase_					)       ,    expected_num_stages + 1					)

																	# Bit's feature maps are of shape (batch_size, num_channels, height, width)
																	self.assertListEqual(
																	    list(hidden_states[0].shape[-2:]					)       ,    [self.model_tester.image_size // 4, self.model_tester.image_size // 4]       ,    )

										lowercase_		,		lowercase_      :    List[Any]       		= self.model_tester.prepare_config_and_inputs_for_common()
										lowercase_      :    Dict       		= ["""preactivation""", """bottleneck"""]
										for model_class in self.all_model_classes:
																	for layer_type in layers_type:
																								lowercase_      :    Union[str, Any]       		= layer_type
																								lowercase_      :    Optional[Any]       		= True
																								check_hidden_states_output(lowercase_       ,    lowercase_       ,    lowercase_					)

																								# check that output_hidden_states also work using config
																								del inputs_dict["output_hidden_states"]
																								lowercase_      :    Union[str, Any]       		= True

																								check_hidden_states_output(lowercase_       ,    lowercase_       ,    lowercase_					)







			@unittest.skip(reason="""Bit does not use feedforward chunking"""					)
			def 			SCREAMING_SNAKE_CASE_  (			self       :						Tuple					):
										pass







			def 			SCREAMING_SNAKE_CASE_  (			self       :						Optional[Any]					):
										lowercase_      :    Tuple       		= self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_image_classification(*lowercase_					)







			@slow
			def 			SCREAMING_SNAKE_CASE_  (			self       :						Tuple					):
										for model_name in BIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	lowercase_      :    List[str]       		= BitModel.from_pretrained(lowercase_					)
																	self.assertIsNotNone(lowercase_					)




def 							lowerCamelCase			(   )    ->   Optional[Any]:
							lowercase_      :    List[str]       		= Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""				)
							return image





@require_torch
@require_vision
class 					__magic_name__		(					unittest.TestCase):







			@cached_property
			def 			SCREAMING_SNAKE_CASE_  (			self       :						List[str]					):
										return (
										    BitImageProcessor.from_pretrained(BIT_PRETRAINED_MODEL_ARCHIVE_LIST[0]					) if is_vision_available() else None
										)







			@slow
			def 			SCREAMING_SNAKE_CASE_  (			self       :						int					):
										lowercase_      :    List[str]       		= BitForImageClassification.from_pretrained(BIT_PRETRAINED_MODEL_ARCHIVE_LIST[0]					).to(lowercase_					)

										lowercase_      :    int       		= self.default_image_processor
										lowercase_      :    List[Any]       		= prepare_img()
										lowercase_      :    Dict       		= image_processor(images=lowercase_       ,    return_tensors="""pt"""					).to(lowercase_					)

										# forward pass
										with torch.no_grad():
																	lowercase_      :    str       		= model(**lowercase_					)

										# verify the logits
										lowercase_      :    Optional[int]       		= torch.Size((1, 1000)					)
										self.assertEqual(outputs.logits.shape       ,    lowercase_					)

										lowercase_      :    Union[str, Any]       		= torch.tensor([[-0.65_26, -0.52_63, -1.43_98]]					).to(lowercase_					)

										self.assertTrue(torch.allclose(outputs.logits[0, :3]       ,    lowercase_       ,    atol=1E-4					)					)





@require_torch
class 					__magic_name__		(					_UpperCAmelCase,  unittest.TestCase):
			UpperCamelCase__       = (BitBackbone,) if is_torch_available() else ()
			UpperCamelCase__       = BitConfig

			UpperCamelCase__       = False







			def 			SCREAMING_SNAKE_CASE_  (			self       :						List[Any]					):
										lowercase_      :    Union[str, Any]       		= BitModelTester(self					)





