
import asyncio
import os
import re
import sys
import tempfile
import unittest
from contextlib import contextmanager
from copy import deepcopy
from distutils.util import strtobool
from enum import Enum
from importlib.util import find_spec
from pathlib import Path
from unittest.mock import patch

import pyarrow as pa
import pytest
import requests
from packaging import version

from datasets import config


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





def 			lowerCamelCase_ (				lowerCamelCase__	, lowerCamelCase__=False    ):
				try:
								lowerCamelCase_						=			os.environ[key]
				except KeyError:
								# KEY isn't set, default to `default`.
								lowerCamelCase_						=			default
				else:
								# KEY is set, convert it to True or False.
								try:
												lowerCamelCase_						=			strtobool(__lowerCamelCase    )
								except ValueError:
												# More values are supported, but let's keep the message simple.
												raise ValueError(F'If set, {key} must be yes or no.'    )
				return _value


__A				=parse_flag_from_env('''RUN_SLOW''', default=False)
__A				=parse_flag_from_env('''RUN_REMOTE''', default=False)
__A				=parse_flag_from_env('''RUN_LOCAL''', default=True)
__A				=parse_flag_from_env('''RUN_PACKAGED''', default=True)

# Compression
__A				=pytest.mark.skipif(not config.LZ4_AVAILABLE, reason='''test requires lz4''')
__A				=pytest.mark.skipif(not config.PY7ZR_AVAILABLE, reason='''test requires py7zr''')
__A				=pytest.mark.skipif(not config.ZSTANDARD_AVAILABLE, reason='''test requires zstandard''')

# Audio
__A				=pytest.mark.skipif(
    # On Windows and OS X, soundfile installs sndfile
    find_spec('''soundfile''') is None or version.parse(importlib_metadata.version('''soundfile''')) < version.parse('''0.12.0'''),
    reason='''test requires sndfile>=0.12.1: \'pip install \"soundfile>=0.12.1\"\'; ''',
)

# Beam
__A				=pytest.mark.skipif(
    not config.BEAM_AVAILABLE or config.DILL_VERSION >= version.parse('''0.3.2'''),
    reason='''test requires apache-beam and a compatible dill version''',
)

# Dill-cloudpickle compatibility
__A				=pytest.mark.skipif(
    config.DILL_VERSION <= version.parse('''0.3.2'''),
    reason='''test requires dill>0.3.2 for cloudpickle compatibility''',
)

# Windows
__A				=pytest.mark.skipif(
    sys.platform == '''win32''',
    reason='''test should not be run on Windows''',
)





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import faiss  # noqa
				except ImportError:
								lowerCamelCase_						=			unittest.skip("test requires faiss"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import regex  # noqa
				except ImportError:
								lowerCamelCase_						=			unittest.skip("test requires regex"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import elasticsearch  # noqa
				except ImportError:
								lowerCamelCase_						=			unittest.skip("test requires elasticsearch"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import sqlalchemy  # noqa
				except ImportError:
								lowerCamelCase_						=			unittest.skip("test requires sqlalchemy"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not config.TORCH_AVAILABLE:
								lowerCamelCase_						=			unittest.skip("test requires PyTorch"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not config.TF_AVAILABLE:
								lowerCamelCase_						=			unittest.skip("test requires TensorFlow"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not config.JAX_AVAILABLE:
								lowerCamelCase_						=			unittest.skip("test requires JAX"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not config.PIL_AVAILABLE:
								lowerCamelCase_						=			unittest.skip("test requires Pillow"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import transformers  # noqa F401
				except ImportError:
								return unittest.skip("test requires transformers"    )(__lowerCamelCase    )
				else:
								return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import tiktoken  # noqa F401
				except ImportError:
								return unittest.skip("test requires tiktoken"    )(__lowerCamelCase    )
				else:
								return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import spacy  # noqa F401
				except ImportError:
								return unittest.skip("test requires spacy"    )(__lowerCamelCase    )
				else:
								return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):

				def _require_spacy_model(lowerCamelCase__    ):
								try:
												import spacy  # noqa F401

												spacy.load(__lowerCamelCase    )
								except ImportError:
												return unittest.skip("test requires spacy"    )(__lowerCamelCase    )
								except OSError:
												return unittest.skip("test requires spacy model '{}'".format(__lowerCamelCase    )    )(__lowerCamelCase    )
								else:
												return test_case

				return _require_spacy_model





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import pyspark  # noqa F401
				except ImportError:
								return unittest.skip("test requires pyspark"    )(__lowerCamelCase    )
				else:
								return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				try:
								import joblibspark  # noqa F401
				except ImportError:
								return unittest.skip("test requires joblibspark"    )(__lowerCamelCase    )
				else:
								return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not _run_slow_tests or _run_slow_tests == 0:
								lowerCamelCase_						=			unittest.skip("test is slow"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not _run_local_tests or _run_local_tests == 0:
								lowerCamelCase_						=			unittest.skip("test is local"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not _run_packaged_tests or _run_packaged_tests == 0:
								lowerCamelCase_						=			unittest.skip("test is packaged"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				if not _run_remote_tests or _run_remote_tests == 0:
								lowerCamelCase_						=			unittest.skip("test requires remote"    )(__lowerCamelCase    )
				return test_case





def 			lowerCamelCase_ (				*lowerCamelCase__    ):
				def decorate(cls    ):
								for name, fn in cls.__dict__.items():
												if callable(__lowerCamelCase    ) and name.startswith("test"    ):
																for decorator in decorators:
																				lowerCamelCase_						=			decorator(__lowerCamelCase    )
																setattr(cls	, __lowerCamelCase	, __lowerCamelCase    )
								return cls

				return decorate




class     _SCREAMING_SNAKE_CASE     (      A_						):
	pass




class     _SCREAMING_SNAKE_CASE     (      A_						):
	lowerCAmelCase__         = 0
	lowerCAmelCase__         = 1
	lowerCAmelCase__         = 2





@contextmanager
def 			lowerCamelCase_ (				lowerCamelCase__=OfflineSimulationMode.CONNECTION_FAILS	, lowerCamelCase__=1e-16    ):
				lowerCamelCase_						=			requests.Session().request

				def timeout_request(lowerCamelCase__	, lowerCamelCase__	, lowerCamelCase__	, **lowerCamelCase__    ):
								# Change the url to an invalid url so that the connection hangs
								lowerCamelCase_						=			"https://10.255.255.1"
								if kwargs.get("timeout"    ) is None:
												raise RequestWouldHangIndefinitelyError(
												    F'Tried a call to {url} in offline mode with no timeout set. Please set a timeout.'    )
								lowerCamelCase_						=			timeout
								try:
												return online_request(__lowerCamelCase	, __lowerCamelCase	, **__lowerCamelCase    )
								except Exception as e:
												# The following changes in the error are just here to make the offline timeout error prettier
												lowerCamelCase_						=			url
												lowerCamelCase_						=			e.args[0]
												lowerCamelCase_						=			(max_retry_error.args[0].replace("10.255.255.1"	, F'OfflineMock[{url}]'    ),)
												lowerCamelCase_						=			(max_retry_error,)
												raise

				def raise_connection_error(lowerCamelCase__	, lowerCamelCase__	, **lowerCamelCase__    ):
								raise requests.ConnectionError("Offline mode is enabled."	, request=__lowerCamelCase    )

				if mode is OfflineSimulationMode.CONNECTION_FAILS:
								with patch("requests.Session.send"	, __lowerCamelCase    ):
												yield
				elif mode is OfflineSimulationMode.CONNECTION_TIMES_OUT:
								# inspired from https://stackoverflow.com/a/904609
								with patch("requests.Session.request"	, __lowerCamelCase    ):
												yield
				elif mode is OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1:
								with patch("datasets.config.HF_DATASETS_OFFLINE"	, __lowerCamelCase    ):
												yield
				else:
								raise ValueError("Please use a value from the OfflineSimulationMode enum."    )





@contextmanager
def 			lowerCamelCase_ (				*lowerCamelCase__	, **lowerCamelCase__    ):
				lowerCamelCase_						=			str(Path().resolve()    )
				with tempfile.TemporaryDirectory(*__lowerCamelCase	, **__lowerCamelCase    ) as tmp_dir:
								try:
												os.chdir(__lowerCamelCase    )
												yield
								finally:
												os.chdir(__lowerCamelCase    )





@contextmanager
def 			lowerCamelCase_ (				):
				import gc

				gc.collect()
				lowerCamelCase_						=			pa.total_allocated_bytes()
				yield
				assert pa.total_allocated_bytes() - previous_allocated_memory > 0, "Arrow memory didn't increase."





@contextmanager
def 			lowerCamelCase_ (				):
				import gc

				gc.collect()
				lowerCamelCase_						=			pa.total_allocated_bytes()
				yield
				assert pa.total_allocated_bytes() - previous_allocated_memory <= 0, "Arrow memory wasn't expected to increase."





def 			lowerCamelCase_ (				lowerCamelCase__	, lowerCamelCase__    ):
				return deepcopy(__lowerCamelCase    ).integers(0	, 1_0_0	, 1_0    ).tolist() == deepcopy(__lowerCamelCase    ).integers(0	, 1_0_0	, 1_0    ).tolist()





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				import decorator
				from requests.exceptions import HTTPError

				def _wrapper(lowerCamelCase__	, *lowerCamelCase__	, **lowerCamelCase__    ):
								try:
												return func(*__lowerCamelCase	, **__lowerCamelCase    )
								except HTTPError as err:
												if str(__lowerCamelCase    ).startswith("500"    ) or str(__lowerCamelCase    ).startswith("502"    ):
																pytest.xfail(str(__lowerCamelCase    )    )
												raise err

				return decorator.decorator(_wrapper	, __lowerCamelCase    )




class     _SCREAMING_SNAKE_CASE     :



	def __init__(    self						,  lowercase						,  lowercase						,  lowercase )				->       Optional[int]:
					lowerCamelCase_						=			returncode
					lowerCamelCase_						=			stdout
					lowerCamelCase_						=			stderr





async def 			lowerCamelCase_ (				lowerCamelCase__	, lowerCamelCase__    ):
				while True:
								lowerCamelCase_						=			await stream.readline()
								if line:
												callback(__lowerCamelCase    )
								else:
												break





async def 			lowerCamelCase_ (				lowerCamelCase__	, lowerCamelCase__=None	, lowerCamelCase__=None	, lowerCamelCase__=None	, lowerCamelCase__=False	, lowerCamelCase__=False    ):
				if echo:
								print("\nRunning: "	, " ".join(__lowerCamelCase    )    )

				lowerCamelCase_						=			await asyncio.create_subprocess_exec(
				    cmd[0]	, *cmd[1:]	, stdin=__lowerCamelCase	, stdout=asyncio.subprocess.PIPE	, stderr=asyncio.subprocess.PIPE	, env=__lowerCamelCase	, )

				# note: there is a warning for a possible deadlock when using `wait` with huge amounts of data in the pipe
				# https://docs.python.org/3/library/asyncio-subprocess.html#asyncio.asyncio.subprocess.Process.wait
				#
				# If it starts hanging, will need to switch to the following code. The problem is that no data
				# will be seen until it's done and if it hangs for example there will be no debug info.
				# out, err = await p.communicate()
				# return _RunOutput(p.returncode, out, err)

				lowerCamelCase_						=			[]
				lowerCamelCase_						=			[]

				def tee(lowerCamelCase__	, lowerCamelCase__	, lowerCamelCase__	, lowerCamelCase__=""    ):
								lowerCamelCase_						=			line.decode("utf-8"    ).rstrip()
								sink.append(__lowerCamelCase    )
								if not quiet:
												print(__lowerCamelCase	, __lowerCamelCase	, file=__lowerCamelCase    )

    # XXX: the timeout doesn't seem to make any difference here
				await asyncio.wait(
				    [
				        _read_stream(p.stdout	, lambda lowerCamelCase__    : tee(__lowerCamelCase	, __lowerCamelCase	, sys.stdout	, label="stdout:"    )    ),
				        _read_stream(p.stderr	, lambda lowerCamelCase__    : tee(__lowerCamelCase	, __lowerCamelCase	, sys.stderr	, label="stderr:"    )    ),
				    ]	, timeout=__lowerCamelCase	, )
				return _RunOutput(await p.wait()	, __lowerCamelCase	, __lowerCamelCase    )





def 			lowerCamelCase_ (				lowerCamelCase__	, lowerCamelCase__=None	, lowerCamelCase__=None	, lowerCamelCase__=1_8_0	, lowerCamelCase__=False	, lowerCamelCase__=True    ):
				lowerCamelCase_						=			asyncio.get_event_loop()
				lowerCamelCase_						=			loop.run_until_complete(
				    _stream_subprocess(__lowerCamelCase	, env=__lowerCamelCase	, stdin=__lowerCamelCase	, timeout=__lowerCamelCase	, quiet=__lowerCamelCase	, echo=__lowerCamelCase    )    )

				lowerCamelCase_						=			" ".join(__lowerCamelCase    )
				if result.returncode > 0:
								lowerCamelCase_						=			"\n".join(result.stderr    )
								raise RuntimeError(
								    F'\'{cmd_str}\' failed with returncode {result.returncode}\n\n'
								    F'The combined stderr from workers follows:\n{stderr}'    )

				# check that the subprocess actually did run and produced some output, should the test rely on
				# the remote side to do the testing
				if not result.stdout and not result.stderr:
								raise RuntimeError(F'\'{cmd_str}\' produced no output.'    )

				return result





def 			lowerCamelCase_ (				):
				lowerCamelCase_						=			os.environ.get("PYTEST_XDIST_WORKER"	, "gw0"    )
				lowerCamelCase_						=			re.sub(r"^gw"	, ""	, __lowerCamelCase	, 0	, re.M    )
				return int(__lowerCamelCase    )





def 			lowerCamelCase_ (				):
				lowerCamelCase_						=			2_9_5_0_0
				lowerCamelCase_						=			pytest_xdist_worker_id()
				return port + uniq_delta




import warnings
from typing import Dict

import numpy as np

from ..utils import ExplicitEnum, add_end_docstrings, is_tf_available, is_torch_available
from .base import PIPELINE_INIT_ARGS, GenericTensor, Pipeline


if is_tf_available():
   from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

if is_torch_available():
   from ..models.auto.modeling_auto import MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING


def 			UpperCamelCase					(    __lowerCamelCase     : List[Any]    ):
       return 1.0 / (1.0 + np.exp(-_outputs    ))


def 			UpperCamelCase					(    __lowerCamelCase     : int    ):
       snake_case       :   Tuple							=      np.max(_outputs				,					axis=-1				,					keepdims=__lowerCamelCase    )
       snake_case       :   int							=      np.exp(_outputs - maxes    )
       return shifted_exp / shifted_exp.sum(axis=-1				,					keepdims=__lowerCamelCase    )






class   UpperCAmelCase		(						A_ ):
   A__					:		Any						     =     "sigmoid"
   A__					:		str						     =     "softmax"
   A__					:		int						     =     "none"






@add_end_docstrings(
    A_       ,r"\n        return_all_scores (`bool`, *optional*, defaults to `False`):\n            Whether to return all prediction scores or just the one of the predicted class.\n        function_to_apply (`str`, *optional*, defaults to `\"default\"`):\n            The function to apply to the model outputs in order to retrieve the scores. Accepts four different values:\n\n            - `\"default\"`: if the model has a single label, will apply the sigmoid function on the output. If the model\n              has several labels, will apply the softmax function on the output.\n            - `\"sigmoid\"`: Applies the sigmoid function on the output.\n            - `\"softmax\"`: Applies the softmax function on the output.\n            - `\"none\"`: Does not apply any function on the output.\n    "       ,)
class   UpperCAmelCase		(						A_ ):
   A__					:		int						     =     False
   A__					:		Union[str, Any]						     =     ClassificationFunction.NONE



   def __init__(self		:					List[str]    ,						**snake_case__		:					int )		->	str:


          '''simple docstring'''


          super().__init__(**snake_case__ )

          self.check_model_type(
              TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
              if self.framework == "tf"
              else MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING )



   def 				_SCREAMING_SNAKE_CASE   (self		:					List[Any]    ,						snake_case__		:					List[str]=None    ,						snake_case__		:					Optional[Any]=None    ,						snake_case__		:					Union[str, Any]=""    ,						**snake_case__		:					List[str] )		->	Union[str, Any]:


          '''simple docstring'''


          snake_case       :   Dict							=      tokenizer_kwargs

          snake_case       :   List[Any]							=      {}
          if hasattr(self.model.config    ,						"return_all_scores" ) and return_all_scores is None:
                 snake_case       :   Optional[int]							=      self.model.config.return_all_scores

          if isinstance(snake_case__    ,						snake_case__ ) or top_k is None:
                 snake_case       :   List[Any]							=      top_k
                 snake_case       :   str							=      False
          elif return_all_scores is not None:
                 warnings.warn(
                     "`return_all_scores` is now deprecated,  if want a similar functionality use `top_k=None` instead of"
                     " `return_all_scores=True` or `top_k=1` instead of `return_all_scores=False`."    ,						snake_case__    ,						)
                 if return_all_scores:
                        snake_case       :   List[str]							=      None
                 else:
                        snake_case       :   Optional[int]							=      1

          if isinstance(snake_case__    ,						snake_case__ ):
                 snake_case       :   Dict							=      ClassificationFunction[function_to_apply.upper()]

          if function_to_apply is not None:
                 snake_case       :   Optional[int]							=      function_to_apply
          return preprocess_params, {}, postprocess_params



   def __call__(self		:					Dict    ,						*snake_case__		:					List[str]    ,						**snake_case__		:					int )		->	Optional[int]:


          '''simple docstring'''


          snake_case       :   Optional[int]							=      super().__call__(*snake_case__    ,						**snake_case__ )
          # TODO try and retrieve it in a nicer way from _sanitize_parameters.
          snake_case       :   Tuple							=      "top_k" not in kwargs
          if isinstance(args[0]    ,						snake_case__ ) and _legacy:
                 # This pipeline is odd, and return a list when single item is run
                 return [result]
          else:
                 return result



   def 				_SCREAMING_SNAKE_CASE   (self		:					Dict    ,						snake_case__		:					Tuple    ,						**snake_case__		:					Union[str, Any] )		->	Dict[str, GenericTensor]:


          '''simple docstring'''


          snake_case       :   int							=      self.framework
          if isinstance(snake_case__    ,						snake_case__ ):
                 return self.tokenizer(**snake_case__    ,						return_tensors=snake_case__    ,						**snake_case__ )
          elif isinstance(snake_case__    ,						snake_case__ ) and len(snake_case__ ) == 1 and isinstance(inputs[0]    ,						snake_case__ ) and len(inputs[0] ) == 2:
                 # It used to be valid to use a list of list of list for text pairs, keeping this path for BC
                 return self.tokenizer(
                     text=inputs[0][0]    ,						text_pair=inputs[0][1]    ,						return_tensors=snake_case__    ,						**snake_case__ )
          elif isinstance(snake_case__    ,						snake_case__ ):
                 # This is likely an invalid usage of the pipeline attempting to pass text pairs.
                 raise ValueError(
                     "The pipeline received invalid inputs, if you are trying to send text pairs, you can try to send a"
                     " dictionary `{\"text\": \"My text\", \"text_pair\": \"My pair\"}` in order to send a text pair." )
          return self.tokenizer(snake_case__    ,						return_tensors=snake_case__    ,						**snake_case__ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					int    ,						snake_case__		:					Union[str, Any] )		->	int:


          '''simple docstring'''


          return self.model(**snake_case__ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					List[str]    ,						snake_case__		:					Optional[Any]    ,						snake_case__		:					List[str]=None    ,						snake_case__		:					Dict=1    ,						snake_case__		:					Tuple=True )		->	str:


          '''simple docstring'''


          if function_to_apply is None:
                 if self.model.config.problem_type == "multi_label_classification" or self.model.config.num_labels == 1:
                        snake_case       :   Tuple							=      ClassificationFunction.SIGMOID
                 elif self.model.config.problem_type == "single_label_classification" or self.model.config.num_labels > 1:
                        snake_case       :   Tuple							=      ClassificationFunction.SOFTMAX
                 elif hasattr(self.model.config    ,						"function_to_apply" ) and function_to_apply is None:
                        snake_case       :   Tuple							=      self.model.config.function_to_apply
                 else:
                        snake_case       :   int							=      ClassificationFunction.NONE

          snake_case       :   Any							=      model_outputs["logits"][0]
          snake_case       :   List[str]							=      outputs.numpy()

          if function_to_apply == ClassificationFunction.SIGMOID:
                 snake_case       :   Optional[Any]							=      sigmoid(snake_case__ )
          elif function_to_apply == ClassificationFunction.SOFTMAX:
                 snake_case       :   Union[str, Any]							=      softmax(snake_case__ )
          elif function_to_apply == ClassificationFunction.NONE:
                 snake_case       :   Optional[Any]							=      outputs
          else:
                 raise ValueError(f"""Unrecognized `function_to_apply` argument: {function_to_apply}""" )

          if top_k == 1 and _legacy:
                 return {"label": self.model.config.idalabel[scores.argmax().item()], "score": scores.max().item()}

          snake_case       :   Optional[int]							=      [
              {"label": self.model.config.idalabel[i], "score": score.item()} for i, score in enumerate(snake_case__ )
          ]
          if not _legacy:
                 dict_scores.sort(key=lambda snake_case__ : x["score"]    ,						reverse=snake_case__ )
                 if top_k is not None:
                        snake_case       :   Optional[int]							=      dict_scores[:top_k]
          return dict_scores





