

"""simple docstring"""


import doctest
import glob
import importlib
import inspect
import os
import re
from contextlib import contextmanager
from functools import wraps
from unittest.mock import patch

import numpy as np
import pytest
from absl.testing import parameterized

import datasets
from datasets import load_metric

from .utils import for_all_test_methods, local, slow


# mark all tests as integration
a					:Tuple    = pytest.mark.integration


a					:List[Any]    = {"comet"}
a					:Optional[Any]    = importlib.util.find_spec("fairseq") is not None

a					:Dict    = {"code_eval"}
a					:Union[str, Any]    = os.name == "nt"

a					:Union[str, Any]    = {"bertscore", "frugalscore", "perplexity"}
a					:Dict    = importlib.util.find_spec("transformers") is not None




def  _lowercase					(    __lowerCAmelCase							)       ->				List[str]:
      @wraps(__lowerCAmelCase							)
      def wrapper(self  ,					__lowerCAmelCase							):
            if not _has_fairseq and metric_name in REQUIRE_FAIRSEQ:
                  self.skipTest("""\"test requires Fairseq\""""							)
            else:
                  test_case(self  ,					__lowerCAmelCase							)

      return wrapper




def  _lowercase					(    __lowerCAmelCase							)       ->				List[str]:
      @wraps(__lowerCAmelCase							)
      def wrapper(self  ,					__lowerCAmelCase							):
            if not _has_transformers and metric_name in REQUIRE_TRANSFORMERS:
                  self.skipTest("""\"test requires transformers\""""							)
            else:
                  test_case(self  ,					__lowerCAmelCase							)

      return wrapper




def  _lowercase					(    __lowerCAmelCase							)       ->				Dict:
      @wraps(__lowerCAmelCase							)
      def wrapper(self  ,					__lowerCAmelCase							):
            if _on_windows and metric_name in UNSUPPORTED_ON_WINDOWS:
                  self.skipTest("""\"test not supported on Windows\""""							)
            else:
                  test_case(self  ,					__lowerCAmelCase							)

      return wrapper




def  _lowercase					(    )       ->				Optional[int]:
      SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	[metric_dir.split(os.sep							)[-2] for metric_dir in glob.glob("""./metrics/*/"""							)]
      return [{"testcase_name": x, "metric_name": x} for x in metrics if x != "gleu"]  # gleu is unfinished

@parameterized.named_parameters(get_local_metric_names())
@for_all_test_methods(
    UpperCamelCase_   ,	UpperCamelCase_   ,	UpperCamelCase_)
@local
class      __a							(parameterized.TestCase):






     '''simple docstring'''
     _SCREAMING_SNAKE_CASE						:List[Any]       = {}
     _SCREAMING_SNAKE_CASE						:str       = None

     @pytest.mark.filterwarnings("""ignore:metric_module_factory is deprecated:FutureWarning"""  )
     @pytest.mark.filterwarnings("""ignore:load_metric is deprecated:FutureWarning"""  )
     def 					_a			(						self	,    _a  )  ->      Any:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      List[str]      		=	"""[...]"""
           SCREAMING_SNAKE_CASE__   :      Dict      		=	importlib.import_module(
               datasets.load.metric_module_factory(os.path.join("""metrics"""	,    _a  )  ).module_path  )
           SCREAMING_SNAKE_CASE__   :      List[str]      		=	datasets.load.import_main_class(metric_module.__name__	,    dataset=_a  )
           # check parameters
           SCREAMING_SNAKE_CASE__   :      List[Any]      		=	inspect.signature(metric._compute  ).parameters
           self.assertTrue(all(p.kind != p.VAR_KEYWORD for p in parameters.values()  )  )  # no **kwargs
           # run doctest
           with self.patch_intensive_calls(_a	,    metric_module.__name__  ):
                 with self.use_local_metrics():
                       try:
                             SCREAMING_SNAKE_CASE__   :      List[str]      		=	doctest.testmod(_a	,    verbose=_a	,    raise_on_error=_a  )
                       except doctest.UnexpectedException as e:
                             raise e.exc_info[1]  # raise the exception that doctest caught
           self.assertEqual(results.failed	,    0  )
           self.assertGreater(results.attempted	,    1  )

     @slow
     def 					_a			(						self	,    _a  )  ->      Any:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	"""[...]"""
           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	importlib.import_module(
               datasets.load.metric_module_factory(os.path.join("""metrics"""	,    _a  )  ).module_path  )
           # run doctest
           with self.use_local_metrics():
                 SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	doctest.testmod(_a	,    verbose=_a	,    raise_on_error=_a  )
           self.assertEqual(results.failed	,    0  )
           self.assertGreater(results.attempted	,    1  )

     @contextmanager
     def 					_a			(						self	,    _a	,    _a  )  ->      Dict:


           """simple docstring"""


           if metric_name in self.INTENSIVE_CALLS_PATCHER:
                 with self.INTENSIVE_CALLS_PATCHER[metric_name](_a  ):
                       yield
           else:
                 yield

     @contextmanager
     def 					_a			(						self  )  ->      List[str]:


           """simple docstring"""


           def load_local_metric(_a	,    *_a	,    **_a  ):
                 return load_metric(os.path.join("""metrics"""	,    _a  )	,    *_a	,    **_a  )

           with patch("""datasets.load_metric"""  ) as mock_load_metric:
                 SCREAMING_SNAKE_CASE__   :      Any      		=	load_local_metric
                 yield





     @classmethod
     def 					_a			(						cls	,    _a  )  ->      Dict:


           """simple docstring"""


           def wrapper(_a  ):
                 SCREAMING_SNAKE_CASE__   :      Tuple      		=	contextmanager(_a  )
                 SCREAMING_SNAKE_CASE__   :      List[str]      		=	patcher
                 return patcher

           return wrapper





@LocalMetricTest.register_intensive_calls_patcher("""bleurt"""							)
def  _lowercase					(    __lowerCAmelCase							)       ->				Optional[int]:
      import tensorflow.compat.va as tf
      from bleurt.score import Predictor

      tf.flags.DEFINE_string("""sv"""  ,					""""""  ,					""""""							)  # handle pytest cli flags

      class      __a							(UpperCamelCase_):






           '''simple docstring'''

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


                 """simple docstring"""


                 assert len(input_dict["""input_ids"""]  ) == 2
                 return np.array([1.03, 1.04]  )

    # mock predict_fn which is supposed to do a forward pass with a bleurt model
      with patch("""bleurt.score._create_predictor"""							) as mock_create_predictor:
            SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	MockedPredictor()
            yield




@LocalMetricTest.register_intensive_calls_patcher("""bertscore"""							)
def  _lowercase					(    __lowerCAmelCase							)       ->				Tuple:
      import torch

      def bert_cos_score_idf(__lowerCAmelCase  ,					__lowerCAmelCase  ,					*__lowerCAmelCase  ,					**__lowerCAmelCase							):
            return torch.tensor([[1.0, 1.0, 1.0]] * len(__lowerCAmelCase							)							)

      # mock get_model which is supposed to do download a bert model
      # mock bert_cos_score_idf which is supposed to do a forward pass with a bert model
      with patch("""bert_score.scorer.get_model"""							), patch(
          """bert_score.scorer.bert_cos_score_idf"""							) as mock_bert_cos_score_idf:
            SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	bert_cos_score_idf
            yield




@LocalMetricTest.register_intensive_calls_patcher("""comet"""							)
def  _lowercase					(    __lowerCAmelCase							)       ->				List[str]:
      def load_from_checkpoint(__lowerCAmelCase							):

            class      __a							:






                 '''simple docstring'''

                 def 					_a			(						self	,    _a	,    *_a	,    **_a  )  ->      Dict:


                       """simple docstring"""


                       assert len(_a  ) == 2
                       SCREAMING_SNAKE_CASE__   :      Any      		=	[0.19, 0.92]
                       return scores, sum(_a  ) / len(_a  )





            return Model()

      # mock load_from_checkpoint which is supposed to do download a bert model
      # mock load_from_checkpoint which is supposed to do download a bert model
      with patch("""comet.download_model"""							) as mock_download_model:
            SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	None
            with patch("""comet.load_from_checkpoint"""							) as mock_load_from_checkpoint:
                  SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	load_from_checkpoint
                  yield




def  _lowercase					(    )       ->				Optional[Any]:
      SCREAMING_SNAKE_CASE__   :      Dict      		=	load_metric(os.path.join("""metrics"""  ,					"""seqeval"""							)							)
      SCREAMING_SNAKE_CASE__   :      Tuple      		=	"""ERROR"""
      SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	F'''Scheme should be one of [IOB1, IOB2, IOE1, IOE2, IOBES, BILOU], got {wrong_scheme}'''
      with pytest.raises(__lowerCAmelCase  ,					match=re.escape(__lowerCAmelCase							)							):
            metric.compute(predictions=[]  ,					references=[]  ,					scheme=__lowerCAmelCase							)







"""simple docstring"""


import os
from shutil import copyfile
from typing import List, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
  from .tokenization_nllb import NllbTokenizer
else:
  a					:Optional[int]    = None


a					:Optional[Any]    = logging.get_logger(__name__)


a					:Optional[Any]    = {"vocab_file": "sentencepiece.bpe.model", "tokenizer_file": "tokenizer.json"}

a					:Union[str, Any]    = {
    "vocab_file": {
        "facebook/nllb-200-distilled-600M": (
            "https://huggingface.co/facebook/nllb-200-distilled-600M/resolve/main/sentencepiece.bpe.model"
        ),
    },
    "tokenizer_file": {
        "facebook/nllb-200-distilled-600M": (
            "https://huggingface.co/facebook/nllb-200-distilled-600M/resolve/main/tokenizer.json"
        ),
    },
}

a					:Any    = {
    "facebook/nllb-large-en-ro": 1_024,
    "facebook/nllb-200-distilled-600M": 1_024,
}

# fmt: off
a					:Tuple    = ["ace_Arab", "ace_Latn", "acm_Arab", "acq_Arab", "aeb_Arab", "afr_Latn", "ajp_Arab", "aka_Latn", "amh_Ethi", "apc_Arab", "arb_Arab", "ars_Arab", "ary_Arab", "arz_Arab", "asm_Beng", "ast_Latn", "awa_Deva", "ayr_Latn", "azb_Arab", "azj_Latn", "bak_Cyrl", "bam_Latn", "ban_Latn", "bel_Cyrl", "bem_Latn", "ben_Beng", "bho_Deva", "bjn_Arab", "bjn_Latn", "bod_Tibt", "bos_Latn", "bug_Latn", "bul_Cyrl", "cat_Latn", "ceb_Latn", "ces_Latn", "cjk_Latn", "ckb_Arab", "crh_Latn", "cym_Latn", "dan_Latn", "deu_Latn", "dik_Latn", "dyu_Latn", "dzo_Tibt", "ell_Grek", "eng_Latn", "epo_Latn", "est_Latn", "eus_Latn", "ewe_Latn", "fao_Latn", "pes_Arab", "fij_Latn", "fin_Latn", "fon_Latn", "fra_Latn", "fur_Latn", "fuv_Latn", "gla_Latn", "gle_Latn", "glg_Latn", "grn_Latn", "guj_Gujr", "hat_Latn", "hau_Latn", "heb_Hebr", "hin_Deva", "hne_Deva", "hrv_Latn", "hun_Latn", "hye_Armn", "ibo_Latn", "ilo_Latn", "ind_Latn", "isl_Latn", "ita_Latn", "jav_Latn", "jpn_Jpan", "kab_Latn", "kac_Latn", "kam_Latn", "kan_Knda", "kas_Arab", "kas_Deva", "kat_Geor", "knc_Arab", "knc_Latn", "kaz_Cyrl", "kbp_Latn", "kea_Latn", "khm_Khmr", "kik_Latn", "kin_Latn", "kir_Cyrl", "kmb_Latn", "kon_Latn", "kor_Hang", "kmr_Latn", "lao_Laoo", "lvs_Latn", "lij_Latn", "lim_Latn", "lin_Latn", "lit_Latn", "lmo_Latn", "ltg_Latn", "ltz_Latn", "lua_Latn", "lug_Latn", "luo_Latn", "lus_Latn", "mag_Deva", "mai_Deva", "mal_Mlym", "mar_Deva", "min_Latn", "mkd_Cyrl", "plt_Latn", "mlt_Latn", "mni_Beng", "khk_Cyrl", "mos_Latn", "mri_Latn", "zsm_Latn", "mya_Mymr", "nld_Latn", "nno_Latn", "nob_Latn", "npi_Deva", "nso_Latn", "nus_Latn", "nya_Latn", "oci_Latn", "gaz_Latn", "ory_Orya", "pag_Latn", "pan_Guru", "pap_Latn", "pol_Latn", "por_Latn", "prs_Arab", "pbt_Arab", "quy_Latn", "ron_Latn", "run_Latn", "rus_Cyrl", "sag_Latn", "san_Deva", "sat_Beng", "scn_Latn", "shn_Mymr", "sin_Sinh", "slk_Latn", "slv_Latn", "smo_Latn", "sna_Latn", "snd_Arab", "som_Latn", "sot_Latn", "spa_Latn", "als_Latn", "srd_Latn", "srp_Cyrl", "ssw_Latn", "sun_Latn", "swe_Latn", "swh_Latn", "szl_Latn", "tam_Taml", "tat_Cyrl", "tel_Telu", "tgk_Cyrl", "tgl_Latn", "tha_Thai", "tir_Ethi", "taq_Latn", "taq_Tfng", "tpi_Latn", "tsn_Latn", "tso_Latn", "tuk_Latn", "tum_Latn", "tur_Latn", "twi_Latn", "tzm_Tfng", "uig_Arab", "ukr_Cyrl", "umb_Latn", "urd_Arab", "uzn_Latn", "vec_Latn", "vie_Latn", "war_Latn", "wol_Latn", "xho_Latn", "ydd_Hebr", "yor_Latn", "yue_Hant", "zho_Hans", "zho_Hant", "zul_Latn"]

class      __a							(UpperCamelCase_):






     '''simple docstring'''
     _SCREAMING_SNAKE_CASE						:Optional[Any]       = VOCAB_FILES_NAMES
     _SCREAMING_SNAKE_CASE						:List[str]       = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _SCREAMING_SNAKE_CASE						:str       = PRETRAINED_VOCAB_FILES_MAP
     _SCREAMING_SNAKE_CASE						:int       = ["""input_ids""", """attention_mask"""]
     _SCREAMING_SNAKE_CASE						:Tuple       = NllbTokenizer

     _SCREAMING_SNAKE_CASE						:List[int]       = []
     _SCREAMING_SNAKE_CASE						:List[int]       = []

     def __init__(						self	,    _a=None	,    _a=None	,    _a="<s>"	,    _a="</s>"	,    _a="</s>"	,    _a="<s>"	,    _a="<unk>"	,    _a="<pad>"	,    _a="<mask>"	,    _a=None	,    _a=None	,    _a=None	,    _a=False	,    **_a	,    )  ->      str:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	AddedToken(_a	,    lstrip=_a	,    rstrip=_a  ) if isinstance(_a	,    _a  ) else mask_token
           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	legacy_behaviour
           super().__init__(
               vocab_file=_a	,    tokenizer_file=_a	,    bos_token=_a	,    eos_token=_a	,    sep_token=_a	,    cls_token=_a	,    unk_token=_a	,    pad_token=_a	,    mask_token=_a	,    src_lang=_a	,    tgt_lang=_a	,    additional_special_tokens=_a	,    legacy_behaviour=_a	,    **_a	,    )

           SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	vocab_file
           SCREAMING_SNAKE_CASE__   :      str      		=	False if not self.vocab_file else True

           SCREAMING_SNAKE_CASE__   :      Dict      		=	FAIRSEQ_LANGUAGE_CODES.copy()

           if additional_special_tokens is not None:
                 # Only add those special tokens if they are not already there.
                 _additional_special_tokens.extend(
                     [t for t in additional_special_tokens if t not in _additional_special_tokens]  )

           self.add_special_tokens({"""additional_special_tokens""": _additional_special_tokens}  )
           SCREAMING_SNAKE_CASE__   :      List[str]      		=	{
               lang_code: self.convert_tokens_to_ids(_a  ) for lang_code in FAIRSEQ_LANGUAGE_CODES
           }

           SCREAMING_SNAKE_CASE__   :      Dict      		=	src_lang if src_lang is not None else """eng_Latn"""
           SCREAMING_SNAKE_CASE__   :      List[str]      		=	self.convert_tokens_to_ids(self._src_lang  )
           SCREAMING_SNAKE_CASE__   :      Dict      		=	tgt_lang
           self.set_src_lang_special_tokens(self._src_lang  )

     @property
     def 					_a			(						self  )  ->      str:


           """simple docstring"""


           return self._src_lang

     @src_lang.setter
     def 					_a			(						self	,    _a  )  ->      None:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      int      		=	new_src_lang
           self.set_src_lang_special_tokens(self._src_lang  )

     def 					_a			(						self	,    _a	,    _a = None  )  ->      List[int]:


           """simple docstring"""


           if token_ids_a is None:
                 return self.prefix_tokens + token_ids_a + self.suffix_tokens
           # We don't expect to process pairs, but leave the pair logic for API consistency
           return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens

     def 					_a			(						self	,    _a	,    _a = None  )  ->      List[int]:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      int      		=	[self.sep_token_id]
           SCREAMING_SNAKE_CASE__   :      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 + sep + token_ids_a + sep  ) * [0]

     def 					_a			(						self	,    _a	,    _a	,    _a	,    _a	,    **_a  )  ->      Tuple:


           """simple docstring"""


           if src_lang is None or tgt_lang is None:
                 raise ValueError("""Translation requires a `src_lang` and a `tgt_lang` for this model"""  )
           SCREAMING_SNAKE_CASE__   :      Dict      		=	src_lang
           SCREAMING_SNAKE_CASE__   :      Dict      		=	self(_a	,    add_special_tokens=_a	,    return_tensors=_a	,    **_a  )
           SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	self.convert_tokens_to_ids(_a  )
           SCREAMING_SNAKE_CASE__   :      List[Any]      		=	tgt_lang_id
           return inputs

     def 					_a			(						self	,    _a	,    _a = "eng_Latn"	,    _a = None	,    _a = "fra_Latn"	,    **_a	,    )  ->      BatchEncoding:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      Tuple      		=	src_lang
           SCREAMING_SNAKE_CASE__   :      Dict      		=	tgt_lang
           return super().prepare_seqaseq_batch(_a	,    _a	,    **_a  )

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


           """simple docstring"""


           return self.set_src_lang_special_tokens(self.src_lang  )

     def 					_a			(						self  )  ->      str:


           """simple docstring"""


           return self.set_tgt_lang_special_tokens(self.tgt_lang  )

     def 					_a			(						self	,    _a  )  ->      None:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      List[Any]      		=	self.convert_tokens_to_ids(_a  )

           if self.legacy_behaviour:
                 SCREAMING_SNAKE_CASE__   :      str      		=	[]
                 SCREAMING_SNAKE_CASE__   :      Dict      		=	[self.eos_token_id, self.cur_lang_code]
           else:
                 SCREAMING_SNAKE_CASE__   :      Dict      		=	[self.cur_lang_code]
                 SCREAMING_SNAKE_CASE__   :      Dict      		=	[self.eos_token_id]

           SCREAMING_SNAKE_CASE__   :      Optional[Any]      		=	self.convert_ids_to_tokens(self.prefix_tokens  )
           SCREAMING_SNAKE_CASE__   :      int      		=	self.convert_ids_to_tokens(self.suffix_tokens  )

           SCREAMING_SNAKE_CASE__   :      int      		=	processors.TemplateProcessing(
               single=prefix_tokens_str + ["""$A"""] + suffix_tokens_str	,    pair=prefix_tokens_str + ["""$A""", """$B"""] + suffix_tokens_str	,    special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str	,    self.prefix_tokens + self.suffix_tokens  )  )	,    )

     def 					_a			(						self	,    _a  )  ->      None:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__   :      str      		=	self.convert_tokens_to_ids(_a  )
           if self.legacy_behaviour:
                 SCREAMING_SNAKE_CASE__   :      List[Any]      		=	[]
                 SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	[self.eos_token_id, self.cur_lang_code]
           else:
                 SCREAMING_SNAKE_CASE__   :      Optional[int]      		=	[self.cur_lang_code]
                 SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	[self.eos_token_id]

           SCREAMING_SNAKE_CASE__   :      Any      		=	self.convert_ids_to_tokens(self.prefix_tokens  )
           SCREAMING_SNAKE_CASE__   :      Union[str, Any]      		=	self.convert_ids_to_tokens(self.suffix_tokens  )

           SCREAMING_SNAKE_CASE__   :      Tuple      		=	processors.TemplateProcessing(
               single=prefix_tokens_str + ["""$A"""] + suffix_tokens_str	,    pair=prefix_tokens_str + ["""$A""", """$B"""] + suffix_tokens_str	,    special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str	,    self.prefix_tokens + self.suffix_tokens  )  )	,    )





     def 					_a			(						self	,    _a	,    _a = None  )  ->      Tuple[str]:


           """simple docstring"""


           if not self.can_save_slow_tokenizer:
                 raise ValueError(
                     """Your fast tokenizer does not have the necessary information to save the vocabulary for a slow """
                     """tokenizer."""  )

           if not os.path.isdir(_a  ):
                 logger.error(f'''Vocabulary path ({save_directory}) should be a directory.'''  )
                 return
           SCREAMING_SNAKE_CASE__   :      Dict      		=	os.path.join(
               _a	,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]  )

           if os.path.abspath(self.vocab_file  ) != os.path.abspath(_a  ):
                 copyfile(self.vocab_file	,    _a  )

           return (out_vocab_file,)




