

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 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Union[str, Any]      )			->      List[Any]:




				'''simple docstring'''





				A__        =			filter(lambda SCREAMING_SNAKE_CASE_      : p.requires_grad	,       model.parameters()      )
				A__        =			sum([np.prod(p.size()      ) for p in model_parameters]      )
				return params


lowerCAmelCase__	   = logging.getLogger(__name__)



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




				'''simple docstring'''





				if metric == "rouge2":
								A__        =			"{val_avg_rouge2:.4f}-{step_count}"
				elif metric == "bleu":
								A__        =			"{val_avg_bleu:.4f}-{step_count}"
				elif metric == "em":
								A__        =			"{val_avg_em:.4f}-{step_count}"
				elif metric == "loss":
								A__        =			"{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__        =			ModelCheckpoint(
				    dirpath=SCREAMING_SNAKE_CASE_	,       filename=SCREAMING_SNAKE_CASE_	,       monitor=F'val_{metric}'	,       mode="max"	,       save_top_k=1	,       every_n_epochs=1	,       )

				return checkpoint_callback



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Optional[Any]	,       SCREAMING_SNAKE_CASE_:							Tuple      )			->      List[Any]:




				'''simple docstring'''





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





class 					a__						(					pl.Callback			):
			"""simple docstring"""







			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase       )  ->						List[str]:
							'''simple docstring'''






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


			@rank_zero_only
			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase=True       )  ->						None:
							'''simple docstring'''






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

							if not save_generations:
											return

							if "preds" in metrics:
											A__        =			"\n".join(metrics["preds"]       )
											generations_file.open("w+"       ).write(lowercase       )


			@rank_zero_only
			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase       )  ->						Optional[Any]:
							'''simple docstring'''






							try:
											A__        =			pl_module.model.model.num_parameters()
							except AttributeError:
											A__        =			pl_module.model.num_parameters()

							A__        =			count_trainable_parameters(lowercase       )
							# 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      UpperCamelCase		(    self      ,      lowercase      ,      lowercase       )  ->						int:
							'''simple docstring'''






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


			@rank_zero_only
			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase       )  ->						str:
							'''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")




from __future__ import annotations

import copy
import tempfile
import unittest

from transformers import CONFIG_MAPPING, AutoConfig, BertConfig, GPTaConfig, TaConfig, TapasConfig, is_tf_available
from transformers.testing_utils import (
    DUMMY_UNKNOWN_IDENTIFIER,
    SMALL_MODEL_IDENTIFIER,
    RequestCounter,
    require_tensorflow_probability,
    require_tf,
    slow,
)

from ..bert.test_modeling_bert import BertModelTester


if is_tf_available():
	from transformers import (
	    TFAutoModel,
	    TFAutoModelForCausalLM,
	    TFAutoModelForMaskedLM,
	    TFAutoModelForPreTraining,
	    TFAutoModelForQuestionAnswering,
	    TFAutoModelForSeqaSeqLM,
	    TFAutoModelForSequenceClassification,
	    TFAutoModelForTableQuestionAnswering,
	    TFAutoModelForTokenClassification,
	    TFAutoModelWithLMHead,
	    TFBertForMaskedLM,
	    TFBertForPreTraining,
	    TFBertForQuestionAnswering,
	    TFBertForSequenceClassification,
	    TFBertModel,
	    TFFunnelBaseModel,
	    TFFunnelModel,
	    TFGPTaLMHeadModel,
	    TFRobertaForMaskedLM,
	    TFTaForConditionalGeneration,
	    TFTapasForQuestionAnswering,
	)
	from transformers.models.auto.modeling_tf_auto import (
	    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
	    TF_MODEL_FOR_MASKED_LM_MAPPING,
	    TF_MODEL_FOR_PRETRAINING_MAPPING,
	    TF_MODEL_FOR_QUESTION_ANSWERING_MAPPING,
	    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
	    TF_MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING,
	    TF_MODEL_MAPPING,
	)
	from transformers.models.bert.modeling_tf_bert import TF_BERT_PRETRAINED_MODEL_ARCHIVE_LIST
	from transformers.models.gpta.modeling_tf_gpta import TF_GPT2_PRETRAINED_MODEL_ARCHIVE_LIST
	from transformers.models.ta.modeling_tf_ta import TF_T5_PRETRAINED_MODEL_ARCHIVE_LIST
	from transformers.models.tapas.modeling_tf_tapas import TF_TAPAS_PRETRAINED_MODEL_ARCHIVE_LIST

class 					a__						(					snake_case			):
			"""simple docstring"""







			__lowerCamelCase        =       'new-model'


if is_tf_available():

	class 					a__						(					snake_case			):
				"""simple docstring"""







				__lowerCamelCase        =       NewModelConfig

@require_tf
class 					a__						(					unittest.TestCase			):
			"""simple docstring"""







			@slow
			def      UpperCamelCase		(    self       )  ->						List[Any]:
							'''simple docstring'''






							A__        =			"bert-base-cased"
							A__        =			AutoConfig.from_pretrained(lowercase       )
							self.assertIsNotNone(lowercase       )
							self.assertIsInstance(lowercase      ,      lowercase       )

							A__        =			TFAutoModel.from_pretrained(lowercase       )
							self.assertIsNotNone(lowercase       )
							self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			def      UpperCamelCase		(    self       )  ->						Optional[Any]:
							'''simple docstring'''






							A__        =			"bert-base-cased"
							A__        =			AutoConfig.from_pretrained(lowercase       )
							self.assertIsNotNone(lowercase       )
							self.assertIsInstance(lowercase      ,      lowercase       )

							A__        =			TFAutoModelForPreTraining.from_pretrained(lowercase       )
							self.assertIsNotNone(lowercase       )
							self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			def      UpperCamelCase		(    self       )  ->						Dict:
							'''simple docstring'''






							for model_name in TF_GPT2_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											A__        =			AutoConfig.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )

											A__        =			TFAutoModelForCausalLM.from_pretrained(lowercase       )
											A__    ,				A__        =			TFAutoModelForCausalLM.from_pretrained(lowercase      ,      output_loading_info=lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			def      UpperCamelCase		(    self       )  ->						Optional[int]:
							'''simple docstring'''






							for model_name in TF_BERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											A__        =			AutoConfig.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )

											A__        =			TFAutoModelWithLMHead.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






							for model_name in TF_BERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											A__        =			AutoConfig.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )

											A__        =			TFAutoModelForMaskedLM.from_pretrained(lowercase       )
											A__    ,				A__        =			TFAutoModelForMaskedLM.from_pretrained(lowercase      ,      output_loading_info=lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			def      UpperCamelCase		(    self       )  ->						str:
							'''simple docstring'''






							for model_name in TF_T5_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											A__        =			AutoConfig.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )

											A__        =			TFAutoModelForSeqaSeqLM.from_pretrained(lowercase       )
											A__    ,				A__        =			TFAutoModelForSeqaSeqLM.from_pretrained(lowercase      ,      output_loading_info=lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			def      UpperCamelCase		(    self       )  ->						Dict:
							'''simple docstring'''






							for model_name in ["bert-base-uncased"]:
											A__        =			AutoConfig.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )

											A__        =			TFAutoModelForSequenceClassification.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






							for model_name in ["bert-base-uncased"]:
											A__        =			AutoConfig.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )

											A__        =			TFAutoModelForQuestionAnswering.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			@slow
			@require_tensorflow_probability
			def      UpperCamelCase		(    self       )  ->						List[str]:
							'''simple docstring'''






							for model_name in TF_TAPAS_PRETRAINED_MODEL_ARCHIVE_LIST[5:6]:
											A__        =			AutoConfig.from_pretrained(lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )

											A__        =			TFAutoModelForTableQuestionAnswering.from_pretrained(lowercase       )
											A__    ,				A__        =			TFAutoModelForTableQuestionAnswering.from_pretrained(
											    lowercase      ,      output_loading_info=lowercase       )
											self.assertIsNotNone(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			def      UpperCamelCase		(    self       )  ->						Dict:
							'''simple docstring'''






							A__        =			TFAutoModelWithLMHead.from_pretrained(lowercase       )
							self.assertIsInstance(lowercase      ,      lowercase       )
							self.assertEqual(model.num_parameters()      ,      14410       )
							self.assertEqual(model.num_parameters(only_trainable=lowercase       )      ,      14410       )


			def      UpperCamelCase		(    self       )  ->						Any:
							'''simple docstring'''






							A__        =			TFAutoModelWithLMHead.from_pretrained(lowercase       )
							self.assertIsInstance(lowercase      ,      lowercase       )
							self.assertEqual(model.num_parameters()      ,      14410       )
							self.assertEqual(model.num_parameters(only_trainable=lowercase       )      ,      14410       )


			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






							A__        =			TFAutoModel.from_pretrained("sgugger/funnel-random-tiny"       )
							self.assertIsInstance(lowercase      ,      lowercase       )

							A__        =			copy.deepcopy(model.config       )
							A__        =			["FunnelBaseModel"]
							A__        =			TFAutoModel.from_config(lowercase       )
							self.assertIsInstance(lowercase      ,      lowercase       )

							with tempfile.TemporaryDirectory() as tmp_dir:
											model.save_pretrained(lowercase       )
											A__        =			TFAutoModel.from_pretrained(lowercase       )
											self.assertIsInstance(lowercase      ,      lowercase       )


			def      UpperCamelCase		(    self       )  ->						Optional[Any]:
							'''simple docstring'''






							try:
											AutoConfig.register("new-model"      ,      lowercase       )

											A__        =			[
											    TFAutoModel,
											    TFAutoModelForCausalLM,
											    TFAutoModelForMaskedLM,
											    TFAutoModelForPreTraining,
											    TFAutoModelForQuestionAnswering,
											    TFAutoModelForSequenceClassification,
											    TFAutoModelForTokenClassification,
											]

											for auto_class in auto_classes:
															with self.subTest(auto_class.__name__       ):
																			# Wrong config class will raise an error
																			with self.assertRaises(lowercase       ):
																							auto_class.register(lowercase      ,      lowercase       )
																			auto_class.register(lowercase      ,      lowercase       )
																			# Trying to register something existing in the Transformers library will raise an error
																			with self.assertRaises(lowercase       ):
																							auto_class.register(lowercase      ,      lowercase       )

																			# Now that the config is registered, it can be used as any other config with the auto-API
																			A__        =			BertModelTester(self       ).get_config()
																			A__        =			NewModelConfig(**tiny_config.to_dict()       )
																			A__        =			auto_class.from_config(lowercase       )
																			self.assertIsInstance(lowercase      ,      lowercase       )

																			with tempfile.TemporaryDirectory() as tmp_dir:
																							model.save_pretrained(lowercase       )
																							A__        =			auto_class.from_pretrained(lowercase       )
																							self.assertIsInstance(lowercase      ,      lowercase       )

							finally:
											if "new-model" in CONFIG_MAPPING._extra_content:
															del CONFIG_MAPPING._extra_content["new-model"]
											for mapping in (
											    TF_MODEL_MAPPING,
											    TF_MODEL_FOR_PRETRAINING_MAPPING,
											    TF_MODEL_FOR_QUESTION_ANSWERING_MAPPING,
											    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
											    TF_MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING,
											    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
											    TF_MODEL_FOR_MASKED_LM_MAPPING,
											):
															if NewModelConfig in mapping._extra_content:
																			del mapping._extra_content[NewModelConfig]


			def      UpperCamelCase		(    self       )  ->						str:
							'''simple docstring'''






							with self.assertRaisesRegex(
							    lowercase      ,      "bert-base is not a local folder and is not a valid model identifier"       ):
											A__        =			TFAutoModel.from_pretrained("bert-base"       )


			def      UpperCamelCase		(    self       )  ->						Optional[Any]:
							'''simple docstring'''






							with self.assertRaisesRegex(
							    lowercase      ,      R"aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"       ):
											A__        =			TFAutoModel.from_pretrained(lowercase      ,      revision="aaaaaa"       )


			def      UpperCamelCase		(    self       )  ->						Union[str, Any]:
							'''simple docstring'''






							with self.assertRaisesRegex(
							    lowercase      ,      "hf-internal-testing/config-no-model does not appear to have a file named pytorch_model.bin"      ,      ):
											A__        =			TFAutoModel.from_pretrained("hf-internal-testing/config-no-model"       )


			def      UpperCamelCase		(    self       )  ->						Union[str, Any]:
							'''simple docstring'''






							with self.assertRaisesRegex(lowercase      ,      "Use `from_pt=True` to load this model"       ):
											A__        =			TFAutoModel.from_pretrained("hf-internal-testing/tiny-bert-pt-only"       )


			def      UpperCamelCase		(    self       )  ->						Any:
							'''simple docstring'''






							A__        =			TFAutoModel.from_pretrained("hf-internal-testing/tiny-random-bert"       )
							with RequestCounter() as counter:
											A__        =			TFAutoModel.from_pretrained("hf-internal-testing/tiny-random-bert"       )
											self.assertEqual(counter.get_request_count      ,      0       )
											self.assertEqual(counter.head_request_count      ,      1       )
											self.assertEqual(counter.other_request_count      ,      0       )

							# With a sharded checkpoint
							A__        =			TFAutoModel.from_pretrained("ArthurZ/tiny-random-bert-sharded"       )
							with RequestCounter() as counter:
											A__        =			TFAutoModel.from_pretrained("ArthurZ/tiny-random-bert-sharded"       )
											self.assertEqual(counter.get_request_count      ,      0       )
											self.assertEqual(counter.head_request_count      ,      1       )
											self.assertEqual(counter.other_request_count      ,      0       )

