





import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
	import torch

	from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
	import pyanvml.pyanvml as nvml


_lowerCamelCase       =	logging.get_logger(__name__)




class      UpperCamelCase_				(   UpperCamelCase__			):
							lowerCamelCase_							=    42
							lowerCamelCase_							=    42
							lowerCamelCase_							=    "PyTorch"


							@property
							def    _snake_case     (							self			:Any						)      ->      str:



									"""simple docstring"""



									return torch.__version__


							def    _snake_case     (							self			:int							, __A			:str							, __A			:int							, __A			:int						)      ->      float:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=					self._prepare_inference_func(__A							, __A							, __A						)
									return self._measure_speed(_inference						)


							def    _snake_case     (							self			:Tuple							, __A			:str							, __A			:int							, __A			:int						)      ->      [Memory, Optional[MemorySummary]]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=					self._prepare_inference_func(__A							, __A							, __A						)
									return self._measure_memory(_inference						)


							def    _snake_case     (							self			:Dict							, __A			:str							, __A			:int							, __A			:int						)      ->      float:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=					self._prepare_train_func(__A							, __A							, __A						)
									return self._measure_speed(_train						)


							def    _snake_case     (							self			:str							, __A			:str							, __A			:int							, __A			:int						)      ->      [Memory, Optional[MemorySummary]]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=					self._prepare_train_func(__A							, __A							, __A						)
									return self._measure_memory(_train						)


							def    _snake_case     (							self			:Optional[Any]							, __A			:str							, __A			:int							, __A			:int						)      ->      Callable[[], None]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=					self.config_dict[model_name]

									if self.args.torchscript:
											SCREAMING_SNAKE_CASE__								=					True

									SCREAMING_SNAKE_CASE__								=					(
									    hasattr(__A							, """architectures"""						)
									    and isinstance(config.architectures							, __A						)
									    and len(config.architectures						) > 0
									)
									if not self.args.only_pretrain_model and has_model_class_in_config:
											try:
													SCREAMING_SNAKE_CASE__								=					config.architectures[0]
													SCREAMING_SNAKE_CASE__								=					__import__("""transformers"""							, fromlist=[model_class]						)
													SCREAMING_SNAKE_CASE__								=					getattr(__A							, __A						)
													SCREAMING_SNAKE_CASE__								=					model_cls(__A						)
											except ImportError:
													raise ImportError(
													    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
													    """ set `--only_pretrain_model` or `args.only_pretrain_model=True`."""						)
									else:
											SCREAMING_SNAKE_CASE__								=					MODEL_MAPPING[config.__class__](__A						)

									model.eval()
									model.to(self.args.device						)

									# encoder-decoder has vocab size saved differently
									SCREAMING_SNAKE_CASE__								=					config.vocab_size if hasattr(__A							, """vocab_size"""						) else config.encoder.vocab_size
									SCREAMING_SNAKE_CASE__								=					torch.randint(__A							, (batch_size, sequence_length)							, dtype=torch.long							, device=self.args.device						)

									if self.args.fpaa:
											logger.info("""Running training in Mixed Precision..."""						)
											if not self.args.is_gpu:
													raise ValueError("""Mixed precision is possible only for GPU."""						)
											# amp seems to have memory leaks so that memory usage
											# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
											model.half()

									if self.args.torchscript:
											with torch.no_grad():
													SCREAMING_SNAKE_CASE__								=					torch.jit.trace(__A							, __A						)
									else:
											SCREAMING_SNAKE_CASE__								=					model

									def encoder_decoder_forward():
											with torch.no_grad():
													SCREAMING_SNAKE_CASE__								=					inference_model(__A							, decoder_input_ids=__A						)
											return outputs

									def encoder_forward():
											with torch.no_grad():
													SCREAMING_SNAKE_CASE__								=					inference_model(__A						)
											return outputs

									SCREAMING_SNAKE_CASE__								=					encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
									return _forward


							def    _snake_case     (							self			:Union[str, Any]							, __A			:str							, __A			:int							, __A			:int						)      ->      Callable[[], None]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=					self.config_dict[model_name]

									SCREAMING_SNAKE_CASE__								=					(
									    hasattr(__A							, """architectures"""						)
									    and isinstance(config.architectures							, __A						)
									    and len(config.architectures						) > 0
									)
									if not self.args.only_pretrain_model and has_model_class_in_config:
											try:
													SCREAMING_SNAKE_CASE__								=					config.architectures[0]
													SCREAMING_SNAKE_CASE__								=					__import__("""transformers"""							, fromlist=[model_class]						)
													SCREAMING_SNAKE_CASE__								=					getattr(__A							, __A						)
													SCREAMING_SNAKE_CASE__								=					model_cls(__A						)
											except ImportError:
													raise ImportError(
													    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
													    """ set `--only_pretrain_model` or `args.only_pretrain_model=True`."""						)
									else:
											SCREAMING_SNAKE_CASE__								=					MODEL_WITH_LM_HEAD_MAPPING[config.__class__](__A						)

									if self.args.torchscript:
											raise NotImplementedError("""Training for torchscript is currently not implemented"""						)
									else:
											SCREAMING_SNAKE_CASE__								=					model

									model.train()
									model.to(self.args.device						)

									# encoder-decoder has vocab size saved differently
									SCREAMING_SNAKE_CASE__								=					config.vocab_size if hasattr(__A							, """vocab_size"""						) else config.encoder.vocab_size
									SCREAMING_SNAKE_CASE__								=					torch.randint(__A							, (batch_size, sequence_length)							, dtype=torch.long							, device=self.args.device						)

									if self.args.fpaa:
											logger.info("""Running training in Mixed Precision..."""						)
											if not self.args.is_gpu:
													raise ValueError("""Mixed precision is possible only for GPU."""						)

											# amp seems to have memory leaks so that memory usage
											# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
											model.half()

									def compute_loss_and_backprob_encoder():
											SCREAMING_SNAKE_CASE__								=					train_model(__A							, labels=__A						)[0]
											loss.backward()
											return loss

									def compute_loss_and_backprob_encoder_decoder():
											SCREAMING_SNAKE_CASE__								=					train_model(__A							, decoder_input_ids=__A							, labels=__A						)[0]
											loss.backward()
											return loss

									SCREAMING_SNAKE_CASE__								=					(
									    compute_loss_and_backprob_encoder_decoder
									    if config.is_encoder_decoder
									    else compute_loss_and_backprob_encoder
									)
									return _train


							def    _snake_case     (							self			:List[Any]							, __A			:Optional[int]						)      ->      float:



									"""simple docstring"""



									try:
											if self.args.is_tpu or self.args.torchscript:
													# run additional 10 times to stabilize compilation for tpu and torchscript
													logger.info("""Do inference on TPU or torchscript. Running model 5 times to stabilize compilation"""						)
													timeit.repeat(
													    __A							, repeat=1							, number=5							, )

											# as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
											SCREAMING_SNAKE_CASE__								=					timeit.repeat(
											    __A							, repeat=self.args.repeat							, number=10							, )

											if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
													import torch_xla.debug.metrics as met

													self.print_fn(met.metrics_report()						)

											return min(__A						) / 1_0.0
									except RuntimeError as e:
											self.print_fn(f'''Doesn\'t fit on GPU. {e}'''						)
											return "N/A"


							def    _snake_case     (							self			:List[Any]							, __A			:Callable[[], None]						)      ->      [Memory, MemorySummary]:



									"""simple docstring"""



									try:
											if self.args.trace_memory_line_by_line:
													SCREAMING_SNAKE_CASE__								=					start_memory_tracing("""transformers"""						)

											if self.args.is_tpu:
													# tpu
													raise NotImplementedError(
													    """Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with"""
													    """ `--no-memory` or `args.memory=False`"""						)
											elif self.args.is_gpu:
													if not is_pyanvml_available():
															logger.warning(
															    """py3nvml not installed, we won't log GPU memory usage. """
															    """Install py3nvml (pip install py3nvml) to log information about GPU."""						)
															SCREAMING_SNAKE_CASE__								=					"""N/A"""
													else:
															logger.info(
															    """Measuring total GPU usage on GPU device. Make sure to not have additional processes running"""
															    """ on the same GPU."""						)
															# init nvml
															nvml.nvmlInit()
															func()
															SCREAMING_SNAKE_CASE__								=					nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx						)
															SCREAMING_SNAKE_CASE__								=					nvml.nvmlDeviceGetMemoryInfo(__A						)
															SCREAMING_SNAKE_CASE__								=					meminfo.used
															SCREAMING_SNAKE_CASE__								=					Memory(__A						)
															# shutdown nvml
															nvml.nvmlShutdown()
											else:
													# cpu
													SCREAMING_SNAKE_CASE__								=					measure_peak_memory_cpu(__A						)
													SCREAMING_SNAKE_CASE__								=					Memory(__A						) if isinstance(__A							, __A						) else memory_bytes

											if self.args.trace_memory_line_by_line:
													SCREAMING_SNAKE_CASE__								=					stop_memory_tracing(__A						)
											else:
													SCREAMING_SNAKE_CASE__								=					None

											return memory, summary
									except RuntimeError as e:
											self.print_fn(f'''Doesn\'t fit on GPU. {e}'''						)
											return "N/A", None






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			(	UpperCAmelCase_							):


  '''simple docstring'''
  a_							:				List[str]					    =["""image_processor""", """tokenizer"""]
  a_							:				Optional[Any]					    ="""LayoutLMv2ImageProcessor"""
  a_							:				Any					    =("""LayoutLMv2Tokenizer""", """LayoutLMv2TokenizerFast""")







  def __init__(					self					:    Union[str, Any]					, UpperCamelCase					:    str=None					, UpperCamelCase					:    List[Any]=None					, **UpperCamelCase					:    List[str]							):


    '''simple docstring'''





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

    _snake_case      : Optional[Any]														=		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__(UpperCamelCase					, UpperCamelCase							)







  def __call__(					self					:    int					, UpperCamelCase					:    Tuple					, UpperCamelCase					:    Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None					, UpperCamelCase					:    Optional[Union[PreTokenizedInput, List[PreTokenizedInput]]] = None					, UpperCamelCase					:    Union[List[List[int]], List[List[List[int]]]] = None					, UpperCamelCase					:    Optional[Union[List[int], List[List[int]]]] = None					, UpperCamelCase					:    bool = True					, UpperCamelCase					:    Union[bool, str, PaddingStrategy] = False					, UpperCamelCase					:    Union[bool, str, TruncationStrategy] = False					, UpperCamelCase					:    Optional[int] = None					, UpperCamelCase					:    int = 0					, UpperCamelCase					:    Optional[int] = None					, UpperCamelCase					:    Optional[bool] = None					, UpperCamelCase					:    Optional[bool] = None					, UpperCamelCase					:    bool = False					, UpperCamelCase					:    bool = False					, UpperCamelCase					:    bool = False					, UpperCamelCase					:    bool = False					, UpperCamelCase					:    bool = True					, UpperCamelCase					:    Optional[Union[str, TensorType]] = None					, **UpperCamelCase					:    Optional[Any]					, ):


    '''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
    _snake_case      : Optional[Any]														=		self.image_processor(images=UpperCamelCase					, return_tensors=UpperCamelCase							)

    # second, apply the tokenizer
    if text is not None and self.image_processor.apply_ocr and text_pair is None:
      if isinstance(UpperCamelCase					, UpperCamelCase							):
        _snake_case      : List[Any]														=		[text]  # add batch dimension (as the image processor always adds a batch dimension)
      _snake_case      : Dict														=		features['words']

    _snake_case      : Optional[int]														=		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=UpperCamelCase					, add_special_tokens=UpperCamelCase					, padding=UpperCamelCase					, truncation=UpperCamelCase					, max_length=UpperCamelCase					, stride=UpperCamelCase					, pad_to_multiple_of=UpperCamelCase					, return_token_type_ids=UpperCamelCase					, return_attention_mask=UpperCamelCase					, return_overflowing_tokens=UpperCamelCase					, return_special_tokens_mask=UpperCamelCase					, return_offsets_mapping=UpperCamelCase					, return_length=UpperCamelCase					, verbose=UpperCamelCase					, return_tensors=UpperCamelCase					, **UpperCamelCase					, )

    # add pixel values
    _snake_case      : Optional[int]														=		features.pop('pixel_values'							)
    if return_overflowing_tokens is True:
      _snake_case      : Any														=		self.get_overflowing_images(UpperCamelCase					, encoded_inputs['overflow_to_sample_mapping']							)
    _snake_case      : int														=		images

    return encoded_inputs







  def UpperCamelCase_       (					self					:    List[Any]					, UpperCamelCase					:    List[str]					, UpperCamelCase					:    List[str]							):


    '''simple docstring'''





    _snake_case      : Optional[Any]														=		[]
    for sample_idx in overflow_to_sample_mapping:
      images_with_overflow.append(images[sample_idx]							)

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

    return images_with_overflow







  def UpperCamelCase_       (					self					:    List[Any]					, *UpperCamelCase					:    Any					, **UpperCamelCase					:    int							):


    '''simple docstring'''





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







  def UpperCamelCase_       (					self					:    Any					, *UpperCamelCase					:    Optional[int]					, **UpperCamelCase					:    Any							):


    '''simple docstring'''





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







  @property
  def UpperCamelCase_       (					self					:    str							):


    '''simple docstring'''





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







  @property
  def UpperCamelCase_       (					self					:    Dict							):


    '''simple docstring'''





    warnings.warn(
        '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'					, UpperCamelCase					, )
    return self.image_processor_class







  @property
  def UpperCamelCase_       (					self					:    Tuple							):


    '''simple docstring'''





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




