




import inspect
import unittest

from transformers import MobileNetVaConfig
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_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 transformers import MobileNetVaForImageClassification, MobileNetVaModel
		from transformers.models.mobilenet_va.modeling_mobilenet_va import MOBILENET_V1_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
		from PIL import Image

		from transformers import MobileNetVaImageProcessor




class 					snake_case__							(_UpperCamelCase     ):
				"""simple docstring"""



				def 							__UpperCAmelCase							(   self       :  Optional[Any]							)      ->  int:
							a					=    self.config_class(**self.inputs_dict							)
							self.parent.assertTrue(hasattr(__lowerCamelCase      ,   "tf_padding"							)							)
							self.parent.assertTrue(hasattr(__lowerCamelCase      ,   "depth_multiplier"							)							)




class 					snake_case__							:
				"""simple docstring"""



				def __init__(   self       :  Dict      ,   __lowerCamelCase       :  Optional[int]      ,   __lowerCamelCase       :  Any=13      ,   __lowerCamelCase       :  List[str]=3      ,   __lowerCamelCase       :  Optional[int]=32      ,   __lowerCamelCase       :  Dict=0.25      ,   __lowerCamelCase       :  List[str]=8      ,   __lowerCamelCase       :  Any=True      ,   __lowerCamelCase       :  int=10_24      ,   __lowerCamelCase       :  List[Any]=32      ,   __lowerCamelCase       :  Optional[Any]="relu6"      ,   __lowerCamelCase       :  List[str]=0.1      ,   __lowerCamelCase       :  Union[str, Any]=0.02      ,   __lowerCamelCase       :  int=True      ,   __lowerCamelCase       :  Optional[int]=True      ,   __lowerCamelCase       :  str=10      ,   __lowerCamelCase       :  Any=None      ,   )      ->  Tuple:
							a					=    parent
							a					=    batch_size
							a					=    num_channels
							a					=    image_size
							a					=    depth_multiplier
							a					=    min_depth
							a					=    tf_padding
							a					=    int(last_hidden_size * depth_multiplier							)
							a					=    output_stride
							a					=    hidden_act
							a					=    classifier_dropout_prob
							a					=    use_labels
							a					=    is_training
							a					=    num_labels
							a					=    initializer_range
							a					=    scope





				def 							__UpperCAmelCase							(   self       :  str							)      ->  Optional[Any]:
							a					=    floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]							)

							a					=    None
							a					=    None
							if self.use_labels:
										a					=    ids_tensor([self.batch_size]      ,   self.num_labels							)
										a					=    ids_tensor([self.batch_size, self.image_size, self.image_size]      ,   self.num_labels							)

							a					=    self.get_config()

							return config, pixel_values, labels, pixel_labels





				def 							__UpperCAmelCase							(   self       :  Optional[int]							)      ->  List[Any]:
							return MobileNetVaConfig(
							    num_channels=self.num_channels      ,   image_size=self.image_size      ,   depth_multiplier=self.depth_multiplier      ,   min_depth=self.min_depth      ,   tf_padding=self.tf_padding      ,   hidden_act=self.hidden_act      ,   classifier_dropout_prob=self.classifier_dropout_prob      ,   initializer_range=self.initializer_range      ,   )





				def 							__UpperCAmelCase							(   self       :  Any      ,   __lowerCamelCase       :  int      ,   __lowerCamelCase       :  Tuple      ,   __lowerCamelCase       :  Optional[int]      ,   __lowerCamelCase       :  List[Any]							)      ->  List[str]:
							a					=    MobileNetVaModel(config=__lowerCamelCase							)
							model.to(__lowerCamelCase							)
							model.eval()
							a					=    model(__lowerCamelCase							)
							self.parent.assertEqual(
							    result.last_hidden_state.shape      ,   (
							        self.batch_size,
							        self.last_hidden_size,
							        self.image_size // self.output_stride,
							        self.image_size // self.output_stride,
							    )      ,   )





				def 							__UpperCAmelCase							(   self       :  List[str]      ,   __lowerCamelCase       :  Optional[int]      ,   __lowerCamelCase       :  Dict      ,   __lowerCamelCase       :  List[Any]      ,   __lowerCamelCase       :  str							)      ->  Union[str, Any]:
							a					=    self.num_labels
							a					=    MobileNetVaForImageClassification(__lowerCamelCase							)
							model.to(__lowerCamelCase							)
							model.eval()
							a					=    model(__lowerCamelCase      ,   labels=__lowerCamelCase							)
							self.parent.assertEqual(result.logits.shape      ,   (self.batch_size, self.num_labels)							)





				def 							__UpperCAmelCase							(   self       :  str							)      ->  Any:
							a					=    self.prepare_config_and_inputs()
							a					,	a					,	a					,	a					=    config_and_inputs
							a					=    {"pixel_values": pixel_values}
							return config, inputs_dict




@require_torch
class 					snake_case__							(_UpperCamelCase					, _UpperCamelCase					, unittest.TestCase     ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_ :							Optional[Any]											=      (MobileNetVaModel, MobileNetVaForImageClassification) if is_torch_available() else ()
				SCREAMING_SNAKE_CASE_ :							Any											=      (
				    {"""feature-extraction""": MobileNetVaModel, """image-classification""": MobileNetVaForImageClassification}
				    if is_torch_available()
				    else {}
				)

				SCREAMING_SNAKE_CASE_ :							List[str]											=      False
				SCREAMING_SNAKE_CASE_ :							Dict											=      False
				SCREAMING_SNAKE_CASE_ :							Optional[int]											=      False
				SCREAMING_SNAKE_CASE_ :							Optional[int]											=      False





				def 							__UpperCAmelCase							(   self       :  Union[str, Any]							)      ->  Dict:
							a					=    MobileNetVaModelTester(self							)
							a					=    MobileNetVaConfigTester(self      ,   config_class=__lowerCamelCase      ,   has_text_modality=__lowerCamelCase							)





				def 							__UpperCAmelCase							(   self       :  int							)      ->  Tuple:
							self.config_tester.run_common_tests()





				@unittest.skip(reason="MobileNetV1 does not use inputs_embeds"							)
				def 							__UpperCAmelCase							(   self       :  List[Any]							)      ->  Optional[Any]:
							pass





				@unittest.skip(reason="MobileNetV1 does not support input and output embeddings"							)
				def 							__UpperCAmelCase							(   self       :  Any							)      ->  List[str]:
							pass





				@unittest.skip(reason="MobileNetV1 does not output attentions"							)
				def 							__UpperCAmelCase							(   self       :  List[Any]							)      ->  int:
							pass





				def 							__UpperCAmelCase							(   self       :  Tuple							)      ->  List[str]:
							a					,	a					=    self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
										a					=    model_class(__lowerCamelCase							)
										a					=    inspect.signature(model.forward							)
										# signature.parameters is an OrderedDict => so arg_names order is deterministic
										a					=    [*signature.parameters.keys()]

										a					=    ["pixel_values"]
										self.assertListEqual(arg_names[:1]      ,   __lowerCamelCase							)





				def 							__UpperCAmelCase							(   self       :  str							)      ->  int:
							a					=    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*__lowerCamelCase							)





				def 							__UpperCAmelCase							(   self       :  Dict							)      ->  Optional[Any]:
							def check_hidden_states_output(__lowerCamelCase       :  List[str]      ,   __lowerCamelCase       :  Optional[Any]      ,   __lowerCamelCase       :  Optional[int]							):
										a					=    model_class(__lowerCamelCase							)
										model.to(__lowerCamelCase							)
										model.eval()

										with torch.no_grad():
													a					=    model(**self._prepare_for_class(__lowerCamelCase      ,   __lowerCamelCase							)							)

										a					=    outputs.hidden_states

										a					=    26
										self.assertEqual(len(__lowerCamelCase							)      ,   __lowerCamelCase							)

							a					,	a					=    self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
										a					=    True
										check_hidden_states_output(__lowerCamelCase      ,   __lowerCamelCase      ,   __lowerCamelCase							)

										# check that output_hidden_states also work using config
										del inputs_dict["output_hidden_states"]
										a					=    True

										check_hidden_states_output(__lowerCamelCase      ,   __lowerCamelCase      ,   __lowerCamelCase							)





				def 							__UpperCAmelCase							(   self       :  Union[str, Any]							)      ->  Dict:
							a					=    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_image_classification(*__lowerCamelCase							)





				@slow
				def 							__UpperCAmelCase							(   self       :  Tuple							)      ->  Optional[int]:
							for model_name in MOBILENET_V1_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
										a					=    MobileNetVaModel.from_pretrained(__lowerCamelCase							)
										self.assertIsNotNone(__lowerCamelCase							)




def     __magic_name__       (				):






			'''simple docstring'''







			a					=    Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"     )
			return image






@require_torch
@require_vision
class 					snake_case__							(unittest.TestCase     ):
				"""simple docstring"""



				@cached_property
				def 							__UpperCAmelCase							(   self       :  List[str]							)      ->  Any:
							return (
							    MobileNetVaImageProcessor.from_pretrained("google/mobilenet_v1_1.0_224"							) if is_vision_available() else None
							)





				@slow
				def 							__UpperCAmelCase							(   self       :  str							)      ->  Dict:
							a					=    MobileNetVaForImageClassification.from_pretrained("google/mobilenet_v1_1.0_224"							).to(__lowerCamelCase							)

							a					=    self.default_image_processor
							a					=    prepare_img()
							a					=    image_processor(images=__lowerCamelCase      ,   return_tensors="pt"							).to(__lowerCamelCase							)

							# forward pass
							with torch.no_grad():
										a					=    model(**__lowerCamelCase							)

							# verify the logits
							a					=    torch.Size((1, 10_01)							)
							self.assertEqual(outputs.logits.shape      ,   __lowerCamelCase							)

							a					=    torch.tensor([-4.1_739, -1.1_233, 3.1_205]							).to(__lowerCamelCase							)

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












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


__lowerCAmelCase	:       Union[str, Any]     =	['\nclass', '\ndef', '\n#', '\n@', '\nprint', '\nif']




class 					snake_case__							(_UpperCamelCase     ):
				"""simple docstring"""



				def __init__(   self       :  Any      ,   __lowerCamelCase       :  List[str]      ,   __lowerCamelCase       :  Union[str, Any]      ,   __lowerCamelCase       :  Union[str, Any]=None      ,   __lowerCamelCase       :  Any=1							)      ->  Union[str, Any]:
							a					=    tokenizer
							a					=    dataset
							a					=    len(__lowerCamelCase							) if n_tasks is None else n_tasks
							a					=    n_copies





				def __iter__(   self       :  Tuple							)      ->  str:
							a					=    []
							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()							)
							a					=    self.tokenizer(__lowerCamelCase      ,   padding=__lowerCamelCase      ,   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 					snake_case__							(_UpperCamelCase     ):
				"""simple docstring"""



				def __init__(   self       :  int      ,   __lowerCamelCase       :  Dict      ,   __lowerCamelCase       :  Dict      ,   __lowerCamelCase       :  Tuple							)      ->  Optional[Any]:
							a					=    start_length
							a					=    eof_strings
							a					=    tokenizer





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




def     __magic_name__       (				A	:      List[Any]     ):






			'''simple docstring'''







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




def     __magic_name__       (				A	:      Union[str, Any],						A	:      Optional[Any],						A	:      List[Any],						A	:      Optional[Any],						A	:      List[str],						A	:      List[Any]=20,						**A	:      Union[str, Any]     ):






			'''simple docstring'''







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

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

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

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




def     __magic_name__       (				):






			'''simple docstring'''







			a					=    HfArgumentParser(A     )
			a					=    parser.parse_args()

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

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

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

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

			# Generation settings
			a					=    {
			    "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,						A,						A     )]     ),
			}

			# Load evaluation dataset and metric
			a					=    load_dataset("openai_humaneval"     )
			a					=    load_metric("code_eval"     )

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

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

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

			a					,	a					=    accelerator.prepare(A,						A     )

			a					=    complete_code(
			    A,						A,						A,						A,						n_tasks=A,						batch_size=args.batch_size,						**A,						)

			if accelerator.is_main_process:
						a					=    []

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

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


# 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()






