




"""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_      			=       pytest.mark.integration


A_      			=       {'''comet'''}
A_      			=       importlib.util.find_spec('''fairseq''') is not None

A_      			=       {'''code_eval'''}
A_      			=       os.name == '''nt'''

A_      			=       {'''bertscore''', '''frugalscore''', '''perplexity'''}
A_      			=       importlib.util.find_spec('''transformers''') is not None


def       UpperCAmelCase__ (snake_case__							:      List[Any]					):
  """simple docstring"""

  @wraps(snake_case__					)
  def wrapper(self							:      Dict	,     snake_case__							:      Optional[Any]					):
    if not _has_fairseq and metric_name in REQUIRE_FAIRSEQ:
      self.skipTest("""\"test requires Fairseq\""""					)
    else:
      test_case(self	,     snake_case__					)

  return wrapper


def       UpperCAmelCase__ (snake_case__							:      Any					):
  """simple docstring"""

  @wraps(snake_case__					)
  def wrapper(self							:      Optional[Any]	,     snake_case__							:      str					):
    if not _has_transformers and metric_name in REQUIRE_TRANSFORMERS:
      self.skipTest("""\"test requires transformers\""""					)
    else:
      test_case(self	,     snake_case__					)

  return wrapper


def       UpperCAmelCase__ (snake_case__							:      str					):
  """simple docstring"""

  @wraps(snake_case__					)
  def wrapper(self							:      Dict	,     snake_case__							:      Optional[Any]					):
    if _on_windows and metric_name in UNSUPPORTED_ON_WINDOWS:
      self.skipTest("""\"test not supported on Windows\""""					)
    else:
      test_case(self	,     snake_case__					)

  return wrapper


def       UpperCAmelCase__ ():
  """simple docstring"""

  _snake_case							:	Optional[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(
    __a ,    __a ,    __a				)
@local
class       lowercase(  parameterized.TestCase				):







 '''simple docstring'''





 lowercase__				      =      {}
 lowercase__				      =      None


 @pytest.mark.filterwarnings("""ignore:metric_module_factory is deprecated:FutureWarning"""	)
 @pytest.mark.filterwarnings("""ignore:load_metric is deprecated:FutureWarning"""	)
 def 				UpperCamelCase_     (       self:	List[str],   a_:	List[str]	):




   '''simple docstring'''



   _snake_case							:	Any			=						"""[...]"""
   _snake_case							:	str			=						importlib.import_module(
       datasets.load.metric_module_factory(os.path.join("""metrics""",   a_	)	).module_path	)
   _snake_case							:	List[str]			=						datasets.load.import_main_class(metric_module.__name__,   dataset=a_	)
   # check parameters
   _snake_case							:	str			=						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:
         _snake_case							:	int			=						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 				UpperCamelCase_     (       self:	Optional[int],   a_:	List[Any]	):




   '''simple docstring'''



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


 @contextmanager
 def 				UpperCamelCase_     (       self:	List[Any],   a_:	Tuple,   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 				UpperCamelCase_     (       self:	str	):




   '''simple docstring'''



   def load_local_metric(a_:	Any,   *a_:	List[Any],   **a_:	Optional[Any]	):
     return load_metric(os.path.join("""metrics""",   a_	),   *a_,   **a_	)

   with patch("""datasets.load_metric"""	) as mock_load_metric:
     _snake_case							:	str			=						load_local_metric
     yield


 @classmethod
 def 				UpperCamelCase_     (       cls:	List[str],   a_:	Union[str, Any]	):




   '''simple docstring'''



   def wrapper(a_:	int	):
     _snake_case							:	str			=						contextmanager(a_	)
     _snake_case							:	Union[str, Any]			=						patcher
     return patcher

   return wrapper







@LocalMetricTest.register_intensive_calls_patcher("""bleurt"""					)
def       UpperCAmelCase__ (snake_case__							:      Any					):
  """simple docstring"""

  import tensorflow.compat.va as tf
  from bleurt.score import Predictor

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






  class       lowercase(  __a				):







   '''simple docstring'''





   def 				UpperCamelCase_     (       self:	int,   a_:	Union[str, 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:
    _snake_case							:	Union[str, Any]			=						MockedPredictor()
    yield


@LocalMetricTest.register_intensive_calls_patcher("""bertscore"""					)
def       UpperCAmelCase__ (snake_case__							:      Optional[Any]					):
  """simple docstring"""

  import torch

  def bert_cos_score_idf(snake_case__							:      Optional[Any]	,     snake_case__							:      List[Any]	,     *snake_case__							:      List[str]	,     **snake_case__							:      int					):
    return torch.tensor([[1.0, 1.0, 1.0]] * len(snake_case__					)					)

  # 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:
    _snake_case							:	Union[str, Any]			=						bert_cos_score_idf
    yield


@LocalMetricTest.register_intensive_calls_patcher("""comet"""					)
def       UpperCAmelCase__ (snake_case__							:      Optional[int]					):
  """simple docstring"""

  def load_from_checkpoint(snake_case__							:      Dict					):






    class       lowercase:







     '''simple docstring'''





     def 				UpperCamelCase_     (       self:	Optional[int],   a_:	Union[str, Any],   *a_:	Optional[Any],   **a_:	Dict	):




       '''simple docstring'''



       assert len(a_	) == 2
       _snake_case							:	Optional[int]			=						[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:
    _snake_case							:	Any			=						None
    with patch("""comet.load_from_checkpoint"""					) as mock_load_from_checkpoint:
      _snake_case							:	Union[str, Any]			=						load_from_checkpoint
      yield






def       UpperCAmelCase__ ():
  """simple docstring"""

  _snake_case							:	Dict			=						load_metric(os.path.join("""metrics"""	,     """seqeval"""					)					)
  _snake_case							:	List[str]			=						"""ERROR"""
  _snake_case							:	int			=						F"Scheme should be one of [IOB1, IOB2, IOE1, IOE2, IOBES, BILOU], got {wrong_scheme}"
  with pytest.raises(snake_case__	,     match=re.escape(snake_case__					)					):
    metric.compute(predictions=[]	,     references=[]	,     scheme=snake_case__					)








"""simple docstring"""


import inspect
import unittest

from huggingface_hub import hf_hub_download

from transformers import ConvNextConfig, UperNetConfig
from transformers.testing_utils import require_torch, require_torch_multi_gpu, 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, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch

     from transformers import UperNetForSemanticSegmentation
     from transformers.models.upernet.modeling_upernet import UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from PIL import Image

     from transformers import AutoImageProcessor






class       lowercase:







 '''simple docstring'''





 def __init__(       self:	Dict,   a_:	Union[str, Any],   a_:	Tuple=13,   a_:	Dict=32,   a_:	Optional[Any]=3,   a_:	Optional[Any]=4,   a_:	Optional[int]=[10, 20, 30, 40],   a_:	Any=[2, 2, 3, 2],   a_:	Dict=True,   a_:	Dict=True,   a_:	List[str]=37,   a_:	Dict="gelu",   a_:	List[str]=10,   a_:	Union[str, Any]=0.02,   a_:	Any=["stage2", "stage3", "stage4"],   a_:	Optional[int]=3,   a_:	Tuple=None,   ):




   '''simple docstring'''



   _snake_case							:	Dict			=						parent
   _snake_case							:	Dict			=						batch_size
   _snake_case							:	Optional[Any]			=						image_size
   _snake_case							:	int			=						num_channels
   _snake_case							:	Tuple			=						num_stages
   _snake_case							:	int			=						hidden_sizes
   _snake_case							:	List[str]			=						depths
   _snake_case							:	str			=						is_training
   _snake_case							:	Dict			=						use_labels
   _snake_case							:	List[str]			=						intermediate_size
   _snake_case							:	Optional[int]			=						hidden_act
   _snake_case							:	Any			=						type_sequence_label_size
   _snake_case							:	List[str]			=						initializer_range
   _snake_case							:	Union[str, Any]			=						out_features
   _snake_case							:	Dict			=						num_labels
   _snake_case							:	int			=						scope
   _snake_case							:	Dict			=						num_stages


 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   _snake_case							:	Optional[Any]			=						floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]	)

   _snake_case							:	Optional[int]			=						None
   if self.use_labels:
     _snake_case							:	Optional[Any]			=						ids_tensor([self.batch_size],   self.type_sequence_label_size	)

   _snake_case							:	Tuple			=						self.get_config()

   return config, pixel_values, labels


 def 				UpperCamelCase_     (       self:	Tuple	):




   '''simple docstring'''



   return ConvNextConfig(
       num_channels=self.num_channels,   num_stages=self.num_stages,   hidden_sizes=self.hidden_sizes,   depths=self.depths,   is_training=self.is_training,   intermediate_size=self.intermediate_size,   hidden_act=self.hidden_act,   out_features=self.out_features,   )


 def 				UpperCamelCase_     (       self:	List[Any]	):




   '''simple docstring'''



   return UperNetConfig(
       backbone_config=self.get_backbone_config(),   hidden_size=512,   pool_scales=[1, 2, 3, 6],   use_auxiliary_head=a_,   auxiliary_loss_weight=0.4,   auxiliary_in_channels=40,   auxiliary_channels=256,   auxiliary_num_convs=1,   auxiliary_concat_input=a_,   loss_ignore_index=255,   num_labels=self.num_labels,   )


 def 				UpperCamelCase_     (       self:	Tuple,   a_:	List[Any],   a_:	Dict,   a_:	Tuple	):




   '''simple docstring'''



   _snake_case							:	List[Any]			=						UperNetForSemanticSegmentation(config=a_	)
   model.to(a_	)
   model.eval()
   _snake_case							:	Tuple			=						model(a_	)
   self.parent.assertEqual(
       result.logits.shape,   (self.batch_size, self.num_labels, self.image_size, self.image_size)	)


 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   _snake_case							:	Any			=						self.prepare_config_and_inputs()
   (
       (
       _snake_case
   )	,      (
       _snake_case
   )	,      (
       _snake_case
   )	,      
   )							:	List[Any]			=						config_and_inputs
   _snake_case							:	Any			=						{"""pixel_values""": pixel_values}
   return config, inputs_dict







@require_torch
class       lowercase(  __a ,    __a ,    unittest.TestCase				):







 '''simple docstring'''





 lowercase__				      =      (UperNetForSemanticSegmentation,) if is_torch_available() else ()
 lowercase__				      =      {"image-segmentation": UperNetForSemanticSegmentation} if is_torch_available() else {}
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False


 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   _snake_case							:	List[str]			=						UperNetModelTester(self	)
   _snake_case							:	Dict			=						ConfigTester(self,   config_class=a_,   has_text_modality=a_,   hidden_size=37	)


 def 				UpperCamelCase_     (       self:	Any	):




   '''simple docstring'''



   self.create_and_test_config_common_properties()
   self.config_tester.create_and_test_config_to_json_string()
   self.config_tester.create_and_test_config_to_json_file()
   self.config_tester.create_and_test_config_from_and_save_pretrained()
   self.config_tester.create_and_test_config_with_num_labels()
   self.config_tester.check_config_can_be_init_without_params()
   self.config_tester.check_config_arguments_init()


 def 				UpperCamelCase_     (       self:	List[Any]	):




   '''simple docstring'''



   return


 def 				UpperCamelCase_     (       self:	Union[str, Any]	):




   '''simple docstring'''



   _snake_case	,      _snake_case							:	Any			=						self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
     _snake_case							:	Dict			=						model_class(a_	)
     _snake_case							:	Optional[int]			=						inspect.signature(model.forward	)
     # signature.parameters is an OrderedDict => so arg_names order is deterministic
     _snake_case							:	Tuple			=						[*signature.parameters.keys()]

     _snake_case							:	Any			=						["""pixel_values"""]
     self.assertListEqual(arg_names[:1],   a_	)


 def 				UpperCamelCase_     (       self:	Any	):




   '''simple docstring'''



   _snake_case							:	Union[str, Any]			=						self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_for_semantic_segmentation(*a_	)


 @unittest.skip(reason="""UperNet does not use inputs_embeds"""	)
 def 				UpperCamelCase_     (       self:	Any	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""UperNet does not support input and output embeddings"""	)
 def 				UpperCamelCase_     (       self:	Dict	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""UperNet does not have a base model"""	)
 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""UperNet does not have a base model"""	)
 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   pass


 @require_torch_multi_gpu
 @unittest.skip(reason="""UperNet has some layers using `add_module` which doesn't work well with `nn.DataParallel`"""	)
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   pass


 @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""	)
 def 				UpperCamelCase_     (       self:	Dict	):




   '''simple docstring'''



   pass


 def 				UpperCamelCase_     (       self:	str	):




   '''simple docstring'''



   def check_hidden_states_output(a_:	Dict,   a_:	List[str],   a_:	Optional[int]	):
     _snake_case							:	Optional[Any]			=						model_class(a_	)
     model.to(a_	)
     model.eval()

     with torch.no_grad():
       _snake_case							:	Any			=						model(**self._prepare_for_class(a_,   a_	)	)

     _snake_case							:	Any			=						outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

     _snake_case							:	List[str]			=						self.model_tester.num_stages
     self.assertEqual(len(a_	),   expected_num_stages + 1	)

     # ConvNext's feature maps are of shape (batch_size, num_channels, height, width)
     self.assertListEqual(
         list(hidden_states[0].shape[-2:]	),   [self.model_tester.image_size // 4, self.model_tester.image_size // 4],   )

   _snake_case	,      _snake_case							:	str			=						self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
     _snake_case							:	int			=						True
     check_hidden_states_output(a_,   a_,   a_	)

     # check that output_hidden_states also work using config
     del inputs_dict["output_hidden_states"]
     _snake_case							:	Optional[int]			=						True

     check_hidden_states_output(a_,   a_,   a_	)


 def 				UpperCamelCase_     (       self:	str	):




   '''simple docstring'''



   _snake_case	,      _snake_case							:	int			=						self.model_tester.prepare_config_and_inputs_for_common()

   _snake_case							:	Tuple			=						_config_zero_init(a_	)
   _snake_case							:	Dict			=						_config_zero_init(configs_no_init.backbone_config	)
   for model_class in self.all_model_classes:
     _snake_case							:	Optional[int]			=						model_class(config=a_	)
     for name, param in model.named_parameters():
       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="""UperNet does not have tied weights"""	)
 def 				UpperCamelCase_     (       self:	Union[str, Any]	):




   '''simple docstring'''



   pass


 @slow
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   for model_name in UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
     _snake_case							:	int			=						UperNetForSemanticSegmentation.from_pretrained(a_	)
     self.assertIsNotNone(a_	)







def       UpperCAmelCase__ ():
  """simple docstring"""

  _snake_case							:	Union[str, Any]			=						hf_hub_download(
      repo_id="""hf-internal-testing/fixtures_ade20k"""	,     repo_type="""dataset"""	,     filename="""ADE_val_00000001.jpg"""					)
  _snake_case							:	List[Any]			=						Image.open(snake_case__					).convert("""RGB"""					)
  return image






@require_torch
@require_vision
@slow
class       lowercase(  unittest.TestCase				):







 '''simple docstring'''





 def 				UpperCamelCase_     (       self:	Tuple	):




   '''simple docstring'''



   _snake_case							:	str			=						AutoImageProcessor.from_pretrained("""openmmlab/upernet-swin-tiny"""	)
   _snake_case							:	Any			=						UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-swin-tiny"""	).to(a_	)

   _snake_case							:	Dict			=						prepare_img()
   _snake_case							:	str			=						processor(images=a_,   return_tensors="""pt"""	).to(a_	)

   with torch.no_grad():
     _snake_case							:	Tuple			=						model(**a_	)

   _snake_case							:	Tuple			=						torch.Size((1, model.config.num_labels, 512, 512)	)
   self.assertEqual(outputs.logits.shape,   a_	)

   _snake_case							:	int			=						torch.tensor(
       [[-7.5_958, -7.5_958, -7.4_302], [-7.5_958, -7.5_958, -7.4_302], [-7.4_797, -7.4_797, -7.3_068]]	).to(a_	)
   self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3],   a_,   atol=1E-4	)	)


 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   _snake_case							:	Optional[Any]			=						AutoImageProcessor.from_pretrained("""openmmlab/upernet-convnext-tiny"""	)
   _snake_case							:	Optional[int]			=						UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-convnext-tiny"""	).to(a_	)

   _snake_case							:	List[str]			=						prepare_img()
   _snake_case							:	Tuple			=						processor(images=a_,   return_tensors="""pt"""	).to(a_	)

   with torch.no_grad():
     _snake_case							:	Optional[Any]			=						model(**a_	)

   _snake_case							:	Union[str, Any]			=						torch.Size((1, model.config.num_labels, 512, 512)	)
   self.assertEqual(outputs.logits.shape,   a_	)

   _snake_case							:	Optional[Any]			=						torch.tensor(
       [[-8.8_110, -8.8_110, -8.6_521], [-8.8_110, -8.8_110, -8.6_521], [-8.7_746, -8.7_746, -8.6_130]]	).to(a_	)
   self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3],   a_,   atol=1E-4	)	)


