


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


UpperCAmelCase_				:			Any   =	get_tests_dir('fixtures/dummy-config.json')
class 			SCREAMING_SNAKE_CASE__  (    unittest.TestCase      ):
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Dict:
             a_ :		str							=  0
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					)		->      int:
             self.assertIsNotNone(transformers.models.auto.__spec__					)
             self.assertIsNotNone(importlib.util.find_spec('transformers.models.auto'					)					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      List[str]:
             a_ :		List[str]							=  AutoConfig.from_pretrained('bert-base-uncased'					)
             self.assertIsInstance(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					)		->      Optional[Any]:
             a_ :		Tuple							=  AutoConfig.from_pretrained(SCREAMING_SNAKE_CASE__					)
             self.assertIsInstance(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      List[Any]:
             a_ :		List[str]							=  AutoConfig.from_pretrained(SCREAMING_SNAKE_CASE__					)
             self.assertIsInstance(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					)		->      Tuple:
             a_ :		int							=  AutoConfig.for_model('roberta'					)
             self.assertIsInstance(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					)		->      int:
             with tempfile.TemporaryDirectory() as tmp_dir:
                    # This model name contains bert and roberta, but roberta ends up being picked.
                    a_ :		str							=  os.path.join(SCREAMING_SNAKE_CASE__					,							'fake-roberta'					)
                    os.makedirs(SCREAMING_SNAKE_CASE__					,							exist_ok=SCREAMING_SNAKE_CASE__					)
                    with open(os.path.join(SCREAMING_SNAKE_CASE__					,							'config.json'					)					,							'w'					) as f:
                           f.write(json.dumps({}					)					)
                    a_ :		Optional[int]							=  AutoConfig.from_pretrained(SCREAMING_SNAKE_CASE__					)
                    self.assertEqual(type(SCREAMING_SNAKE_CASE__					)					,							SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					)		->      int:
             try:
                    AutoConfig.register('custom'					,							SCREAMING_SNAKE_CASE__					)
                    # Wrong model type will raise an error
                    with self.assertRaises(SCREAMING_SNAKE_CASE__					):
                           AutoConfig.register('model'					,							SCREAMING_SNAKE_CASE__					)
                    # Trying to register something existing in the Transformers library will raise an error
                    with self.assertRaises(SCREAMING_SNAKE_CASE__					):
                           AutoConfig.register('bert'					,							SCREAMING_SNAKE_CASE__					)

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

             finally:
                    if "custom" in CONFIG_MAPPING._extra_content:
                           del CONFIG_MAPPING._extra_content["custom"]
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					)		->      Any:
             with self.assertRaisesRegex(
                 SCREAMING_SNAKE_CASE__					,							'bert-base is not a local folder and is not a valid model identifier'					):
                    a_ :		int							=  AutoConfig.from_pretrained('bert-base'					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					)		->      Tuple:
             with self.assertRaisesRegex(
                 SCREAMING_SNAKE_CASE__					,							r'aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'					):
                    a_ :		int							=  AutoConfig.from_pretrained(SCREAMING_SNAKE_CASE__					,							revision='aaaaaa'					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      List[Any]:
             with self.assertRaisesRegex(
                 SCREAMING_SNAKE_CASE__					,							'hf-internal-testing/no-config-test-repo does not appear to have a file named config.json.'					,							):
                    a_ :		Tuple							=  AutoConfig.from_pretrained('hf-internal-testing/no-config-test-repo'					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					)		->      Tuple:
             # If remote code is not set, we will time out when asking whether to load the model.
             with self.assertRaises(SCREAMING_SNAKE_CASE__					):
                    a_ :		Union[str, Any]							=  AutoConfig.from_pretrained('hf-internal-testing/test_dynamic_model'					)
             # If remote code is disabled, we can't load this config.
             with self.assertRaises(SCREAMING_SNAKE_CASE__					):
                    a_ :		Optional[int]							=  AutoConfig.from_pretrained('hf-internal-testing/test_dynamic_model'					,							trust_remote_code=SCREAMING_SNAKE_CASE__					)

             a_ :		List[Any]							=  AutoConfig.from_pretrained('hf-internal-testing/test_dynamic_model'					,							trust_remote_code=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(config.__class__.__name__					,							'NewModelConfig'					)

             # Test config can be reloaded.
             with tempfile.TemporaryDirectory() as tmp_dir:
                    config.save_pretrained(SCREAMING_SNAKE_CASE__					)
                    a_ :		Optional[int]							=  AutoConfig.from_pretrained(SCREAMING_SNAKE_CASE__					,							trust_remote_code=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(reloaded_config.__class__.__name__					,							'NewModelConfig'					)

      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					)		->      Any:
             class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
                   snake_case__    :					Union[str, Any]					       =						'''new-model'''

             try:
                    AutoConfig.register('new-model'					,							SCREAMING_SNAKE_CASE__					)
                    # If remote code is not set, the default is to use local
                    a_ :		Optional[int]							=  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.
                    a_ :		Union[str, Any]							=  AutoConfig.from_pretrained('hf-internal-testing/test_dynamic_model'					,							trust_remote_code=SCREAMING_SNAKE_CASE__					)
                    self.assertEqual(config.__class__.__name__					,							'NewModelConfigLocal'					)

                    # If remote is enabled, we load from the Hub
                    a_ :		List[Any]							=  AutoConfig.from_pretrained('hf-internal-testing/test_dynamic_model'					,							trust_remote_code=SCREAMING_SNAKE_CASE__					)
                    self.assertEqual(config.__class__.__name__					,							'NewModelConfig'					)

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






import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only
from utils_rag import save_json







def 	SCREAMING_SNAKE_CASE_						(      __A			:      Any    )     ->		Optional[int]:

       """simple docstring"""

       a_ :		str							=  filter(lambda __A    : p.requires_grad  ,   model.parameters()    )
       a_ :		List[Any]							=  sum([np.prod(p.size()    ) for p in model_parameters]    )
       return params


UpperCAmelCase_				:			Dict   =	logging.getLogger(__name__)







def 	SCREAMING_SNAKE_CASE_						(      __A			:      int  ,   __A			:      int    )     ->		List[str]:

       """simple docstring"""

       if metric == "rouge2":
              a_ :		Dict							=  '{val_avg_rouge2:.4f}-{step_count}'
       elif metric == "bleu":
              a_ :		Dict							=  '{val_avg_bleu:.4f}-{step_count}'
       elif metric == "em":
              a_ :		Union[str, Any]							=  '{val_avg_em:.4f}-{step_count}'
       elif metric == "loss":
              a_ :		str							=  '{val_avg_loss:.4f}-{step_count}'
       else:
              raise NotImplementedError(
                  F"""seq2seq callbacks only support rouge2 and bleu, got {metric}, You can make your own by adding to this"""
                  ' function.'    )

       a_ :		Dict							=  ModelCheckpoint(
           dirpath=__A  ,   filename=__A  ,   monitor=F"""val_{metric}"""  ,   mode='max'  ,   save_top_k=1  ,   every_n_epochs=1  ,   )

       return checkpoint_callback







def 	SCREAMING_SNAKE_CASE_						(      __A			:      int  ,   __A			:      List[Any]    )     ->		int:

       """simple docstring"""

       return EarlyStopping(
           monitor=F"""val_{metric}"""  ,   mode='min' if 'loss' in metric else 'max'  ,   patience=__A  ,   verbose=__A  ,   )
class 			SCREAMING_SNAKE_CASE__  (    pl.Callback      ):
      def 	SCREAMING_SNAKE_CASE							(	self    :	Dict					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	List[str]					)		->      Any:
             a_ :		int							=  {F"""lr_group_{i}""": param['lr'] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups					)}
             pl_module.logger.log_metrics(SCREAMING_SNAKE_CASE__					)
      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[str]					,							SCREAMING_SNAKE_CASE__    :	pl.Trainer					,							SCREAMING_SNAKE_CASE__    :	pl.LightningModule					,							SCREAMING_SNAKE_CASE__    :	str					,							SCREAMING_SNAKE_CASE__    :	List[str]=True					)		->      None:
             logger.info(F"""***** {type_path} results at step {trainer.global_step:05d} *****"""					)
             a_ :		List[str]							=  trainer.callback_metrics
             trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ['log', 'progress_bar', 'preds']}					)
             # Log results
             a_ :		Optional[int]							=  Path(pl_module.hparams.output_dir					)
             if type_path == "test":
                    a_ :		Tuple							=  od / 'test_results.txt'
                    a_ :		Optional[int]							=  od / 'test_generations.txt'
             else:
                    # this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
                    # If people want this it will be easy enough to add back.
                    a_ :		str							=  od / F"""{type_path}_results/{trainer.global_step:05d}.txt"""
                    a_ :		Optional[Any]							=  od / F"""{type_path}_generations/{trainer.global_step:05d}.txt"""
                    results_file.parent.mkdir(exist_ok=SCREAMING_SNAKE_CASE__					)
                    generations_file.parent.mkdir(exist_ok=SCREAMING_SNAKE_CASE__					)
             with open(SCREAMING_SNAKE_CASE__					,							'a+'					) as writer:
                    for key in sorted(SCREAMING_SNAKE_CASE__					):
                           if key in ["log", "progress_bar", "preds"]:
                                  continue
                           a_ :		int							=  metrics[key]
                           if isinstance(SCREAMING_SNAKE_CASE__					,							torch.Tensor					):
                                  a_ :		List[str]							=  val.item()
                           a_ :		int							=  F"""{key}: {val:.6f}\n"""
                           writer.write(SCREAMING_SNAKE_CASE__					)

             if not save_generations:
                    return

             if "preds" in metrics:
                    a_ :		Optional[Any]							=  '\n'.join(metrics['preds']					)
                    generations_file.open('w+'					).write(SCREAMING_SNAKE_CASE__					)
      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					,							SCREAMING_SNAKE_CASE__    :	Dict					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]					)		->      Dict:
             try:
                    a_ :		Any							=  pl_module.model.model.num_parameters()
             except AttributeError:
                    a_ :		List[str]							=  pl_module.model.num_parameters()

             a_ :		Any							=  count_trainable_parameters(SCREAMING_SNAKE_CASE__					)
             # mp stands for million parameters
             trainer.logger.log_metrics({'n_params': npars, 'mp': npars / 1E6, 'grad_mp': n_trainable_pars / 1E6}					)
      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	pl.Trainer					,							SCREAMING_SNAKE_CASE__    :	pl.LightningModule					)		->      Optional[int]:
             save_json(pl_module.metrics					,							pl_module.metrics_save_path					)
             return self._write_logs(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							'test'					)

      @rank_zero_only
      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					,							SCREAMING_SNAKE_CASE__    :	pl.Trainer					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]					)		->      Optional[Any]:
             save_json(pl_module.metrics					,							pl_module.metrics_save_path					)
             # Uncommenting this will save val generations
             # return self._write_logs(trainer, pl_module, "valid")


