





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


lowerCAmelCase_						   =    logging.getLogger(__name__)


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

  _snake_case   :				Dict														=	ModelCheckpoint(
      dirpath=lowerCAmelCase    ,							filename=lowerCAmelCase    ,							monitor=F"""val_{metric}"""    ,							mode='max'    ,							save_top_k=3    ,							every_n_epochs=1    ,							)
  return checkpoint_callback


def     lowerCamelCase_	(      lowerCAmelCase:	Union[str, Any]    ,							lowerCAmelCase:	Tuple				)->		List[str]:
  return EarlyStopping(
      monitor=F"""val_{metric}"""    ,							mode='min' if 'loss' in metric else 'max'    ,							patience=lowerCAmelCase    ,							verbose=lowerCAmelCase    ,							)







class        _lowerCAmelCase    (			pl.Callback							):


  '''simple docstring'''







  def UpperCamelCase_ (      self      :					List[str]      ,				UpperCamelCase      :					str      ,				UpperCamelCase      :					Any							):


    '''simple docstring'''





    _snake_case   :				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 UpperCamelCase_ (      self      :					str      ,				UpperCamelCase      :					pl.Trainer      ,				UpperCamelCase      :					pl.LightningModule      ,				UpperCamelCase      :					str      ,				UpperCamelCase      :					Dict=True							):


    '''simple docstring'''





    logger.info(f"""***** {type_path} results at step {trainer.global_step:05d} *****"""							)
    _snake_case   :				Optional[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
    _snake_case   :				Tuple														=	Path(pl_module.hparams.output_dir							)
    if type_path == "test":
      _snake_case   :				Tuple														=	od / 'test_results.txt'
      _snake_case   :				Union[str, 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.
      _snake_case   :				Optional[Any]														=	od / f"""{type_path}_results/{trainer.global_step:05d}.txt"""
      _snake_case   :				Optional[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
        _snake_case   :				Any														=	metrics[key]
        if isinstance(UpperCamelCase      ,				torch.Tensor							):
          _snake_case   :				Optional[int]														=	val.item()
        _snake_case   :				Union[str, Any]														=	f"""{key}: {val:.6f}\n"""
        writer.write(UpperCamelCase							)

    if not save_generations:
      return

    if "preds" in metrics:
      _snake_case   :				Any														=	'\n'.join(metrics['preds']							)
      generations_file.open('w+'							).write(UpperCamelCase							)







  @rank_zero_only
  def UpperCamelCase_ (      self      :					Tuple      ,				UpperCamelCase      :					List[Any]      ,				UpperCamelCase      :					Dict							):


    '''simple docstring'''





    try:
      _snake_case   :				int														=	pl_module.model.model.num_parameters()
    except AttributeError:
      _snake_case   :				Dict														=	pl_module.model.num_parameters()

    _snake_case   :				Optional[Any]														=	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 UpperCamelCase_ (      self      :					str      ,				UpperCamelCase      :					pl.Trainer      ,				UpperCamelCase      :					pl.LightningModule							):


    '''simple docstring'''





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







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











import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_lxmert import LxmertTokenizer


lowerCAmelCase_						   =    {"""vocab_file""": """vocab.txt""", """tokenizer_file""": """tokenizer.json"""}

lowerCAmelCase_						   =    {
    """vocab_file""": {
        """unc-nlp/lxmert-base-uncased""": """https://huggingface.co/unc-nlp/lxmert-base-uncased/resolve/main/vocab.txt""",
    },
    """tokenizer_file""": {
        """unc-nlp/lxmert-base-uncased""": (
            """https://huggingface.co/unc-nlp/lxmert-base-uncased/resolve/main/tokenizer.json"""
        ),
    },
}

lowerCAmelCase_						   =    {
    """unc-nlp/lxmert-base-uncased""": 512,
}

lowerCAmelCase_						   =    {
    """unc-nlp/lxmert-base-uncased""": {"""do_lower_case""": True},
}



class        _lowerCAmelCase    (			UpperCAmelCase_							):


  '''simple docstring'''
  a_							:       List[Any]      					=VOCAB_FILES_NAMES
  a_							:       Tuple      					=PRETRAINED_VOCAB_FILES_MAP
  a_							:       Optional[Any]      					=PRETRAINED_INIT_CONFIGURATION
  a_							:       Any      					=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  a_							:       Any      					=LxmertTokenizer







  def __init__(      self      :					Any      ,				UpperCamelCase      :					Union[str, Any]=None      ,				UpperCamelCase      :					Dict=None      ,				UpperCamelCase      :					List[str]=True      ,				UpperCamelCase      :					List[str]="[UNK]"      ,				UpperCamelCase      :					List[Any]="[SEP]"      ,				UpperCamelCase      :					List[Any]="[PAD]"      ,				UpperCamelCase      :					Optional[Any]="[CLS]"      ,				UpperCamelCase      :					Optional[int]="[MASK]"      ,				UpperCamelCase      :					Optional[int]=True      ,				UpperCamelCase      :					str=None      ,				**UpperCamelCase      :					List[str]      ,				):


    '''simple docstring'''





    super().__init__(
        UpperCamelCase      ,				tokenizer_file=UpperCamelCase      ,				do_lower_case=UpperCamelCase      ,				unk_token=UpperCamelCase      ,				sep_token=UpperCamelCase      ,				pad_token=UpperCamelCase      ,				cls_token=UpperCamelCase      ,				mask_token=UpperCamelCase      ,				tokenize_chinese_chars=UpperCamelCase      ,				strip_accents=UpperCamelCase      ,				**UpperCamelCase      ,				)

    _snake_case   :				Tuple														=	json.loads(self.backend_tokenizer.normalizer.__getstate__()							)
    if (
        normalizer_state.get('lowercase'      ,				UpperCamelCase							) != do_lower_case
        or normalizer_state.get('strip_accents'      ,				UpperCamelCase							) != strip_accents
        or normalizer_state.get('handle_chinese_chars'      ,				UpperCamelCase							) != tokenize_chinese_chars
    ):
      _snake_case   :				List[Any]														=	getattr(UpperCamelCase      ,				normalizer_state.pop('type'							)							)
      _snake_case   :				Optional[int]														=	do_lower_case
      _snake_case   :				Dict														=	strip_accents
      _snake_case   :				Optional[int]														=	tokenize_chinese_chars
      _snake_case   :				Optional[Any]														=	normalizer_class(**UpperCamelCase							)

    _snake_case   :				int														=	do_lower_case







  def UpperCamelCase_ (      self      :					int      ,				UpperCamelCase      :					List[str]      ,				UpperCamelCase      :					str=None							):


    '''simple docstring'''





    _snake_case   :				List[Any]														=	[self.cls_token_id] + token_ids_a + [self.sep_token_id]

    if token_ids_a:
      output += token_ids_a + [self.sep_token_id]

    return output







  def UpperCamelCase_ (      self      :					List[str]      ,				UpperCamelCase      :					List[int]      ,				UpperCamelCase      :					Optional[List[int]] = None							):


    '''simple docstring'''





    _snake_case   :				Tuple														=	[self.sep_token_id]
    _snake_case   :				List[str]														=	[self.cls_token_id]
    if token_ids_a is None:
      return len(cls + token_ids_a + sep							) * [0]
    return len(cls + token_ids_a + sep							) * [0] + len(token_ids_a + sep							) * [1]







  def UpperCamelCase_ (      self      :					int      ,				UpperCamelCase      :					str      ,				UpperCamelCase      :					Optional[str] = None							):


    '''simple docstring'''





    _snake_case   :				int														=	self._tokenizer.model.save(UpperCamelCase      ,				name=UpperCamelCase							)
    return tuple(UpperCamelCase							)




