






import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


sys.path.append(str(Path(__file__).parent.parent.parent.parent / """utils"""))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402



class 							SCREAMING_SNAKE_CASE__						(    unittest.TestCase						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[Any]	):
												snake_case__			:				str       =  0



					def 	_lowercase					( self					:  Dict	):
												snake_case__			:				Optional[int]       =  AutoImageProcessor.from_pretrained("openai/clip-vit-base-patch32"	)
												self.assertIsInstance(__A							,		__A	)



					def 	_lowercase					( self					:  Union[str, Any]	):
												with tempfile.TemporaryDirectory() as tmpdirname:
																			snake_case__			:				List[Any]       =  Path(__A	) / "preprocessor_config.json"
																			snake_case__			:				int       =  Path(__A	) / "config.json"
																			json.dump(
																			    {"image_processor_type": "CLIPImageProcessor", "processor_class": "CLIPProcessor"}							,		open(__A							,		"w"	)							,		)
																			json.dump({"model_type": "clip"}							,		open(__A							,		"w"	)	)

																			snake_case__			:				Any       =  AutoImageProcessor.from_pretrained(__A	)
																			self.assertIsInstance(__A							,		__A	)



					def 	_lowercase					( self					:  Union[str, Any]	):
												# Ensure we can load the image processor from the feature extractor config
												with tempfile.TemporaryDirectory() as tmpdirname:
																			snake_case__			:				Dict       =  Path(__A	) / "preprocessor_config.json"
																			snake_case__			:				int       =  Path(__A	) / "config.json"
																			json.dump(
																			    {"feature_extractor_type": "CLIPFeatureExtractor", "processor_class": "CLIPProcessor"}							,		open(__A							,		"w"	)							,		)
																			json.dump({"model_type": "clip"}							,		open(__A							,		"w"	)	)

																			snake_case__			:				List[str]       =  AutoImageProcessor.from_pretrained(__A	)
																			self.assertIsInstance(__A							,		__A	)



					def 	_lowercase					( self					:  List[str]	):
												with tempfile.TemporaryDirectory() as tmpdirname:
																			snake_case__			:				List[str]       =  CLIPConfig()

																			# Create a dummy config file with image_proceesor_type
																			snake_case__			:				Dict       =  Path(__A	) / "preprocessor_config.json"
																			snake_case__			:				str       =  Path(__A	) / "config.json"
																			json.dump(
																			    {"image_processor_type": "CLIPImageProcessor", "processor_class": "CLIPProcessor"}							,		open(__A							,		"w"	)							,		)
																			json.dump({"model_type": "clip"}							,		open(__A							,		"w"	)	)

																			# remove image_processor_type to make sure config.json alone is enough to load image processor locally
																			snake_case__			:				Tuple       =  AutoImageProcessor.from_pretrained(__A	).to_dict()

																			config_dict.pop("image_processor_type"	)
																			snake_case__			:				Tuple       =  CLIPImageProcessor(**__A	)

																			# save in new folder
																			model_config.save_pretrained(__A	)
																			config.save_pretrained(__A	)

																			snake_case__			:				List[Any]       =  AutoImageProcessor.from_pretrained(__A	)

																			# make sure private variable is not incorrectly saved
																			snake_case__			:				Dict       =  json.loads(config.to_json_string()	)
																			self.assertTrue("_processor_class" not in dict_as_saved	)

												self.assertIsInstance(__A							,		__A	)



					def 	_lowercase					( self					:  Tuple	):
												with tempfile.TemporaryDirectory() as tmpdirname:
																			snake_case__			:				List[str]       =  Path(__A	) / "preprocessor_config.json"
																			json.dump(
																			    {"image_processor_type": "CLIPImageProcessor", "processor_class": "CLIPProcessor"}							,		open(__A							,		"w"	)							,		)

																			snake_case__			:				Dict       =  AutoImageProcessor.from_pretrained(__A	)
																			self.assertIsInstance(__A							,		__A	)



					def 	_lowercase					( self					:  Union[str, Any]	):
												with self.assertRaisesRegex(
												    __A							,		"clip-base is not a local folder and is not a valid model identifier"	):
																			snake_case__			:				Union[str, Any]       =  AutoImageProcessor.from_pretrained("clip-base"	)



					def 	_lowercase					( self					:  Any	):
												with self.assertRaisesRegex(
												    __A							,		R"aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"	):
																			snake_case__			:				List[Any]       =  AutoImageProcessor.from_pretrained(__A							,		revision="aaaaaa"	)



					def 	_lowercase					( self					:  List[str]	):
												with self.assertRaisesRegex(
												    __A							,		"hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."							,		):
																			snake_case__			:				List[Any]       =  AutoImageProcessor.from_pretrained("hf-internal-testing/config-no-model"	)



					def 	_lowercase					( self					:  Tuple	):
												# If remote code is not set, we will time out when asking whether to load the model.
												with self.assertRaises(__A	):
																			snake_case__			:				int       =  AutoImageProcessor.from_pretrained("hf-internal-testing/test_dynamic_image_processor"	)
												# If remote code is disabled, we can't load this config.
												with self.assertRaises(__A	):
																			snake_case__			:				Union[str, Any]       =  AutoImageProcessor.from_pretrained(
																			    "hf-internal-testing/test_dynamic_image_processor"							,		trust_remote_code=__A	)

												snake_case__			:				Optional[int]       =  AutoImageProcessor.from_pretrained(
												    "hf-internal-testing/test_dynamic_image_processor"							,		trust_remote_code=__A	)
												self.assertEqual(image_processor.__class__.__name__							,		"NewImageProcessor"	)

												# Test image processor can be reloaded.
												with tempfile.TemporaryDirectory() as tmp_dir:
																			image_processor.save_pretrained(__A	)
																			snake_case__			:				Dict       =  AutoImageProcessor.from_pretrained(__A							,		trust_remote_code=__A	)
												self.assertEqual(reloaded_image_processor.__class__.__name__							,		"NewImageProcessor"	)



					def 	_lowercase					( self					:  str	):
												try:
																			AutoConfig.register("custom"							,		__A	)
																			AutoImageProcessor.register(__A							,		__A	)
																			# Trying to register something existing in the Transformers library will raise an error
																			with self.assertRaises(__A	):
																										AutoImageProcessor.register(__A							,		__A	)

																			with tempfile.TemporaryDirectory() as tmpdirname:
																										snake_case__			:				Tuple       =  Path(__A	) / "preprocessor_config.json"
																										snake_case__			:				List[Any]       =  Path(__A	) / "config.json"
																										json.dump(
																										    {"feature_extractor_type": "CLIPFeatureExtractor", "processor_class": "CLIPProcessor"}							,		open(__A							,		"w"	)							,		)
																										json.dump({"model_type": "clip"}							,		open(__A							,		"w"	)	)

																										snake_case__			:				int       =  CustomImageProcessor.from_pretrained(__A	)

																			# Now that the config is registered, it can be used as any other config with the auto-API
																			with tempfile.TemporaryDirectory() as tmp_dir:
																										image_processor.save_pretrained(__A	)
																										snake_case__			:				Optional[int]       =  AutoImageProcessor.from_pretrained(__A	)
																										self.assertIsInstance(__A							,		__A	)

												finally:
																			if "custom" in CONFIG_MAPPING._extra_content:
																										del CONFIG_MAPPING._extra_content["custom"]
																			if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
																										del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]






					def 	_lowercase					( self					:  int	):



												class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





																	"""simple docstring"""






																	a_     =		True

												try:
																			AutoConfig.register("custom"							,		__A	)
																			AutoImageProcessor.register(__A							,		__A	)
																			# If remote code is not set, the default is to use local
																			snake_case__			:				List[Any]       =  AutoImageProcessor.from_pretrained("hf-internal-testing/test_dynamic_image_processor"	)
																			self.assertEqual(image_processor.__class__.__name__							,		"NewImageProcessor"	)
																			self.assertTrue(image_processor.is_local	)

																			# If remote code is disabled, we load the local one.
																			snake_case__			:				List[str]       =  AutoImageProcessor.from_pretrained(
																			    "hf-internal-testing/test_dynamic_image_processor"							,		trust_remote_code=__A	)
																			self.assertEqual(image_processor.__class__.__name__							,		"NewImageProcessor"	)
																			self.assertTrue(image_processor.is_local	)

																			# If remote is enabled, we load from the Hub
																			snake_case__			:				Optional[int]       =  AutoImageProcessor.from_pretrained(
																			    "hf-internal-testing/test_dynamic_image_processor"							,		trust_remote_code=__A	)
																			self.assertEqual(image_processor.__class__.__name__							,		"NewImageProcessor"	)
																			self.assertTrue(not hasattr(__A							,		"is_local"	)	)

												finally:
																			if "custom" in CONFIG_MAPPING._extra_content:
																										del CONFIG_MAPPING._extra_content["custom"]
																			if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
																										del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]











import os

import pytest
from attr import dataclass


__lowerCamelCase :      Any										=				"""us-east-1"""  # defaults region



@dataclass
class 							SCREAMING_SNAKE_CASE__						:





					"""simple docstring"""






					a_     =		42
					a_     =		"arn:aws:iam::558105141721:role/sagemaker_execution_role"
					a_     =		{
					    "task_name": "mnli",
					    "per_device_train_batch_size": 1_6,
					    "per_device_eval_batch_size": 1_6,
					    "do_train": True,
					    "do_eval": True,
					    "do_predict": True,
					    "output_dir": "/opt/ml/model",
					    "overwrite_output_dir": True,
					    "max_steps": 5_0_0,
					    "save_steps": 5_5_0_0,
					}
					a_     =		{**hyperparameters, "max_steps": 1_0_0_0}



					@property
					def 	_lowercase					( self					:  List[Any]	):
						if self.framework == "pytorch":
							return [
							    {"Name": "train_runtime", "Regex": r"train_runtime.*=\D*(.*?)$"},
							    {"Name": "eval_accuracy", "Regex": r"eval_accuracy.*=\D*(.*?)$"},
							    {"Name": "eval_loss", "Regex": r"eval_loss.*=\D*(.*?)$"},
							]
						else:
							return [
							    {"Name": "train_runtime", "Regex": r"train_runtime.*=\D*(.*?)$"},
							    {"Name": "eval_accuracy", "Regex": r"loss.*=\D*(.*?)]?$"},
							    {"Name": "eval_loss", "Regex": r"sparse_categorical_accuracy.*=\D*(.*?)]?$"},
							]



					@property
					def 	_lowercase					( self					:  Any	):
						return f'''{self.framework}-transfromers-test'''



					@property
					def 	_lowercase					( self					:  Optional[Any]	):
						return f'''./tests/sagemaker/scripts/{self.framework}'''






					@property
					def 	_lowercase					( self					:  Union[str, Any]	):
						if self.framework == "pytorch":
							return "763104351884.dkr.ecr.us-east-1.amazonaws.com/huggingface-pytorch-training:1.7.1-transformers4.6.1-gpu-py36-cu110-ubuntu18.04"
						else:
							return "763104351884.dkr.ecr.us-east-1.amazonaws.com/huggingface-tensorflow-training:2.4.1-transformers4.6.1-gpu-py37-cu110-ubuntu18.04"







@pytest.fixture(scope="class"       )
def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							Optional[Any]       ):
	snake_case__			:				Optional[int]       =  SageMakerTestEnvironment(framework=request.cls.framework       )



