

import warnings
from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxConfigWithPast, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


lowerCamelCase						=		logging.get_logger(__name__)

lowerCamelCase						=		{
    '''facebook/bart-large''': '''https://huggingface.co/facebook/bart-large/resolve/main/config.json''',
    # See all BART models at https://huggingface.co/models?filter=bart
}





class  _a							(    _a):
	_a  :  Any				   =							"""bart"""
	_a  :  int				   =							["""past_key_values"""]
	_a  :  Optional[int]				   =							{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}




	def __init__( self						:		List[str]		, _SCREAMING_SNAKE_CASE						:		Optional[int]=5_0265		, _SCREAMING_SNAKE_CASE						:		List[Any]=1024		, _SCREAMING_SNAKE_CASE						:		Optional[int]=12		, _SCREAMING_SNAKE_CASE						:		str=4096		, _SCREAMING_SNAKE_CASE						:		Any=16		, _SCREAMING_SNAKE_CASE						:		Dict=12		, _SCREAMING_SNAKE_CASE						:		Tuple=4096		, _SCREAMING_SNAKE_CASE						:		Optional[int]=16		, _SCREAMING_SNAKE_CASE						:		Any=0.0		, _SCREAMING_SNAKE_CASE						:		List[str]=0.0		, _SCREAMING_SNAKE_CASE						:		Optional[Any]="gelu"		, _SCREAMING_SNAKE_CASE						:		Tuple=1024		, _SCREAMING_SNAKE_CASE						:		List[Any]=0.1		, _SCREAMING_SNAKE_CASE						:		Optional[int]=0.0		, _SCREAMING_SNAKE_CASE						:		str=0.0		, _SCREAMING_SNAKE_CASE						:		Optional[int]=0.02		, _SCREAMING_SNAKE_CASE						:		Optional[int]=0.0		, _SCREAMING_SNAKE_CASE						:		List[str]=False		, _SCREAMING_SNAKE_CASE						:		Dict=True		, _SCREAMING_SNAKE_CASE						:		Any=3		, _SCREAMING_SNAKE_CASE						:		Optional[Any]=1		, _SCREAMING_SNAKE_CASE						:		Any=0		, _SCREAMING_SNAKE_CASE						:		Tuple=2		, _SCREAMING_SNAKE_CASE						:		Dict=True		, _SCREAMING_SNAKE_CASE						:		Tuple=2		, _SCREAMING_SNAKE_CASE						:		int=2		, **_SCREAMING_SNAKE_CASE						:		List[Any]		, )->	Dict:
			lowerCAmelCase__       :      Optional[int]   					= vocab_size
			lowerCAmelCase__       :      List[str]   					= max_position_embeddings
			lowerCAmelCase__       :      Tuple   					= d_model
			lowerCAmelCase__       :      Tuple   					= encoder_ffn_dim
			lowerCAmelCase__       :      Dict   					= encoder_layers
			lowerCAmelCase__       :      Tuple   					= encoder_attention_heads
			lowerCAmelCase__       :      Optional[Any]   					= decoder_ffn_dim
			lowerCAmelCase__       :      List[str]   					= decoder_layers
			lowerCAmelCase__       :      Any   					= decoder_attention_heads
			lowerCAmelCase__       :      Dict   					= dropout
			lowerCAmelCase__       :      Optional[int]   					= attention_dropout
			lowerCAmelCase__       :      Tuple   					= activation_dropout
			lowerCAmelCase__       :      Any   					= activation_function
			lowerCAmelCase__       :      Optional[Any]   					= init_std
			lowerCAmelCase__       :      List[str]   					= encoder_layerdrop
			lowerCAmelCase__       :      Optional[int]   					= decoder_layerdrop
			lowerCAmelCase__       :      str   					= classifier_dropout
			lowerCAmelCase__       :      Optional[Any]   					= use_cache
			lowerCAmelCase__       :      List[str]   					= encoder_layers
			lowerCAmelCase__       :      Optional[Any]   					= scale_embedding  # scale factor will be sqrt(d_model) if True

			super().__init__(
			    num_labels=__lowerCamelCase		, pad_token_id=__lowerCamelCase		, bos_token_id=__lowerCamelCase		, eos_token_id=__lowerCamelCase		, is_encoder_decoder=__lowerCamelCase		, decoder_start_token_id=__lowerCamelCase		, forced_eos_token_id=__lowerCamelCase		, **__lowerCamelCase		, )

			# ensure backward compatibility for BART CNN models
			if self.forced_bos_token_id is None and kwargs.get('''force_bos_token_to_be_generated'''		, __lowerCamelCase		):
					lowerCAmelCase__       :      List[str]   					= self.bos_token_id
					warnings.warn(
					    F'Please make sure the config includes `forced_bos_token_id={self.bos_token_id}` in future versions. '
					    '''The config can simply be saved and uploaded again to be fixed.'''		)





class  _a							(    _a):




	@property
	def 							UpperCAmelCase__( self						:		Any		)->	List[str]:
			if self.task in ["default", "seq2seq-lm"]:
					lowerCAmelCase__       :      List[Any]   					= OrderedDict(
					    [
					        ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
					        ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
					    ]		)

					if self.use_past:
							lowerCAmelCase__       :      int   					= {0: """batch"""}
							lowerCAmelCase__       :      Union[str, Any]   					= {0: """batch""", 1: """past_decoder_sequence + sequence"""}
					else:
							lowerCAmelCase__       :      int   					= {0: """batch""", 1: """decoder_sequence"""}
							lowerCAmelCase__       :      Any   					= {0: """batch""", 1: """decoder_sequence"""}

					if self.use_past:
							self.fill_with_past_key_values_(__lowerCamelCase		, direction='''inputs'''		)
			elif self.task == "causal-lm":
					# TODO: figure this case out.
					lowerCAmelCase__       :      Optional[int]   					= OrderedDict(
					    [
					        ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
					        ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
					    ]		)
					if self.use_past:
							lowerCAmelCase__       :      Tuple   					= self.num_layers
							for i in range(__lowerCamelCase		):
									lowerCAmelCase__       :      Union[str, Any]   					= {0: """batch""", 2: """past_sequence + sequence"""}
									lowerCAmelCase__       :      Union[str, Any]   					= {0: """batch""", 2: """past_sequence + sequence"""}
			else:
					lowerCAmelCase__       :      int   					= OrderedDict(
					    [
					        ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
					        ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
					        ('''decoder_input_ids''', {0: '''batch''', 1: '''decoder_sequence'''}),
					        ('''decoder_attention_mask''', {0: '''batch''', 1: '''decoder_sequence'''}),
					    ]		)

			return common_inputs




	@property
	def 							UpperCAmelCase__( self						:		Tuple		)->	Union[str, Any]:
			if self.task in ["default", "seq2seq-lm"]:
					lowerCAmelCase__       :      List[str]   					= super().outputs
			else:
					lowerCAmelCase__       :      str   					= super(__lowerCamelCase		, self		).outputs
					if self.use_past:
							lowerCAmelCase__       :      int   					= self.num_layers
							for i in range(__lowerCamelCase		):
									lowerCAmelCase__       :      int   					= {0: """batch""", 2: """past_sequence + sequence"""}
									lowerCAmelCase__       :      List[Any]   					= {0: """batch""", 2: """past_sequence + sequence"""}
			return common_outputs




	def 							UpperCAmelCase__( self						:		Optional[Any]		, _SCREAMING_SNAKE_CASE						:		PreTrainedTokenizer		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		bool = False		, _SCREAMING_SNAKE_CASE						:		Optional[TensorType] = None		, )->	Union[str, Any]:
			lowerCAmelCase__       :      Optional[Any]   					= self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
			    __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		)

			# Generate decoder inputs
			lowerCAmelCase__       :      Any   					= seq_length if not self.use_past else 1
			lowerCAmelCase__       :      Tuple   					= self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
			    __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		)
			lowerCAmelCase__       :      Optional[Any]   					= {F'decoder_{name}': tensor for name, tensor in decoder_inputs.items()}
			lowerCAmelCase__       :      Optional[int]   					= dict(**__lowerCamelCase		, **__lowerCamelCase		)

			if self.use_past:
					if not is_torch_available():
							raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''		)
					else:
							import torch
					lowerCAmelCase__       :      Tuple   					= common_inputs["""input_ids"""].shape
					lowerCAmelCase__       :      str   					= common_inputs["""decoder_input_ids"""].shape[1]
					lowerCAmelCase__       :      Optional[int]   					= self.num_attention_heads
					lowerCAmelCase__       :      List[Any]   					= (
					    batch,
					    num_encoder_attention_heads,
					    encoder_seq_length,
					    self._config.hidden_size // num_encoder_attention_heads,
					)
					lowerCAmelCase__       :      str   					= decoder_seq_length + 3
					lowerCAmelCase__       :      List[str]   					= (
					    batch,
					    num_decoder_attention_heads,
					    decoder_past_length,
					    self._config.hidden_size // num_decoder_attention_heads,
					)

					lowerCAmelCase__       :      Optional[Any]   					= torch.cat(
					    [common_inputs['''decoder_attention_mask'''], torch.ones(__lowerCamelCase		, __lowerCamelCase		)]		, dim=1		)

					lowerCAmelCase__       :      Union[str, Any]   					= []
					# If the number of encoder and decoder layers are present in the model configuration, both are considered
					lowerCAmelCase__       :      Optional[int]   					= self.num_layers
					lowerCAmelCase__       :      List[str]   					= min(__lowerCamelCase		, __lowerCamelCase		)
					lowerCAmelCase__       :      List[Any]   					= max(__lowerCamelCase		, __lowerCamelCase		) - min_num_layers
					lowerCAmelCase__       :      Tuple   					= """encoder""" if num_encoder_layers > num_decoder_layers else """decoder"""

					for _ in range(__lowerCamelCase		):
							common_inputs["past_key_values"].append(
							    (
							        torch.zeros(__lowerCamelCase		),
							        torch.zeros(__lowerCamelCase		),
							        torch.zeros(__lowerCamelCase		),
							        torch.zeros(__lowerCamelCase		),
							    )		)
					# TODO: test this.
					lowerCAmelCase__       :      List[str]   					= encoder_shape if remaining_side_name == """encoder""" else decoder_shape
					for _ in range(__lowerCamelCase		, __lowerCamelCase		):
							common_inputs["past_key_values"].append((torch.zeros(__lowerCamelCase		), torch.zeros(__lowerCamelCase		))		)
			return common_inputs




	def 							UpperCAmelCase__( self						:		int		, _SCREAMING_SNAKE_CASE						:		PreTrainedTokenizer		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		bool = False		, _SCREAMING_SNAKE_CASE						:		Optional[TensorType] = None		, )->	Tuple:
			lowerCAmelCase__       :      List[str]   					= self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
			    __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		)

			if self.use_past:
					if not is_torch_available():
							raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''		)
					else:
							import torch
					lowerCAmelCase__       :      int   					= common_inputs["""input_ids"""].shape
					# Not using the same length for past_key_values
					lowerCAmelCase__       :      List[str]   					= seqlen + 2
					lowerCAmelCase__       :      Union[str, Any]   					= self.num_layers
					lowerCAmelCase__       :      Union[str, Any]   					= self.num_attention_heads
					lowerCAmelCase__       :      Dict   					= (
					    batch,
					    num_encoder_attention_heads,
					    past_key_values_length,
					    self._config.hidden_size // num_encoder_attention_heads,
					)

					lowerCAmelCase__       :      Optional[int]   					= common_inputs["""attention_mask"""].dtype
					lowerCAmelCase__       :      int   					= torch.cat(
					    [common_inputs['''attention_mask'''], torch.ones(__lowerCamelCase		, __lowerCamelCase		, dtype=__lowerCamelCase		)]		, dim=1		)
					lowerCAmelCase__       :      List[Any]   					= [
					    (torch.zeros(__lowerCamelCase		), torch.zeros(__lowerCamelCase		)) for _ in range(__lowerCamelCase		)
					]
			return common_inputs




	def 							UpperCAmelCase__( self						:		Tuple		, _SCREAMING_SNAKE_CASE						:		PreTrainedTokenizer		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		bool = False		, _SCREAMING_SNAKE_CASE						:		Optional[TensorType] = None		, )->	str:
			# Copied from OnnxConfig.generate_dummy_inputs
			# Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
			# If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
			lowerCAmelCase__       :      List[Any]   					= compute_effective_axis_dimension(
			    __lowerCamelCase		, fixed_dimension=OnnxConfig.default_fixed_batch		, num_token_to_add=0		)

			# If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
			lowerCAmelCase__       :      Optional[int]   					= tokenizer.num_special_tokens_to_add(__lowerCamelCase		)
			lowerCAmelCase__       :      str   					= compute_effective_axis_dimension(
			    __lowerCamelCase		, fixed_dimension=OnnxConfig.default_fixed_sequence		, num_token_to_add=__lowerCamelCase		)

			# Generate dummy inputs according to compute batch and sequence
			lowerCAmelCase__       :      str   					= [""" """.join([tokenizer.unk_token]		) * seq_length] * batch_size
			lowerCAmelCase__       :      List[Any]   					= dict(tokenizer(__lowerCamelCase		, return_tensors=__lowerCamelCase		)		)
			return common_inputs




	def 							UpperCAmelCase__( self						:		Optional[int]		, _SCREAMING_SNAKE_CASE						:		PreTrainedTokenizer		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		int = -1		, _SCREAMING_SNAKE_CASE						:		bool = False		, _SCREAMING_SNAKE_CASE						:		Optional[TensorType] = None		, )->	str:
			if self.task in ["default", "seq2seq-lm"]:
					lowerCAmelCase__       :      str   					= self._generate_dummy_inputs_for_default_and_seqaseq_lm(
					    __lowerCamelCase		, batch_size=__lowerCamelCase		, seq_length=__lowerCamelCase		, is_pair=__lowerCamelCase		, framework=__lowerCamelCase		)

			elif self.task == "causal-lm":
					lowerCAmelCase__       :      List[Any]   					= self._generate_dummy_inputs_for_causal_lm(
					    __lowerCamelCase		, batch_size=__lowerCamelCase		, seq_length=__lowerCamelCase		, is_pair=__lowerCamelCase		, framework=__lowerCamelCase		)
			else:
					lowerCAmelCase__       :      List[Any]   					= self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
					    __lowerCamelCase		, batch_size=__lowerCamelCase		, seq_length=__lowerCamelCase		, is_pair=__lowerCamelCase		, framework=__lowerCamelCase		)

			return common_inputs




	def 							UpperCAmelCase__( self						:		List[str]		, _SCREAMING_SNAKE_CASE						:		List[str]		, _SCREAMING_SNAKE_CASE						:		List[Any]		, _SCREAMING_SNAKE_CASE						:		Union[str, Any]		, _SCREAMING_SNAKE_CASE						:		Dict		)->	List[str]:
			if self.task in ["default", "seq2seq-lm"]:
					lowerCAmelCase__       :      Tuple   					= super()._flatten_past_key_values_(__lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		)
			else:
					lowerCAmelCase__       :      List[str]   					= super(__lowerCamelCase		, self		)._flatten_past_key_values_(
					    __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		, __lowerCamelCase		)








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_   (	_a	):




		"""simple docstring"""






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


lowerCamelCase						=		logging.getLogger(__name__)






def 					lowerCamelCase_   (	_a     ,       _a	):




		"""simple docstring"""






		if metric == "rouge2":
				lowerCAmelCase__       :      Optional[int]   					= '''{val_avg_rouge2:.4f}-{step_count}'''
		elif metric == "bleu":
				lowerCAmelCase__       :      Optional[int]   					= '''{val_avg_bleu:.4f}-{step_count}'''
		elif metric == "em":
				lowerCAmelCase__       :      List[Any]   					= '''{val_avg_em:.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.'''	)

		lowerCAmelCase__       :      Dict   					= ModelCheckpoint(
		    dirpath=_a     ,       filename=_a     ,       monitor=f'val_{metric}'     ,       mode='''max'''     ,       save_top_k=3     ,       every_n_epochs=1     ,       )
		return checkpoint_callback






def 					lowerCamelCase_   (	_a     ,       _a	):




		"""simple docstring"""






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





class  _a							(    pl.Callback):




	def 							UpperCAmelCase__( self						:		Optional[Any]		, _SCREAMING_SNAKE_CASE						:		int		, _SCREAMING_SNAKE_CASE						:		Any		)->	Optional[int]:
			lowerCAmelCase__       :      Dict   					= {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 							UpperCAmelCase__( 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} *****'		)
			lowerCAmelCase__       :      List[Any]   					= 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__       :      List[Any]   					= Path(pl_module.hparams.output_dir		)
			if type_path == "test":
					lowerCAmelCase__       :      Optional[int]   					= od / '''test_results.txt'''
					lowerCAmelCase__       :      Tuple   					= 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__       :      int   					= od / F'{type_path}_results/{trainer.global_step:05d}.txt'
					lowerCAmelCase__       :      int   					= 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
							lowerCAmelCase__       :      Optional[int]   					= metrics[key]
							if isinstance(_SCREAMING_SNAKE_CASE		, torch.Tensor		):
									lowerCAmelCase__       :      List[str]   					= val.item()
							lowerCAmelCase__       :      List[str]   					= F'{key}: {val:.6f}\n'
							writer.write(_SCREAMING_SNAKE_CASE		)

			if not save_generations:
					return

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




	@rank_zero_only
	def 							UpperCAmelCase__( self						:		str		, _SCREAMING_SNAKE_CASE						:		Union[str, Any]		, _SCREAMING_SNAKE_CASE						:		Optional[int]		)->	Optional[int]:
			try:
					lowerCAmelCase__       :      Tuple   					= pl_module.model.model.num_parameters()
			except AttributeError:
					lowerCAmelCase__       :      Optional[Any]   					= pl_module.model.num_parameters()

			lowerCAmelCase__       :      Dict   					= 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 							UpperCAmelCase__( self						:		List[Any]		, _SCREAMING_SNAKE_CASE						:		pl.Trainer		, _SCREAMING_SNAKE_CASE						:		pl.LightningModule		)->	Optional[Any]:
			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 							UpperCAmelCase__( self						:		Optional[int]		, _SCREAMING_SNAKE_CASE						:		pl.Trainer		, _SCREAMING_SNAKE_CASE						:		List[Any]		)->	List[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")




