






'''simple docstring'''

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 						__UpperCamelCase				(							_lowercase					) ->			List[Any]:
	_lowercase					:       Union[str, Any]       =		filter(lambda _lowercase					: p.requires_grad,				model.parameters()					)
	_lowercase					:       Optional[int]       =		sum([np.prod(p.size()					) for p in model_parameters]					)
	return params


_A				:    Dict     						=logging.getLogger(__name__)







def 						__UpperCamelCase				(							_lowercase,				_lowercase					) ->			str:
	if metric == "rouge2":
		_lowercase					:       Union[str, Any]       =		"""{val_avg_rouge2:.4f}-{step_count}"""
	elif metric == "bleu":
		_lowercase					:       List[Any]       =		"""{val_avg_bleu:.4f}-{step_count}"""
	elif metric == "em":
		_lowercase					:       Optional[int]       =		"""{val_avg_em:.4f}-{step_count}"""
	elif metric == "loss":
		_lowercase					:       Union[str, Any]       =		"""{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.'					)

	_lowercase					:       List[str]       =		ModelCheckpoint(
	    dirpath=_lowercase,				filename=_lowercase,				monitor=f'''val_{metric}''',				mode='max',				save_top_k=1,				every_n_epochs=1,				)

	return checkpoint_callback







def 						__UpperCamelCase				(							_lowercase,				_lowercase					) ->			List[Any]:
	return EarlyStopping(
	    monitor=f'''val_{metric}''',				mode='min' if 'loss' in metric else 'max',				patience=_lowercase,				verbose=_lowercase,				)






class 		lowerCamelCase__      (						pl.Callback     ):




					'''simple docstring'''



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



						'''simple docstring'''





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




					@rank_zero_only
					def 						__UpperCAmelCase     (  self				:	str     ,  UpperCamelCase_				:	pl.Trainer     ,  UpperCamelCase_				:	pl.LightningModule     ,  UpperCamelCase_				:	str     ,  UpperCamelCase_				:	str=True					)  ->       Dict:



						'''simple docstring'''





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

						if not save_generations:
							return

						if "preds" in metrics:
							_lowercase					:       List[str]       =		"""\n""".join(metrics['preds']					)
							generations_file.open('w+'					).write(_a					)




					@rank_zero_only
					def 						__UpperCAmelCase     (  self				:	str     ,  UpperCamelCase_				:	int     ,  UpperCamelCase_				:	Optional[int]					)  ->       int:



						'''simple docstring'''





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

						_lowercase					:       Dict       =		count_trainable_parameters(_a					)
						# 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				:	Tuple     ,  UpperCamelCase_				:	pl.Trainer     ,  UpperCamelCase_				:	pl.LightningModule					)  ->       str:



						'''simple docstring'''





						save_json(pl_module.metrics     ,  pl_module.metrics_save_path					)
						return self._write_logs(_a     ,  _a     ,  'test'					)




					@rank_zero_only
					def 						__UpperCAmelCase     (  self				:	Any     ,  UpperCamelCase_				:	pl.Trainer     ,  UpperCamelCase_				:	Dict					)  ->       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")





'''simple docstring'''

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_A				:    Dict     						=logging.get_logger(__name__)

_A				:    Dict     						={
    # See all MEGATRON_BERT models at https://huggingface.co/models?filter=bert
}



class 		lowerCamelCase__      (						A     ):




					'''simple docstring'''



					A_						    =						"""megatron-bert"""




					def __init__(  self				:	int     ,  UpperCamelCase_				:	int=2_9056     ,  UpperCamelCase_				:	Optional[int]=1024     ,  UpperCamelCase_				:	Optional[Any]=24     ,  UpperCamelCase_				:	List[Any]=16     ,  UpperCamelCase_				:	Optional[int]=4096     ,  UpperCamelCase_				:	Optional[Any]="gelu"     ,  UpperCamelCase_				:	List[Any]=0.1     ,  UpperCamelCase_				:	Union[str, Any]=0.1     ,  UpperCamelCase_				:	int=512     ,  UpperCamelCase_				:	Dict=2     ,  UpperCamelCase_				:	Union[str, Any]=0.02     ,  UpperCamelCase_				:	Any=1E-12     ,  UpperCamelCase_				:	Tuple=0     ,  UpperCamelCase_				:	Optional[int]="absolute"     ,  UpperCamelCase_				:	Optional[Any]=True     ,  **UpperCamelCase_				:	Any     ,  )  ->       List[Any]:



						'''simple docstring'''





						super().__init__(pad_token_id=UpperCamelCase_     ,  **UpperCamelCase_					)

						_lowercase					:       Dict       =		vocab_size
						_lowercase					:       Any       =		hidden_size
						_lowercase					:       Union[str, Any]       =		num_hidden_layers
						_lowercase					:       Dict       =		num_attention_heads
						_lowercase					:       Dict       =		hidden_act
						_lowercase					:       Optional[Any]       =		intermediate_size
						_lowercase					:       Optional[int]       =		hidden_dropout_prob
						_lowercase					:       Optional[Any]       =		attention_probs_dropout_prob
						_lowercase					:       Any       =		max_position_embeddings
						_lowercase					:       str       =		type_vocab_size
						_lowercase					:       Optional[Any]       =		initializer_range
						_lowercase					:       List[str]       =		layer_norm_eps
						_lowercase					:       List[Any]       =		position_embedding_type
						_lowercase					:       Optional[Any]       =		use_cache
