
# 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   =			get_logger()

_A   =			None







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





     def __init__(self				:   Union[str, Any]    ,   _A				:   Dict=None    ,   _A				:   Any=None    ,   **_A				:   str       )						->						List[Any]:
         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								= 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								= 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								= str(jax.devices()[0]       )
         snake_case								= jnp_array_kwargs





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

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





     def 		UpperCAmelCase(self				:   Any    ,   _A				:   List[str]       )						->						str:
         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 		UpperCAmelCase(self				:   List[Any]    ,   _A				:   Optional[Any]       )						->						Any:
         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								= {}

         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								= {"dtype": jnp.intaa}
             else:
                 snake_case								= {"dtype": jnp.intaa}
         elif isinstance(_A    ,   (np.number, np.ndarray)       ) and np.issubdtype(value.dtype    ,   np.floating       ):
             snake_case								= {"dtype": jnp.floataa}
         elif config.PIL_AVAILABLE and "PIL" in sys.modules:
             import PIL.Image

             if isinstance(_A    ,   PIL.Image.Image       ):
                 snake_case								= 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								= 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 		UpperCAmelCase(self				:   str    ,   _A				:   Any       )						->						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								= 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 		UpperCAmelCase(self				:   int    ,   _A				:   dict       )						->						Optional[int]:
         return map_nested(self._recursive_tensorize    ,   _A    ,   map_list=_A       )





     def 		UpperCAmelCase(self				:   Any    ,   _A				:   pa.Table       )						->						Mapping:
         snake_case								= self.numpy_arrow_extractor().extract_row(_A       )
         snake_case								= self.python_features_decoder.decode_row(_A       )
         return self.recursive_tensorize(_A       )





     def 		UpperCAmelCase(self				:   List[Any]    ,   _A				:   pa.Table       )						->						"jax.Array":
         snake_case								= self.numpy_arrow_extractor().extract_column(_A       )
         snake_case								= self.python_features_decoder.decode_column(_A    ,   pa_table.column_names[0]       )
         snake_case								= self.recursive_tensorize(_A       )
         snake_case								= self._consolidate(_A       )
         return column





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



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 					lowerCamelCase		(  A_  ):
     UpperCAmelCase__    :							Dict									=		["image_processor", "tokenizer"]
     UpperCAmelCase__    :							Dict									=		"LayoutLMv2ImageProcessor"
     UpperCAmelCase__    :							Optional[Any]									=		("LayoutXLMTokenizer", "LayoutXLMTokenizerFast")





     def __init__(self				:   str    ,   _A				:   Any=None    ,   _A				:   Tuple=None    ,   **_A				:   Optional[Any]       )						->						Optional[int]:
         if "feature_extractor" in kwargs:
             warnings.warn(
                 "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
                 " instead."    ,   _A    ,   )
             snake_case								= kwargs.pop("feature_extractor"       )

         snake_case								= 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				:   int    ,   _A				:   List[str]    ,   _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				:   Dict    ,   )						->						BatchEncoding:
         # 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
         snake_case								= 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       ):
                 snake_case								= [text]  # add batch dimension (as the image processor always adds a batch dimension)
             snake_case								= features["words"]

         snake_case								= 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
         snake_case								= features.pop("pixel_values"       )
         if return_overflowing_tokens is True:
             snake_case								= self.get_overflowing_images(_A    ,   encoded_inputs["overflow_to_sample_mapping"]       )
         snake_case								= images

         return encoded_inputs





     def 		UpperCAmelCase(self				:   Dict    ,   _A				:   Dict    ,   _A				:   List[str]       )						->						Optional[int]:
         # in case there's an overflow, ensure each `input_ids` sample is mapped to its corresponding image
         snake_case								= []
         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 		UpperCAmelCase(self				:   Tuple    ,   *_A				:   int    ,   **_A				:   Dict       )						->						str:
         return self.tokenizer.batch_decode(*_A    ,   **_A       )





     def 		UpperCAmelCase(self				:   str    ,   *_A				:   List[Any]    ,   **_A				:   List[Any]       )						->						Optional[Any]:
         return self.tokenizer.decode(*_A    ,   **_A       )





     @property
     def 		UpperCAmelCase(self				:   Tuple       )						->						Optional[int]:
         return ["input_ids", "bbox", "attention_mask", "image"]





     @property
     def 		UpperCAmelCase(self				:   List[Any]       )						->						int:
         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 		UpperCAmelCase(self				:   Dict       )						->						Union[str, Any]:
         warnings.warn(
             "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."    ,   _A    ,   )
         return self.image_processor

