
"""simple docstring"""






import json
import os
import re
import warnings
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__UpperCAmelCase 					=    logging.get_logger(__name__)

__UpperCAmelCase 					=    {
    'source_spm': 'source.spm',
    'target_spm': 'target.spm',
    'vocab': 'vocab.json',
    'target_vocab_file': 'target_vocab.json',
    'tokenizer_config_file': 'tokenizer_config.json',
}

__UpperCAmelCase 					=    {
    'source_spm': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/source.spm'
    },
    'target_spm': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/target.spm'
    },
    'vocab': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/vocab.json'
    },
    'tokenizer_config_file': {
        'Helsinki-NLP/opus-mt-en-de': (
            'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/tokenizer_config.json'
        )
    },
}

__UpperCAmelCase 					=    {'Helsinki-NLP/opus-mt-en-de': 5_12}
__UpperCAmelCase 					=    {}




class      _SCREAMING_SNAKE_CASE   (     __SCREAMING_SNAKE_CASE				):
			UpperCAmelCase_    :Tuple												=		VOCAB_FILES_NAMES
			UpperCAmelCase_    :Any												=		PRETRAINED_VOCAB_FILES_MAP
			UpperCAmelCase_    :Union[str, Any]												=		PRETRAINED_INIT_CONFIGURATION
			UpperCAmelCase_    :Any												=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			UpperCAmelCase_    :List[str]												=		["input_ids", "attention_mask"]
			UpperCAmelCase_    :Optional[int]												=		re.compile(">>.+<<"				)  # type: re.Pattern






			def __init__(   self			,  __A			,  __A			,  __A			,  __A=None			,  __A=None			,  __A=None			,  __A="<unk>"			,  __A="</s>"			,  __A="<pad>"			,  __A=512			,  __A = None			,  __A=False			,  **__A			,  )     ->	None:
						lowerCAmelCase_       :Optional[int]     	=    {} if sp_model_kwargs is None else sp_model_kwargs

						super().__init__(
						    # bos_token=bos_token,  unused. Start decoding with config.decoder_start_token_id
						    source_lang=_A			,  target_lang=_A			,  unk_token=_A			,  eos_token=_A			,  pad_token=_A			,  model_max_length=_A			,  sp_model_kwargs=self.sp_model_kwargs			,  target_vocab_file=_A			,  separate_vocabs=_A			,  **_A			,  )
						assert Path(_A			).exists(), f"""cannot find spm source {source_spm}"""

						lowerCAmelCase_       :List[str]     	=    separate_vocabs
						lowerCAmelCase_       :List[str]     	=    load_json(_A			)
						if self.unk_token not in self.encoder:
									raise KeyError("""<unk> token must be in vocab"""			)
						assert self.pad_token in self.encoder

						if separate_vocabs:
									lowerCAmelCase_       :Tuple     	=    load_json(_A			)
									lowerCAmelCase_       :Tuple     	=    {v: k for k, v in self.target_encoder.items()}
									lowerCAmelCase_       :Optional[int]     	=    []
						else:
									lowerCAmelCase_       :Any     	=    {v: k for k, v in self.encoder.items()}
									lowerCAmelCase_       :Optional[Any]     	=    [k for k in self.encoder if k.startswith(""">>"""			) and k.endswith("""<<"""			)]

						lowerCAmelCase_       :int     	=    source_lang
						lowerCAmelCase_       :Any     	=    target_lang
						lowerCAmelCase_       :Any     	=    [source_spm, target_spm]

						# load SentencePiece model for pre-processing
						lowerCAmelCase_       :Dict     	=    load_spm(_A			,  self.sp_model_kwargs			)
						lowerCAmelCase_       :Optional[Any]     	=    load_spm(_A			,  self.sp_model_kwargs			)
						lowerCAmelCase_       :List[Any]     	=    self.spm_source
						lowerCAmelCase_       :str     	=    self.encoder

						# Multilingual target side: default to using first supported language code.

						self._setup_normalizer()






			def 					__lowerCAmelCase					(   self			)     ->	Dict:
						try:
									from sacremoses import MosesPunctNormalizer

									lowerCAmelCase_       :int     	=    MosesPunctNormalizer(self.source_lang			).normalize
						except (ImportError, FileNotFoundError):
									warnings.warn("""Recommended: pip install sacremoses."""			)
									lowerCAmelCase_       :Union[str, Any]     	=    lambda __A			: x






			def 					__lowerCAmelCase					(   self			,  __A			)     ->	str:
						return self.punc_normalizer(_A			) if x else ""






			def 					__lowerCAmelCase					(   self			,  __A			)     ->	List[Any]:
						return self.current_encoder.get(_A			,  self.current_encoder[self.unk_token]			)






			def 					__lowerCAmelCase					(   self			,  __A			)     ->	Dict:
						lowerCAmelCase_       :Optional[int]     	=    self.language_code_re.match(_A			)
						lowerCAmelCase_       :Optional[int]     	=    [match.group(0			)] if match else []
						return code, self.language_code_re.sub(""""""			,  _A			)






			def 					__lowerCAmelCase					(   self			,  __A			)     ->	List[str]:
						lowerCAmelCase_	,			lowerCAmelCase_       :Optional[int]     	=    self.remove_language_code(_A			)
						lowerCAmelCase_       :Any     	=    self.current_spm.encode(_A			,  out_type=_A			)
						return code + pieces






			def 					__lowerCAmelCase					(   self			,  __A			)     ->	str:
						return self.decoder.get(_A			,  self.unk_token			)






			def 					__lowerCAmelCase					(   self			,  __A			,  **__A			)     ->	str:
						return super().batch_decode(_A			,  **_A			)






			def 					__lowerCAmelCase					(   self			,  __A			,  **__A			)     ->	List[Any]:
						return super().decode(_A			,  **_A			)






			def 					__lowerCAmelCase					(   self			,  __A			)     ->	str:
						lowerCAmelCase_       :int     	=    self.spm_source if self._decode_use_source_tokenizer else self.spm_target
						lowerCAmelCase_       :Optional[int]     	=    []
						lowerCAmelCase_       :str     	=    """"""
						for token in tokens:
									# make sure that special tokens are not decoded using sentencepiece model
									if token in self.all_special_tokens:
												out_string += sp_model.decode_pieces(_A			) + token + " "
												lowerCAmelCase_       :Tuple     	=    []
									else:
												current_sub_tokens.append(_A			)
						out_string += sp_model.decode_pieces(_A			)
						return out_string.strip()






			def 					__lowerCAmelCase					(   self			,  __A			,  __A=None			)     ->	List[int]:
						if token_ids_a is None:
									return token_ids_a + [self.eos_token_id]
						# We don't expect to process pairs, but leave the pair logic for API consistency
						return token_ids_a + token_ids_a + [self.eos_token_id]






			def 					__lowerCAmelCase					(   self			)     ->	Tuple:
						lowerCAmelCase_       :Optional[Any]     	=    self.spm_source
						lowerCAmelCase_       :Dict     	=    self.encoder






			def 					__lowerCAmelCase					(   self			)     ->	Optional[int]:
						lowerCAmelCase_       :Tuple     	=    self.spm_target
						if self.separate_vocabs:
									lowerCAmelCase_       :Any     	=    self.target_encoder






			@property
			def 					__lowerCAmelCase					(   self			)     ->	int:
						return len(self.encoder			)






			def 					__lowerCAmelCase					(   self			,  __A			,  __A = None			)     ->	Tuple[str]:
						if not os.path.isdir(_A			):
									logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""			)
									return
						lowerCAmelCase_       :Any     	=    []

						if self.separate_vocabs:
									lowerCAmelCase_       :List[Any]     	=    os.path.join(
									    _A			,  (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab"""]			,  )
									lowerCAmelCase_       :Optional[int]     	=    os.path.join(
									    _A			,  (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""target_vocab_file"""]			,  )
									save_json(self.encoder			,  _A			)
									save_json(self.target_encoder			,  _A			)
									saved_files.append(_A			)
									saved_files.append(_A			)
						else:
									lowerCAmelCase_       :Tuple     	=    os.path.join(
									    _A			,  (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab"""]			)
									save_json(self.encoder			,  _A			)
									saved_files.append(_A			)

						for spm_save_filename, spm_orig_path, spm_model in zip(
						    [VOCAB_FILES_NAMES["""source_spm"""], VOCAB_FILES_NAMES["""target_spm"""]]			,  self.spm_files			,  [self.spm_source, self.spm_target]			,  ):
									lowerCAmelCase_       :Union[str, Any]     	=    os.path.join(
									    _A			,  (filename_prefix + """-""" if filename_prefix else """""") + spm_save_filename			)
									if os.path.abspath(_A			) != os.path.abspath(_A			) and os.path.isfile(_A			):
												copyfile(_A			,  _A			)
												saved_files.append(_A			)
									elif not os.path.isfile(_A			):
												with open(_A			,  """wb"""			) as fi:
															lowerCAmelCase_       :str     	=    spm_model.serialized_model_proto()
															fi.write(_A			)
												saved_files.append(_A			)

						return tuple(_A			)






			def 					__lowerCAmelCase					(   self			)     ->	Dict:
						return self.get_src_vocab()






			def 					__lowerCAmelCase					(   self			)     ->	List[Any]:
						return dict(self.encoder			,  **self.added_tokens_encoder			)






			def 					__lowerCAmelCase					(   self			)     ->	Any:
						return dict(self.target_encoder			,  **self.added_tokens_decoder			)






			def __getstate__(   self			)     ->	Dict:
						lowerCAmelCase_       :List[str]     	=    self.__dict__.copy()
						state.update(
						    {k: None for k in ["""spm_source""", """spm_target""", """current_spm""", """punc_normalizer""", """target_vocab_file"""]}			)
						return state






			def __setstate__(   self			,  __A			)     ->	None:
						lowerCAmelCase_       :Tuple     	=    d

						# for backward compatibility
						if not hasattr(self			,  """sp_model_kwargs"""			):
									lowerCAmelCase_       :Tuple     	=    {}

						lowerCAmelCase_	,			lowerCAmelCase_       :Tuple     	=    (load_spm(_A			,  self.sp_model_kwargs			) for f in self.spm_files)
						lowerCAmelCase_       :str     	=    self.spm_source
						self._setup_normalizer()






			def 					__lowerCAmelCase					(   self			,  *__A			,  **__A			)     ->	Any:
						return 1






			def 					__lowerCAmelCase					(   self			,  __A			)     ->	Tuple:
						lowerCAmelCase_       :Optional[Any]     	=    set(self.all_special_ids			)  # call it once instead of inside list comp
						all_special_ids.remove(self.unk_token_id			)  # <unk> is only sometimes special
						return [1 if x in all_special_ids else 0 for x in seq]






			def 					__lowerCAmelCase					(   self			,  __A			,  __A = None			,  __A = False			)     ->	List[int]:
						if already_has_special_tokens:
									return self._special_token_mask(_A			)
						elif token_ids_a is None:
									return self._special_token_mask(_A			) + [1]
						else:
									return self._special_token_mask(token_ids_a + token_ids_a			) + [1]






def 		_snake_case			(      lowercase__						:    Optional[Any] ,	lowercase__						:    Tuple						)    ->							str:


			'''simple docstring'''





			lowerCAmelCase_       :Dict     	=    sentencepiece.SentencePieceProcessor(**__lowerCamelCase						)
			spm.Load(__lowerCamelCase						)
			return spm






def 		_snake_case			(      lowercase__						:    Optional[int] ,	lowercase__						:    List[str]						)    ->							Tuple:


			'''simple docstring'''





			with open(__lowerCamelCase ,	"""w"""						) as f:
						json.dump(__lowerCamelCase ,	__lowerCamelCase ,	indent=2						)






def 		_snake_case			(      lowercase__						:    Dict						)    ->							str:


			'''simple docstring'''





			with open(__lowerCamelCase ,	"""r"""						) as f:
						return json.load(__lowerCamelCase						)








import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
   import torch


__UpperCAmelCase   						=     logging.get_logger(__name__)





class 				UpperCamelCase__		:






       """simple docstring"""


       def __init__(							self ,     _A ,     _A=21 ,     _A=[] ,     _A=None	)				->				List[Any]:
          SCREAMING_SNAKE_CASE_	=      model
          SCREAMING_SNAKE_CASE_	=      trace_batch_nums
          SCREAMING_SNAKE_CASE_	=      abort_after_batch_num

          # keep a LIFO buffer of frames to dump as soon as inf/nan is encountered to give context to the problem emergence
          SCREAMING_SNAKE_CASE_	=      collections.deque([] ,     _A	)
          SCREAMING_SNAKE_CASE_	=      []
          SCREAMING_SNAKE_CASE_	=      0
          SCREAMING_SNAKE_CASE_	=      0
          SCREAMING_SNAKE_CASE_	=      False
          SCREAMING_SNAKE_CASE_	=      '''                 '''

          self.analyse_model()

          self.register_forward_hook()



       def     _UpperCamelCase	(							self ,     _A=None	)				->				Any:
          if frame is not None:
             self.expand_frame(_A	)
          self.frames.append('''\n'''.join(self.frame	)	)
          SCREAMING_SNAKE_CASE_	=      []  # start a new frame



       def     _UpperCamelCase	(							self ,     _A	)				->				List[Any]:
          self.frame.append(_A	)



       def     _UpperCamelCase	(							self	)				->				Optional[Any]:
          print('''\n'''.join(self.frames	)	)
          SCREAMING_SNAKE_CASE_	=      []



       def     _UpperCamelCase	(							self	)				->				int:
          SCREAMING_SNAKE_CASE_	=      []



       def     _UpperCamelCase	(							self	)				->				List[str]:
          print(F'''\nDetected inf/nan during batch_number={self.batch_number}'''	)
          print(F'''Last {len(self.frames	)} forward frames:'''	)
          print(F'''{"abs min":8} {"abs max":8} metadata'''	)
          print('''\n'''.join(self.frames	)	)
          print('''\n\n'''	)
          SCREAMING_SNAKE_CASE_	=      []



       def     _UpperCamelCase	(							self	)				->				Optional[int]:
          # extract the fully qualified module names, to be able to report at run time. e.g.:
          # encoder.block.2.layer.0.SelfAttention.o
          #
          # for shared weights only the first shared module name will be registered
          SCREAMING_SNAKE_CASE_	=      {m: name for name, m in self.model.named_modules()}
          # self.longest_module_name = max(len(v) for v in self.module_names.values())



       def     _UpperCamelCase	(							self ,     _A ,     _A	)				->				Union[str, Any]:
          if torch.is_tensor(_A	):
             self.expand_frame(get_abs_min_max(_A ,     _A	)	)
             if detect_overflow(_A ,     _A	):
                SCREAMING_SNAKE_CASE_	=      True
          elif var is None:
             self.expand_frame(F'''{"None":>17} {ctx}'''	)
          else:
             self.expand_frame(F'''{"not a tensor":>17} {ctx}'''	)



       def     _UpperCamelCase	(							self	)				->				List[str]:
          self.expand_frame(F'''\n\n{self.prefix} *** Starting batch number={self.batch_number} ***'''	)
          self.expand_frame(F'''{"abs min":8} {"abs max":8} metadata'''	)



       def     _UpperCamelCase	(							self	)				->				List[Any]:
          self.expand_frame(F'''{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n'''	)



       def     _UpperCamelCase	(							self ,     _A ,     _A ,     _A	)				->				Union[str, Any]:
          self.expand_frame(F'''{self.prefix} {self.module_names[module]} {module.__class__.__name__}'''	)

          # params
          for name, p in module.named_parameters(recurse=_A	):
             self.analyse_variable(_A ,     _A	)

          # inputs
          if isinstance(_A ,     _A	):
             for i, x in enumerate(_A	):
                self.analyse_variable(_A ,     F'''input[{i}]'''	)
          else:
             self.analyse_variable(_A ,     '''input'''	)

          # outputs
          if isinstance(_A ,     _A	):
             for i, x in enumerate(_A	):
                # possibly a tuple of tuples
                if isinstance(_A ,     _A	):
                   for j, y in enumerate(_A	):
                      self.analyse_variable(_A ,     F'''output[{i}][{j}]'''	)
                else:
                   self.analyse_variable(_A ,     F'''output[{i}]'''	)
          else:
             self.analyse_variable(_A ,     '''output'''	)

          self.save_frame()



       def     _UpperCamelCase	(							self	)				->				List[Any]:
          self.model.apply(self._register_forward_hook	)



       def     _UpperCamelCase	(							self ,     _A	)				->				Tuple:
          module.register_forward_hook(self.forward_hook	)




       def     _UpperCamelCase	(							self ,     _A ,     _A ,     _A	)				->				Any:
          # - input is a tuple of packed inputs (could be non-Tensors)
          # - output could be a Tensor or a tuple of Tensors and non-Tensors

          SCREAMING_SNAKE_CASE_	=      False

          SCREAMING_SNAKE_CASE_	=      True if self.batch_number in self.trace_batch_nums else False
          if trace_mode:
             self.reset_saved_frames()

          if self.total_calls == 0:
             self.batch_start_frame()
          self.total_calls += 1

          # count batch numbers - the very first forward hook of the batch will be called when the
          # batch completes - i.e. it gets called very last - we know this batch has finished
          if module == self.model:
             self.batch_number += 1
             SCREAMING_SNAKE_CASE_	=      True

          self.create_frame(_A ,     _A ,     _A	)

          # if last_frame_of_batch:
          #     self.batch_end_frame()

          if trace_mode:
             self.trace_frames()

          if last_frame_of_batch:
             self.batch_start_frame()

          if self.detected_overflow and not trace_mode:
             self.dump_saved_frames()

             # now we can abort, as it's pointless to continue running
             raise ValueError(
                 '''DebugUnderflowOverflow: inf/nan detected, aborting as there is no point running further. '''
                 '''Please scroll up above this traceback to see the activation values prior to this event.'''	)

          # abort after certain batch if requested to do so
          if self.abort_after_batch_num is not None and self.batch_number > self.abort_after_batch_num:
             raise ValueError(
                 F'''DebugUnderflowOverflow: aborting after {self.batch_number} batches due to'''
                 F''' `abort_after_batch_num={self.abort_after_batch_num}` arg'''	)







def  A__ (       __lowerCamelCase,  __lowerCamelCase						):
   SCREAMING_SNAKE_CASE_	=      var.abs()
   return F'''{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}'''


def  A__ (       __lowerCamelCase,  __lowerCamelCase						):
   SCREAMING_SNAKE_CASE_	=      False
   if torch.isnan(__lowerCamelCase						).any().item():
      SCREAMING_SNAKE_CASE_	=      True
      print(F'''{ctx} has nans'''						)
   if torch.isinf(__lowerCamelCase						).any().item():
      SCREAMING_SNAKE_CASE_	=      True
      print(F'''{ctx} has infs'''						)

   # if needed to monitor large elements can enable the following
   if 0:  # and detected:
      SCREAMING_SNAKE_CASE_	=      var[torch.ge(var.abs(),  1_00						)]
      if naaa.numel() > 0:
         print(F'''{ctx}:  n100={naaa.numel()}'''						)
      SCREAMING_SNAKE_CASE_	=      var[torch.ge(var.abs(),  10_00						)]
      if naaaa.numel() > 0:
         print(F'''{ctx}: n1000={naaaa.numel()}'''						)
      SCREAMING_SNAKE_CASE_	=      var[torch.ge(var.abs(),  1_00_00						)]
      if naaaaa.numel() > 0:
         print(F'''{ctx}: n10000={naaaaa.numel()}'''						)

   if 0:
      print(F'''min={var.min():9.2e} max={var.max():9.2e}'''						)

   if 0:
      print(F'''min={var.min():9.2e} max={var.max():9.2e} var={var.var():9.2e} mean={var.mean():9.2e} ({ctx})'''						)

   return detected





class 				UpperCamelCase__		(	__SCREAMING_SNAKE_CASE	):






       """simple docstring"""


       UpperCAmelCase_				       ="underflow_overflow"
       UpperCAmelCase_				       ="tpu_metrics_debug"


