



import inspect
import unittest

import requests

from transformers import MgpstrConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
   import torch
   from torch import nn

   from transformers import MgpstrForSceneTextRecognition


if is_vision_available():
   from PIL import Image

   from transformers import MgpstrProcessor


class   lowerCamelCase__  :



 """simple docstring"""

 def __init__(						self				:				List[str] ,	__lowerCAmelCase				:				List[str] ,	__lowerCAmelCase				:				int=False ,	__lowerCAmelCase				:				int=13 ,	__lowerCAmelCase				:				Optional[int]=(32, 1_28) ,	__lowerCAmelCase				:				Optional[int]=4 ,	__lowerCAmelCase				:				List[Any]=3 ,	__lowerCAmelCase				:				Optional[int]=27 ,	__lowerCAmelCase				:				int=38 ,	__lowerCAmelCase				:				Optional[int]=99 ,	__lowerCAmelCase				:				Tuple=99 ,	__lowerCAmelCase				:				Optional[int]=32 ,	__lowerCAmelCase				:				List[Any]=5 ,	__lowerCAmelCase				:				Optional[Any]=4 ,	__lowerCAmelCase				:				Optional[Any]=4.0 ,	__lowerCAmelCase				:				Optional[int]=2_57 ,	__lowerCAmelCase				:				Tuple=None ,	)      ->  int:
    _A						     =					parent
    _A						     =					is_training
    _A						     =					batch_size
    _A						     =					image_size
    _A						     =					patch_size
    _A						     =					num_channels
    _A						     =					max_token_length
    _A						     =					num_character_labels
    _A						     =					num_bpe_labels
    _A						     =					num_wordpiece_labels
    _A						     =					hidden_size
    _A						     =					num_hidden_layers
    _A						     =					num_attention_heads
    _A						     =					mlp_ratio
    _A						     =					patch_embeds_hidden_size
    _A						     =					output_hidden_states

 def 						snake_case_			(						self				:				str			)      ->  str:
    _A						     =					floats_tensor([self.batch_size, self.num_channels, self.image_size[0], self.image_size[1]]			)
    _A						     =					self.get_config()
    return config, pixel_values

 def 						snake_case_			(						self				:				Optional[int]			)      ->  List[str]:
    return MgpstrConfig(
        image_size=self.image_size ,	patch_size=self.patch_size ,	num_channels=self.num_channels ,	max_token_length=self.max_token_length ,	num_character_labels=self.num_character_labels ,	num_bpe_labels=self.num_bpe_labels ,	num_wordpiece_labels=self.num_wordpiece_labels ,	hidden_size=self.hidden_size ,	num_hidden_layers=self.num_hidden_layers ,	num_attention_heads=self.num_attention_heads ,	mlp_ratio=self.mlp_ratio ,	output_hidden_states=self.output_hidden_states ,	)

 def 						snake_case_			(						self				:				Union[str, Any] ,	__lowerCAmelCase				:				Union[str, Any] ,	__lowerCAmelCase				:				List[Any]			)      ->  Dict:
    _A						     =					MgpstrForSceneTextRecognition(__lowerCAmelCase			)
    model.to(__lowerCAmelCase			)
    model.eval()
    with torch.no_grad():
       _A						     =					model(__lowerCAmelCase			)
    self.parent.assertEqual(
        generated_ids[0][0].shape ,	(self.batch_size, self.max_token_length, self.num_character_labels)			)

 def 						snake_case_			(						self				:				Tuple			)      ->  Tuple:
    _A						     =					self.prepare_config_and_inputs()
    _A						     =					config_and_inputs
    _A						     =					{'pixel_values': pixel_values}
    return config, inputs_dict


@require_torch
class   lowerCamelCase__  (					a_       ,     a_       ,     unittest.TestCase):



 """simple docstring"""

 a__				:   List[Any]   =     (MgpstrForSceneTextRecognition,) if is_torch_available() else ()
 a__				:   Any   =     {'feature-extraction': MgpstrForSceneTextRecognition} if is_torch_available() else {}
 a__				:   Dict   =     False

 a__				:   Union[str, Any]   =     False
 a__				:   List[str]   =     False
 a__				:   Optional[Any]   =     False
 a__				:   Dict   =     False

 def 						snake_case_			(						self				:				Dict			)      ->  List[Any]:
    _A						     =					MgpstrModelTester(self			)
    _A						     =					ConfigTester(self ,	config_class=__lowerCAmelCase ,	has_text_modality=__lowerCAmelCase			)

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

 def 						snake_case_			(						self				:				int			)      ->  Any:
    _A						     =					self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_model(*__lowerCAmelCase			)

 @unittest.skip(reason='''MgpstrModel does not use inputs_embeds'''			)
 def 						snake_case_			(						self				:				Tuple			)      ->  Dict:
    pass

 def 						snake_case_			(						self				:				Dict			)      ->  Dict:
    _A						     =					self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
       _A						     =					model_class(__lowerCAmelCase			)
       self.assertIsInstance(model.get_input_embeddings() ,	(nn.Module)			)
       _A						     =					model.get_output_embeddings()
       self.assertTrue(x is None or isinstance(__lowerCAmelCase ,	nn.Linear			)			)

 def 						snake_case_			(						self				:				List[Any]			)      ->  Tuple:
    _A						     =					self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
       _A						     =					model_class(__lowerCAmelCase			)
       _A						     =					inspect.signature(model.forward			)
       # signature.parameters is an OrderedDict => so arg_names order is deterministic
       _A						     =					[*signature.parameters.keys()]

       _A						     =					['pixel_values']
       self.assertListEqual(arg_names[:1] ,	__lowerCAmelCase			)

 @unittest.skip(reason='''MgpstrModel does not support feedforward chunking'''			)
 def 						snake_case_			(						self				:				Optional[Any]			)      ->  Optional[Any]:
    pass

 def 						snake_case_			(						self				:				str			)      ->  List[Any]:
    _A						     =					self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
       if not model_class.supports_gradient_checkpointing:
          continue

       _A						     =					True
       _A						     =					model_class(__lowerCAmelCase			)
       self.assertTrue(model.is_gradient_checkpointing			)

 def 						snake_case_			(						self				:				Tuple			)      ->  Tuple:
    def check_hidden_states_output(__lowerCAmelCase				:				int ,	__lowerCAmelCase				:				Optional[Any] ,	__lowerCAmelCase				:				Optional[int]			):
       _A						     =					model_class(__lowerCAmelCase			)
       model.to(__lowerCAmelCase			)
       model.eval()

       with torch.no_grad():
          _A						     =					model(**self._prepare_for_class(__lowerCAmelCase ,	__lowerCAmelCase			)			)

       _A						     =					outputs.hidden_states

       _A						     =					getattr(
           self.model_tester ,	'''expected_num_hidden_layers''' ,	self.model_tester.num_hidden_layers + 1			)
       self.assertEqual(len(__lowerCAmelCase			) ,	__lowerCAmelCase			)

       self.assertListEqual(
           list(hidden_states[0].shape[-2:]			) ,	[self.model_tester.patch_embeds_hidden_size, self.model_tester.hidden_size] ,	)

    _A						     =					self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
       _A						     =					True
       check_hidden_states_output(__lowerCAmelCase ,	__lowerCAmelCase ,	__lowerCAmelCase			)

       # check that output_hidden_states also work using config
       del inputs_dict["output_hidden_states"]
       _A						     =					True

       check_hidden_states_output(__lowerCAmelCase ,	__lowerCAmelCase ,	__lowerCAmelCase			)

 def 						snake_case_			(						self				:				List[Any]			)      ->  Optional[Any]:
    _A						     =					self.model_tester.prepare_config_and_inputs_for_common()

    _A						     =					_config_zero_init(__lowerCAmelCase			)
    for model_class in self.all_model_classes:
       _A						     =					model_class(config=__lowerCAmelCase			)
       for name, param in model.named_parameters():
          if isinstance(__lowerCAmelCase ,	(nn.Linear, nn.Convad, nn.LayerNorm)			):
             if param.requires_grad:
                self.assertIn(
                    ((param.data.mean() * 1E9).round() / 1E9).item() ,	[0.0, 1.0] ,	msg=f'''Parameter {name} of model {model_class} seems not properly initialized''' ,	)

 @unittest.skip(reason='''Retain_grad is tested in individual model tests'''			)
 def 						snake_case_			(						self				:				List[str]			)      ->  str:
    pass

def   SCREAMING_SNAKE_CASE_       (  )     ->    int:

   '''simple docstring'''





   _A						     =					'https://i.postimg.cc/ZKwLg2Gw/367-14.png'
   _A						     =					Image.open(requests.get(lowercase_  ,      stream=lowercase_							).raw							).convert('''RGB'''							)
   return im


@require_vision
@require_torch
class   lowerCamelCase__  (					unittest.TestCase):



 """simple docstring"""

 @slow
 def 						snake_case_			(						self				:				Dict			)      ->  Dict:
    _A						     =					'alibaba-damo/mgp-str-base'
    _A						     =					MgpstrForSceneTextRecognition.from_pretrained(__lowerCAmelCase			).to(__lowerCAmelCase			)
    _A						     =					MgpstrProcessor.from_pretrained(__lowerCAmelCase			)

    _A						     =					prepare_img()
    _A						     =					processor(images=__lowerCAmelCase ,	return_tensors='''pt'''			).pixel_values.to(__lowerCAmelCase			)

    # forward pass
    with torch.no_grad():
       _A						     =					model(__lowerCAmelCase			)

    # verify the logits
    self.assertEqual(outputs.logits[0].shape ,	torch.Size((1, 27, 38)			)			)

    _A						     =					processor.batch_decode(outputs.logits			)
    _A						     =					'ticket'

    self.assertEqual(out_strs['''generated_text'''][0] ,	__lowerCAmelCase			)

    _A						     =					torch.tensor(
        [[[-39.53_97, -44.40_24, -36.18_44], [-61.47_09, -63.86_39, -58.34_54], [-74.02_25, -68.54_94, -71.21_64]]] ,	device=__lowerCAmelCase ,	)

    self.assertTrue(torch.allclose(outputs.logits[0][:, 1:4, 1:4] ,	__lowerCAmelCase ,	atol=1E-4			)			)








from typing import List

from packaging import version
from sklearn.metrics import fa_score

import datasets
from datasets.config import PY_VERSION


if PY_VERSION < version.parse("""3.8"""):
   import importlib_metadata
else:
   import importlib.metadata as importlib_metadata


# TODO(Patrick/Anton)
UpperCAmelCase_		   =				"""\
"""

UpperCAmelCase_		   =				"""\
    XTREME-S is a benchmark to evaluate universal cross-lingual speech representations in many languages.
    XTREME-S covers four task families: speech recognition, classification, speech-to-text translation and retrieval.
"""

UpperCAmelCase_		   =				"""
Compute XTREME-S evaluation metric associated to each XTREME-S dataset.
Args:
    predictions: list of predictions to score.
        Each translation should be tokenized into a list of tokens.
    references: list of lists of references for each translation.
        Each reference should be tokenized into a list of tokens.
    bleu_kwargs: optional dict of keywords to be passed when computing 'bleu'.
        Keywords include Dict can be one of 'smooth_method', 'smooth_value', 'force', 'lowercase',
        'tokenize', 'use_effective_order'.
    wer_kwargs: optional dict of keywords to be passed when computing 'wer' and 'cer'.
        Keywords include 'concatenate_texts'.
Returns: depending on the XTREME-S task, one or several of:
    \"accuracy\": Accuracy - for 'fleurs-lang_id', 'minds14'
    \"f1\": F1 score - for 'minds14'
    \"wer\": Word error rate - for 'mls', 'fleurs-asr', 'voxpopuli', 'babel'
    \"cer\": Character error rate - for 'mls', 'fleurs-asr', 'voxpopuli', 'babel'
    \"bleu\": BLEU score according to the `sacrebleu` metric - for 'covost2'
Examples:

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'mls')  # 'mls', 'voxpopuli', 'fleurs-asr' or 'babel'
    >>> references = [\"it is sunny here\", \"paper and pen are essentials\"]
    >>> predictions = [\"it's sunny\", \"paper pen are essential\"]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'wer': 0.56, 'cer': 0.27}

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'covost2')
    >>> references = [\"bonjour paris\", \"il est necessaire de faire du sport de temps en temp\"]
    >>> predictions = [\"bonjour paris\", \"il est important de faire du sport souvent\"]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'bleu': 31.65}

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'fleurs-lang_id')
    >>> references = [0, 1, 0, 0, 1]
    >>> predictions = [0, 1, 1, 0, 0]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'accuracy': 0.6}

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'minds14')
    >>> references = [0, 1, 0, 0, 1]
    >>> predictions = [0, 1, 1, 0, 0]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'f1': 0.58, 'accuracy': 0.6}
"""

UpperCAmelCase_		   =				["""fleurs-asr""", """mls""", """voxpopuli""", """babel""", """covost2""", """fleurs-lang_id""", """minds14"""]
UpperCAmelCase_		   =				""""""

try:
   from jiwer import transforms as tr

   UpperCAmelCase_		   =				True
except ImportError:
   UpperCAmelCase_		   =				False

if _jiwer_available and version.parse(importlib_metadata.version("""jiwer""")) < version.parse("""2.3.0"""):


   class   lowerCamelCase__  (					tr.AbstractTransform):



    """simple docstring"""

    def __init__(						self				:				str ,	__lowerCAmelCase				:				str = " "			)      ->  Any:
       _A						     =					sentence_delimiter

    def 						snake_case_			(						self				:				List[str] ,	__lowerCAmelCase				:				str			)      ->  Optional[int]:
       return list(__lowerCAmelCase			)

    def 						snake_case_			(						self				:				Union[str, Any] ,	__lowerCAmelCase				:				List[str]			)      ->  Optional[Any]:
       _A						     =					[]
       for sent_idx, sentence in enumerate(__lowerCAmelCase			):
          chars.extend(self.process_string(__lowerCAmelCase			)			)
          if self.sentence_delimiter is not None and self.sentence_delimiter != "" and sent_idx < len(__lowerCAmelCase			) - 1:
             chars.append(self.sentence_delimiter			)
       return chars
   UpperCAmelCase_		   =				tr.Compose(
       [tr.RemoveMultipleSpaces(), tr.Strip(), SentencesToListOfCharacters(SENTENCE_DELIMITER)]
   )
elif _jiwer_available:
   UpperCAmelCase_		   =				tr.Compose(
       [
           tr.RemoveMultipleSpaces(),
           tr.Strip(),
           tr.ReduceToSingleSentence(SENTENCE_DELIMITER),
           tr.ReduceToListOfListOfChars(),
       ]
   )
else:
   UpperCAmelCase_		   =				None

def   SCREAMING_SNAKE_CASE_       (  _snake_case  :Dict  ,      _snake_case  :Union[str, Any]							)     ->    int:
   return float((preds == labels).mean()							)

def   SCREAMING_SNAKE_CASE_       (  _snake_case  :int  ,      _snake_case  :Dict							)     ->    Optional[Any]:
   return {
       "f1": float(fa_score(y_true=_snake_case  ,      y_pred=_snake_case  ,      average='''macro'''							)							),
       "accuracy": simple_accuracy(_snake_case  ,      _snake_case							),
   }

def   SCREAMING_SNAKE_CASE_       (  _snake_case  :Union[str, Any]  ,      _snake_case  :Union[str, Any]  ,      _snake_case  :Tuple="exp"  ,      _snake_case  :Dict=None  ,      _snake_case  :List[str]=False  ,      _snake_case  :Union[str, Any]=False  ,      _snake_case  :List[Any]=None  ,      _snake_case  :Tuple=False  ,      )     ->    Tuple:
   # xtreme-s can only have one label
   _A						     =					[[label] for label in labels]
   _A						     =					list(_snake_case							)
   try:
      import sacrebleu as scb
   except ImportError:
      raise ValueError(
          '''sacrebleu has to be installed in order to apply the bleu metric for covost2.'''
          '''You can install it via `pip install sacrebleu`.'''							)

   if version.parse(scb.__version__							) < version.parse('''1.4.12'''							):
      raise ImportWarning(
          '''To use `sacrebleu`, the module `sacrebleu>=1.4.12` is required, and the current version of `sacrebleu` doesn\'t match this condition.\n'''
          '''You can install it with `pip install "sacrebleu>=1.4.12"`.'''							)

   _A						     =					len(labels[0]							)
   if any(len(_snake_case							) != references_per_prediction for refs in labels							):
      raise ValueError('''Sacrebleu requires the same number of references for each prediction'''							)
   _A						     =					[[refs[i] for refs in labels] for i in range(_snake_case							)]
   _A						     =					scb.corpus_bleu(
       _snake_case  ,      _snake_case  ,      smooth_method=_snake_case  ,      smooth_value=_snake_case  ,      force=_snake_case  ,      lowercase=_snake_case  ,      use_effective_order=_snake_case  ,      **({'''tokenize''': tokenize} if tokenize else {})  ,      )
   return {"bleu": output.score}

def   SCREAMING_SNAKE_CASE_       (  _snake_case  :Optional[Any]  ,      _snake_case  :Optional[int]  ,      _snake_case  :List[Any]  ,      _snake_case  :str							)     ->    Tuple:
   try:
      from jiwer import compute_measures
   except ImportError:
      raise ValueError(
          F'''jiwer has to be installed in order to apply the wer metric for {config_name}.'''
          '''You can install it via `pip install jiwer`.'''							)

   if concatenate_texts:
      _A						     =					compute_measures(_snake_case  ,      _snake_case							)['''wer''']

      _A						     =					compute_measures(_snake_case  ,      _snake_case  ,      truth_transform=_snake_case  ,      hypothesis_transform=_snake_case							)['''wer''']
      return {"wer": wer, "cer": cer}
   else:

      def compute_score(_snake_case  :Tuple  ,      _snake_case  :Any  ,      _snake_case  :str="wer"							):
         _A						     =					0
         _A						     =					0
         for prediction, reference in zip(_snake_case  ,      _snake_case							):
            if score_type == "wer":
               _A						     =					compute_measures(_snake_case  ,      _snake_case							)
            elif score_type == "cer":
               _A						     =					compute_measures(
                   _snake_case  ,      _snake_case  ,      truth_transform=_snake_case  ,      hypothesis_transform=_snake_case							)
            incorrect += measures["substitutions"] + measures["deletions"] + measures["insertions"]
            total += measures["substitutions"] + measures["deletions"] + measures["hits"]
         return incorrect / total

      return {"wer": compute_score(_snake_case  ,      _snake_case  ,      '''wer'''							), "cer": compute_score(_snake_case  ,      _snake_case  ,      '''cer'''							)}


@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION       ,     _KWARGS_DESCRIPTION)
class   lowerCamelCase__  (					datasets.Metric):



 """simple docstring"""

 def 						snake_case_			(						self				:				str			)      ->  List[Any]:
    if self.config_name not in _CONFIG_NAMES:
       raise KeyError(f'''You should supply a configuration name selected in {_CONFIG_NAMES}'''			)

    _A						     =					'''int64''' if self.config_name in ['''fleurs-lang_id''', '''minds14'''] else '''string'''

    return datasets.MetricInfo(
        description=_DESCRIPTION ,	citation=_CITATION ,	inputs_description=_KWARGS_DESCRIPTION ,	features=datasets.Features(
            {'''predictions''': datasets.Value(__lowerCAmelCase			), '''references''': datasets.Value(__lowerCAmelCase			)}			) ,	codebase_urls=[] ,	reference_urls=[] ,	format='''numpy''' ,	)

 def 						snake_case_			(						self				:				Dict ,	__lowerCAmelCase				:				str ,	__lowerCAmelCase				:				Any ,	__lowerCAmelCase				:				Union[str, Any]=None ,	__lowerCAmelCase				:				Any=None			)      ->  Union[str, Any]:
    _A						     =					bleu_kwargs if bleu_kwargs is not None else {}
    _A						     =					wer_kwargs if wer_kwargs is not None else {}

    if self.config_name == "fleurs-lang_id":
       return {"accuracy": simple_accuracy(__lowerCAmelCase ,	__lowerCAmelCase			)}
    elif self.config_name == "minds14":
       return fa_and_simple_accuracy(__lowerCAmelCase ,	__lowerCAmelCase			)
    elif self.config_name == "covost2":
       _A						     =					bleu_kwargs.pop('''smooth_method''' ,	'''exp'''			)
       _A						     =					bleu_kwargs.pop('''smooth_value''' ,	__lowerCAmelCase			)
       _A						     =					bleu_kwargs.pop('''force''' ,	__lowerCAmelCase			)
       _A						     =					bleu_kwargs.pop('''lowercase''' ,	__lowerCAmelCase			)
       _A						     =					bleu_kwargs.pop('''tokenize''' ,	__lowerCAmelCase			)
       _A						     =					bleu_kwargs.pop('''use_effective_order''' ,	__lowerCAmelCase			)
       return bleu(
           preds=__lowerCAmelCase ,	labels=__lowerCAmelCase ,	smooth_method=__lowerCAmelCase ,	smooth_value=__lowerCAmelCase ,	force=__lowerCAmelCase ,	lowercase=__lowerCAmelCase ,	tokenize=__lowerCAmelCase ,	use_effective_order=__lowerCAmelCase ,	)
    elif self.config_name in ["fleurs-asr", "mls", "voxpopuli", "babel"]:
       _A						     =					wer_kwargs.pop('''concatenate_texts''' ,	__lowerCAmelCase			)
       return wer_and_cer(__lowerCAmelCase ,	__lowerCAmelCase ,	__lowerCAmelCase ,	self.config_name			)
    else:
       raise KeyError(f'''You should supply a configuration name selected in {_CONFIG_NAMES}'''			)



