



"""simple docstring"""


import json
from typing import List, Optional, Tuple

from tokenizers import normalizers
from tokenizers.pre_tokenizers import BertPreTokenizer, PreTokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roformer import RoFormerTokenizer
from .tokenization_utils import JiebaPreTokenizer


__A   :							List[str]										=   logging.get_logger(__name__)

__A   :							Optional[Any]										=   {"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer.json"}

__A   :							Dict										=   {
    "vocab_file": {
        "junnyu/roformer_chinese_small": "https://huggingface.co/junnyu/roformer_chinese_small/resolve/main/vocab.txt",
        "junnyu/roformer_chinese_base": "https://huggingface.co/junnyu/roformer_chinese_base/resolve/main/vocab.txt",
        "junnyu/roformer_chinese_char_small": (
            "https://huggingface.co/junnyu/roformer_chinese_char_small/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_chinese_char_base": (
            "https://huggingface.co/junnyu/roformer_chinese_char_base/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_small_discriminator": (
            "https://huggingface.co/junnyu/roformer_small_discriminator/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_small_generator": (
            "https://huggingface.co/junnyu/roformer_small_generator/resolve/main/vocab.txt"
        ),
    }
}

__A   :							Optional[Any]										=   {
    "junnyu/roformer_chinese_small": 1536,
    "junnyu/roformer_chinese_base": 1536,
    "junnyu/roformer_chinese_char_small": 512,
    "junnyu/roformer_chinese_char_base": 512,
    "junnyu/roformer_small_discriminator": 128,
    "junnyu/roformer_small_generator": 128,
}


__A   :							Union[str, Any]										=   {
    "junnyu/roformer_chinese_small": {"do_lower_case": True},
    "junnyu/roformer_chinese_base": {"do_lower_case": True},
    "junnyu/roformer_chinese_char_small": {"do_lower_case": True},
    "junnyu/roformer_chinese_char_base": {"do_lower_case": True},
    "junnyu/roformer_small_discriminator": {"do_lower_case": True},
    "junnyu/roformer_small_generator": {"do_lower_case": True},
}
class 			_a (					lowerCAmelCase):
							"""simple docstring"""
							UpperCamelCase__			      =	VOCAB_FILES_NAMES
							UpperCamelCase__			      =	PRETRAINED_VOCAB_FILES_MAP
							UpperCamelCase__			      =	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCamelCase__			      =	PRETRAINED_INIT_CONFIGURATION
							UpperCamelCase__			      =	RoFormerTokenizer



							def __init__(		self     :    Optional[Any]				,   __UpperCamelCase     :    List[str]=None				,   __UpperCamelCase     :    List[str]=None				,   __UpperCamelCase     :    List[Any]=True				,   __UpperCamelCase     :    int="[UNK]"				,   __UpperCamelCase     :    int="[SEP]"				,   __UpperCamelCase     :    List[Any]="[PAD]"				,   __UpperCamelCase     :    Dict="[CLS]"				,   __UpperCamelCase     :    Optional[int]="[MASK]"				,   __UpperCamelCase     :    Optional[Any]=True				,   __UpperCamelCase     :    Optional[Any]=None				,   **__UpperCamelCase     :    Tuple				,   )->int:
											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				,   )

											_UpperCAmelCase					=    json.loads(self.backend_tokenizer.normalizer.__getstate__()   )
											if (
											    pre_tok_state.get('''lowercase'''				,   __UpperCamelCase   ) != do_lower_case
											    or pre_tok_state.get('''strip_accents'''				,   __UpperCamelCase   ) != strip_accents
											):
															_UpperCAmelCase					=    getattr(__UpperCamelCase				,   pre_tok_state.pop('''type'''   )   )
															_UpperCAmelCase					=    do_lower_case
															_UpperCAmelCase					=    strip_accents
															_UpperCAmelCase					=    pre_tok_class(**__UpperCamelCase   )

											_UpperCAmelCase					=    do_lower_case



							def __getstate__(		self     :    str   )->Union[str, Any]:
											_UpperCAmelCase					=    self.__dict__.copy()
											_UpperCAmelCase					=    BertPreTokenizer()
											return state



							def __setstate__(		self     :    Union[str, Any]				,   __UpperCamelCase     :    List[Any]   )->int:
											_UpperCAmelCase					=    d
											_UpperCAmelCase					=    self.__dict__['''_tokenizer'''].get_vocab()
											_UpperCAmelCase					=    PreTokenizer.custom(JiebaPreTokenizer(__UpperCamelCase   )   )



							def  lowercase__						(		self     :    Optional[int]				,   __UpperCamelCase     :    Dict				,   __UpperCamelCase     :    Any=None   )->List[Any]:
											_UpperCAmelCase					=    [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  lowercase__						(		self     :    str				,   __UpperCamelCase     :    List[int]				,   __UpperCamelCase     :    Optional[List[int]] = None   )->str:
											_UpperCAmelCase					=    [self.sep_token_id]
											_UpperCAmelCase					=    [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  lowercase__						(		self     :    str				,   __UpperCamelCase     :    str				,   __UpperCamelCase     :    Optional[str] = None   )->Union[str, Any]:
											_UpperCAmelCase					=    self._tokenizer.model.save(__UpperCamelCase				,   name=__UpperCamelCase   )
											return tuple(__UpperCamelCase   )



							def  lowercase__						(		self     :    Any				,   __UpperCamelCase     :    Optional[Any]				,   __UpperCamelCase     :    str=None				,   __UpperCamelCase     :    Union[str, Any]=None				,   __UpperCamelCase     :    str=False				,   **__UpperCamelCase     :    Dict				,   )->Any:
											_UpperCAmelCase					=    BertPreTokenizer()
											return super().save_pretrained(__UpperCamelCase				,   __UpperCamelCase				,   __UpperCamelCase				,   __UpperCamelCase				,   **__UpperCamelCase   )








"""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  lowercase	(      _SCREAMING_SNAKE_CASE   :						Optional[Any]				):






				'''simple docstring'''



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


__A   :							Dict										=   logging.getLogger(__name__)







def  lowercase	(      _SCREAMING_SNAKE_CASE   :						Optional[Any]					,     _SCREAMING_SNAKE_CASE   :						Any				):






				'''simple docstring'''



				if metric == "rouge2":
								_UpperCAmelCase					=    '''{val_avg_rouge2:.4f}-{step_count}'''
				elif metric == "bleu":
								_UpperCAmelCase					=    '''{val_avg_bleu:.4f}-{step_count}'''
				elif metric == "em":
								_UpperCAmelCase					=    '''{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.'''				)

				_UpperCAmelCase					=    ModelCheckpoint(
				    dirpath=_SCREAMING_SNAKE_CASE					,     filename=_SCREAMING_SNAKE_CASE					,     monitor=f'val_{metric}'					,     mode='''max'''					,     save_top_k=3					,     every_n_epochs=1					,     )
				return checkpoint_callback







def  lowercase	(      _SCREAMING_SNAKE_CASE   :						str					,     _SCREAMING_SNAKE_CASE   :						List[str]				):






				'''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  lowercase__						(		self     :    List[str]				,   __UpperCamelCase     :    str				,   __UpperCamelCase     :    str   )->Tuple:
							_UpperCAmelCase					=    {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  lowercase__						(		self     :    Union[str, Any]				,   __UpperCamelCase     :    pl.Trainer				,   __UpperCamelCase     :    pl.LightningModule				,   __UpperCamelCase     :    str				,   __UpperCamelCase     :    Tuple=True   )->None:
							logger.info(F'***** {type_path} results at step {trainer.global_step:05d} *****'   )
							_UpperCAmelCase					=    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
							_UpperCAmelCase					=    Path(pl_module.hparams.output_dir   )
							if type_path == "test":
											_UpperCAmelCase					=    od / '''test_results.txt'''
											_UpperCAmelCase					=    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.
											_UpperCAmelCase					=    od / F'{type_path}_results/{trainer.global_step:05d}.txt'
											_UpperCAmelCase					=    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
															_UpperCAmelCase					=    metrics[key]
															if isinstance(__UpperCamelCase				,   torch.Tensor   ):
																			_UpperCAmelCase					=    val.item()
															_UpperCAmelCase					=    F'{key}: {val:.6f}\n'
															writer.write(__UpperCamelCase   )

							if not save_generations:
											return

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



			@rank_zero_only
			def  lowercase__						(		self     :    Optional[int]				,   __UpperCamelCase     :    int				,   __UpperCamelCase     :    Dict   )->Union[str, Any]:
							try:
											_UpperCAmelCase					=    pl_module.model.model.num_parameters()
							except AttributeError:
											_UpperCAmelCase					=    pl_module.model.num_parameters()

							_UpperCAmelCase					=    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  lowercase__						(		self     :    str				,   __UpperCamelCase     :    pl.Trainer				,   __UpperCamelCase     :    pl.LightningModule   )->List[Any]:
							save_json(pl_module.metrics				,   pl_module.metrics_save_path   )
							return self._write_logs(__UpperCamelCase				,   __UpperCamelCase				,   '''test'''   )



			@rank_zero_only
			def  lowercase__						(		self     :    Optional[Any]				,   __UpperCamelCase     :    pl.Trainer				,   __UpperCamelCase     :    List[str]   )->Optional[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")



