"""simple docstring"""



# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


_UpperCamelCase								=		"Create a default config file for Accelerate with only a few flags set."

def  _A(		lowerCAmelCase="no"			,  lowerCAmelCase = default_json_config_file			,  lowerCAmelCase = False							):
							A__       :     Dict								=			Path(lowerCAmelCase							)
							path.parent.mkdir(parents=lowerCAmelCase			,  exist_ok=lowerCAmelCase							)
							if path.exists():
														print(
														    F'''Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`.'''							)
														return False
							A__       :     Optional[Any]								=			mixed_precision.lower()
							if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
														raise ValueError(
														    F'''`mixed_precision` should be one of \'no\', \'fp16\', \'bf16\', or \'fp8\'. Received {mixed_precision}'''							)
							A__       :     Any								=			{
							    """compute_environment""": """LOCAL_MACHINE""",
							    """mixed_precision""": mixed_precision,
							}
							if torch.cuda.is_available():
														A__       :     Any								=			torch.cuda.device_count()
														A__       :     Union[str, Any]								=			num_gpus
														A__       :     Union[str, Any]								=			False
														if num_gpus > 1:
																					A__       :     Optional[Any]								=			"""MULTI_GPU"""
														else:
																					A__       :     Dict								=			"""NO"""
							elif is_xpu_available() and use_xpu:
														A__       :     str								=			torch.xpu.device_count()
														A__       :     str								=			num_xpus
														A__       :     List[Any]								=			False
														if num_xpus > 1:
																					A__       :     int								=			"""MULTI_XPU"""
														else:
																					A__       :     Optional[int]								=			"""NO"""
							elif is_npu_available():
														A__       :     Optional[Any]								=			torch.npu.device_count()
														A__       :     int								=			num_npus
														A__       :     Any								=			False
														if num_npus > 1:
																					A__       :     int								=			"""MULTI_NPU"""
														else:
																					A__       :     List[Any]								=			"""NO"""
							else:
														A__       :     Optional[Any]								=			0
														A__       :     Optional[int]								=			True
														A__       :     List[Any]								=			1
														A__       :     Optional[Any]								=			"""NO"""
							A__       :     Optional[int]								=			ClusterConfig(**lowerCAmelCase							)
							config.to_json_file(lowerCAmelCase							)
							return path

def  _A(		lowerCAmelCase			,  lowerCAmelCase							):
							A__       :     Dict								=			parser.add_parser("""default"""			,  parents=lowerCAmelCase			,  help=lowerCAmelCase			,  formatter_class=lowerCAmelCase							)
							parser.add_argument(
							    """--config_file"""			,  default=lowerCAmelCase			,  help=(
							        """The path to use to store the config file. Will default to a file named default_config.yaml in the cache """
							        """location, which is the content of the environment `HF_HOME` suffixed with 'accelerate', or if you don't have """
							        """such an environment variable, your cache directory ('~/.cache' or the content of `XDG_CACHE_HOME`) suffixed """
							        """with 'huggingface'."""
							    )			,  dest="""save_location"""			,  )

							parser.add_argument(
							    """--mixed_precision"""			,  choices=["""no""", """fp16""", """bf16"""]			,  type=lowerCAmelCase			,  help="""Whether or not to use mixed precision training. """
							    """Choose between FP16 and BF16 (bfloat16) training. """
							    """BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later."""			,  default="""no"""			,  )
							parser.set_defaults(func=lowerCAmelCase							)
							return parser



def  _A(		lowerCAmelCase							):
							A__       :     Optional[Any]								=			write_basic_config(args.mixed_precision			,  args.save_location							)
							if config_file:
														print(F'''accelerate configuration saved at {config_file}'''							)






"""simple docstring"""



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







			'''simple docstring'''





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										A__       :     List[str]								=			0





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										A__       :     List[str]								=			AutoImageProcessor.from_pretrained("""openai/clip-vit-base-patch32"""							)
										self.assertIsInstance(snake_case_   ,      snake_case_							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with tempfile.TemporaryDirectory() as tmpdirname:
																	A__       :     Optional[Any]								=			Path(snake_case_							) / """preprocessor_config.json"""
																	A__       :     Any								=			Path(snake_case_							) / """config.json"""
																	json.dump(
																	    {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}   ,      open(snake_case_   ,      """w"""							)   ,      )
																	json.dump({"""model_type""": """clip"""}   ,      open(snake_case_   ,      """w"""							)							)

																	A__       :     Any								=			AutoImageProcessor.from_pretrained(snake_case_							)
																	self.assertIsInstance(snake_case_   ,      snake_case_							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with tempfile.TemporaryDirectory() as tmpdirname:
																	A__       :     str								=			Path(snake_case_							) / """preprocessor_config.json"""
																	A__       :     Union[str, Any]								=			Path(snake_case_							) / """config.json"""
																	json.dump(
																	    {"""feature_extractor_type""": """CLIPFeatureExtractor""", """processor_class""": """CLIPProcessor"""}   ,      open(snake_case_   ,      """w"""							)   ,      )
																	json.dump({"""model_type""": """clip"""}   ,      open(snake_case_   ,      """w"""							)							)

																	A__       :     Optional[int]								=			AutoImageProcessor.from_pretrained(snake_case_							)
																	self.assertIsInstance(snake_case_   ,      snake_case_							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with tempfile.TemporaryDirectory() as tmpdirname:
																	A__       :     Union[str, Any]								=			CLIPConfig()

																	# Create a dummy config file with image_proceesor_type
																	A__       :     Dict								=			Path(snake_case_							) / """preprocessor_config.json"""
																	A__       :     Optional[int]								=			Path(snake_case_							) / """config.json"""
																	json.dump(
																	    {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}   ,      open(snake_case_   ,      """w"""							)   ,      )
																	json.dump({"""model_type""": """clip"""}   ,      open(snake_case_   ,      """w"""							)							)

																	# remove image_processor_type to make sure config.json alone is enough to load image processor locally
																	A__       :     List[str]								=			AutoImageProcessor.from_pretrained(snake_case_							).to_dict()

																	config_dict.pop("""image_processor_type"""							)
																	A__       :     str								=			CLIPImageProcessor(**snake_case_							)

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

																	A__       :     Optional[int]								=			AutoImageProcessor.from_pretrained(snake_case_							)

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

										self.assertIsInstance(snake_case_   ,      snake_case_							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with tempfile.TemporaryDirectory() as tmpdirname:
																	A__       :     List[Any]								=			Path(snake_case_							) / """preprocessor_config.json"""
																	json.dump(
																	    {"""image_processor_type""": """CLIPImageProcessor""", """processor_class""": """CLIPProcessor"""}   ,      open(snake_case_   ,      """w"""							)   ,      )

																	A__       :     List[str]								=			AutoImageProcessor.from_pretrained(snake_case_							)
																	self.assertIsInstance(snake_case_   ,      snake_case_							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with self.assertRaisesRegex(
										    snake_case_   ,      """clip-base is not a local folder and is not a valid model identifier"""							):
																	A__       :     Optional[Any]								=			AutoImageProcessor.from_pretrained("""clip-base"""							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with self.assertRaisesRegex(
										    snake_case_   ,      r"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""							):
																	A__       :     Any								=			AutoImageProcessor.from_pretrained(snake_case_   ,      revision="""aaaaaa"""							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with self.assertRaisesRegex(
										    snake_case_   ,      """hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""   ,      ):
																	A__       :     Optional[Any]								=			AutoImageProcessor.from_pretrained("""hf-internal-testing/config-no-model"""							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										with self.assertRaises(snake_case_							):
																	A__       :     str								=			AutoImageProcessor.from_pretrained("""hf-internal-testing/test_dynamic_image_processor"""							)
										# If remote code is disabled, we can't load this config.
										with self.assertRaises(snake_case_							):
																	A__       :     int								=			AutoImageProcessor.from_pretrained(
																	    """hf-internal-testing/test_dynamic_image_processor"""   ,      trust_remote_code=snake_case_							)

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

										# Test image processor can be reloaded.
										with tempfile.TemporaryDirectory() as tmp_dir:
																	image_processor.save_pretrained(snake_case_							)
																	A__       :     Optional[Any]								=			AutoImageProcessor.from_pretrained(snake_case_   ,      trust_remote_code=snake_case_							)
										self.assertEqual(reloaded_image_processor.__class__.__name__   ,      """NewImageProcessor"""							)





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


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

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

																								A__       :     Optional[int]								=			CustomImageProcessor.from_pretrained(snake_case_							)

																	# 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(snake_case_							)
																								A__       :     int								=			AutoImageProcessor.from_pretrained(snake_case_							)
																								self.assertIsInstance(snake_case_   ,      snake_case_							)

										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 		lowerCamelCase					(			self							):







										'''simple docstring'''


										class 						__UpperCAmelCase    (__A     ):







													'''simple docstring'''
													_UpperCamelCase  :						Union[str, Any]					     =				True

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

																	# If remote is enabled, we load from the Hub
																	A__       :     Optional[Any]								=			AutoImageProcessor.from_pretrained(
																	    """hf-internal-testing/test_dynamic_image_processor"""   ,      trust_remote_code=snake_case_							)
																	self.assertEqual(image_processor.__class__.__name__   ,      """NewImageProcessor"""							)
																	self.assertTrue(not hasattr(snake_case_   ,      """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]





