






import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class  _UpperCAmelCase   (			A__      ):

				"""simple docstring"""




				@require_torch
				def 							lowercase__  (	self		:       Union[str, Any]    ):

								'''simple docstring'''
								# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
								# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
								# while running an external program

								# python one-liner segments

								# this must be loaded before socket.socket is monkey-patched
								lowercase__  					=     '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

								lowercase__  					=     '''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

								lowercase__  					=     '''
import socket
def offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")
socket.socket = offline_socket
        '''

								# Force fetching the files so that we can use the cache
								lowercase__  					=     '''hf-internal-testing/tiny-random-bert'''
								BertConfig.from_pretrained(lowerCamelCase    )
								BertModel.from_pretrained(lowerCamelCase    )
								BertTokenizer.from_pretrained(lowerCamelCase    )
								pipeline(task='''fill-mask''',    model=lowerCamelCase    )

								# baseline - just load from_pretrained with normal network
								lowercase__  					=     [sys.executable, '''-c''', '''\n'''.join([load, run, mock]    )]

								# should succeed
								lowercase__  					=     self.get_env()
								# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
								lowercase__  					=     '''1'''
								lowercase__  					=     subprocess.run(lowerCamelCase,    env=lowerCamelCase,    check=lowerCamelCase,    capture_output=lowerCamelCase    )
								self.assertEqual(result.returncode,    0,    result.stderr    )
								self.assertIn('''success''',    result.stdout.decode()    )




				@require_torch
				def 							lowercase__  (	self		:       List[str]    ):

								'''simple docstring'''
								# python one-liner segments
								# this must be loaded before socket.socket is monkey-patched
								lowercase__  					=     '''
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        '''

								lowercase__  					=     '''
mname = "hf-internal-testing/tiny-random-bert"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task="fill-mask", model=mname)
print("success")
        '''

								lowercase__  					=     '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")
socket.socket = offline_socket
        '''

								# Force fetching the files so that we can use the cache
								lowercase__  					=     '''hf-internal-testing/tiny-random-bert'''
								BertConfig.from_pretrained(lowerCamelCase    )
								BertModel.from_pretrained(lowerCamelCase    )
								BertTokenizer.from_pretrained(lowerCamelCase    )
								pipeline(task='''fill-mask''',    model=lowerCamelCase    )

								# baseline - just load from_pretrained with normal network
								lowercase__  					=     [sys.executable, '''-c''', '''\n'''.join([load, run, mock]    )]

								# should succeed
								lowercase__  					=     self.get_env()
								lowercase__  					=     subprocess.run(lowerCamelCase,    env=lowerCamelCase,    check=lowerCamelCase,    capture_output=lowerCamelCase    )
								self.assertEqual(result.returncode,    0,    result.stderr    )
								self.assertIn('''success''',    result.stdout.decode()    )




				@require_torch
				def 							lowercase__  (	self		:       Tuple    ):

								'''simple docstring'''
								# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
								# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
								# while running an external program

								# python one-liner segments

								# this must be loaded before socket.socket is monkey-patched
								lowercase__  					=     '''
from transformers import BertConfig, BertModel, BertTokenizer
        '''

								lowercase__  					=     '''
mname = "hf-internal-testing/tiny-random-bert-sharded"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print("success")
        '''

								lowercase__  					=     '''
import socket
def offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")
socket.socket = offline_socket
        '''

								# baseline - just load from_pretrained with normal network
								lowercase__  					=     [sys.executable, '''-c''', '''\n'''.join([load, run]    )]

								# should succeed
								lowercase__  					=     self.get_env()
								lowercase__  					=     subprocess.run(lowerCamelCase,    env=lowerCamelCase,    check=lowerCamelCase,    capture_output=lowerCamelCase    )
								self.assertEqual(result.returncode,    0,    result.stderr    )
								self.assertIn('''success''',    result.stdout.decode()    )

								# next emulate no network
								lowercase__  					=     [sys.executable, '''-c''', '''\n'''.join([load, mock, run]    )]

								# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
								# env["TRANSFORMERS_OFFLINE"] = "0"
								# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
								# self.assertEqual(result.returncode, 1, result.stderr)

								# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
								lowercase__  					=     '''1'''
								lowercase__  					=     subprocess.run(lowerCamelCase,    env=lowerCamelCase,    check=lowerCamelCase,    capture_output=lowerCamelCase    )
								self.assertEqual(result.returncode,    0,    result.stderr    )
								self.assertIn('''success''',    result.stdout.decode()    )




				@require_torch
				def 							lowercase__  (	self		:       str    ):

								'''simple docstring'''
								lowercase__  					=     '''
from transformers import pipeline
        '''
								lowercase__  					=     '''
mname = "hf-internal-testing/tiny-random-bert"
pipe = pipeline(model=mname)
        '''

								lowercase__  					=     '''
import socket
def offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")
socket.socket = offline_socket
        '''
								lowercase__  					=     self.get_env()
								lowercase__  					=     '''1'''
								lowercase__  					=     [sys.executable, '''-c''', '''\n'''.join([load, mock, run]    )]
								lowercase__  					=     subprocess.run(lowerCamelCase,    env=lowerCamelCase,    check=lowerCamelCase,    capture_output=lowerCamelCase    )
								self.assertEqual(result.returncode,    1,    result.stderr    )
								self.assertIn(
								    '''You cannot infer task automatically within `pipeline` when using offline mode''',    result.stderr.decode().replace('''\n''',    ''''''    ),    )




				@require_torch
				def 							lowercase__  (	self		:       str    ):

								'''simple docstring'''
								lowercase__  					=     '''
from transformers import AutoModel
        '''
								lowercase__  					=     '''
mname = "hf-internal-testing/test_dynamic_model"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print("success")
        '''

								# baseline - just load from_pretrained with normal network
								lowercase__  					=     [sys.executable, '''-c''', '''\n'''.join([load, run]    )]

								# should succeed
								lowercase__  					=     self.get_env()
								lowercase__  					=     subprocess.run(lowerCamelCase,    env=lowerCamelCase,    check=lowerCamelCase,    capture_output=lowerCamelCase    )
								self.assertEqual(result.returncode,    0,    result.stderr    )
								self.assertIn('''success''',    result.stdout.decode()    )

								# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
								lowercase__  					=     '''1'''
								lowercase__  					=     subprocess.run(lowerCamelCase,    env=lowerCamelCase,    check=lowerCamelCase,    capture_output=lowerCamelCase    )
								self.assertEqual(result.returncode,    0,    result.stderr    )
								self.assertIn('''success''',    result.stdout.decode()    )












import numpy as np

import datasets


A__      :     int       =					'\nCompute the Mahalanobis Distance\n\nMahalonobis distance is the distance between a point and a distribution.\nAnd not between two distinct points. It is effectively a multivariate equivalent of the Euclidean distance.\nIt was introduced by Prof. P. C. Mahalanobis in 1936\nand has been used in various statistical applications ever since\n[source: https://www.machinelearningplus.com/statistics/mahalanobis-distance/]\n'

A__      :     Optional[int]       =					'\\n@article{de2000mahalanobis,\n  title={The mahalanobis distance},\n  author={De Maesschalck, Roy and Jouan-Rimbaud, Delphine and Massart, D{\'e}sir{\'e} L},\n  journal={Chemometrics and intelligent laboratory systems},\n  volume={50},\n  number={1},\n  pages={1--18},\n  year={2000},\n  publisher={Elsevier}\n}\n'

A__      :     Optional[int]       =					'\nArgs:\n    X: List of datapoints to be compared with the `reference_distribution`.\n    reference_distribution: List of datapoints from the reference distribution we want to compare to.\nReturns:\n    mahalanobis: The Mahalonobis distance for each datapoint in `X`.\nExamples:\n\n    >>> mahalanobis_metric = datasets.load_metric("mahalanobis")\n    >>> results = mahalanobis_metric.compute(reference_distribution=[[0, 1], [1, 0]], X=[[0, 1]])\n    >>> print(results)\n    {\'mahalanobis\': array([0.5])}\n'
@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION						,_KWARGS_DESCRIPTION      )
class  _UpperCAmelCase   (			datasets.Metric      ):

				"""simple docstring"""




				def 							lowercase__  (	self		:       str    ):

								'''simple docstring'''
								return datasets.MetricInfo(
								    description=_DESCRIPTION,    citation=_CITATION,    inputs_description=_KWARGS_DESCRIPTION,    features=datasets.Features(
								        {
								            '''X''': datasets.Sequence(datasets.Value('''float''',    id='''sequence'''    ),    id='''X'''    ),
								        }    ),    )




				def 							lowercase__  (	self		:       Dict,    lowerCamelCase		:       int,    lowerCamelCase		:       Optional[int]    ):

								'''simple docstring'''
								# convert to numpy arrays
								lowercase__  					=     np.array(lowerCamelCase    )
								lowercase__  					=     np.array(lowerCamelCase    )

								# Assert that arrays are 2D
								if len(X.shape    ) != 2:
												raise ValueError('''Expected `X` to be a 2D vector'''    )
								if len(reference_distribution.shape    ) != 2:
												raise ValueError('''Expected `reference_distribution` to be a 2D vector'''    )
								if reference_distribution.shape[0] < 2:
												raise ValueError(
												    '''Expected `reference_distribution` to be a 2D vector with more than one element in the first dimension'''    )

								# Get mahalanobis distance for each prediction
								lowercase__  					=     X - np.mean(lowerCamelCase    )
								lowercase__  					=     np.cov(reference_distribution.T    )
								try:
												lowercase__  					=     np.linalg.inv(lowerCamelCase    )
								except np.linalg.LinAlgError:
												lowercase__  					=     np.linalg.pinv(lowerCamelCase    )
								lowercase__  					=     np.dot(lowerCamelCase,    lowerCamelCase    )
								lowercase__  					=     np.dot(lowerCamelCase,    X_minus_mu.T    ).diagonal()

								return {"mahalanobis": mahal_dist}




