
"""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,
)


_A     :			Tuple  						=			'hf-internal-testing/tiny-random-bert'
_A     :			Dict  						=			os.path.join(TRANSFORMERS_CACHE, """models--hf-internal-testing--tiny-random-bert""")
_A     :			Optional[Any]  						=			'9b8c223d42b2188cb49d29af482996f9d0f3e5a6'

class 						a__		( unittest.TestCase		):
   def      __magic_name__     (     self							):
          lowercase       :							List[str]									=				cached_file(A_      ,       A_							)
          # Should have downloaded the file in here
          self.assertTrue(os.path.isdir(A_							)							)
          # Cache should contain at least those three subfolders:
          for subfolder in ["blobs", "refs", "snapshots"]:
                 self.assertTrue(os.path.isdir(os.path.join(A_      ,       A_							)							)							)
          with open(os.path.join(A_      ,       "refs"      ,       "main"							)							) as f:
                 lowercase       :							Optional[int]									=				f.read()
          self.assertEqual(A_      ,       os.path.join(A_      ,       "snapshots"      ,       A_      ,       A_							)							)
          self.assertTrue(os.path.isfile(A_							)							)

          # File is cached at the same place the second time.
          lowercase       :							Dict									=				cached_file(A_      ,       A_							)
          self.assertEqual(A_      ,       A_							)

          # Using a specific revision to test the full commit hash.
          lowercase       :							Optional[Any]									=				cached_file(A_      ,       A_      ,       revision="9b8c223"							)
          self.assertEqual(A_      ,       os.path.join(A_      ,       "snapshots"      ,       A_      ,       A_							)							)
   def      __magic_name__     (     self							):
          with self.assertRaisesRegex(A_      ,       "is not a valid model identifier"							):
                 lowercase       :							int									=				cached_file("tiny-random-bert"      ,       A_							)

          with self.assertRaisesRegex(A_      ,       "is not a valid git identifier"							):
                 lowercase       :							int									=				cached_file(A_      ,       A_      ,       revision="aaaa"							)

          with self.assertRaisesRegex(A_      ,       "does not appear to have a file named"							):
                 lowercase       :							str									=				cached_file(A_      ,       "conf"							)
   def      __magic_name__     (     self							):
          with self.assertRaisesRegex(A_      ,       "does not appear to have a file named"							):
                 lowercase       :							int									=				cached_file(A_      ,       "conf"							)

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

          lowercase       :							Dict									=				cached_file(A_      ,       "conf"      ,       _raise_exceptions_for_missing_entries=A_							)
          self.assertIsNone(A_							)

          lowercase       :							List[str]									=				cached_file(A_      ,       "conf"      ,       local_files_only=A_      ,       _raise_exceptions_for_missing_entries=A_							)
          self.assertIsNone(A_							)

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

          # Under the mock environment we get a 500 error when trying to reach the tokenizer.
          with mock.patch("requests.Session.request"      ,       return_value=A_							) as mock_head:
                 lowercase       :							List[Any]									=				cached_file(A_      ,       "conf"      ,       _raise_exceptions_for_connection_errors=A_							)
                 self.assertIsNone(A_							)
                 # This check we did call the fake head request
                 mock_head.assert_called()
   def      __magic_name__     (     self							):
          self.assertTrue(has_file("hf-internal-testing/tiny-bert-pt-only"      ,       A_							)							)
          self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"      ,       A_							)							)
          self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"      ,       A_							)							)
   def      __magic_name__     (     self							):
          # `get_file_from_repo` returns None if the file does not exist
          self.assertIsNone(get_file_from_repo("bert-base-cased"      ,       "ahah.txt"							)							)

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

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

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

   def      __magic_name__     (     self							):
          with tempfile.TemporaryDirectory() as tmp_dir:
                 lowercase       :							List[str]									=				Path(A_							) / "a.txt"
                 filename.touch()
                 self.assertEqual(get_file_from_repo(A_      ,       "a.txt"							)      ,       str(A_							)							)

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






"""simple docstring"""




import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType




class 			__snake_case   (lowerCamelCase  ):
							__a					 =       ['''image_processor''', '''tokenizer''']
							__a					 =       '''LayoutLMv2ImageProcessor'''
							__a					 =       ('''LayoutXLMTokenizer''', '''LayoutXLMTokenizerFast''')




							def __init__(     self:    Any     ,							A_:    Optional[Any]=None     ,							A_:    Dict=None     ,							**A_:    Any   ):
										if "feature_extractor" in kwargs:
													warnings.warn(
													    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
													    """ instead."""     ,							A_     ,							)
													__lowerCamelCase          =      kwargs.pop("""feature_extractor"""   )

										__lowerCamelCase          =      image_processor if image_processor is not None else feature_extractor
										if image_processor is None:
													raise ValueError("""You need to specify an `image_processor`."""   )
										if tokenizer is None:
													raise ValueError("""You need to specify a `tokenizer`."""   )

										super().__init__(A_     ,							A_   )




							def __call__(     self:    List[str]     ,							A_:    Any     ,							A_:    Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None     ,							A_:    Optional[Union[PreTokenizedInput, List[PreTokenizedInput]]] = None     ,							A_:    Union[List[List[int]], List[List[List[int]]]] = None     ,							A_:    Optional[Union[List[int], List[List[int]]]] = None     ,							A_:    bool = True     ,							A_:    Union[bool, str, PaddingStrategy] = False     ,							A_:    Union[bool, str, TruncationStrategy] = None     ,							A_:    Optional[int] = None     ,							A_:    int = 0     ,							A_:    Optional[int] = None     ,							A_:    Optional[bool] = None     ,							A_:    Optional[bool] = None     ,							A_:    bool = False     ,							A_:    bool = False     ,							A_:    bool = False     ,							A_:    bool = False     ,							A_:    bool = True     ,							A_:    Optional[Union[str, TensorType]] = None     ,							**A_:    Any     ,							):
										# verify input
										if self.image_processor.apply_ocr and (boxes is not None):
													raise ValueError(
													    """You cannot provide bounding boxes """
													    """if you initialized the image processor with apply_ocr set to True."""   )

										if self.image_processor.apply_ocr and (word_labels is not None):
													raise ValueError(
													    """You cannot provide word labels if you initialized the image processor with apply_ocr set to True."""   )

										if return_overflowing_tokens is True and return_offsets_mapping is False:
													raise ValueError("""You cannot return overflowing tokens without returning the offsets mapping."""   )

										# first, apply the image processor
										__lowerCamelCase          =      self.image_processor(images=A_     ,							return_tensors=A_   )

										# second, apply the tokenizer
										if text is not None and self.image_processor.apply_ocr and text_pair is None:
													if isinstance(A_     ,							A_   ):
																__lowerCamelCase          =      [text]  # add batch dimension (as the image processor always adds a batch dimension)
													__lowerCamelCase          =      features["""words"""]

										__lowerCamelCase          =      self.tokenizer(
										    text=text if text is not None else features["""words"""]     ,							text_pair=text_pair if text_pair is not None else None     ,							boxes=boxes if boxes is not None else features["""boxes"""]     ,							word_labels=A_     ,							add_special_tokens=A_     ,							padding=A_     ,							truncation=A_     ,							max_length=A_     ,							stride=A_     ,							pad_to_multiple_of=A_     ,							return_token_type_ids=A_     ,							return_attention_mask=A_     ,							return_overflowing_tokens=A_     ,							return_special_tokens_mask=A_     ,							return_offsets_mapping=A_     ,							return_length=A_     ,							verbose=A_     ,							return_tensors=A_     ,							**A_     ,							)

										# add pixel values
										__lowerCamelCase          =      features.pop("""pixel_values"""   )
										if return_overflowing_tokens is True:
													__lowerCamelCase          =      self.get_overflowing_images(A_     ,							encoded_inputs["""overflow_to_sample_mapping"""]   )
										__lowerCamelCase          =      images

										return encoded_inputs




							def   __a   (     self:    Optional[Any]     ,							A_:    Any     ,							A_:    Tuple   ):
										# in case there's an overflow, ensure each `input_ids` sample is mapped to its corresponding image
										__lowerCamelCase          =      []
										for sample_idx in overflow_to_sample_mapping:
													images_with_overflow.append(images[sample_idx]   )

										if len(A_   ) != len(A_   ):
													raise ValueError(
													    """Expected length of images to be the same as the length of `overflow_to_sample_mapping`, but got"""
													    f' {len(A_   )} and {len(A_   )}'   )

										return images_with_overflow




							def   __a   (     self:    Optional[Any]     ,							*A_:    str     ,							**A_:    Dict   ):
										return self.tokenizer.batch_decode(*A_     ,							**A_   )




							def   __a   (     self:    Union[str, Any]     ,							*A_:    Optional[Any]     ,							**A_:    Tuple   ):
										return self.tokenizer.decode(*A_     ,							**A_   )




							@property
							def   __a   (     self:    Union[str, Any]   ):
										return ["input_ids", "bbox", "attention_mask", "image"]




							@property
							def   __a   (     self:    Optional[Any]   ):
										warnings.warn(
										    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""     ,							A_     ,							)
										return self.image_processor_class




							@property
							def   __a   (     self:    List[Any]   ):
										warnings.warn(
										    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""     ,							A_     ,							)
										return self.image_processor




