"""simple docstring"""







# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING, Dict, Optional

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.logging import get_logger
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
	import jax
	import jaxlib

_a					:					Optional[int]=			get_logger()

_a					:					Optional[dict]=			None

class 		UpperCamelCase      (   TensorFormatter[Mapping, """jax.Array""", Mapping]						):



				def __init__(self    :				Dict     ,  _A    :				Optional[int]=None     ,  _A    :				Optional[int]=None     ,  **_A    :				Union[str, Any])   ->     Tuple:
					super().__init__(features=_A)
					import jax
					from jaxlib.xla_client import Device

					if isinstance(_A     ,  _A):
						raise ValueError(
						    f"Expected {device} to be a `str` not {type(_A)}, as `jaxlib.xla_extension.Device` "
						    'is not serializable neither with `pickle` nor with `dill`. Instead you can surround '
						    'the device with `str()` to get its string identifier that will be internally mapped '
						    'to the actual `jaxlib.xla_extension.Device`.')
					__snake_case      :							Optional[Any]						=			device if isinstance(_A     ,  _A) else str(jax.devices()[0])
					# using global variable since `jaxlib.xla_extension.Device` is not serializable neither
					# with `pickle` nor with `dill`, so we need to use a global variable instead
					global DEVICE_MAPPING
					if DEVICE_MAPPING is None:
						__snake_case      :							Optional[Any]						=			self._map_devices_to_str()
					if self.device not in list(DEVICE_MAPPING.keys()):
						logger.warning(
						    f"Device with string identifier {self.device} not listed among the available "
						    f"devices: {list(DEVICE_MAPPING.keys())}, so falling back to the default "
						    f"device: {str(jax.devices()[0])}.")
						__snake_case      :							int						=			str(jax.devices()[0])
					__snake_case      :							str						=			jnp_array_kwargs



				@staticmethod
				def  _lowercase					()   ->     Dict[str, "jaxlib.xla_extension.Device"]:
					import jax

					return {str(_A): device for device in jax.devices()}



				def  _lowercase					(self    :				Tuple     ,  _A    :				List[Any])   ->     Optional[Any]:
					import jax
					import jax.numpy as jnp

					if isinstance(_A     ,  _A) and column:
						if all(
						    isinstance(_A     ,  jax.Array) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column):
							return jnp.stack(_A     ,  axis=0)
					return column



				def  _lowercase					(self    :				Tuple     ,  _A    :				List[str])   ->     int:
					import jax
					import jax.numpy as jnp

					if isinstance(_A     ,  (str, bytes, type(_A))):
						return value
					elif isinstance(_A     ,  (np.character, np.ndarray)) and np.issubdtype(value.dtype     ,  np.character):
						return value.tolist()

					__snake_case      :							int						=			{}

					if isinstance(_A     ,  (np.number, np.ndarray)) and np.issubdtype(value.dtype     ,  np.integer):
						# the default int precision depends on the jax config
						# see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
						if jax.config.jax_enable_xaa:
							__snake_case      :							Dict						=			{'dtype': jnp.intaa}
						else:
							__snake_case      :							List[str]						=			{'dtype': jnp.intaa}
					elif isinstance(_A     ,  (np.number, np.ndarray)) and np.issubdtype(value.dtype     ,  np.floating):
						__snake_case      :							List[Any]						=			{'dtype': jnp.floataa}
					elif config.PIL_AVAILABLE and "PIL" in sys.modules:
						import PIL.Image

						if isinstance(_A     ,  PIL.Image.Image):
							__snake_case      :							Tuple						=			np.asarray(_A)

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
					global DEVICE_MAPPING
					if DEVICE_MAPPING is None:
						__snake_case      :							List[Any]						=			self._map_devices_to_str()

					with jax.default_device(DEVICE_MAPPING[self.device]):
						# calling jnp.array on a np.ndarray does copy the data
						# see https://github.com/google/jax/issues/4486
						return jnp.array(_A     ,  **{**default_dtype, **self.jnp_array_kwargs})



				def  _lowercase					(self    :				Union[str, Any]     ,  _A    :				str)   ->     Union[str, Any]:
					import jax

					# support for torch, tf, jax etc.
					if config.TORCH_AVAILABLE and "torch" in sys.modules:
						import torch

						if isinstance(_A     ,  torch.Tensor):
							return self._tensorize(data_struct.detach().cpu().numpy()[()])
					if hasattr(_A     ,  '__array__') and not isinstance(_A     ,  jax.Array):
						__snake_case      :							Optional[int]						=			data_struct.__array__()
					# support for nested types like struct of list of struct
					if isinstance(_A     ,  np.ndarray):
						if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
							return self._consolidate([self.recursive_tensorize(_A) for substruct in data_struct])
					elif isinstance(_A     ,  (list, tuple)):
						return self._consolidate([self.recursive_tensorize(_A) for substruct in data_struct])
					return self._tensorize(_A)



				def  _lowercase					(self    :				Union[str, Any]     ,  _A    :				dict)   ->     List[str]:
					return map_nested(self._recursive_tensorize     ,  _A     ,  map_list=_A)



				def  _lowercase					(self    :				Union[str, Any]     ,  _A    :				pa.Table)   ->     Mapping:
					__snake_case      :							Any						=			self.numpy_arrow_extractor().extract_row(_A)
					__snake_case      :							Optional[int]						=			self.python_features_decoder.decode_row(_A)
					return self.recursive_tensorize(_A)



				def  _lowercase					(self    :				Any     ,  _A    :				pa.Table)   ->     "jax.Array":
					__snake_case      :							int						=			self.numpy_arrow_extractor().extract_column(_A)
					__snake_case      :							int						=			self.python_features_decoder.decode_column(_A     ,  pa_table.column_names[0])
					__snake_case      :							Optional[Any]						=			self.recursive_tensorize(_A)
					__snake_case      :							str						=			self._consolidate(_A)
					return column






				def  _lowercase					(self    :				Dict     ,  _A    :				pa.Table)   ->     Mapping:
					__snake_case      :							List[str]						=			self.numpy_arrow_extractor().extract_batch(_A)
					__snake_case      :							Optional[Any]						=			self.python_features_decoder.decode_batch(_A)
					__snake_case      :							Dict						=			self.recursive_tensorize(_A)
					for column_name in batch:
						__snake_case      :							Optional[Any]						=			self._consolidate(batch[column_name])
					return batch






"""simple docstring"""







import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch

class 		UpperCamelCase      (   lowercase						):



				@require_torch
				def  _lowercase					(self    :				Union[str, Any])   ->     Optional[Any]:
					# 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
					__snake_case      :							Any						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

					__snake_case      :							Tuple						=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

					__snake_case      :							int						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

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

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

					# should succeed
					__snake_case      :							Union[str, Any]						=			self.get_env()
					# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
					__snake_case      :							str						=			'1'
					__snake_case      :							Union[str, Any]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())



				@require_torch
				def  _lowercase					(self    :				Union[str, Any])   ->     Union[str, Any]:
					# python one-liner segments
					# this must be loaded before socket.socket is monkey-patched
					__snake_case      :							Optional[Any]						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

					__snake_case      :							Optional[Any]						=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

					__snake_case      :							Union[str, Any]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

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

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

					# should succeed
					__snake_case      :							int						=			self.get_env()
					__snake_case      :							Tuple						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())



				@require_torch
				def  _lowercase					(self    :				int)   ->     Any:
					# 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
					__snake_case      :							int						=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

					__snake_case      :							int						=			'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

					__snake_case      :							Optional[int]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

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

					# should succeed
					__snake_case      :							Optional[int]						=			self.get_env()
					__snake_case      :							Dict						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())

					# next emulate no network
					__snake_case      :							Optional[Any]						=			[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
					__snake_case      :							Union[str, Any]						=			'1'
					__snake_case      :							str						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())



				@require_torch
				def  _lowercase					(self    :				str)   ->     Dict:
					__snake_case      :							Dict						=			'\nfrom transformers import pipeline\n        '
					__snake_case      :							List[Any]						=			'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

					__snake_case      :							List[str]						=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
					__snake_case      :							str						=			self.get_env()
					__snake_case      :							Tuple						=			'1'
					__snake_case      :							Tuple						=			[sys.executable, '-c', '\n'.join([load, mock, run])]
					__snake_case      :							Optional[Any]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					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    :				int)   ->     Optional[Any]:
					__snake_case      :							int						=			'\nfrom transformers import AutoModel\n        '
					__snake_case      :							str						=			'\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

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

					# should succeed
					__snake_case      :							str						=			self.get_env()
					__snake_case      :							Dict						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					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
					__snake_case      :							List[str]						=			'1'
					__snake_case      :							Optional[int]						=			subprocess.run(_A     ,  env=_A     ,  check=_A     ,  capture_output=_A)
					self.assertEqual(result.returncode     ,  0     ,  result.stderr)
					self.assertIn('success'     ,  result.stdout.decode())





