
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       __A(				unittest.TestCase       ):

					def    lowercase__      (	self					:     Tuple   ):
												lowerCamelCase_							  =					0

					def    lowercase__      (	self					:     Dict   ):
												lowerCamelCase_							  =					AutoImageProcessor.from_pretrained("""openai/clip-vit-base-patch32"""   )
												self.assertIsInstance(__UpperCamelCase   ,					__UpperCamelCase   )

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

																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(__UpperCamelCase   )
																			self.assertIsInstance(__UpperCamelCase   ,					__UpperCamelCase   )

					def    lowercase__      (	self					:     Any   ):
												# Ensure we can load the image processor from the feature extractor config
												with tempfile.TemporaryDirectory() as tmpdirname:
																			lowerCamelCase_							  =					Path(__UpperCamelCase   ) / """preprocessor_config.json"""
																			lowerCamelCase_							  =					Path(__UpperCamelCase   ) / """config.json"""
																			json.dump(
																			    {"""feature_extractor_type""": """CLIPFeatureExtractor""", """processor_class""": """CLIPProcessor"""}   ,					open(__UpperCamelCase   ,					"""w"""   )   ,					)
																			json.dump({"""model_type""": """clip"""}   ,					open(__UpperCamelCase   ,					"""w"""   )   )

																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(__UpperCamelCase   )
																			self.assertIsInstance(__UpperCamelCase   ,					__UpperCamelCase   )

					def    lowercase__      (	self					:     Optional[Any]   ):
												with tempfile.TemporaryDirectory() as tmpdirname:
																			lowerCamelCase_							  =					CLIPConfig()

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

																			# remove image_processor_type to make sure config.json alone is enough to load image processor locally
																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(__UpperCamelCase   ).to_dict()

																			config_dict.pop("""image_processor_type"""   )
																			lowerCamelCase_							  =					CLIPImageProcessor(**__UpperCamelCase   )

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

																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(__UpperCamelCase   )

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

												self.assertIsInstance(__UpperCamelCase   ,					__UpperCamelCase   )

					def    lowercase__      (	self					:     Optional[Any]   ):
												with tempfile.TemporaryDirectory() as tmpdirname:
																			lowerCamelCase_							  =					Path(__UpperCamelCase   ) / """preprocessor_config.json"""
																			json.dump(
																			    {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}   ,					open(__UpperCamelCase   ,					"""w"""   )   ,					)

																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(__UpperCamelCase   )
																			self.assertIsInstance(__UpperCamelCase   ,					__UpperCamelCase   )

					def    lowercase__      (	self					:     Any   ):
												with self.assertRaisesRegex(
												    __UpperCamelCase   ,					"""clip-base is not a local folder and is not a valid model identifier"""   ):
																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained("""clip-base"""   )

					def    lowercase__      (	self					:     Any   ):
												with self.assertRaisesRegex(
												    __UpperCamelCase   ,					R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""   ):
																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(__UpperCamelCase   ,					revision="""aaaaaa"""   )

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

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

												lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(
												    """hf-internal-testing/test_dynamic_image_processor"""   ,					trust_remote_code=__UpperCamelCase   )
												self.assertEqual(image_processor.__class__.__name__   ,					"""NewImageProcessor"""   )

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

					def    lowercase__      (	self					:     List[Any]   ):
												try:
																			AutoConfig.register("""custom"""   ,					__UpperCamelCase   )
																			AutoImageProcessor.register(__UpperCamelCase   ,					__UpperCamelCase   )
																			# Trying to register something existing in the Transformers library will raise an error
																			with self.assertRaises(__UpperCamelCase   ):
																										AutoImageProcessor.register(__UpperCamelCase   ,					__UpperCamelCase   )

																			with tempfile.TemporaryDirectory() as tmpdirname:
																										lowerCamelCase_							  =					Path(__UpperCamelCase   ) / """preprocessor_config.json"""
																										lowerCamelCase_							  =					Path(__UpperCamelCase   ) / """config.json"""
																										json.dump(
																										    {"""feature_extractor_type""": """CLIPFeatureExtractor""", """processor_class""": """CLIPProcessor"""}   ,					open(__UpperCamelCase   ,					"""w"""   )   ,					)
																										json.dump({"""model_type""": """clip"""}   ,					open(__UpperCamelCase   ,					"""w"""   )   )

																										lowerCamelCase_							  =					CustomImageProcessor.from_pretrained(__UpperCamelCase   )

																			# 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(__UpperCamelCase   )
																										lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(__UpperCamelCase   )
																										self.assertIsInstance(__UpperCamelCase   ,					__UpperCamelCase   )

												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					:     List[str]   ):





												class       __A(				UpperCAmelCase       ):
																	SCREAMING_SNAKE_CASE			    =    True

												try:
																			AutoConfig.register("""custom"""   ,					__UpperCamelCase   )
																			AutoImageProcessor.register(__UpperCamelCase   ,					__UpperCamelCase   )
																			# If remote code is not set, the default is to use local
																			lowerCamelCase_							  =					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.
																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(
																			    """hf-internal-testing/test_dynamic_image_processor"""   ,					trust_remote_code=__UpperCamelCase   )
																			self.assertEqual(image_processor.__class__.__name__   ,					"""NewImageProcessor"""   )
																			self.assertTrue(image_processor.is_local   )

																			# If remote is enabled, we load from the Hub
																			lowerCamelCase_							  =					AutoImageProcessor.from_pretrained(
																			    """hf-internal-testing/test_dynamic_image_processor"""   ,					trust_remote_code=__UpperCamelCase   )
																			self.assertEqual(image_processor.__class__.__name__   ,					"""NewImageProcessor"""   )
																			self.assertTrue(not hasattr(__UpperCamelCase   ,					"""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 socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
		from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
		import torch_xla.core.xla_model as xm



def      __lowerCAmelCase						(							UpperCAmelCase__				:      Any       )  ->    List[str]:
							if is_torch_version("""<"""			,				"""2.0.0"""       ) or not hasattr(UpperCAmelCase__			,				"""_dynamo"""       ):
														return False
							return isinstance(UpperCAmelCase__			,				torch._dynamo.eval_frame.OptimizedModule       )



def      __lowerCAmelCase						(							UpperCAmelCase__				:      str			,				UpperCAmelCase__				:      bool = True       )  ->    str:
							lowerCamelCase_							  =					(torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

							lowerCamelCase_							  =					is_compiled_module(UpperCAmelCase__       )
							if is_compiled:
														lowerCamelCase_							  =					model
														lowerCamelCase_							  =					model._orig_mod

							if is_deepspeed_available():
														options += (DeepSpeedEngine,)

							while isinstance(UpperCAmelCase__			,				UpperCAmelCase__       ):
														lowerCamelCase_							  =					model.module

							if not keep_fpaa_wrapper:
														lowerCamelCase_							  =					getattr(UpperCAmelCase__			,				"""forward"""       )
														lowerCamelCase_							  =					model.__dict__.pop("""_original_forward"""			,				UpperCAmelCase__       )
														if original_forward is not None:
																					while hasattr(UpperCAmelCase__			,				"""__wrapped__"""       ):
																												lowerCamelCase_							  =					forward.__wrapped__
																												if forward == original_forward:
																																			break
																					lowerCamelCase_							  =					forward
														if getattr(UpperCAmelCase__			,				"""_converted_to_transformer_engine"""			,				UpperCAmelCase__       ):
																					convert_model(UpperCAmelCase__			,				to_transformer_engine=UpperCAmelCase__       )

							if is_compiled:
														lowerCamelCase_							  =					model
														lowerCamelCase_							  =					compiled_model

							return model



def      __lowerCAmelCase						(							)  ->    Any:
							PartialState().wait_for_everyone()



def      __lowerCAmelCase						(							UpperCAmelCase__				:      Union[str, Any]			,				UpperCAmelCase__				:      Tuple       )  ->    Dict:
							if PartialState().distributed_type == DistributedType.TPU:
														xm.save(UpperCAmelCase__			,				UpperCAmelCase__       )
							elif PartialState().local_process_index == 0:
														torch.save(UpperCAmelCase__			,				UpperCAmelCase__       )



@contextmanager
def      __lowerCAmelCase						(							**UpperCAmelCase__				:      Any       )  ->    Dict:
							for key, value in kwargs.items():
														lowerCamelCase_							  =					str(UpperCAmelCase__       )

							yield

							for key in kwargs:
														if key.upper() in os.environ:
																					del os.environ[key.upper()]



def      __lowerCAmelCase						(							UpperCAmelCase__				:      Union[str, Any]       )  ->    str:
							if not hasattr(UpperCAmelCase__			,				"""__qualname__"""       ) and not hasattr(UpperCAmelCase__			,				"""__name__"""       ):
														lowerCamelCase_							  =					getattr(UpperCAmelCase__			,				"""__class__"""			,				UpperCAmelCase__       )
							if hasattr(UpperCAmelCase__			,				"""__qualname__"""       ):
														return obj.__qualname__
							if hasattr(UpperCAmelCase__			,				"""__name__"""       ):
														return obj.__name__
							return str(UpperCAmelCase__       )



def      __lowerCAmelCase						(							UpperCAmelCase__				:      str			,				UpperCAmelCase__				:      Union[str, Any]       )  ->    Dict:
							for key, value in source.items():
														if isinstance(UpperCAmelCase__			,				UpperCAmelCase__       ):
																					lowerCamelCase_							  =					destination.setdefault(UpperCAmelCase__			,				{}       )
																					merge_dicts(UpperCAmelCase__			,				UpperCAmelCase__       )
														else:
																					lowerCamelCase_							  =					value

							return destination



def      __lowerCAmelCase						(							UpperCAmelCase__				:      int = None       )  ->    bool:
							if port is None:
														lowerCamelCase_							  =					2_9_5_0_0
							with socket.socket(socket.AF_INET			,				socket.SOCK_STREAM       ) as s:
														return s.connect_ex(("""localhost""", port)       ) == 0





