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 import save_json







def     A__	(       __A						):







    '''simple docstring'''
    _lowerCamelCase  :		Dict			    =    filter(lambda __A						: p.requires_grad ,			model.parameters()						)
    _lowerCamelCase  :		Any			    =    sum([np.prod(p.size()						) for p in model_parameters]						)
    return params


lowerCAmelCase      :				List[str]     			=logging.getLogger(__name__)




class 						__snake_case						(			pl.Callback		):





      '''simple docstring'''






      def 						_SCREAMING_SNAKE_CASE     (       self  :					List[str]     ,	_UpperCamelCase  :					int     ,	_UpperCamelCase  :					Tuple)					->Dict:


          """simple docstring"""


          _lowerCamelCase  :		int			    =    {F"""lr_group_{i}""": param['''lr'''] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups)}
          pl_module.logger.log_metrics(UpperCamelCase_)



      @rank_zero_only
      def 						_SCREAMING_SNAKE_CASE     (       self  :					Union[str, Any]     ,	_UpperCamelCase  :					pl.Trainer     ,	_UpperCamelCase  :					pl.LightningModule     ,	_UpperCamelCase  :					str     ,	_UpperCamelCase  :					List[Any]=True)					->str:


          """simple docstring"""


          logger.info(F"""***** {type_path} results at step {trainer.global_step:05d} *****""")
          _lowerCamelCase  :		int			    =    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
          _lowerCamelCase  :		Union[str, Any]			    =    Path(pl_module.hparams.output_dir)
          if type_path == "test":
              _lowerCamelCase  :		Tuple			    =    od / '''test_results.txt'''
              _lowerCamelCase  :		Optional[Any]			    =    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.
              _lowerCamelCase  :		str			    =    od / F"""{type_path}_results/{trainer.global_step:05d}.txt"""
              _lowerCamelCase  :		List[Any]			    =    od / F"""{type_path}_generations/{trainer.global_step:05d}.txt"""
              results_file.parent.mkdir(exist_ok=UpperCamelCase_)
              generations_file.parent.mkdir(exist_ok=UpperCamelCase_)
          with open(UpperCamelCase_     ,	"""a+""") as writer:
              for key in sorted(UpperCamelCase_):
                  if key in ["log", "progress_bar", "preds"]:
                      continue
                  _lowerCamelCase  :		int			    =    metrics[key]
                  if isinstance(UpperCamelCase_     ,	torch.Tensor):
                      _lowerCamelCase  :		Dict			    =    val.item()
                  _lowerCamelCase  :		List[str]			    =    F"""{key}: {val:.6f}\n"""
                  writer.write(UpperCamelCase_)

          if not save_generations:
              return

          if "preds" in metrics:
              _lowerCamelCase  :		Dict			    =    '''\n'''.join(metrics["""preds"""])
              generations_file.open("""w+""").write(UpperCamelCase_)



      @rank_zero_only
      def 						_SCREAMING_SNAKE_CASE     (       self  :					Tuple     ,	_UpperCamelCase  :					List[Any]     ,	_UpperCamelCase  :					Dict)					->Union[str, Any]:


          """simple docstring"""


          try:
              _lowerCamelCase  :		Union[str, Any]			    =    pl_module.model.model.num_parameters()
          except AttributeError:
              _lowerCamelCase  :		List[str]			    =    pl_module.model.num_parameters()

          _lowerCamelCase  :		List[str]			    =    count_trainable_parameters(UpperCamelCase_)
          # 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  :					Dict     ,	_UpperCamelCase  :					pl.Trainer     ,	_UpperCamelCase  :					pl.LightningModule)					->List[str]:


          """simple docstring"""


          save_json(pl_module.metrics     ,	pl_module.metrics_save_path)
          return self._write_logs(UpperCamelCase_     ,	UpperCamelCase_     ,	"""test""")




      @rank_zero_only
      def 						_SCREAMING_SNAKE_CASE     (       self  :					Optional[Any]     ,	_UpperCamelCase  :					pl.Trainer     ,	_UpperCamelCase  :					Any)					->List[Any]:


          """simple docstring"""


          save_json(pl_module.metrics     ,	pl_module.metrics_save_path)
          # Uncommenting this will save val generations
          # return self._write_logs(trainer, pl_module, "valid")







def     A__	(       __A ,			__A ,			__A=1 ,			__A=False						):







    '''simple docstring'''
    if metric == "rouge2":
        _lowerCamelCase  :		int			    =    '''{val_avg_rouge2:.4f}-{step_count}'''
    elif metric == "bleu":
        _lowerCamelCase  :		Tuple			    =    '''{val_avg_bleu:.4f}-{step_count}'''
    elif metric == "loss":
        _lowerCamelCase  :		Tuple			    =    '''{val_avg_loss:.4f}-{step_count}'''
    else:
        raise NotImplementedError(
            F"""seq2seq callbacks only support rouge2, bleu and loss, got {metric}, You can make your own by adding to"""
            """ this function."""						)

    _lowerCamelCase  :		Optional[Any]			    =    ModelCheckpoint(
        dirpath=_snake_case ,			filename=_snake_case ,			monitor=F"""val_{metric}""" ,			mode="""min""" if """loss""" in metric else """max""" ,			save_top_k=_snake_case ,			)
    return checkpoint_callback







def     A__	(       __A ,			__A						):







    '''simple docstring'''
    return EarlyStopping(
        monitor=F"""val_{metric}""" ,			mode="""min""" if """loss""" in metric else """max""" ,			patience=_snake_case ,			verbose=_snake_case ,			)








"""simple docstring"""





import unittest

from transformers import ErnieMConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
 import torch

 from transformers import (
     ErnieMForInformationExtraction,
     ErnieMForMultipleChoice,
     ErnieMForQuestionAnswering,
     ErnieMForSequenceClassification,
     ErnieMForTokenClassification,
     ErnieMModel,
 )
 from transformers.models.ernie_m.modeling_ernie_m import ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST



class 		snake_case_:






     def __init__(						self	:       Optional[int]       ,					UpperCamelCase_	:       Dict       ,					UpperCamelCase_	:       Optional[int]=1_3       ,					UpperCamelCase_	:       Dict=7       ,					UpperCamelCase_	:       Dict=True       ,					UpperCamelCase_	:       str=True       ,					UpperCamelCase_	:       int=True       ,					UpperCamelCase_	:       Optional[Any]=9_9       ,					UpperCamelCase_	:       Optional[int]=3_2       ,					UpperCamelCase_	:       Any=5       ,					UpperCamelCase_	:       Any=4       ,					UpperCamelCase_	:       Tuple=3_7       ,					UpperCamelCase_	:       Tuple="gelu"       ,					UpperCamelCase_	:       Union[str, Any]=0.1       ,					UpperCamelCase_	:       List[Any]=0.1       ,					UpperCamelCase_	:       int=5_1_2       ,					UpperCamelCase_	:       int=1_6       ,					UpperCamelCase_	:       List[Any]=2       ,					UpperCamelCase_	:       str=0.02       ,					UpperCamelCase_	:       int=3       ,					UpperCamelCase_	:       str=4       ,					UpperCamelCase_	:       Union[str, Any]=None       ,					):
           lowerCAmelCase       :						List[str]							= parent
           lowerCAmelCase       :						str							= batch_size
           lowerCAmelCase       :						Dict							= seq_length
           lowerCAmelCase       :						Dict							= is_training
           lowerCAmelCase       :						Tuple							= use_input_mask
           lowerCAmelCase       :						str							= use_labels
           lowerCAmelCase       :						Optional[Any]							= vocab_size
           lowerCAmelCase       :						Optional[Any]							= hidden_size
           lowerCAmelCase       :						Optional[int]							= num_hidden_layers
           lowerCAmelCase       :						List[Any]							= num_attention_heads
           lowerCAmelCase       :						Tuple							= intermediate_size
           lowerCAmelCase       :						Tuple							= hidden_act
           lowerCAmelCase       :						Optional[int]							= hidden_dropout_prob
           lowerCAmelCase       :						str							= attention_probs_dropout_prob
           lowerCAmelCase       :						str							= max_position_embeddings
           lowerCAmelCase       :						List[str]							= type_vocab_size
           lowerCAmelCase       :						str							= type_sequence_label_size
           lowerCAmelCase       :						int							= initializer_range
           lowerCAmelCase       :						Optional[Any]							= num_labels
           lowerCAmelCase       :						Any							= num_choices
           lowerCAmelCase       :						str							= scope






     def   lowerCamelCase__  (						self	:       Dict					):
           lowerCAmelCase       :						str							= ids_tensor([self.batch_size, self.seq_length]       ,					self.vocab_size					)

           lowerCAmelCase       :						List[Any]							= None
           if self.use_input_mask:
                 lowerCAmelCase       :						str							= random_attention_mask([self.batch_size, self.seq_length]					)

           lowerCAmelCase       :						List[Any]							= None
           lowerCAmelCase       :						str							= None
           lowerCAmelCase       :						str							= None
           if self.use_labels:
                 lowerCAmelCase       :						Tuple							= ids_tensor([self.batch_size]       ,					self.type_sequence_label_size					)
                 lowerCAmelCase       :						Dict							= ids_tensor([self.batch_size, self.seq_length]       ,					self.num_labels					)
                 lowerCAmelCase       :						List[Any]							= ids_tensor([self.batch_size]       ,					self.num_choices					)

           lowerCAmelCase       :						Any							= self.get_config()

           return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels






     def   lowerCamelCase__  (						self	:       str					):
           lowerCAmelCase       :						Tuple							= ids_tensor([self.batch_size, self.seq_length]       ,					self.vocab_size					)

           lowerCAmelCase       :						Optional[Any]							= None
           if self.use_input_mask:
                 lowerCAmelCase       :						Union[str, Any]							= random_attention_mask([self.batch_size, self.seq_length]					)
           lowerCAmelCase       :						Dict							= self.get_config()

           return config, input_ids, input_mask






     def   lowerCamelCase__  (						self	:       Union[str, Any]					):
           return ErnieMConfig(
               vocab_size=self.vocab_size       ,					hidden_size=self.hidden_size       ,					num_hidden_layers=self.num_hidden_layers       ,					num_attention_heads=self.num_attention_heads       ,					intermediate_size=self.intermediate_size       ,					hidden_act=self.hidden_act       ,					hidden_dropout_prob=self.hidden_dropout_prob       ,					attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,					max_position_embeddings=self.max_position_embeddings       ,					type_vocab_size=self.type_vocab_size       ,					initializer_range=self.initializer_range       ,					)






     def   lowerCamelCase__  (						self	:       Union[str, Any]       ,					UpperCamelCase_	:       Tuple       ,					UpperCamelCase_	:       Union[str, Any]       ,					UpperCamelCase_	:       Optional[Any]       ,					UpperCamelCase_	:       Dict       ,					UpperCamelCase_	:       Union[str, Any]       ,					UpperCamelCase_	:       Optional[Any]					):
           lowerCAmelCase       :						List[str]							= ErnieMModel(config=UpperCamelCase_					)
           model.to(UpperCamelCase_					)
           model.eval()
           lowerCAmelCase       :						List[Any]							= model(UpperCamelCase_       ,					return_dict=UpperCamelCase_					)
           self.parent.assertEqual(result.last_hidden_state.shape       ,					(self.batch_size, self.seq_length, self.hidden_size)					)






     def   lowerCamelCase__  (						self	:       Union[str, Any]       ,					UpperCamelCase_	:       Union[str, Any]       ,					UpperCamelCase_	:       Optional[Any]       ,					UpperCamelCase_	:       int       ,					UpperCamelCase_	:       List[Any]       ,					UpperCamelCase_	:       List[Any]       ,					UpperCamelCase_	:       Optional[Any]					):
           lowerCAmelCase       :						Optional[int]							= ErnieMForQuestionAnswering(config=UpperCamelCase_					)
           model.to(UpperCamelCase_					)
           model.eval()
           lowerCAmelCase       :						List[Any]							= model(
               UpperCamelCase_       ,					attention_mask=UpperCamelCase_       ,					start_positions=UpperCamelCase_       ,					end_positions=UpperCamelCase_       ,					)
           self.parent.assertEqual(result.start_logits.shape       ,					(self.batch_size, self.seq_length)					)
           self.parent.assertEqual(result.end_logits.shape       ,					(self.batch_size, self.seq_length)					)






     def   lowerCamelCase__  (						self	:       Dict       ,					UpperCamelCase_	:       Optional[int]       ,					UpperCamelCase_	:       Any       ,					UpperCamelCase_	:       List[str]       ,					UpperCamelCase_	:       Dict       ,					UpperCamelCase_	:       Optional[Any]       ,					UpperCamelCase_	:       Union[str, Any]					):
           lowerCAmelCase       :						Tuple							= ErnieMForInformationExtraction(config=UpperCamelCase_					)
           model.to(UpperCamelCase_					)
           model.eval()
           lowerCAmelCase       :						Tuple							= torch.ones_like(UpperCamelCase_       ,					dtype=torch.floataa					)
           lowerCAmelCase       :						str							= model(
               UpperCamelCase_       ,					attention_mask=UpperCamelCase_       ,					start_positions=UpperCamelCase_       ,					end_positions=UpperCamelCase_       ,					)
           self.parent.assertEqual(result.start_logits.shape       ,					(self.batch_size, self.seq_length)					)
           self.parent.assertEqual(result.end_logits.shape       ,					(self.batch_size, self.seq_length)					)






     def   lowerCamelCase__  (						self	:       Optional[int]       ,					UpperCamelCase_	:       Optional[int]       ,					UpperCamelCase_	:       Dict       ,					UpperCamelCase_	:       int       ,					UpperCamelCase_	:       int       ,					UpperCamelCase_	:       Union[str, Any]       ,					UpperCamelCase_	:       Any					):
           lowerCAmelCase       :						Union[str, Any]							= self.num_labels
           lowerCAmelCase       :						Tuple							= ErnieMForSequenceClassification(UpperCamelCase_					)
           model.to(UpperCamelCase_					)
           model.eval()
           lowerCAmelCase       :						Union[str, Any]							= model(UpperCamelCase_       ,					attention_mask=UpperCamelCase_       ,					labels=UpperCamelCase_					)
           self.parent.assertEqual(result.logits.shape       ,					(self.batch_size, self.num_labels)					)






     def   lowerCamelCase__  (						self	:       Optional[int]       ,					UpperCamelCase_	:       Any       ,					UpperCamelCase_	:       List[str]       ,					UpperCamelCase_	:       int       ,					UpperCamelCase_	:       Dict       ,					UpperCamelCase_	:       str       ,					UpperCamelCase_	:       List[str]					):
           lowerCAmelCase       :						Union[str, Any]							= self.num_labels
           lowerCAmelCase       :						List[Any]							= ErnieMForTokenClassification(config=UpperCamelCase_					)
           model.to(UpperCamelCase_					)
           model.eval()
           input_ids.to(UpperCamelCase_					)
           input_mask.to(UpperCamelCase_					)
           token_labels.to(UpperCamelCase_					)

           lowerCAmelCase       :						Optional[Any]							= model(UpperCamelCase_       ,					attention_mask=UpperCamelCase_       ,					labels=UpperCamelCase_					)

           self.parent.assertEqual(result.logits.shape       ,					(self.batch_size, self.seq_length, self.num_labels)					)






     def   lowerCamelCase__  (						self	:       Any       ,					UpperCamelCase_	:       str       ,					UpperCamelCase_	:       Optional[int]       ,					UpperCamelCase_	:       Tuple       ,					UpperCamelCase_	:       Dict       ,					UpperCamelCase_	:       Dict       ,					UpperCamelCase_	:       str					):
           lowerCAmelCase       :						Union[str, Any]							= self.num_choices
           lowerCAmelCase       :						Any							= ErnieMForMultipleChoice(config=UpperCamelCase_					)
           model.to(UpperCamelCase_					)
           model.eval()
           lowerCAmelCase       :						Any							= input_ids.unsqueeze(1					).expand(-1       ,					self.num_choices       ,					-1					).contiguous()
           lowerCAmelCase       :						int							= input_mask.unsqueeze(1					).expand(-1       ,					self.num_choices       ,					-1					).contiguous()
           lowerCAmelCase       :						Any							= model(
               UpperCamelCase_       ,					attention_mask=UpperCamelCase_       ,					labels=UpperCamelCase_       ,					)
           self.parent.assertEqual(result.logits.shape       ,					(self.batch_size, self.num_choices)					)






     def   lowerCamelCase__  (						self	:       Tuple					):
           lowerCAmelCase       :						List[Any]							= self.prepare_config_and_inputs()
           (
               (
               lowerCAmelCase
           ),   (
               lowerCAmelCase
           ),   (
               lowerCAmelCase
           ),   (
               lowerCAmelCase
           ),   (
               lowerCAmelCase
           ),   (
               lowerCAmelCase
           ),   
           )       :						str							= config_and_inputs
           lowerCAmelCase       :						Dict							= {'''input_ids''': input_ids, '''attention_mask''': input_mask}
           return config, inputs_dict



@require_torch
class 		snake_case_(							a__				,       a__				,       unittest.TestCase      ):
     __UpperCamelCase	=       (
         (
             ErnieMModel,
             ErnieMForMultipleChoice,
             ErnieMForQuestionAnswering,
             ErnieMForSequenceClassification,
             ErnieMForTokenClassification,
         )
         if is_torch_available()
         else ()
     )
     __UpperCamelCase	=       ()
     __UpperCamelCase	=       (
         {
             '''feature-extraction''': ErnieMModel,
             '''question-answering''': ErnieMForQuestionAnswering,
             '''text-classification''': ErnieMForSequenceClassification,
             '''token-classification''': ErnieMForTokenClassification,
             '''zero-shot''': ErnieMForSequenceClassification,
         }
         if is_torch_available()
         else {}
     )
     __UpperCamelCase	=       False






     def   lowerCamelCase__  (						self	:       Optional[Any]       ,					UpperCamelCase_	:       str       ,					UpperCamelCase_	:       List[Any]       ,					UpperCamelCase_	:       Tuple       ,					UpperCamelCase_	:       int       ,					UpperCamelCase_	:       Any					):
           if pipeline_test_casse_name == "QAPipelineTests":
                 return True

           return False






     def   lowerCamelCase__  (						self	:       Dict					):
           lowerCAmelCase       :						str							= ErnieMModelTester(self					)
           lowerCAmelCase       :						Optional[Any]							= ConfigTester(self       ,					config_class=UpperCamelCase_       ,					hidden_size=3_7					)






     def   lowerCamelCase__  (						self	:       Tuple					):
           self.config_tester.run_common_tests()






     def   lowerCamelCase__  (						self	:       List[str]					):
           lowerCAmelCase       :						str							= self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_model(*UpperCamelCase_					)






     def   lowerCamelCase__  (						self	:       Union[str, Any]					):
           lowerCAmelCase       :						Optional[Any]							= self.model_tester.prepare_config_and_inputs()
           for type in ["absolute", "relative_key", "relative_key_query"]:
                 lowerCAmelCase       :						Union[str, Any]							= type
                 self.model_tester.create_and_check_model(*UpperCamelCase_					)






     def   lowerCamelCase__  (						self	:       Tuple					):
           lowerCAmelCase       :						Dict							= self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_for_multiple_choice(*UpperCamelCase_					)






     def   lowerCamelCase__  (						self	:       Dict					):
           lowerCAmelCase       :						int							= self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_for_question_answering(*UpperCamelCase_					)






     def   lowerCamelCase__  (						self	:       Optional[int]					):
           lowerCAmelCase       :						Tuple							= self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_for_sequence_classification(*UpperCamelCase_					)






     def   lowerCamelCase__  (						self	:       Tuple					):
           lowerCAmelCase       :						List[str]							= self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_for_information_extraction(*UpperCamelCase_					)






     def   lowerCamelCase__  (						self	:       List[Any]					):
           lowerCAmelCase       :						str							= self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_for_token_classification(*UpperCamelCase_					)






     @slow
     def   lowerCamelCase__  (						self	:       List[str]					):
           for model_name in ERNIE_M_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                 lowerCAmelCase       :						int							= ErnieMModel.from_pretrained(UpperCamelCase_					)
                 self.assertIsNotNone(UpperCamelCase_					)



@require_torch
class 		snake_case_(							unittest.TestCase      ):






     @slow
     def   lowerCamelCase__  (						self	:       Dict					):
           lowerCAmelCase       :						Optional[int]							= ErnieMModel.from_pretrained('''susnato/ernie-m-base_pytorch'''					)
           model.eval()
           lowerCAmelCase       :						Optional[int]							= torch.tensor([[0, 1, 2, 3, 4, 5]]					)
           lowerCAmelCase       :						str							= model(UpperCamelCase_					)[0]

           # TODO Replace vocab size
           lowerCAmelCase       :						Any							= 7_6_8

           lowerCAmelCase       :						int							= torch.Size((1, 6, hidden_size)					)
           self.assertEqual(output.shape       ,					UpperCamelCase_					)

           lowerCAmelCase       :						str							= torch.tensor(
               [[[-0.0_012, 0.1_245, -0.0_214], [-0.0_742, 0.0_244, -0.0_771], [-0.0_333, 0.1_164, -0.1_554]]]					)

           self.assertTrue(torch.allclose(output[:, :3, :3]       ,					UpperCamelCase_       ,					atol=1E-3					)					)


