


import json
import os
import shutil
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoConfig, BertConfig, GPTaConfig
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test


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

from test_module.custom_configuration import CustomConfig  # noqa E402


SCREAMING_SNAKE_CASE						:List[str]			=  {
    '''return_dict''': False,
    '''output_hidden_states''': True,
    '''output_attentions''': True,
    '''torchscript''': True,
    '''torch_dtype''': '''float16''',
    '''use_bfloat16''': True,
    '''tf_legacy_loss''': True,
    '''pruned_heads''': {'''a''': 1},
    '''tie_word_embeddings''': False,
    '''is_decoder''': True,
    '''cross_attention_hidden_size''': 1_28,
    '''add_cross_attention''': True,
    '''tie_encoder_decoder''': True,
    '''max_length''': 50,
    '''min_length''': 3,
    '''do_sample''': True,
    '''early_stopping''': True,
    '''num_beams''': 3,
    '''num_beam_groups''': 3,
    '''diversity_penalty''': 0.5,
    '''temperature''': 2.0,
    '''top_k''': 10,
    '''top_p''': 0.7,
    '''typical_p''': 0.2,
    '''repetition_penalty''': 0.8,
    '''length_penalty''': 0.8,
    '''no_repeat_ngram_size''': 5,
    '''encoder_no_repeat_ngram_size''': 5,
    '''bad_words_ids''': [1, 2, 3],
    '''num_return_sequences''': 3,
    '''chunk_size_feed_forward''': 5,
    '''output_scores''': True,
    '''return_dict_in_generate''': True,
    '''forced_bos_token_id''': 2,
    '''forced_eos_token_id''': 3,
    '''remove_invalid_values''': True,
    '''architectures''': ['''BertModel'''],
    '''finetuning_task''': '''translation''',
    '''id2label''': {0: '''label'''},
    '''label2id''': {'''label''': '''0'''},
    '''tokenizer_class''': '''BertTokenizerFast''',
    '''prefix''': '''prefix''',
    '''bos_token_id''': 6,
    '''pad_token_id''': 7,
    '''eos_token_id''': 8,
    '''sep_token_id''': 9,
    '''decoder_start_token_id''': 10,
    '''exponential_decay_length_penalty''': (5, 1.01),
    '''suppress_tokens''': [0, 1],
    '''begin_suppress_tokens''': 2,
    '''task_specific_params''': {'''translation''': '''some_params'''},
    '''problem_type''': '''regression''',
}


@is_staging_test
class      __lowerCAmelCase  (			unittest.TestCase       ):






  """simple docstring"""





  @classmethod
  def        lowerCAmelCase__	(  cls							:							Dict	) ->			List[Any]:




     """simple docstring"""
     snake_case_            =       TOKEN
     HfFolder.save_token(_lowerCAmelCase	)



  @classmethod
  def        lowerCAmelCase__	(  cls							:							str	) ->			List[str]:




     """simple docstring"""
     try:
        delete_repo(token=cls._token     ,					repo_id="test-config"	)
     except HTTPError:
        pass

     try:
        delete_repo(token=cls._token     ,					repo_id="valid_org/test-config-org"	)
     except HTTPError:
        pass

     try:
        delete_repo(token=cls._token     ,					repo_id="test-dynamic-config"	)
     except HTTPError:
        pass



  def        lowerCAmelCase__	(  self							:							Tuple	) ->			List[str]:




     """simple docstring"""
     snake_case_            =       BertConfig(
         vocab_size=9_9     ,					hidden_size=3_2     ,					num_hidden_layers=5     ,					num_attention_heads=4     ,					intermediate_size=3_7	)
     config.push_to_hub("test-config"     ,					use_auth_token=self._token	)

     snake_case_            =       BertConfig.from_pretrained(F'''{USER}/test-config'''	)
     for k, v in config.to_dict().items():
        if k != "transformers_version":
           self.assertEqual(_lowerCAmelCase     ,					getattr(_lowerCAmelCase     ,					_lowerCAmelCase	)	)

        # Reset repo
     delete_repo(token=self._token     ,					repo_id="test-config"	)

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
        config.save_pretrained(_lowerCAmelCase     ,					repo_id="test-config"     ,					push_to_hub=_lowerCAmelCase     ,					use_auth_token=self._token	)

     snake_case_            =       BertConfig.from_pretrained(F'''{USER}/test-config'''	)
     for k, v in config.to_dict().items():
        if k != "transformers_version":
           self.assertEqual(_lowerCAmelCase     ,					getattr(_lowerCAmelCase     ,					_lowerCAmelCase	)	)



  def        lowerCAmelCase__	(  self							:							int	) ->			Optional[Any]:




     """simple docstring"""
     snake_case_            =       BertConfig(
         vocab_size=9_9     ,					hidden_size=3_2     ,					num_hidden_layers=5     ,					num_attention_heads=4     ,					intermediate_size=3_7	)
     config.push_to_hub("valid_org/test-config-org"     ,					use_auth_token=self._token	)

     snake_case_            =       BertConfig.from_pretrained("valid_org/test-config-org"	)
     for k, v in config.to_dict().items():
        if k != "transformers_version":
           self.assertEqual(_lowerCAmelCase     ,					getattr(_lowerCAmelCase     ,					_lowerCAmelCase	)	)

        # Reset repo
     delete_repo(token=self._token     ,					repo_id="valid_org/test-config-org"	)

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
        config.save_pretrained(
            _lowerCAmelCase     ,					repo_id="valid_org/test-config-org"     ,					push_to_hub=_lowerCAmelCase     ,					use_auth_token=self._token	)

     snake_case_            =       BertConfig.from_pretrained("valid_org/test-config-org"	)
     for k, v in config.to_dict().items():
        if k != "transformers_version":
           self.assertEqual(_lowerCAmelCase     ,					getattr(_lowerCAmelCase     ,					_lowerCAmelCase	)	)



  def        lowerCAmelCase__	(  self							:							Dict	) ->			int:




     """simple docstring"""
     CustomConfig.register_for_auto_class()
     snake_case_            =       CustomConfig(attribute=4_2	)

     config.push_to_hub("test-dynamic-config"     ,					use_auth_token=self._token	)

     # This has added the proper auto_map field to the config
     self.assertDictEqual(config.auto_map     ,					{"AutoConfig": "custom_configuration.CustomConfig"}	)

     snake_case_            =       AutoConfig.from_pretrained(F'''{USER}/test-dynamic-config'''     ,					trust_remote_code=_lowerCAmelCase	)
     # Can't make an isinstance check because the new_config is from the FakeConfig class of a dynamic module
     self.assertEqual(new_config.__class__.__name__     ,					"CustomConfig"	)
     self.assertEqual(new_config.attribute     ,					4_2	)






class      __lowerCAmelCase  (			unittest.TestCase       ):






  """simple docstring"""





  def        lowerCAmelCase__	(  self							:							Tuple	) ->			Tuple:




     """simple docstring"""
     snake_case_            =       GPTaConfig()

     # attempt to modify each of int/float/bool/str config records and verify they were updated
     snake_case_            =       c.n_embd + 1  # int
     snake_case_            =       c.resid_pdrop + 1.0  # float
     snake_case_            =       not c.scale_attn_weights  # bool
     snake_case_            =       c.summary_type + "foo"  # str
     c.update_from_string(
         F'''n_embd={n_embd},resid_pdrop={resid_pdrop},scale_attn_weights={scale_attn_weights},summary_type={summary_type}'''	)
     self.assertEqual(_lowerCAmelCase     ,					c.n_embd     ,					"mismatch for key: n_embd"	)
     self.assertEqual(_lowerCAmelCase     ,					c.resid_pdrop     ,					"mismatch for key: resid_pdrop"	)
     self.assertEqual(_lowerCAmelCase     ,					c.scale_attn_weights     ,					"mismatch for key: scale_attn_weights"	)
     self.assertEqual(_lowerCAmelCase     ,					c.summary_type     ,					"mismatch for key: summary_type"	)



  def        lowerCAmelCase__	(  self							:							Optional[int]	) ->			Any:




     """simple docstring"""
     snake_case_            =       PretrainedConfig()
     snake_case_            =       [key for key in base_config.__dict__ if key not in config_common_kwargs]
     # If this part of the test fails, you have arguments to addin config_common_kwargs above.
     self.assertListEqual(
         _lowerCAmelCase     ,					["is_encoder_decoder", "_name_or_path", "_commit_hash", "transformers_version"]	)
     snake_case_            =       [key for key, value in config_common_kwargs.items() if value == getattr(_lowerCAmelCase     ,					_lowerCAmelCase	)]
     if len(_lowerCAmelCase	) > 0:
        raise ValueError(
            "The following keys are set with the default values in"
            " `test_configuration_common.config_common_kwargs` pick another value for them:"
            F''' {', '.join(_lowerCAmelCase	)}.'''	)



  def        lowerCAmelCase__	(  self							:							Optional[Any]	) ->			Union[str, Any]:




     """simple docstring"""
     with self.assertRaises(_lowerCAmelCase	):
        # config is in subfolder, the following should not work without specifying the subfolder
        snake_case_            =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert-subfolder"	)

     snake_case_            =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert-subfolder"     ,					subfolder="bert"	)

     self.assertIsNotNone(_lowerCAmelCase	)



  def        lowerCAmelCase__	(  self							:							str	) ->			str:




     """simple docstring"""
     # A mock response for an HTTP head request to emulate server down
     snake_case_            =       mock.Mock()
     snake_case_            =       5_0_0
     snake_case_            =       {}
     snake_case_            =       HTTPError
     snake_case_            =       {}

     # Download this model to make sure it's in the cache.
     snake_case_            =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert"	)

     # Under the mock environment we get a 500 error when trying to reach the model.
     with mock.patch("requests.Session.request"     ,					return_value=_lowerCAmelCase	) as mock_head:
        snake_case_            =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert"	)
        # This check we did call the fake head request
        mock_head.assert_called()



  def        lowerCAmelCase__	(  self							:							int	) ->			Optional[Any]:




     """simple docstring"""
     # This test is for deprecated behavior and can be removed in v5
     snake_case_            =       BertConfig.from_pretrained(
         "https://huggingface.co/hf-internal-testing/tiny-random-bert/resolve/main/config.json"	)



  def        lowerCAmelCase__	(  self							:							Optional[int]	) ->			Tuple:




     """simple docstring"""
     snake_case_            =       AutoConfig.from_pretrained("bert-base-cased"	)
     snake_case_            =       ["config.4.0.0.json"]

     with tempfile.TemporaryDirectory() as tmp_dir:
        configuration.save_pretrained(_lowerCAmelCase	)
        snake_case_            =       2
        json.dump(configuration.to_dict()     ,					open(os.path.join(_lowerCAmelCase     ,					"config.4.0.0.json"	)     ,					"w"	)	)

        # This should pick the new configuration file as the version of Transformers is > 4.0.0
        snake_case_            =       AutoConfig.from_pretrained(_lowerCAmelCase	)
        self.assertEqual(new_configuration.hidden_size     ,					2	)

        # Will need to be adjusted if we reach v42 and this test is still here.
        # Should pick the old configuration file as the version of Transformers is < 4.42.0
        snake_case_            =       ["config.42.0.0.json"]
        snake_case_            =       7_6_8
        configuration.save_pretrained(_lowerCAmelCase	)
        shutil.move(os.path.join(_lowerCAmelCase     ,					"config.4.0.0.json"	)     ,					os.path.join(_lowerCAmelCase     ,					"config.42.0.0.json"	)	)
        snake_case_            =       AutoConfig.from_pretrained(_lowerCAmelCase	)
        self.assertEqual(new_configuration.hidden_size     ,					7_6_8	)



  def        lowerCAmelCase__	(  self							:							str	) ->			List[str]:




     """simple docstring"""
     # This repo has two configuration files, one for v4.0.0 and above with a different hidden size.
     snake_case_            =       "hf-internal-testing/test-two-configs"

     import transformers as new_transformers

     snake_case_            =       "v4.0.0"
     snake_case_     ,     snake_case_            =       new_transformers.models.auto.AutoConfig.from_pretrained(
         _lowerCAmelCase     ,					return_unused_kwargs=_lowerCAmelCase	)
     self.assertEqual(new_configuration.hidden_size     ,					2	)
     # This checks `_configuration_file` ia not kept in the kwargs by mistake.
     self.assertDictEqual(_lowerCAmelCase     ,					{}	)

     # Testing an older version by monkey-patching the version in the module it's used.
     import transformers as old_transformers

     snake_case_            =       "v3.0.0"
     snake_case_            =       old_transformers.models.auto.AutoConfig.from_pretrained(_lowerCAmelCase	)
     self.assertEqual(old_configuration.hidden_size     ,					7_6_8	)








import logging
import os

from .state import PartialState


class      __lowerCAmelCase  (			logging.LoggerAdapter       ):






  """simple docstring"""





  @staticmethod
  def        lowerCAmelCase__	(  _lowerCAmelCase							:							int	) ->			Tuple:




     """simple docstring"""
     snake_case_            =       PartialState()
     return not main_process_only or (main_process_only and state.is_main_process)



  def        lowerCAmelCase__	(  self							:							Union[str, Any]     ,					_lowerCAmelCase							:							List[Any]     ,					_lowerCAmelCase							:							Tuple     ,					*_lowerCAmelCase							:							List[Any]     ,					**_lowerCAmelCase							:							Any	) ->			Optional[int]:




     """simple docstring"""
     if PartialState._shared_state == {}:
        raise RuntimeError(
            "You must initialize the accelerate state by calling either `PartialState()` or `Accelerator()` before using the logging utility."	)
     snake_case_            =       kwargs.pop("main_process_only"     ,					_lowerCAmelCase	)
     snake_case_            =       kwargs.pop("in_order"     ,					_lowerCAmelCase	)

     if self.isEnabledFor(_lowerCAmelCase	):
        if self._should_log(_lowerCAmelCase	):
           snake_case_     ,     snake_case_            =       self.process(_lowerCAmelCase     ,					_lowerCAmelCase	)
           self.logger.log(_lowerCAmelCase     ,					_lowerCAmelCase     ,					*_lowerCAmelCase     ,					**_lowerCAmelCase	)

        elif in_order:
           snake_case_            =       PartialState()
           for i in range(state.num_processes	):
              if i == state.process_index:
                 snake_case_     ,     snake_case_            =       self.process(_lowerCAmelCase     ,					_lowerCAmelCase	)
                 self.logger.log(_lowerCAmelCase     ,					_lowerCAmelCase     ,					*_lowerCAmelCase     ,					**_lowerCAmelCase	)
              state.wait_for_everyone()






def       _lowerCAmelCase    (				lowerCAmelCase_				:str		,    lowerCAmelCase_				:str = None				)->Any:
   '''simple docstring'''






   if log_level is None:
      snake_case_            =       os.environ.get("ACCELERATE_LOG_LEVEL"		,    lowerCAmelCase_				)
   snake_case_            =       logging.getLogger(lowerCAmelCase_				)
   if log_level is not None:
      logger.setLevel(log_level.upper()				)
      logger.root.setLevel(log_level.upper()				)
   return MultiProcessAdapter(lowerCAmelCase_		,    {}				)




