




import baseaa
import io
import json
import os
from copy import deepcopy

from ..optimizer import AcceleratedOptimizer
from ..scheduler import AcceleratedScheduler




class    a__       :






       def __init__(   self:					Optional[int]       ,					_lowerCAmelCase:					Optional[int]				):
         if isinstance(_lowerCAmelCase       ,					_lowerCAmelCase				):
           # Don't modify user's data should they want to reuse it (e.g. in tests), because once we
           # modified it, it will not be accepted here again, since `auto` values would have been overridden
           lowercase    :int      	=						deepcopy(_lowerCAmelCase				)
         elif os.path.exists(_lowerCAmelCase				):
           with io.open(_lowerCAmelCase       ,					"r"       ,					encoding="utf-8"				) as f:
             lowercase    :Tuple      	=						json.load(_lowerCAmelCase				)
         else:
           try:
             lowercase    :str      	=						baseaa.urlsafe_baadecode(_lowerCAmelCase				).decode("utf-8"				)
             lowercase    :int      	=						json.loads(_lowerCAmelCase				)
           except (UnicodeDecodeError, AttributeError, ValueError):
             raise ValueError(
                 F"Expected a string path to an existing deepspeed config, or a dictionary, or a base64 encoded string. Received: {config_file_or_dict}"				)

         lowercase    :Optional[int]      	=						config

         self.set_stage_and_offload()






       def 					SCREAMING_SNAKE_CASE    (   self:					Any				):
         # zero stage - this is done as early as possible, before model is created, to allow
         # ``is_deepspeed_zero3_enabled`` query and getting to the early deepspeed config object
         # during ``zero.Init()`` which needs to know the dtype, and some other hparams.
         lowercase    :Optional[Any]      	=						self.get_value("zero_optimization.stage"       ,					-1				)

         # offload
         lowercase    :Optional[int]      	=						False
         if self.is_zeroa() or self.is_zeroa():
           lowercase    :Dict      	=						set(["cpu", "nvme"]				)
           lowercase    :str      	=						set(
               [
                   self.get_value("zero_optimization.offload_optimizer.device"				),
                   self.get_value("zero_optimization.offload_param.device"				),
               ]				)
           if len(offload_devices & offload_devices_valid				) > 0:
             lowercase    :Any      	=						True






       def 					SCREAMING_SNAKE_CASE    (   self:					str       ,					_lowerCAmelCase:					Dict				):
         lowercase    :List[Any]      	=						self.config

         # find the config node of interest if it exists
         lowercase    :int      	=						ds_key_long.split("."				)
         lowercase    :str      	=						nodes.pop()
         for node in nodes:
           lowercase    :Dict      	=						config.get(_lowerCAmelCase				)
           if config is None:
             return None, ds_key

         return config, ds_key






       def 					SCREAMING_SNAKE_CASE    (   self:					Tuple       ,					_lowerCAmelCase:					Optional[int]       ,					_lowerCAmelCase:					Tuple=None				):
         lowercase    :Dict      	=						self.find_config_node(_lowerCAmelCase				)
         if config is None:
           return default
         return config.get(_lowerCAmelCase       ,					_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					Tuple       ,					_lowerCAmelCase:					List[Any]       ,					_lowerCAmelCase:					Dict=False				):
         lowercase    :int      	=						self.config

         # find the config node of interest if it exists
         lowercase    :int      	=						ds_key_long.split("."				)
         for node in nodes:
           lowercase    :Dict      	=						config
           lowercase    :Union[str, Any]      	=						config.get(_lowerCAmelCase				)
           if config is None:
             if must_exist:
               raise ValueError(F"Can't find {ds_key_long} entry in the config: {self.config}"				)
             else:
               return

        # if found remove it
         if parent_config is not None:
           parent_config.pop(_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]       ,					_lowerCAmelCase:					Any				):
         lowercase    :str      	=						self.get_value(_lowerCAmelCase				)
         return False if value is None else bool(_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					Any       ,					_lowerCAmelCase:					List[Any]				):
         lowercase    :str      	=						self.get_value(_lowerCAmelCase				)
         return False if value is None else not bool(_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         return self._stage == 2






       def 					SCREAMING_SNAKE_CASE    (   self:					Union[str, Any]				):
         return self._stage == 3






       def 					SCREAMING_SNAKE_CASE    (   self:					int				):
         return self._offload







class    a__       :






       def __init__(   self:					Optional[Any]       ,					_lowerCAmelCase:					Tuple				):
         lowercase    :List[Any]      	=						engine






       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[Any]       ,					_lowerCAmelCase:					int       ,					**_lowerCAmelCase:					List[Any]				):
         # runs backpropagation and handles mixed precision
         self.engine.backward(_lowerCAmelCase       ,					**_lowerCAmelCase				)

         # Deepspeed's `engine.step` performs the following operations:
         # - gradient accumulation check
         # - gradient clipping
         # - optimizer step
         # - zero grad
         # - checking overflow
         # - lr_scheduler step (only if engine.lr_scheduler is not None)
         self.engine.step()
         # and this plugin overrides the above calls with no-ops when Accelerate runs under
         # Deepspeed, but allows normal functionality for non-Deepspeed cases thus enabling a simple
         # training loop that works transparently under many training regimes.







class    a__       (							lowerCAmelCase):






       def __init__(   self:					List[Any]       ,					_lowerCAmelCase:					Tuple				):
         super().__init__(_lowerCAmelCase       ,					device_placement=_lowerCAmelCase       ,					scaler=_lowerCAmelCase				)
         lowercase    :Optional[int]      	=						hasattr(self.optimizer       ,					"overflow"				)






       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[int]       ,					_lowerCAmelCase:					int=None				):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed






       def 					SCREAMING_SNAKE_CASE    (   self:					Dict				):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Dict				):
         if self.__has_overflow__:
           return self.optimizer.overflow
         return False







class    a__       (							lowerCAmelCase):






       def __init__(   self:					Dict       ,					_lowerCAmelCase:					Optional[int]       ,					_lowerCAmelCase:					List[str]				):
         super().__init__(_lowerCAmelCase       ,					_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed







class    a__       :






       def __init__(   self:					int       ,					_lowerCAmelCase:					List[str]       ,					_lowerCAmelCase:					Tuple=0.0_01       ,					_lowerCAmelCase:					str=0       ,					**_lowerCAmelCase:					int				):
         lowercase    :Dict      	=						params
         lowercase    :Optional[int]      	=						lr
         lowercase    :Any      	=						weight_decay
         lowercase    :Optional[Any]      	=						kwargs







class    a__       :






       def __init__(   self:					List[Any]       ,					_lowerCAmelCase:					Dict       ,					_lowerCAmelCase:					Any=None       ,					_lowerCAmelCase:					Any=0       ,					**_lowerCAmelCase:					List[str]				):
         lowercase    :Optional[int]      	=						optimizer
         lowercase    :List[Any]      	=						total_num_steps
         lowercase    :Union[str, Any]      	=						warmup_num_steps
         lowercase    :Dict      	=						kwargs









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       (							lowerCAmelCase):
       _a 					=						['''image_processor''', '''tokenizer''']
       _a 					=						'''LayoutLMv2ImageProcessor'''
       _a 					=						('''LayoutXLMTokenizer''', '''LayoutXLMTokenizerFast''')






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

         lowercase    :Union[str, 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__(_lowerCAmelCase       ,					_lowerCAmelCase				)






       def __call__(   self:					List[str]       ,					_lowerCAmelCase:					Optional[Any]       ,					_lowerCAmelCase:					Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None       ,					_lowerCAmelCase:					Optional[Union[PreTokenizedInput, List[PreTokenizedInput]]] = None       ,					_lowerCAmelCase:					Union[List[List[int]], List[List[List[int]]]] = None       ,					_lowerCAmelCase:					Optional[Union[List[int], List[List[int]]]] = None       ,					_lowerCAmelCase:					bool = True       ,					_lowerCAmelCase:					Union[bool, str, PaddingStrategy] = False       ,					_lowerCAmelCase:					Union[bool, str, TruncationStrategy] = None       ,					_lowerCAmelCase:					Optional[int] = None       ,					_lowerCAmelCase:					int = 0       ,					_lowerCAmelCase:					Optional[int] = None       ,					_lowerCAmelCase:					Optional[bool] = None       ,					_lowerCAmelCase:					Optional[bool] = None       ,					_lowerCAmelCase:					bool = False       ,					_lowerCAmelCase:					bool = False       ,					_lowerCAmelCase:					bool = False       ,					_lowerCAmelCase:					bool = False       ,					_lowerCAmelCase:					bool = True       ,					_lowerCAmelCase:					Optional[Union[str, TensorType]] = None       ,					**_lowerCAmelCase:					Tuple       ,					):
         # 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
         lowercase    :int      	=						self.image_processor(images=_lowerCAmelCase       ,					return_tensors=_lowerCAmelCase				)

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

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

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

         return encoded_inputs






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

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

         return images_with_overflow






       def 					SCREAMING_SNAKE_CASE    (   self:					Dict       ,					*_lowerCAmelCase:					str       ,					**_lowerCAmelCase:					List[Any]				):
         return self.tokenizer.batch_decode(*_lowerCAmelCase       ,					**_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					str       ,					*_lowerCAmelCase:					Tuple       ,					**_lowerCAmelCase:					Union[str, Any]				):
         return self.tokenizer.decode(*_lowerCAmelCase       ,					**_lowerCAmelCase				)






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[Any]				):
         return ["input_ids", "bbox", "attention_mask", "image"]






       @property
       def 					SCREAMING_SNAKE_CASE    (   self:					List[str]				):
         warnings.warn(
             "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."       ,					_lowerCAmelCase       ,					)
         return self.image_processor_class






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



