


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 						A__							(		UpperCamelCase__  ):
						UpperCAmelCase             =						["image_processor", "tokenizer"]
						UpperCAmelCase             =						"LayoutLMv2ImageProcessor"
						UpperCAmelCase             =						("LayoutXLMTokenizer", "LayoutXLMTokenizerFast")




						def __init__(       self					:      Any				,    _a					:      Optional[Any]=None				,    _a					:      Optional[Any]=None				,    **_a					:      Any  )							->     Tuple:


										"""simple docstring"""





										if "feature_extractor" in kwargs:
														warnings.warn(
														    '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
														    ''' instead.'''				,    _a				,    )
														_SCREAMING_SNAKE_CASE			  =kwargs.pop('''feature_extractor'''  )

										_SCREAMING_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					:      Any				,    _a					:      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					:      Any				,    )							->     BatchEncoding:


										"""simple docstring"""





										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
										_SCREAMING_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  ):
																		_SCREAMING_SNAKE_CASE			  =[text]  # add batch dimension (as the image processor always adds a batch dimension)
														_SCREAMING_SNAKE_CASE			  =features['''words''']

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

										return encoded_inputs




						def 							__UpperCamelCase				(       self					:      Union[str, Any]				,    _a					:      Optional[Any]				,    _a					:      List[str]  )							->     Tuple:


										"""simple docstring"""





										_SCREAMING_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					:      Any  )							->     Dict:


										"""simple docstring"""





										return self.tokenizer.batch_decode(*_a				,    **_a  )




						def 							__UpperCamelCase				(       self					:      Union[str, Any]				,    *_a					:      int				,    **_a					:      List[Any]  )							->     Tuple:


										"""simple docstring"""





										return self.tokenizer.decode(*_a				,    **_a  )




						@property
						def 							__UpperCamelCase				(       self					:      Optional[Any]  )							->     List[Any]:


										"""simple docstring"""





										return ["input_ids", "bbox", "attention_mask", "image"]




						@property
						def 							__UpperCamelCase				(       self					:      int  )							->     Union[str, Any]:


										"""simple docstring"""





										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					:      List[str]  )							->     Optional[int]:


										"""simple docstring"""





										warnings.warn(
										    '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''				,    _a				,    )
										return self.image_processor



import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment







def 						lowerCamelCase(   a__	,a__=()	,a__=None	,a__="no"	,a__="29500"):
				_SCREAMING_SNAKE_CASE			  =False
				_SCREAMING_SNAKE_CASE			  =False
				if any(key.startswith('''KAGGLE''') for key in os.environ.keys()):
								_SCREAMING_SNAKE_CASE			  =True
				elif "IPython" in sys.modules:
								_SCREAMING_SNAKE_CASE			  ='''google.colab''' in str(sys.modules['''IPython'''].get_ipython())

				try:
								_SCREAMING_SNAKE_CASE			  =PrecisionType(mixed_precision.lower())
				except ValueError:
								raise ValueError(
								    f"Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.")

				if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''	,a__) is not None):
								# TPU launch
								import torch_xla.distributed.xla_multiprocessing as xmp

								if len(AcceleratorState._shared_state) > 0:
												raise ValueError(
												    '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
												    '''your training function. Restart your notebook and make sure no cells initializes an '''
												    '''`Accelerator`.''')
								if num_processes is None:
												_SCREAMING_SNAKE_CASE			  =8

								_SCREAMING_SNAKE_CASE			  =PrepareForLaunch(a__	,distributed_type='''TPU''')
								print(f"Launching a training on {num_processes} TPU cores.")
								xmp.spawn(a__	,args=a__	,nprocs=a__	,start_method='''fork''')
				elif in_colab:
								# No need for a distributed launch otherwise as it's either CPU or one GPU.
								if torch.cuda.is_available():
												print('''Launching training on one GPU.''')
								else:
												print('''Launching training on one CPU.''')
								function(*a__)
				else:
								if num_processes is None:
												raise ValueError(
												    '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.''')

								if num_processes > 1:
												# Multi-GPU launch
												from torch.multiprocessing import start_processes
												from torch.multiprocessing.spawn import ProcessRaisedException

												if len(AcceleratorState._shared_state) > 0:
																raise ValueError(
																    '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
																    '''inside your training function. Restart your notebook and make sure no cells initializes an '''
																    '''`Accelerator`.''')

												if torch.cuda.is_initialized():
																raise ValueError(
																    '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
																    '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
																    '''function.''')

												# torch.distributed will expect a few environment variable to be here. We set the ones common to each
												# process here (the other ones will be set be the launcher).
												with patch_environment(
												    world_size=a__	,master_addr='''127.0.01'''	,master_port=a__	,mixed_precision=a__):
																_SCREAMING_SNAKE_CASE			  =PrepareForLaunch(a__	,distributed_type='''MULTI_GPU''')
																print(f"Launching training on {num_processes} GPUs.")
																try:
																				start_processes(a__	,args=a__	,nprocs=a__	,start_method='''fork''')
																except ProcessRaisedException as e:
																				if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																								raise RuntimeError(
																								    '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
																								    '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
																								    '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
																								    '''which one is problematic.''') from e

								else:
												# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
												if is_mps_available():
																_SCREAMING_SNAKE_CASE			  ='''1'''
																print('''Launching training on MPS.''')
												elif torch.cuda.is_available():
																print('''Launching training on one GPU.''')
												else:
																print('''Launching training on CPU.''')
												function(*a__)







def 						lowerCamelCase(   a__	,a__=()	,a__=2):
				from torch.multiprocessing import start_processes

				with tempfile.NamedTemporaryFile() as tmp_file:
								# torch.distributed will expect a few environment variable to be here. We set the ones common to each
								# process here (the other ones will be set be the launcher).
								with patch_environment(
								    world_size=a__	,master_addr='''127.0.01'''	,master_port='''29500'''	,accelerate_mixed_precision='''no'''	,accelerate_debug_rdv_file=tmp_file.name	,accelerate_use_cpu='''yes'''	,):
												_SCREAMING_SNAKE_CASE			  =PrepareForLaunch(a__	,debug=a__)
												start_processes(a__	,args=a__	,nprocs=a__	,start_method='''fork''')