




'''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
UpperCamelCase_    	=				pytest.mark.integration


UpperCamelCase_    	=				{"comet"}
UpperCamelCase_    	=				importlib.util.find_spec("fairseq") is not None

UpperCamelCase_    	=				{"code_eval"}
UpperCamelCase_    	=				os.name == "nt"

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




def   lowercase__(			__UpperCamelCase:					Tuple    ):

      """simple docstring"""






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

      return wrapper




def   lowercase__(			__UpperCamelCase:					str    ):

      """simple docstring"""






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

      return wrapper




def   lowercase__(			__UpperCamelCase:					Optional[int]    ):

      """simple docstring"""






      @wraps(__UpperCamelCase    )
      def wrapper(self:					Optional[int]      ,__UpperCamelCase:					Tuple    ):
            if _on_windows and metric_name in UNSUPPORTED_ON_WINDOWS:
                  self.skipTest('"test not supported on Windows"'    )
            else:
                  test_case(self      ,__UpperCamelCase    )

      return wrapper




def   lowercase__(			):

      """simple docstring"""






      SCREAMING_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(
    SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE ,      SCREAMING_SNAKE_CASE )
@local
class 							_a				(			parameterized.TestCase ):





       '''simple docstring'''






       A						:				Any        =						{}
       A						:				List[Any]        =						None







       @pytest.mark.filterwarnings('ignore:metric_module_factory is deprecated:FutureWarning'  )
       @pytest.mark.filterwarnings('ignore:load_metric is deprecated:FutureWarning'  )
       def        UpperCamelCase_							(      self,							A  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      Any    					=						'[...]'
             SCREAMING_SNAKE_CASE   :      List[Any]    					=						importlib.import_module(
                 datasets.load.metric_module_factory(os.path.join('metrics',							A  )  ).module_path  )
             SCREAMING_SNAKE_CASE   :      int    					=						datasets.load.import_main_class(metric_module.__name__,							dataset=A  )
             # check parameters
             SCREAMING_SNAKE_CASE   :      int    					=						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   :      Any    					=						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,							A  ):







             '''simple docstring'''







             SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						'[...]'
             SCREAMING_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():
                   SCREAMING_SNAKE_CASE   :      Tuple    					=						doctest.testmod(A,							verbose=A,							raise_on_error=A  )
             self.assertEqual(results.failed,							0  )
             self.assertGreater(results.attempted,							1  )







       @contextmanager
       def        UpperCamelCase_							(      self,							A,							A  ):







             '''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  ):







             '''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   :      Tuple    					=						load_local_metric
                   yield







       @classmethod
       def        UpperCamelCase_							(      cls,							A  ):







             '''simple docstring'''







             def wrapper(A  ):
                   SCREAMING_SNAKE_CASE   :      str    					=						contextmanager(A  )
                   SCREAMING_SNAKE_CASE   :      int    					=						patcher
                   return patcher

             return wrapper




@LocalMetricTest.register_intensive_calls_patcher('bleurt'    )
def   lowercase__(			__UpperCamelCase:					Union[str, Any]    ):

      """simple docstring"""






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

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






      class 							_a				(			SCREAMING_SNAKE_CASE ):





             '''simple docstring'''






             def        UpperCamelCase_							(      self,							A  ):







                   '''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__(			__UpperCamelCase:					Union[str, Any]    ):

      """simple docstring"""






      import torch

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

      # 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   :      str    					=						bert_cos_score_idf
            yield




@LocalMetricTest.register_intensive_calls_patcher('comet'    )
def   lowercase__(			__UpperCamelCase:					Dict    ):

      """simple docstring"""






      def load_from_checkpoint(__UpperCamelCase:					Any    ):






            class 							_a				:





                   '''simple docstring'''






                   def        UpperCamelCase_							(      self,							A,							*A,							**A  ):







                         '''simple docstring'''







                         assert len(A  ) == 2
                         SCREAMING_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:
            SCREAMING_SNAKE_CASE   :      Any    					=						None
            with patch('comet.load_from_checkpoint'    ) as mock_load_from_checkpoint:
                  SCREAMING_SNAKE_CASE   :      Dict    					=						load_from_checkpoint
                  yield





def   lowercase__(			):

      """simple docstring"""






      SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						load_metric(os.path.join('metrics'      ,'seqeval'    )    )
      SCREAMING_SNAKE_CASE   :      List[Any]    					=						'ERROR'
      SCREAMING_SNAKE_CASE   :      Union[str, Any]    					=						f"Scheme should be one of [IOB1, IOB2, IOE1, IOE2, IOBES, BILOU], got {wrong_scheme}"
      with pytest.raises(__UpperCamelCase      ,match=re.escape(__UpperCamelCase    )    ):
            metric.compute(predictions=[]      ,references=[]      ,scheme=__UpperCamelCase    )





import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402

class 							UpperCAmelCase_						(     unittest.TestCase  ):
							"""simple docstring"""







							def 							__lowercase							(     self			)       ->       List[Any]:
												_a      :       int   =						0


							def 							__lowercase							(     self			)       ->       List[str]:
												_a      :       Dict   =						AutoImageProcessor.from_pretrained('''openai/clip-vit-base-patch32'''			)
												self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Tuple   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       Optional[Any]   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Optional[Any]:
												# Ensure we can load the image processor from the feature extractor config
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Optional[int]   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       Any   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Optional[Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Dict   =						CLIPConfig()

																	# Create a dummy config file with image_proceesor_type
																	_a      :       Tuple   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       List[str]   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	# remove image_processor_type to make sure config.json alone is enough to load image processor locally
																	_a      :       Tuple   =						AutoImageProcessor.from_pretrained(_a			).to_dict()

																	config_dict.pop('''image_processor_type'''			)
																	_a      :       Tuple   =						CLIPImageProcessor(**_a			)

																	# save in new folder
																	model_config.save_pretrained(_a			)
																	config.save_pretrained(_a			)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)

																	# make sure private variable is not incorrectly saved
																	_a      :       Optional[int]   =						json.loads(config.to_json_string()			)
																	self.assertTrue('''_processor_class''' not in dict_as_saved			)

												self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Dict:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Optional[int]   =						Path(_a			) / '''preprocessor_config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Any:
												with self.assertRaisesRegex(
												    _a			,				'''clip-base is not a local folder and is not a valid model identifier'''			):
																	_a      :       Dict   =						AutoImageProcessor.from_pretrained('''clip-base'''			)


							def 							__lowercase							(     self			)       ->       List[Any]:
												with self.assertRaisesRegex(
												    _a			,				R'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''			):
																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			,				revision='''aaaaaa'''			)


							def 							__lowercase							(     self			)       ->       Dict:
												with self.assertRaisesRegex(
												    _a			,				'''hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'''			,				):
																	_a      :       Optional[int]   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/config-no-model'''			)


							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												# If remote code is not set, we will time out when asking whether to load the model.
												with self.assertRaises(_a			):
																	_a      :       str   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''			)
												# If remote code is disabled, we can't load this config.
												with self.assertRaises(_a			):
																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)

												_a      :       Union[str, Any]   =						AutoImageProcessor.from_pretrained(
												    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
												self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)

												# Test image processor can be reloaded.
												with tempfile.TemporaryDirectory() as tmp_dir:
																	image_processor.save_pretrained(_a			)
																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			,				trust_remote_code=_a			)
												self.assertEqual(reloaded_image_processor.__class__.__name__			,				'''NewImageProcessor'''			)


							def 							__lowercase							(     self			)       ->       Dict:
												try:
																	AutoConfig.register('''custom'''			,				_a			)
																	AutoImageProcessor.register(_a			,				_a			)
																	# Trying to register something existing in the Transformers library will raise an error
																	with self.assertRaises(_a			):
																						AutoImageProcessor.register(_a			,				_a			)

																	with tempfile.TemporaryDirectory() as tmpdirname:
																						_a      :       int   =						Path(_a			) / '''preprocessor_config.json'''
																						_a      :       int   =						Path(_a			) / '''config.json'''
																						json.dump(
																						    {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																						json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																						_a      :       int   =						CustomImageProcessor.from_pretrained(_a			)

																	# Now that the config is registered, it can be used as any other config with the auto-API
																	with tempfile.TemporaryDirectory() as tmp_dir:
																						image_processor.save_pretrained(_a			)
																						_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			)
																						self.assertIsInstance(_a			,				_a			)

												finally:
																	if "custom" in CONFIG_MAPPING._extra_content:
																						del CONFIG_MAPPING._extra_content["custom"]
																	if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
																						del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]




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

												class 							UpperCAmelCase_						(     __lowercase  ):
																			"""simple docstring"""







																			UpperCAmelCase__			:		Tuple             =       True

												try:
																	AutoConfig.register('''custom'''			,				_a			)
																	AutoImageProcessor.register(_a			,				_a			)
																	# If remote code is not set, the default is to use local
																	_a      :       str   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(image_processor.is_local			)

																	# If remote code is disabled, we load the local one.
																	_a      :       int   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(image_processor.is_local			)

																	# If remote is enabled, we load from the Hub
																	_a      :       Dict   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(not hasattr(_a			,				'''is_local'''			)			)

												finally:
																	if "custom" in CONFIG_MAPPING._extra_content:
																						del CONFIG_MAPPING._extra_content["custom"]
																	if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
																						del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]



