






'''simple docstring'''




import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format='''%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s''',
    datefmt='''%m/%d/%Y %H:%M:%S''',
    level=logging.INFO,
)
__lowerCamelCase :       List[Any] 				=       logging.getLogger(__name__)

def 						__UpperCAmelCase							(			__magic_name__    )->     List[str]:



			"""simple docstring"""



			snake_case_					:       Any =							git.Repo(search_parent_directories=lowerCAmelCase_    )
			snake_case_					:       Optional[Any] =							{
			    "repo_id": str(lowerCAmelCase_    ),
			    "repo_sha": str(repo.head.object.hexsha    ),
			    "repo_branch": str(repo.active_branch    ),
			}

			with open(os.path.join(lowerCAmelCase_  ,"git_log.json"    )  ,"w"    ) as f:
						json.dump(lowerCAmelCase_  ,lowerCAmelCase_  ,indent=4    )

def 						__UpperCAmelCase							(			__magic_name__    )->     int:



			"""simple docstring"""



			if params.n_gpu <= 0:
						snake_case_					:       List[Any] =							0
						snake_case_					:       Optional[Any] =							-1
						snake_case_					:       Optional[int] =							True
						snake_case_					:       List[Any] =							False
						return

			assert torch.cuda.is_available()

			logger.info("Initializing GPUs"    )
			if params.n_gpu > 1:
						assert params.local_rank != -1

						snake_case_					:       Tuple =							int(os.environ["WORLD_SIZE"]    )
						snake_case_					:       Union[str, Any] =							int(os.environ["N_GPU_NODE"]    )
						snake_case_					:       Any =							int(os.environ["RANK"]    )

						# number of nodes / node ID
						snake_case_					:       Tuple =							params.world_size // params.n_gpu_per_node
						snake_case_					:       Tuple =							params.global_rank // params.n_gpu_per_node
						snake_case_					:       Optional[Any] =							True

						assert params.n_nodes == int(os.environ["N_NODES"]    )
						assert params.node_id == int(os.environ["NODE_RANK"]    )

			# local job (single GPU)
			else:
						assert params.local_rank == -1

						snake_case_					:       str =							1
						snake_case_					:       Optional[Any] =							0
						snake_case_					:       Any =							0
						snake_case_					:       Union[str, Any] =							0
						snake_case_					:       Tuple =							1
						snake_case_					:       Optional[Any] =							1
						snake_case_					:       int =							False

			# sanity checks
			assert params.n_nodes >= 1
			assert 0 <= params.node_id < params.n_nodes
			assert 0 <= params.local_rank <= params.global_rank < params.world_size
			assert params.world_size == params.n_nodes * params.n_gpu_per_node

			# define whether this is the master process / if we are in multi-node distributed mode
			snake_case_					:       Any =							params.node_id == 0 and params.local_rank == 0
			snake_case_					:       int =							params.n_nodes > 1

			# summary
			snake_case_					:       Dict =							F'''--- Global rank: {params.global_rank} - '''
			logger.info(PREFIX + "Number of nodes: %i" % params.n_nodes    )
			logger.info(PREFIX + "Node ID        : %i" % params.node_id    )
			logger.info(PREFIX + "Local rank     : %i" % params.local_rank    )
			logger.info(PREFIX + "World size     : %i" % params.world_size    )
			logger.info(PREFIX + "GPUs per node  : %i" % params.n_gpu_per_node    )
			logger.info(PREFIX + "Master         : %s" % str(params.is_master    )    )
			logger.info(PREFIX + "Multi-node     : %s" % str(params.multi_node    )    )
			logger.info(PREFIX + "Multi-GPU      : %s" % str(params.multi_gpu    )    )
			logger.info(PREFIX + "Hostname       : %s" % socket.gethostname()    )

			# set GPU device
			torch.cuda.set_device(params.local_rank    )

			# initialize multi-GPU
			if params.multi_gpu:
						logger.info("Initializing PyTorch distributed"    )
						torch.distributed.init_process_group(
						    init_method="env://"  ,backend="nccl"  ,)






def 						__UpperCAmelCase							(			__magic_name__    )->     Dict:



			"""simple docstring"""



			np.random.seed(args.seed    )
			torch.manual_seed(args.seed    )
			if args.n_gpu > 0:
						torch.cuda.manual_seed_all(args.seed    )









"""simple docstring"""






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

import transformers
import transformers.models.auto
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.bert.configuration_bert import BertConfig
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


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

from test_module.custom_configuration import CustomConfig  # noqa E402


snake_case   					=				get_tests_dir('''fixtures/dummy-config.json''')



class     UpperCAmelCase    (	unittest.TestCase  ):



   def    __UpperCAmelCase    (       self    :	Any				):



     """simple docstring"""
     _snake_case   	=      0



   def    __UpperCAmelCase    (       self    :	Optional[Any]				):



     """simple docstring"""
     self.assertIsNotNone(transformers.models.auto.__spec__				)
     self.assertIsNotNone(importlib.util.find_spec('''transformers.models.auto'''				)				)



   def    __UpperCAmelCase    (       self    :	List[str]				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained('''bert-base-uncased'''				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Union[str, Any]				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	int				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Union[str, Any]				):



     """simple docstring"""
     _snake_case   	=      AutoConfig.for_model('''roberta'''				)
     self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with tempfile.TemporaryDirectory() as tmp_dir:
       # This model name contains bert and roberta, but roberta ends up being picked.
       _snake_case   	=      os.path.join(__lowerCamelCase							,       '''fake-roberta'''				)
       os.makedirs(__lowerCamelCase							,       exist_ok=__lowerCamelCase				)
       with open(os.path.join(__lowerCamelCase							,       '''config.json'''				)							,       '''w'''				) as f:
         f.write(json.dumps({}				)				)
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
       self.assertEqual(type(__lowerCamelCase				)							,       __lowerCamelCase				)



   def    __UpperCAmelCase    (       self    :	Optional[Any]				):



     """simple docstring"""
     try:
       AutoConfig.register('''custom'''							,       __lowerCamelCase				)
       # Wrong model type will raise an error
       with self.assertRaises(__lowerCamelCase				):
         AutoConfig.register('''model'''							,       __lowerCamelCase				)
       # Trying to register something existing in the Transformers library will raise an error
       with self.assertRaises(__lowerCamelCase				):
         AutoConfig.register('''bert'''							,       __lowerCamelCase				)

       # Now that the config is registered, it can be used as any other config with the auto-API
       _snake_case   	=      CustomConfig()
       with tempfile.TemporaryDirectory() as tmp_dir:
         config.save_pretrained(__lowerCamelCase				)
         _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase				)
         self.assertIsInstance(__lowerCamelCase							,       __lowerCamelCase				)

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



   def    __UpperCAmelCase    (       self    :	Optional[int]				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       '''bert-base is not a local folder and is not a valid model identifier'''				):
       _snake_case   	=      AutoConfig.from_pretrained('''bert-base'''				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       R'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''				):
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase							,       revision='''aaaaaa'''				)



   def    __UpperCAmelCase    (       self    :	Dict				):



     """simple docstring"""
     with self.assertRaisesRegex(
         __lowerCamelCase							,       '''hf-internal-testing/no-config-test-repo does not appear to have a file named config.json.'''							,       ):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/no-config-test-repo'''				)



   def    __UpperCAmelCase    (       self    :	str				):



     """simple docstring"""
     # If remote code is not set, we will time out when asking whether to load the model.
     with self.assertRaises(__lowerCamelCase				):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''				)
     # If remote code is disabled, we can't load this config.
     with self.assertRaises(__lowerCamelCase				):
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)

     _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
     self.assertEqual(config.__class__.__name__							,       '''NewModelConfig'''				)

     # Test config can be reloaded.
     with tempfile.TemporaryDirectory() as tmp_dir:
       config.save_pretrained(__lowerCamelCase				)
       _snake_case   	=      AutoConfig.from_pretrained(__lowerCamelCase							,       trust_remote_code=__lowerCamelCase				)
     self.assertEqual(reloaded_config.__class__.__name__							,       '''NewModelConfig'''				)





   def    __UpperCAmelCase    (       self    :	Union[str, Any]				):



     """simple docstring"""



     class     UpperCAmelCase    (	__SCREAMING_SNAKE_CASE  ):
        A__		:	List[Any]           =  '''new-model'''

     try:
       AutoConfig.register('''new-model'''							,       __lowerCamelCase				)
       # If remote code is not set, the default is to use local
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfigLocal'''				)

       # If remote code is disabled, we load the local one.
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfigLocal'''				)

       # If remote is enabled, we load from the Hub
       _snake_case   	=      AutoConfig.from_pretrained('''hf-internal-testing/test_dynamic_model'''							,       trust_remote_code=__lowerCamelCase				)
       self.assertEqual(config.__class__.__name__							,       '''NewModelConfig'''				)

     finally:
       if "new-model" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["new-model"]





