
"""simple docstring"""




import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map


def     __magic_name__      ( __snake_case							:			Union[dict, list, tuple, torch.Tensor]    )					->							List[Any]:
       lowercase       :							str									=				[]
       if isinstance(lowercase__	,					lowercase__    ):
              for v in tree.values():
                     shapes.extend(_fetch_dims(lowercase__    )    )
       elif isinstance(lowercase__	,					(list, tuple)    ):
              for t in tree:
                     shapes.extend(_fetch_dims(lowercase__    )    )
       elif isinstance(lowercase__	,					torch.Tensor    ):
              shapes.append(tree.shape    )
       else:
              raise ValueError("Not supported"    )

       return shapes


@torch.jit.ignore
def     __magic_name__      ( __snake_case							:			int	,					__snake_case							:			Tuple[int, ...]    )					->							Optional[int]:
       lowercase       :							Dict									=				[]
       for d in reversed(lowercase__    ):
              idx.append(flat_idx % d    )
              lowercase       :							Tuple									=				flat_idx // d

       return tuple(reversed(lowercase__    )    )


@torch.jit.ignore
def     __magic_name__      ( __snake_case							:			Sequence[int]	,					__snake_case							:			Sequence[int]	,					__snake_case							:			Sequence[int]	,					__snake_case							:			Optional[Sequence[bool]] = None	,					__snake_case							:			Optional[Sequence[bool]] = None	,					)					->							Union[str, Any]:

       # start_edges and end_edges both indicate whether, starting from any given
       # dimension, the start/end index is at the top/bottom edge of the
       # corresponding tensor, modeled as a tree
       def reduce_edge_list(__snake_case							:			List[bool]    ) -> None:
              lowercase       :							Optional[int]									=				True
              for i in range(len(lowercase__    )    ):
                     lowercase       :							int									=				-1 * (i + 1)
                     l[reversed_idx] &= tally
                     lowercase       :							Dict									=				l[reversed_idx]

       if start_edges is None:
              lowercase       :							str									=				[s == 0 for s in start]
              reduce_edge_list(lowercase__    )
       if end_edges is None:
              lowercase       :							List[str]									=				[e == (d - 1) for e, d in zip(lowercase__	,					lowercase__    )]
              reduce_edge_list(lowercase__    )

       # Base cases. Either start/end are empty and we're done, or the final,
       # one-dimensional tensor can be simply sliced
       if len(lowercase__    ) == 0:
              return [()]
       elif len(lowercase__    ) == 1:
              return [(slice(start[0]	,					end[0] + 1    ),)]

       lowercase       :							Union[str, Any]									=				[]
       lowercase       :							Dict									=				[]

       # Dimensions common to start and end can be selected directly
       for s, e in zip(lowercase__	,					lowercase__    ):
              if s == e:
                     path_list.append(slice(lowercase__	,					s + 1    )    )
              else:
                     break

       lowercase       :							Optional[Any]									=				tuple(lowercase__    )
       lowercase       :							Dict									=				len(lowercase__    )

       # start == end, and we're done
       if divergence_idx == len(lowercase__    ):
              return [path]

       def upper() -> Tuple[Tuple[slice, ...], ...]:
              assert start_edges is not None
              assert end_edges is not None

              lowercase       :							Any									=				start[divergence_idx]
              return tuple(
                  path + (slice(lowercase__	,					sdi + 1    ),) + s
                  for s in _get_minimal_slice_set(
                      start[divergence_idx + 1 :]	,					[d - 1 for d in dims[divergence_idx + 1 :]]	,					dims[divergence_idx + 1 :]	,					start_edges=start_edges[divergence_idx + 1 :]	,					end_edges=[True for _ in end_edges[divergence_idx + 1 :]]	,					)    )

       def lower() -> Tuple[Tuple[slice, ...], ...]:
              assert start_edges is not None
              assert end_edges is not None

              lowercase       :							str									=				end[divergence_idx]
              return tuple(
                  path + (slice(lowercase__	,					edi + 1    ),) + s
                  for s in _get_minimal_slice_set(
                      [0 for _ in start[divergence_idx + 1 :]]	,					end[divergence_idx + 1 :]	,					dims[divergence_idx + 1 :]	,					start_edges=[True for _ in start_edges[divergence_idx + 1 :]]	,					end_edges=end_edges[divergence_idx + 1 :]	,					)    )

       # If both start and end are at the edges of the subtree rooted at
       # divergence_idx, we can just select the whole subtree at once
       if start_edges[divergence_idx] and end_edges[divergence_idx]:
              slices.append(path + (slice(start[divergence_idx]	,					end[divergence_idx] + 1    ),)    )
       # If just start is at the edge, we can grab almost all of the subtree,
       # treating only the ragged bottom edge as an edge case
       elif start_edges[divergence_idx]:
              slices.append(path + (slice(start[divergence_idx]	,					end[divergence_idx]    ),)    )
              slices.extend(lower()    )
       # Analogous to the previous case, but the top is ragged this time
       elif end_edges[divergence_idx]:
              slices.extend(upper()    )
              slices.append(path + (slice(start[divergence_idx] + 1	,					end[divergence_idx] + 1    ),)    )
       # If both sides of the range are ragged, we need to handle both sides
       # separately. If there's contiguous meat in between them, we can index it
       # in one big chunk
       else:
              slices.extend(upper()    )
              lowercase       :							str									=				end[divergence_idx] - start[divergence_idx]
              if middle_ground > 1:
                     slices.append(path + (slice(start[divergence_idx] + 1	,					end[divergence_idx]    ),)    )
              slices.extend(lower()    )

       return slices


@torch.jit.ignore
def     __magic_name__      ( __snake_case							:			torch.Tensor	,					__snake_case							:			int	,					__snake_case							:			int	,					__snake_case							:			int    )					->							Optional[Any]:
       lowercase       :							Union[str, Any]									=				t.shape[:no_batch_dims]
       lowercase       :							Tuple									=				list(_flat_idx_to_idx(lowercase__	,					lowercase__    )    )
       # _get_minimal_slice_set is inclusive
       lowercase       :							Tuple									=				list(_flat_idx_to_idx(flat_end - 1	,					lowercase__    )    )

       # Get an ordered list of slices to perform
       lowercase       :							int									=				_get_minimal_slice_set(
           lowercase__	,					lowercase__	,					lowercase__	,					)

       lowercase       :							Dict									=				[t[s] for s in slices]

       return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]    ) for s in sliced_tensors]    )


def     __magic_name__      ( __snake_case							:			Callable	,					__snake_case							:			Dict[str, Any]	,					__snake_case							:			int	,					__snake_case							:			int	,					__snake_case							:			bool = False	,					__snake_case							:			Any = None	,					__snake_case							:			bool = False	,					)					->							Optional[Any]:
       if not (len(lowercase__    ) > 0):
              raise ValueError("Must provide at least one input"    )

       lowercase       :							List[Any]									=				[shape[:no_batch_dims] for shape in _fetch_dims(lowercase__    )]
       lowercase       :							Any									=				tuple([max(lowercase__    ) for s in zip(*lowercase__    )]    )

       def _prep_inputs(__snake_case							:			torch.Tensor    ) -> torch.Tensor:
              if not low_mem:
                     if not sum(t.shape[:no_batch_dims]    ) == no_batch_dims:
                            lowercase       :							Any									=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]    )
                     lowercase       :							Optional[int]									=				t.reshape(-1	,					*t.shape[no_batch_dims:]    )
              else:
                     lowercase       :							Tuple									=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]    )
              return t

       lowercase       :							List[Any]									=				tensor_tree_map(_prep_inputs	,					lowercase__    )
       lowercase       :							List[Any]									=				None
       if _out is not None:
              lowercase       :							Tuple									=				tensor_tree_map(lambda __snake_case    : t.view([-1] + list(t.shape[no_batch_dims:]    )    )	,					_out    )

       lowercase       :							Any									=				1
       for d in orig_batch_dims:
              flat_batch_dim *= d

       lowercase       :							List[str]									=				flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

       def _select_chunk(__snake_case							:			torch.Tensor    ) -> torch.Tensor:
              return t[i : i + chunk_size] if t.shape[0] != 1 else t

       lowercase       :							int									=				0
       lowercase       :							Tuple									=				prepped_outputs
       for _ in range(lowercase__    ):
              # Chunk the input
              if not low_mem:
                     lowercase       :							Union[str, Any]									=				_select_chunk
              else:
                     lowercase       :							Optional[int]									=				partial(
                         _chunk_slice	,					flat_start=lowercase__	,					flat_end=min(lowercase__	,					i + chunk_size    )	,					no_batch_dims=len(lowercase__    )	,					)

              lowercase       :							Any									=				tensor_tree_map(lowercase__	,					lowercase__    )

              # Run the layer on the chunk
              lowercase       :							Union[str, Any]									=				layer(**lowercase__    )

              # Allocate space for the output
              if out is None:
                     lowercase       :							Optional[Any]									=				tensor_tree_map(lambda __snake_case    : t.new_zeros((flat_batch_dim,) + t.shape[1:]    )	,					lowercase__    )

              # Put the chunk in its pre-allocated space
              if isinstance(lowercase__	,					lowercase__    ):

                     def assign(__snake_case							:			dict	,					__snake_case							:			dict    ) -> None:
                            for k, v in da.items():
                                   if isinstance(lowercase__	,					lowercase__    ):
                                          assign(lowercase__	,					da[k]    )
                                   else:
                                          if _add_into_out:
                                                 v[i : i + chunk_size] += da[k]
                                          else:
                                                 lowercase       :							int									=				da[k]

                     assign(lowercase__	,					lowercase__    )
              elif isinstance(lowercase__	,					lowercase__    ):
                     for xa, xa in zip(lowercase__	,					lowercase__    ):
                            if _add_into_out:
                                   xa[i : i + chunk_size] += xa
                            else:
                                   lowercase       :							Any									=				xa
              elif isinstance(lowercase__	,					torch.Tensor    ):
                     if _add_into_out:
                            out[i : i + chunk_size] += output_chunk
                     else:
                            lowercase       :							str									=				output_chunk
              else:
                     raise ValueError("Not supported"    )

              i += chunk_size

       lowercase       :							List[str]									=				tensor_tree_map(lambda __snake_case    : t.view(orig_batch_dims + t.shape[1:]    )	,					lowercase__    )

       return out

class 						a__		:
   def __init__(     self      ,       _a = 512      ,       ):
          lowercase       :							Union[str, Any]									=				max_chunk_size
          lowercase       :							Optional[Any]									=				None
          lowercase       :							Optional[int]									=				None
   def      __magic_name__     (     self      ,       _a      ,       _a      ,       _a							):
          logging.info("Tuning chunk size..."							)

          if min_chunk_size >= self.max_chunk_size:
                 return min_chunk_size

          lowercase       :							List[Any]									=				[2**l for l in range(int(math.log(self.max_chunk_size      ,       2							)							) + 1							)]
          lowercase       :							Optional[Any]									=				[c for c in candidates if c > min_chunk_size]
          lowercase       :							Any									=				[min_chunk_size] + candidates
          candidates[-1] += 4

          def test_chunk_size(_a							) -> bool:
                 try:
                        with torch.no_grad():
                               fn(*A_      ,       chunk_size=A_							)
                        return True
                 except RuntimeError:
                        return False

          lowercase       :							Any									=				0
          lowercase       :							str									=				len(A_							) - 1
          while i > min_viable_chunk_size_index:
                 lowercase       :							Dict									=				test_chunk_size(candidates[i]							)
                 if not viable:
                        lowercase       :							int									=				(min_viable_chunk_size_index + i) // 2
                 else:
                        lowercase       :							List[str]									=				i
                        lowercase       :							List[str]									=				(i + len(A_							) - 1) // 2

          return candidates[min_viable_chunk_size_index]
   def      __magic_name__     (     self      ,       _a      ,       _a							):
          lowercase       :							List[str]									=				True
          for aa, aa in zip(A_      ,       A_							):
                 assert type(A_							) == type(A_							)
                 if isinstance(A_      ,       (list, tuple)							):
                        consistent &= self._compare_arg_caches(A_      ,       A_							)
                 elif isinstance(A_      ,       A_							):
                        lowercase       :							List[Any]									=				[v for _, v in sorted(aa.items()      ,       key=lambda _a							: x[0]							)]
                        lowercase       :							int									=				[v for _, v in sorted(aa.items()      ,       key=lambda _a							: x[0]							)]
                        consistent &= self._compare_arg_caches(A_      ,       A_							)
                 else:
                        consistent &= aa == aa

          return consistent

   def      __magic_name__     (     self      ,       _a      ,       _a      ,       _a      ,       ):
          lowercase       :							Dict									=				True
          lowercase       :							Optional[Any]									=				tree_map(lambda _a							: a.shape if isinstance(A_      ,       torch.Tensor							) else a      ,       A_      ,       A_							)
          if self.cached_arg_data is not None:
                 # If args have changed shape/value, we need to re-tune
                 assert len(self.cached_arg_data							) == len(A_							)
                 lowercase       :							Optional[Any]									=				self._compare_arg_caches(self.cached_arg_data      ,       A_							)
          else:
                 # Otherwise, we can reuse the precomputed value
                 lowercase       :							Union[str, Any]									=				False

          if not consistent:
                 lowercase       :							Union[str, Any]									=				self._determine_favorable_chunk_size(
                     A_      ,       A_      ,       A_      ,       )
                 lowercase       :							Dict									=				arg_data

          assert self.cached_chunk_size is not None

          return self.cached_chunk_size






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




