


"""simple docstring"""



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

import numpy as np
import pyarrow as pa

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


if TYPE_CHECKING:
  import torch







class _lowerCAmelCase				(      TensorFormatter[Mapping, '''torch.Tensor''', Mapping]	):







   def __init__(	self						,	UpperCamelCase__=None						,	**UpperCamelCase__						)	->		Dict:

       '''simple docstring'''




       super().__init__(features=UpperCamelCase__						)
       snake_case    :      Optional[Any]      					=    torch_tensor_kwargs
       import torch  # noqa import torch at initialization







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Optional[int]:

       '''simple docstring'''




       import torch

       if isinstance(UpperCamelCase__						,	UpperCamelCase__						) and column:
           if all(
               isinstance(UpperCamelCase__						,	torch.Tensor						) and x.shape == column[0].shape and x.dtype == column[0].dtype
               for x in column						):
               return torch.stack(UpperCamelCase__						)
       return column







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Any:

       '''simple docstring'''




       import torch

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

       snake_case    :      List[Any]      					=    {}

       if isinstance(UpperCamelCase__						,	(np.number, np.ndarray)						) and np.issubdtype(value.dtype						,	np.integer						):
           snake_case    :      Union[str, Any]      					=    {"dtype": torch.intaa}
       elif isinstance(UpperCamelCase__						,	(np.number, np.ndarray)						) and np.issubdtype(value.dtype						,	np.floating						):
           snake_case    :      Dict      					=    {"dtype": torch.floataa}
       elif config.PIL_AVAILABLE and "PIL" in sys.modules:
           import PIL.Image

           if isinstance(UpperCamelCase__						,	PIL.Image.Image						):
               snake_case    :      Tuple      					=    np.asarray(UpperCamelCase__						)
       return torch.tensor(UpperCamelCase__						,	**{**default_dtype, **self.torch_tensor_kwargs}						)







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Tuple:

       '''simple docstring'''




       import torch

       # support for torch, tf, jax etc.
       if hasattr(UpperCamelCase__						,	"__array__"						) and not isinstance(UpperCamelCase__						,	torch.Tensor						):
           snake_case    :      List[Any]      					=    data_struct.__array__()
       # support for nested types like struct of list of struct
       if isinstance(UpperCamelCase__						,	np.ndarray						):
           if data_struct.dtype == object:  # torch tensors cannot be instantied from an array of objects
               return self._consolidate([self.recursive_tensorize(UpperCamelCase__						) for substruct in data_struct]						)
       elif isinstance(UpperCamelCase__						,	(list, tuple)						):
           return self._consolidate([self.recursive_tensorize(UpperCamelCase__						) for substruct in data_struct]						)
       return self._tensorize(UpperCamelCase__						)







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		str:

       '''simple docstring'''




       return map_nested(self._recursive_tensorize						,	UpperCamelCase__						,	map_list=UpperCamelCase__						)







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Mapping:

       '''simple docstring'''




       snake_case    :      List[Any]      					=    self.numpy_arrow_extractor().extract_row(UpperCamelCase__						)
       snake_case    :      List[Any]      					=    self.python_features_decoder.decode_row(UpperCamelCase__						)
       return self.recursive_tensorize(UpperCamelCase__						)







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		"torch.Tensor":

       '''simple docstring'''




       snake_case    :      int      					=    self.numpy_arrow_extractor().extract_column(UpperCamelCase__						)
       snake_case    :      int      					=    self.python_features_decoder.decode_column(UpperCamelCase__						,	pa_table.column_names[0]						)
       snake_case    :      Optional[Any]      					=    self.recursive_tensorize(UpperCamelCase__						)
       snake_case    :      str      					=    self._consolidate(UpperCamelCase__						)
       return column







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Mapping:

       '''simple docstring'''




       snake_case    :      str      					=    self.numpy_arrow_extractor().extract_batch(UpperCamelCase__						)
       snake_case    :      int      					=    self.python_features_decoder.decode_batch(UpperCamelCase__						)
       snake_case    :      Union[str, Any]      					=    self.recursive_tensorize(UpperCamelCase__						)
       for column_name in batch:
           snake_case    :      Optional[int]      					=    self._consolidate(batch[column_name]						)
       return batch









"""simple docstring"""



import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


__snake_case   			=    """hf-internal-testing/tiny-random-bert"""
__snake_case   			=    os.path.join(TRANSFORMERS_CACHE, """models--hf-internal-testing--tiny-random-bert""")
__snake_case   			=    """9b8c223d42b2188cb49d29af482996f9d0f3e5a6"""







class _lowerCAmelCase				(      unittest.TestCase	):







   def  lowerCamelCase		(	self						)	->		str:

       '''simple docstring'''




       snake_case    :      int      					=    cached_file(UpperCamelCase__						,	UpperCamelCase__						)
       # Should have downloaded the file in here
       self.assertTrue(os.path.isdir(UpperCamelCase__						)						)
       # Cache should contain at least those three subfolders:
       for subfolder in ["blobs", "refs", "snapshots"]:
           self.assertTrue(os.path.isdir(os.path.join(UpperCamelCase__						,	UpperCamelCase__						)						)						)
       with open(os.path.join(UpperCamelCase__						,	"refs"						,	"main"						)						) as f:
           snake_case    :      Dict      					=    f.read()
       self.assertEqual(UpperCamelCase__						,	os.path.join(UpperCamelCase__						,	"snapshots"						,	UpperCamelCase__						,	UpperCamelCase__						)						)
       self.assertTrue(os.path.isfile(UpperCamelCase__						)						)

       # File is cached at the same place the second time.
       snake_case    :      List[str]      					=    cached_file(UpperCamelCase__						,	UpperCamelCase__						)
       self.assertEqual(UpperCamelCase__						,	UpperCamelCase__						)

       # Using a specific revision to test the full commit hash.
       snake_case    :      Any      					=    cached_file(UpperCamelCase__						,	UpperCamelCase__						,	revision="9b8c223"						)
       self.assertEqual(UpperCamelCase__						,	os.path.join(UpperCamelCase__						,	"snapshots"						,	UpperCamelCase__						,	UpperCamelCase__						)						)







   def  lowerCamelCase		(	self						)	->		Any:

       '''simple docstring'''




       with self.assertRaisesRegex(UpperCamelCase__						,	"is not a valid model identifier"						):
           snake_case    :      Optional[Any]      					=    cached_file("tiny-random-bert"						,	UpperCamelCase__						)

       with self.assertRaisesRegex(UpperCamelCase__						,	"is not a valid git identifier"						):
           snake_case    :      Optional[Any]      					=    cached_file(UpperCamelCase__						,	UpperCamelCase__						,	revision="aaaa"						)

       with self.assertRaisesRegex(UpperCamelCase__						,	"does not appear to have a file named"						):
           snake_case    :      List[Any]      					=    cached_file(UpperCamelCase__						,	"conf"						)







   def  lowerCamelCase		(	self						)	->		Tuple:

       '''simple docstring'''




       with self.assertRaisesRegex(UpperCamelCase__						,	"does not appear to have a file named"						):
           snake_case    :      Tuple      					=    cached_file(UpperCamelCase__						,	"conf"						)

       with open(os.path.join(UpperCamelCase__						,	"refs"						,	"main"						)						) as f:
           snake_case    :      Any      					=    f.read()
       self.assertTrue(os.path.isfile(os.path.join(UpperCamelCase__						,	".no_exist"						,	UpperCamelCase__						,	"conf"						)						)						)

       snake_case    :      Optional[Any]      					=    cached_file(UpperCamelCase__						,	"conf"						,	_raise_exceptions_for_missing_entries=UpperCamelCase__						)
       self.assertIsNone(UpperCamelCase__						)

       snake_case    :      Any      					=    cached_file(UpperCamelCase__						,	"conf"						,	local_files_only=UpperCamelCase__						,	_raise_exceptions_for_missing_entries=UpperCamelCase__						)
       self.assertIsNone(UpperCamelCase__						)

       snake_case    :      Any      					=    mock.Mock()
       snake_case    :      List[Any]      					=    500
       snake_case    :      int      					=    {}
       snake_case    :      Optional[int]      					=    HTTPError
       snake_case    :      Tuple      					=    {}

       # Under the mock environment we get a 500 error when trying to reach the tokenizer.
       with mock.patch("requests.Session.request"						,	return_value=UpperCamelCase__						) as mock_head:
           snake_case    :      Tuple      					=    cached_file(UpperCamelCase__						,	"conf"						,	_raise_exceptions_for_connection_errors=UpperCamelCase__						)
           self.assertIsNone(UpperCamelCase__						)
           # This check we did call the fake head request
           mock_head.assert_called()







   def  lowerCamelCase		(	self						)	->		Any:

       '''simple docstring'''




       self.assertTrue(has_file("hf-internal-testing/tiny-bert-pt-only"						,	UpperCamelCase__						)						)
       self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"						,	UpperCamelCase__						)						)
       self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"						,	UpperCamelCase__						)						)







   def  lowerCamelCase		(	self						)	->		str:

       '''simple docstring'''




       self.assertIsNone(get_file_from_repo("bert-base-cased"						,	"ahah.txt"						)						)

       # The function raises if the repository does not exist.
       with self.assertRaisesRegex(UpperCamelCase__						,	"is not a valid model identifier"						):
           get_file_from_repo("bert-base-case"						,	UpperCamelCase__						)

       # The function raises if the revision does not exist.
       with self.assertRaisesRegex(UpperCamelCase__						,	"is not a valid git identifier"						):
           get_file_from_repo("bert-base-cased"						,	UpperCamelCase__						,	revision="ahaha"						)

       snake_case    :      int      					=    get_file_from_repo("bert-base-cased"						,	UpperCamelCase__						)
       # The name is the cached name which is not very easy to test, so instead we load the content.
       snake_case    :      str      					=    json.loads(open(UpperCamelCase__						,	"r"						).read()						)
       self.assertEqual(config["hidden_size"]						,	768						)







   def  lowerCamelCase		(	self						)	->		Optional[Any]:

       '''simple docstring'''




       with tempfile.TemporaryDirectory() as tmp_dir:
           snake_case    :      int      					=    Path(UpperCamelCase__						) / "a.txt"
           filename.touch()
           self.assertEqual(get_file_from_repo(UpperCamelCase__						,	"a.txt"						)						,	str(UpperCamelCase__						)						)

           self.assertIsNone(get_file_from_repo(UpperCamelCase__						,	"b.txt"						)						)





