



import collections

from .utils import ExplicitEnum, is_torch_available, logging


if is_torch_available():
    import torch


__magic_name__        =  logging.get_logger(__name__)

class 					lowercase						:




       '''simple docstring'''







       def __init__(						self		,       _snake_case		,       _snake_case=21		,       _snake_case=[]		,       _snake_case=None   )					->		Optional[Any]:

           """simple docstring"""
           UpperCAmelCase     =    model
           UpperCAmelCase     =    trace_batch_nums
           UpperCAmelCase     =    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
           UpperCAmelCase     =    collections.deque([]		,       _snake_case   )
           UpperCAmelCase     =    []
           UpperCAmelCase     =    0
           UpperCAmelCase     =    0
           UpperCAmelCase     =    False
           UpperCAmelCase     =    '''                 '''

           self.analyse_model()

           self.register_forward_hook()






       def 				snake_case_ (						self		,       _snake_case=None   )					->		Dict:

           """simple docstring"""
           if frame is not None:
               self.expand_frame(_snake_case   )
           self.frames.append('''\n'''.join(self.frame   )   )
           UpperCAmelCase     =    []  # start a new frame






       def 				snake_case_ (						self		,       _snake_case   )					->		Optional[Any]:

           """simple docstring"""
           self.frame.append(_snake_case   )






       def 				snake_case_ (						self   )					->		str:

           """simple docstring"""
           print('''\n'''.join(self.frames   )   )
           UpperCAmelCase     =    []






       def 				snake_case_ (						self   )					->		str:

           """simple docstring"""
           UpperCAmelCase     =    []






       def 				snake_case_ (						self   )					->		Optional[Any]:

           """simple docstring"""
           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'''   )
           UpperCAmelCase     =    []






       def 				snake_case_ (						self   )					->		List[Any]:

           """simple docstring"""
           # 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
           UpperCAmelCase     =    {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 				snake_case_ (						self		,       _snake_case		,       _snake_case   )					->		Optional[int]:

           """simple docstring"""
           if torch.is_tensor(_snake_case   ):
               self.expand_frame(get_abs_min_max(_snake_case		,       _snake_case   )   )
               if detect_overflow(_snake_case		,       _snake_case   ):
                   UpperCAmelCase     =    True
           elif var is None:
               self.expand_frame(f"""{'None':>17} {ctx}"""   )
           else:
               self.expand_frame(f"""{'not a tensor':>17} {ctx}"""   )






       def 				snake_case_ (						self   )					->		Union[str, Any]:

           """simple docstring"""
           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 				snake_case_ (						self   )					->		Optional[int]:

           """simple docstring"""
           self.expand_frame(f"""{self.prefix} *** Finished batch number={self.batch_number-1} ***\n\n"""   )






       def 				snake_case_ (						self		,       _snake_case		,       _snake_case		,       _snake_case   )					->		Optional[int]:

           """simple docstring"""
           self.expand_frame(f"""{self.prefix} {self.module_names[module]} {module.__class__.__name__}"""   )

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

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

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

           self.save_frame()






       def 				snake_case_ (						self   )					->		str:

           """simple docstring"""
           self.model.apply(self._register_forward_hook   )






       def 				snake_case_ (						self		,       _snake_case   )					->		Dict:

           """simple docstring"""
           module.register_forward_hook(self.forward_hook   )






       def 				snake_case_ (						self		,       _snake_case		,       _snake_case		,       _snake_case   )					->		List[Any]:

           """simple docstring"""
           # - input is a tuple of packed inputs (could be non-Tensors)
           # - output could be a Tensor or a tuple of Tensors and non-Tensors

           UpperCAmelCase     =    False

           UpperCAmelCase     =    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
               UpperCAmelCase     =    True

           self.create_frame(_snake_case		,       _snake_case		,       _snake_case   )

           # 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 					_lowerCAmelCase   (     A__:	str		,						A__:	Optional[Any]		):


    '''simple docstring'''


    UpperCAmelCase     =    var.abs()
    return F"""{abs_var.min():8.2e} {abs_var.max():8.2e} {ctx}"""




def 					_lowerCAmelCase   (     A__:	str		,						A__:	Dict		):


    '''simple docstring'''


    UpperCAmelCase     =    False
    if torch.isnan(lowercase_		).any().item():
        UpperCAmelCase     =    True
        print(F"""{ctx} has nans"""		)
    if torch.isinf(lowercase_		).any().item():
        UpperCAmelCase     =    True
        print(F"""{ctx} has infs"""		)

    # if needed to monitor large elements can enable the following
    if 0:  # and detected:
        UpperCAmelCase     =    var[torch.ge(var.abs()		,						100		)]
        if naaa.numel() > 0:
            print(F"""{ctx}:  n100={naaa.numel()}"""		)
        UpperCAmelCase     =    var[torch.ge(var.abs()		,						1000		)]
        if naaaa.numel() > 0:
            print(F"""{ctx}: n1000={naaaa.numel()}"""		)
        UpperCAmelCase     =    var[torch.ge(var.abs()		,						1_0000		)]
        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 					lowercase						(	__snake_case ):




       '''simple docstring'''







       __SCREAMING_SNAKE_CASE										=					'underflow_overflow'
       __SCREAMING_SNAKE_CASE										=					'tpu_metrics_debug'







import unittest

import pytest

from transformers import (
    MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    MBartaaTokenizerFast,
    MBartConfig,
    MBartForConditionalGeneration,
    TranslationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, require_tf, require_torch, slow

from .test_pipelines_common import ANY







@is_pipeline_test
class  __lowerCamelCase				(	unittest.TestCase	):
      lowerCamelCase_	:      Tuple          =					MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
      lowerCamelCase_	:      int          =					TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING







      def 	lowerCAmelCase_    (     self							,  lowerCamelCase							,  lowerCamelCase							,  lowerCamelCase   )						->       Any:
        if isinstance(model.config							,  lowerCamelCase   ):
          snake_case_						,				snake_case_       						=      list(tokenizer.lang_code_to_id.keys()   )[:2]
          snake_case_       						=      TranslationPipeline(model=lowerCamelCase							,  tokenizer=lowerCamelCase							,  src_lang=lowerCamelCase							,  tgt_lang=lowerCamelCase   )
        else:
          snake_case_       						=      TranslationPipeline(model=lowerCamelCase							,  tokenizer=lowerCamelCase   )
        return translator, ["Some string", "Some other text"]







      def 	lowerCAmelCase_    (     self							,  lowerCamelCase							,  lowerCamelCase   )						->       List[str]:
        snake_case_       						=      translator("""Some string"""   )
        self.assertEqual(lowerCamelCase							,  [{"""translation_text""": ANY(lowerCamelCase   )}]   )

        snake_case_       						=      translator(["""Some string"""]   )
        self.assertEqual(lowerCamelCase							,  [{"""translation_text""": ANY(lowerCamelCase   )}]   )

        snake_case_       						=      translator(["""Some string""", """other string"""]   )
        self.assertEqual(lowerCamelCase							,  [{"""translation_text""": ANY(lowerCamelCase   )}, {"""translation_text""": ANY(lowerCamelCase   )}]   )







      @require_torch
      def 	lowerCAmelCase_    (     self   )						->       List[Any]:
        snake_case_       						=      pipeline("""translation_en_to_ro"""							,  model="""patrickvonplaten/t5-tiny-random"""							,  framework="""pt"""   )
        snake_case_       						=      translator("""This is a test string"""							,  max_length=20   )
        self.assertEqual(
            lowerCamelCase							,  [
                {
                    """translation_text""": (
                        """Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide"""
                        """ Beide Beide"""
                    )
                }
            ]							,  )







      @require_tf
      def 	lowerCAmelCase_    (     self   )						->       Optional[Any]:
        snake_case_       						=      pipeline("""translation_en_to_ro"""							,  model="""patrickvonplaten/t5-tiny-random"""							,  framework="""tf"""   )
        snake_case_       						=      translator("""This is a test string"""							,  max_length=20   )
        self.assertEqual(
            lowerCamelCase							,  [
                {
                    """translation_text""": (
                        """Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide Beide"""
                        """ Beide Beide"""
                    )
                }
            ]							,  )







      @require_torch
      def 	lowerCAmelCase_    (     self   )						->       Union[str, Any]:
        snake_case_       						=      pipeline("""translation_en_to_de"""							,  model="""patrickvonplaten/t5-tiny-random"""							,  framework="""pt"""   )
        snake_case_       						=      translator("""This is a test string"""							,  max_length=20   )
        self.assertEqual(
            lowerCamelCase							,  [
                {
                    """translation_text""": (
                        """monoton monoton monoton monoton monoton monoton monoton monoton monoton monoton urine urine"""
                        """ urine urine urine urine urine urine urine"""
                    )
                }
            ]							,  )







      @require_tf
      def 	lowerCAmelCase_    (     self   )						->       List[Any]:
        snake_case_       						=      pipeline("""translation_en_to_de"""							,  model="""patrickvonplaten/t5-tiny-random"""							,  framework="""tf"""   )
        snake_case_       						=      translator("""This is a test string"""							,  max_length=20   )
        self.assertEqual(
            lowerCamelCase							,  [
                {
                    """translation_text""": (
                        """monoton monoton monoton monoton monoton monoton monoton monoton monoton monoton urine urine"""
                        """ urine urine urine urine urine urine urine"""
                    )
                }
            ]							,  )







class  __lowerCamelCase				(	unittest.TestCase	):







      @require_torch
      @slow
      def 	lowerCAmelCase_    (     self   )						->       Tuple:
        # We don't provide a default for this pair
        with self.assertRaises(lowerCamelCase   ):
          pipeline(task="""translation_cn_to_ar"""   )

        # but we do for this one
        snake_case_       						=      pipeline(task="""translation_en_to_de"""   )
        self.assertEqual(translator._preprocess_params["""src_lang"""]							,  """en"""   )
        self.assertEqual(translator._preprocess_params["""tgt_lang"""]							,  """de"""   )







      @require_torch
      @slow
      def 	lowerCAmelCase_    (     self   )						->       int:
        snake_case_       						=      MBartForConditionalGeneration.from_pretrained("""facebook/mbart-large-50-many-to-many-mmt"""   )
        snake_case_       						=      MBartaaTokenizerFast.from_pretrained("""facebook/mbart-large-50-many-to-many-mmt"""   )

        snake_case_       						=      pipeline(task="""translation"""							,  model=lowerCamelCase							,  tokenizer=lowerCamelCase   )
        # Missing src_lang, tgt_lang
        with self.assertRaises(lowerCamelCase   ):
          translator("""This is a test"""   )

        snake_case_       						=      translator("""This is a test"""							,  src_lang="""en_XX"""							,  tgt_lang="""ar_AR"""   )
        self.assertEqual(lowerCamelCase							,  [{"""translation_text""": """هذا إختبار"""}]   )

        snake_case_       						=      translator("""This is a test"""							,  src_lang="""en_XX"""							,  tgt_lang="""hi_IN"""   )
        self.assertEqual(lowerCamelCase							,  [{"""translation_text""": """यह एक परीक्षण है"""}]   )

        # src_lang, tgt_lang can be defined at pipeline call time
        snake_case_       						=      pipeline(task="""translation"""							,  model=lowerCamelCase							,  tokenizer=lowerCamelCase							,  src_lang="""en_XX"""							,  tgt_lang="""ar_AR"""   )
        snake_case_       						=      translator("""This is a test"""   )
        self.assertEqual(lowerCamelCase							,  [{"""translation_text""": """هذا إختبار"""}]   )







      @require_torch
      def 	lowerCAmelCase_    (     self   )						->       int:
        snake_case_       						=      """patrickvonplaten/t5-tiny-random"""
        snake_case_       						=      pipeline(task="""translation_cn_to_ar"""							,  model=lowerCamelCase   )
        self.assertEqual(translator._preprocess_params["""src_lang"""]							,  """cn"""   )
        self.assertEqual(translator._preprocess_params["""tgt_lang"""]							,  """ar"""   )







      @require_torch
      def 	lowerCAmelCase_    (     self   )						->       List[str]:
        snake_case_       						=      """patrickvonplaten/t5-tiny-random"""
        with pytest.warns(lowerCamelCase							,  match=r""".*translation_en_to_de.*"""   ):
          snake_case_       						=      pipeline(task="""translation"""							,  model=lowerCamelCase   )
        self.assertEqual(translator.task							,  """translation_en_to_de"""   )
        self.assertEqual(translator._preprocess_params["""src_lang"""]							,  """en"""   )
        self.assertEqual(translator._preprocess_params["""tgt_lang"""]							,  """de"""   )







      @require_torch
      def 	lowerCAmelCase_    (     self   )						->       Any:
        with self.assertRaises(lowerCamelCase   ):
          pipeline(task="""translation"""   )