


"""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 lowerCAmelCase_   (					UpperCamelCase__						:			Optional[int]			):






 """simple docstring"""







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

 return wrapper


def lowerCAmelCase_   (					UpperCamelCase__						:			str			):






 """simple docstring"""







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

 return wrapper


def lowerCAmelCase_   (					UpperCamelCase__						:			int			):






 """simple docstring"""







 @wraps(UpperCamelCase__			)
 def wrapper(self						:			Optional[int]					, UpperCamelCase__						:			Optional[Any]			):
  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 lowerCAmelCase_   (					):






 """simple docstring"""







 __lowercase						=						[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  lowerCamelCase__    (		parameterized.TestCase     ):
      a			:  Union[str, Any]           =		{}
      a			:  Any           =		None
      @pytest.mark.filterwarnings("""ignore:metric_module_factory is deprecated:FutureWarning"""				)
      @pytest.mark.filterwarnings("""ignore:load_metric is deprecated:FutureWarning"""				)
      def SCREAMING_SNAKE_CASE_						(		self						:						Any   ,							A_						:						Optional[int]				):






       '''simple docstring'''


       __lowercase						=						"""[...]"""
       __lowercase						=						importlib.import_module(
           datasets.load.metric_module_factory(os.path.join("""metrics"""   ,							A_				)				).module_path				)
       __lowercase						=						datasets.load.import_main_class(metric_module.__name__   ,							dataset=A_				)
       # check parameters
       __lowercase						=						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:
          __lowercase						=						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 SCREAMING_SNAKE_CASE_						(		self						:						Dict   ,							A_						:						Optional[Any]				):






       '''simple docstring'''


       __lowercase						=						"""[...]"""
       __lowercase						=						importlib.import_module(
           datasets.load.metric_module_factory(os.path.join("""metrics"""   ,							A_				)				).module_path				)
       # run doctest
       with self.use_local_metrics():
        __lowercase						=						doctest.testmod(A_   ,							verbose=A_   ,							raise_on_error=A_				)
       self.assertEqual(results.failed   ,							0				)
       self.assertGreater(results.attempted   ,							1				)
      @contextmanager
      def SCREAMING_SNAKE_CASE_						(		self						:						Tuple   ,							A_						:						List[Any]   ,							A_						:						Optional[Any]				):






       '''simple docstring'''


       if metric_name in self.INTENSIVE_CALLS_PATCHER:
        with self.INTENSIVE_CALLS_PATCHER[metric_name](A_				):
         yield
       else:
        yield
      @contextmanager
      def SCREAMING_SNAKE_CASE_						(		self						:						List[str]				):






       '''simple docstring'''


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

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






      @classmethod
      def SCREAMING_SNAKE_CASE_						(		cls						:						List[Any]   ,							A_						:						Any				):






       '''simple docstring'''


       def wrapper(A_						:						Dict				):
        __lowercase						=						contextmanager(A_				)
        __lowercase						=						patcher
        return patcher

       return wrapper






@LocalMetricTest.register_intensive_calls_patcher("""bleurt"""			)
def lowerCAmelCase_   (					UpperCamelCase__						:			str			):






 """simple docstring"""







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

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





 class  lowerCamelCase__    (		_a     ):
       def SCREAMING_SNAKE_CASE_						(		self						:						Union[str, Any]   ,							A_						:						Optional[int]				):






        '''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:
  __lowercase						=						MockedPredictor()
  yield


@LocalMetricTest.register_intensive_calls_patcher("""bertscore"""			)
def lowerCAmelCase_   (					UpperCamelCase__						:			Optional[Any]			):






 """simple docstring"""







 import torch

 def bert_cos_score_idf(UpperCamelCase__						:			Tuple					, UpperCamelCase__						:			List[str]					, *UpperCamelCase__						:			List[Any]					, **UpperCamelCase__						:			str			):
  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:
  __lowercase						=						bert_cos_score_idf
  yield


@LocalMetricTest.register_intensive_calls_patcher("""comet"""			)
def lowerCAmelCase_   (					UpperCamelCase__						:			Any			):






 """simple docstring"""







 def load_from_checkpoint(UpperCamelCase__						:			Dict			):





  class  lowerCamelCase__    :
        def SCREAMING_SNAKE_CASE_						(		self						:						Union[str, Any]   ,							A_						:						List[Any]   ,							*A_						:						Any   ,							**A_						:						Optional[Any]				):






         '''simple docstring'''


         assert len(A_				) == 2
         __lowercase						=						[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:
  __lowercase						=						None
  with patch("""comet.load_from_checkpoint"""			) as mock_load_from_checkpoint:
   __lowercase						=						load_from_checkpoint
   yield


def lowerCAmelCase_   (					):






 """simple docstring"""







 __lowercase						=						load_metric(os.path.join("""metrics"""					, """seqeval"""			)			)
 __lowercase						=						"""ERROR"""
 __lowercase						=						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__			)




"""simple docstring"""

import time
from dataclasses import dataclass
from multiprocessing import Pool
from unittest import TestCase
from unittest.mock import patch

import multiprocess
import numpy as np
import pytest

from datasets.utils.py_utils import (
    NestedDataStructure,
    asdict,
    iflatmap_unordered,
    map_nested,
    temp_seed,
    temporary_assignment,
    zip_dict,
)

from .utils import require_tf, require_torch


def lowerCAmelCase_   (					UpperCamelCase__						:			int			):  # picklable for multiprocessing






 """simple docstring"""







 return x.sum()


def lowerCAmelCase_   (					UpperCamelCase__						:			List[str]			):  # picklable for multiprocessing






 """simple docstring"""







 return i + 1





@dataclass
class  lowerCamelCase__    :
      a			:  int
      a			:  str






class  lowerCamelCase__    (		_a     ):
      def SCREAMING_SNAKE_CASE_						(		self						:						List[str]				):






       '''simple docstring'''


       __lowercase						=						{}
       __lowercase						=						[]
       __lowercase						=						1
       __lowercase						=						[1, 2]
       __lowercase						=						{"""a""": 1, """b""": 2}
       __lowercase						=						{"""a""": [1, 2], """b""": [3, 4]}
       __lowercase						=						{"""a""": {"""1""": 1}, """b""": 2}
       __lowercase						=						{"""a""": 1, """b""": 2, """c""": 3, """d""": 4}
       __lowercase						=						{}
       __lowercase						=						[]
       __lowercase						=						2
       __lowercase						=						[2, 3]
       __lowercase						=						{"""a""": 2, """b""": 3}
       __lowercase						=						{"""a""": [2, 3], """b""": [4, 5]}
       __lowercase						=						{"""a""": {"""1""": 2}, """b""": 3}
       __lowercase						=						{"""a""": 2, """b""": 3, """c""": 4, """d""": 5}
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_				)   ,							A_				)

       __lowercase						=						2
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(map_nested(A_   ,							A_   ,							num_proc=A_				)   ,							A_				)

       __lowercase						=						{"""a""": np.eye(2				), """b""": np.zeros(3				), """c""": np.ones(2				)}
       __lowercase						=						{"""a""": 2, """b""": 0, """c""": 2}
       __lowercase						=						{
           """a""": np.eye(2				).astype(A_				),
           """b""": np.zeros(3				).astype(A_				),
           """c""": np.ones(2				).astype(A_				),
       }
       self.assertEqual(map_nested(A_   ,							A_   ,							map_numpy=A_				)   ,							A_				)
       self.assertEqual(
           {k: v.tolist() for k, v in map_nested(A_   ,							A_   ,							map_numpy=A_				).items()}   ,							{k: v.tolist() for k, v in expected_map_nested_sna_int.items()}   ,							)
       self.assertEqual(map_nested(A_   ,							A_   ,							map_numpy=A_   ,							num_proc=A_				)   ,							A_				)
       self.assertEqual(
           {k: v.tolist() for k, v in map_nested(A_   ,							A_   ,							map_numpy=A_   ,							num_proc=A_				).items()}   ,							{k: v.tolist() for k, v in expected_map_nested_sna_int.items()}   ,							)
       with self.assertRaises(A_				):  # can't pickle a local lambda
        map_nested(lambda A_				: x + 1   ,							A_   ,							num_proc=A_				)
      def SCREAMING_SNAKE_CASE_						(		self						:						Dict				):






       '''simple docstring'''


       __lowercase						=						{"""a""": 1, """b""": 2}
       __lowercase						=						{"""a""": 3, """b""": 4}
       __lowercase						=						{"""a""": 5, """b""": 6}
       __lowercase						=						sorted([("""a""", (1, 3, 5)), ("""b""", (2, 4, 6))]				)
       self.assertEqual(sorted(zip_dict(A_   ,							A_   ,							A_				)				)   ,							A_				)






      def SCREAMING_SNAKE_CASE_						(		self						:						int				):






       '''simple docstring'''


       class  lowerCamelCase__    :
             a			:  int           =		"""bar"""






       __lowercase						=						Foo()
       self.assertEqual(foo.my_attr   ,							"""bar"""				)
       with temporary_assignment(A_   ,							"""my_attr"""   ,							"""BAR"""				):
        self.assertEqual(foo.my_attr   ,							"""BAR"""				)
       self.assertEqual(foo.my_attr   ,							"""bar"""				)






@pytest.mark.parametrize(
    """iterable_length, num_proc, expected_num_proc"""					, [
        (1, None, 1),
        (1, 1, 1),
        (2, None, 1),
        (2, 1, 1),
        (2, 2, 1),
        (2, 3, 1),
        (3, 2, 1),
        (16, 16, 16),
        (16, 17, 16),
        (17, 16, 16),
    ]					, )
def lowerCAmelCase_   (					UpperCamelCase__						:			Tuple					, UpperCamelCase__						:			List[Any]					, UpperCamelCase__						:			int			):






 """simple docstring"""







 with patch("""datasets.utils.py_utils._single_map_nested"""			) as mock_single_map_nested, patch(
     """datasets.parallel.parallel.Pool"""			) as mock_multiprocessing_pool:
  __lowercase						=						{f'''{i}''': i for i in range(UpperCamelCase__			)}
  __lowercase						=						map_nested(lambda UpperCamelCase__			: x + 10					, UpperCamelCase__					, num_proc=UpperCamelCase__					, parallel_min_length=16			)
  if expected_num_proc == 1:
   assert mock_single_map_nested.called
   assert not mock_multiprocessing_pool.called
  else:
   assert not mock_single_map_nested.called
   assert mock_multiprocessing_pool.called
   assert mock_multiprocessing_pool.call_args[0][0] == expected_num_proc





class  lowerCamelCase__    (		_a     ):
      @require_tf
      def SCREAMING_SNAKE_CASE_						(		self						:						List[str]				):






       '''simple docstring'''


       import tensorflow as tf
       from tensorflow.keras import layers

       __lowercase						=						layers.Dense(2				)

       def gen_random_output():
        __lowercase						=						tf.random.uniform((1, 3)				)
        return model(A_				).numpy()

       with temp_seed(4_2   ,							set_tensorflow=A_				):
        __lowercase						=						gen_random_output()
       with temp_seed(4_2   ,							set_tensorflow=A_				):
        __lowercase						=						gen_random_output()
       __lowercase						=						gen_random_output()

       np.testing.assert_equal(A_   ,							A_				)
       self.assertGreater(np.abs(outa - outa				).sum()   ,							0				)
      @require_torch
      def SCREAMING_SNAKE_CASE_						(		self						:						Union[str, Any]				):






       '''simple docstring'''


       import torch

       def gen_random_output():
        __lowercase						=						torch.nn.Linear(3   ,							2				)
        __lowercase						=						torch.rand(1   ,							3				)
        return model(A_				).detach().numpy()

       with temp_seed(4_2   ,							set_pytorch=A_				):
        __lowercase						=						gen_random_output()
       with temp_seed(4_2   ,							set_pytorch=A_				):
        __lowercase						=						gen_random_output()
       __lowercase						=						gen_random_output()

       np.testing.assert_equal(A_   ,							A_				)
       self.assertGreater(np.abs(outa - outa				).sum()   ,							0				)






      def SCREAMING_SNAKE_CASE_						(		self						:						Any				):






       '''simple docstring'''


       def gen_random_output():
        return np.random.rand(1   ,							3				)

       with temp_seed(4_2				):
        __lowercase						=						gen_random_output()
       with temp_seed(4_2				):
        __lowercase						=						gen_random_output()
       __lowercase						=						gen_random_output()

       np.testing.assert_equal(A_   ,							A_				)
       self.assertGreater(np.abs(outa - outa				).sum()   ,							0				)






@pytest.mark.parametrize("""input_data"""					, [{}]			)
def lowerCAmelCase_   (					UpperCamelCase__						:			Optional[int]			):






 """simple docstring"""







 __lowercase						=						NestedDataStructure(UpperCamelCase__			).data
 assert output_data == input_data


@pytest.mark.parametrize(
    """data, expected_output"""					, [
        ({}, []),
        ([], []),
        ("""foo""", ["""foo"""]),
        (["""foo""", """bar"""], ["""foo""", """bar"""]),
        ([["""foo""", """bar"""]], ["""foo""", """bar"""]),
        ([[["""foo"""], ["""bar"""]]], ["""foo""", """bar"""]),
        ([[["""foo"""], """bar"""]], ["""foo""", """bar"""]),
        ({"""a""": 1, """b""": 2}, [1, 2]),
        ({"""a""": [1, 2], """b""": [3, 4]}, [1, 2, 3, 4]),
        ({"""a""": [[1, 2]], """b""": [[3, 4]]}, [1, 2, 3, 4]),
        ({"""a""": [[1, 2]], """b""": [3, 4]}, [1, 2, 3, 4]),
        ({"""a""": [[[1], [2]]], """b""": [[[3], [4]]]}, [1, 2, 3, 4]),
        ({"""a""": [[[1], [2]]], """b""": [[3, 4]]}, [1, 2, 3, 4]),
        ({"""a""": [[[1], [2]]], """b""": [3, 4]}, [1, 2, 3, 4]),
        ({"""a""": [[[1], [2]]], """b""": [3, [4]]}, [1, 2, 3, 4]),
        ({"""a""": {"""1""": 1}, """b""": 2}, [1, 2]),
        ({"""a""": {"""1""": [1]}, """b""": 2}, [1, 2]),
        ({"""a""": {"""1""": [1]}, """b""": [2]}, [1, 2]),
    ]					, )
def lowerCAmelCase_   (					UpperCamelCase__						:			str					, UpperCamelCase__						:			Union[str, Any]			):






 """simple docstring"""







 __lowercase						=						NestedDataStructure(UpperCamelCase__			).flatten()
 assert output == expected_output


def lowerCAmelCase_   (					):






 """simple docstring"""







 __lowercase						=						A(x=1					, y="""foobar"""			)
 __lowercase						=						{"""x""": 1, """y""": """foobar"""}
 assert asdict(UpperCamelCase__			) == expected_output

 __lowercase						=						{"""a""": {"""b""": A(x=10					, y="""foo"""			)}, """c""": [A(x=20					, y="""bar"""			)]}
 __lowercase						=						{"""a""": {"""b""": {"""x""": 10, """y""": """foo"""}}, """c""": [{"""x""": 20, """y""": """bar"""}]}
 assert asdict(UpperCamelCase__			) == expected_output

 with pytest.raises(UpperCamelCase__			):
  asdict([1, A(x=10					, y="""foo"""			)]			)


def lowerCAmelCase_   (					UpperCamelCase__						:			str			):






 """simple docstring"""







 return text.split()


def lowerCAmelCase_   (					UpperCamelCase__						:			Dict			):






 """simple docstring"""







 yield (time.time(), content)
 time.sleep(2			)
 yield (time.time(), content)


def lowerCAmelCase_   (					):






 """simple docstring"""







 with Pool(2			) as pool:
  __lowercase						=						list(iflatmap_unordered(UpperCamelCase__					, _split_text					, kwargs_iterable=[{"""text""": """hello there"""}] * 10			)			)
  assert out.count("""hello"""			) == 10
  assert out.count("""there"""			) == 10
  assert len(UpperCamelCase__			) == 20

 # check multiprocess from pathos (uses dill for pickling)
 with multiprocess.Pool(2			) as pool:
  __lowercase						=						list(iflatmap_unordered(UpperCamelCase__					, _split_text					, kwargs_iterable=[{"""text""": """hello there"""}] * 10			)			)
  assert out.count("""hello"""			) == 10
  assert out.count("""there"""			) == 10
  assert len(UpperCamelCase__			) == 20

 # check that we get items as fast as possible
 with Pool(2			) as pool:
  __lowercase						=						[]
  for yield_time, content in iflatmap_unordered(
      UpperCamelCase__					, _aseconds_generator_of_aitems_with_timing					, kwargs_iterable=[{"""content""": """a"""}, {"""content""": """b"""}]			):
   assert yield_time < time.time() + 0.1, "we should each item directly after it was yielded"
   out.append(UpperCamelCase__			)
  assert out.count("""a"""			) == 2
  assert out.count("""b"""			) == 2
  assert len(UpperCamelCase__			) == 4
