'''simple docstring'''




import importlib.util
import weakref
from functools import partialmethod

from .dependency_versions_check import dep_version_check
from .utils import is_accelerate_available, is_torch_available, logging


if is_torch_available():
						import torch

_lowercase		:					Optional[int]													=		logging.get_logger(__name__)




def 				lowerCamelCase      ( )     -> Dict:
							return importlib.util.find_spec("""deepspeed"""					) is not None


if is_accelerate_available() and is_deepspeed_available():
						from accelerate.utils.deepspeed import HfDeepSpeedConfig as DeepSpeedConfig
else:
						# Inherits from a dummy `object` if accelerate is not available, so that python succeeds to import this file.
						# Deepspeed glue code will never inherit this dummy object as it checks if accelerate is available.
						from builtins import object as DeepSpeedConfig





class        __magic_name__		(       _UpperCAmelCase):







			def __init__(      self							:    Dict							,     lowercase_							:    List[Any]       ):
										# set global weakref object
										set_hf_deepspeed_config(self       )
										dep_version_check("""accelerate"""       )
										dep_version_check("""deepspeed"""       )
										super().__init__(lowercase_       )





class        __magic_name__		(       _UpperCAmelCase):







			def __init__(      self							:    int							,     lowercase_							:    List[Any]       ):
										super().__init__(lowercase_       )
										lowercase_   :     Tuple									=			None
										lowercase_   :     List[str]									=			[]







			def       SCREAMING_SNAKE_CASE_	(      self							:    List[Any]       ):
										if self._dtype is None:
																	raise ValueError("""trainer_config_process() wasn't called yet to tell dtype"""       )
										return self._dtype







			def       SCREAMING_SNAKE_CASE_	(      self							:    Tuple							,     lowercase_							:    Optional[Any]       ):
										lowercase_   :     Tuple									=			self.get_value(lowercase_       )
										if val is None:
																	return False
										else:
																	return val == "auto"







			def       SCREAMING_SNAKE_CASE_	(      self							:    Optional[int]							,     lowercase_							:    int							,     lowercase_							:    Optional[Any]							,     lowercase_							:    Dict=None							,     lowercase_							:    Tuple=True       ):
										lowercase_		,		lowercase_   :     Dict									=			self.find_config_node(lowercase_       )
										if config is None:
																	return

										if config.get(lowercase_       ) == "auto":
																	lowercase_   :     List[str]									=			hf_val
																	return

										if not must_match:
																	return

										lowercase_   :     str									=			config.get(lowercase_       )
										if ds_val is not None and ds_val != hf_val:
																	self.mismatches.append(f'''- ds {ds_key_long}={ds_val} vs hf {hf_key}={hf_val}'''       )

			UpperCamelCase__							=			partialmethod(_UpperCAmelCase,	must_match=_UpperCAmelCase)







			def       SCREAMING_SNAKE_CASE_	(      self							:    Union[str, Any]							,     lowercase_							:    Optional[Any]       ):
										lowercase_   :     str									=			args.world_size * args.per_device_train_batch_size * args.gradient_accumulation_steps
										self.fill_match(
										    """train_micro_batch_size_per_gpu"""							,     args.per_device_train_batch_size							,     """per_device_train_batch_size"""       )
										self.fill_match("""gradient_accumulation_steps"""							,     args.gradient_accumulation_steps							,     """gradient_accumulation_steps"""       )
										self.fill_match("""train_batch_size"""							,     lowercase_							,     """train_batch_size (calculated)"""       )
										self.fill_match("""gradient_clipping"""							,     args.max_grad_norm							,     """max_grad_norm"""       )

										self.fill_match("""optimizer.params.lr"""							,     args.learning_rate							,     """learning_rate"""       )
										self.fill_match("""optimizer.params.betas"""							,     [args.adam_betaa, args.adam_betaa]							,     """adam_beta1+adam_beta2"""       )
										self.fill_match("""optimizer.params.eps"""							,     args.adam_epsilon							,     """adam_epsilon"""       )
										self.fill_match("""optimizer.params.weight_decay"""							,     args.weight_decay							,     """weight_decay"""       )

										self.fill_only("""scheduler.params.warmup_min_lr"""							,     0       )  # not a trainer arg
										self.fill_match("""scheduler.params.warmup_max_lr"""							,     args.learning_rate							,     """learning_rate"""       )
										# total_num_steps - will get set in trainer_config_finalize

										# fp16
										if args.fpaa or args.fpaa_full_eval:
																	lowercase_   :     List[Any]									=			"""apex""" if args.fpaa_backend == """apex""" else """amp"""
										else:
																	lowercase_   :     Tuple									=			None

										if args.save_on_each_node:
																	# deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
																	lowercase_   :     Optional[int]									=			self.config.get("""checkpoint"""							,     {}       )
																	lowercase_   :     Dict									=			args.save_on_each_node

										# amp: similar to the pytorch native amp - it has a bunch of optional params but we won't set
										# any here unless the user did the work
										self.fill_match(
										    """fp16.enabled"""							,     ((args.fpaa or args.fpaa_full_eval) and fpaa_backend == """amp""")							,     """fp16|fp16_full_eval+fp16_backend(amp)"""							,     )

										# apex: delegates amp work to apex (which needs to be available), but it cannot be used with any
										# ZeRO features
										self.fill_match("""amp.enabled"""							,     fpaa_backend == """apex"""							,     """fp16+fp16_backend(apex)"""       )
										self.fill_match("""amp.opt_level"""							,     args.fpaa_opt_level							,     """fp16_opt_level"""       )

										self.fill_match("""bf16.enabled"""							,     (args.bfaa or args.bfaa_full_eval)							,     """bf16|bf16_full_eval"""       )

										# deepspeed's default mode is fp16 unless there is a config that says differently
										if self.is_true("""bf16.enabled"""       ):
																	lowercase_   :     Optional[Any]									=			torch.bfloataa
										elif self.is_false("""fp16.enabled"""       ):
																	lowercase_   :     Tuple									=			torch.floataa
										else:
																	lowercase_   :     int									=			torch.floataa







			def       SCREAMING_SNAKE_CASE_	(      self							:    Union[str, Any]							,     lowercase_							:    Optional[Any]							,     lowercase_							:    Any							,     lowercase_							:    List[str]       ):
										lowercase_   :     Dict									=			[
										    """zero_optimization.reduce_bucket_size""",
										    """zero_optimization.stage3_prefetch_bucket_size""",
										    """zero_optimization.stage3_param_persistence_threshold""",
										]
										lowercase_   :     int									=			[x for x in hidden_size_based_keys if self.is_auto(lowercase_       )]

										if len(lowercase_       ) > 0:
																	if hasattr(model.config							,     """hidden_size"""       ):
																								lowercase_   :     Dict									=			model.config.hidden_size
																	elif hasattr(model.config							,     """hidden_sizes"""       ):
																								# if there are many hidden sizes pick the largest one
																								lowercase_   :     Dict									=			max(model.config.hidden_sizes       )
																	else:
																								raise ValueError(
																								    """The model's config file has neither `hidden_size` nor `hidden_sizes` entry, """
																								    """therefore it's not possible to automatically fill out the following `auto` entries """
																								    f'''in the DeepSpeed config file: {hidden_size_auto_keys}. You can fix that by replacing '''
																								    """`auto` values for these keys with an integer value of your choice."""       )

																	self.fill_only("""zero_optimization.reduce_bucket_size"""							,     hidden_size * hidden_size       )
																	if self.is_zeroa():
																								# automatically assign the optimal config values based on model config
																								self.fill_only("""zero_optimization.stage3_prefetch_bucket_size"""							,     0.9 * hidden_size * hidden_size       )
																								self.fill_only("""zero_optimization.stage3_param_persistence_threshold"""							,     10 * hidden_size       )

        # scheduler
										self.fill_match("""scheduler.params.total_num_steps"""							,     lowercase_							,     """num_training_steps (calculated)"""       )
										self.fill_match("""scheduler.params.warmup_num_steps"""							,     args.get_warmup_steps(lowercase_       )							,     """warmup_steps"""       )

										if len(self.mismatches       ) > 0:
																	lowercase_   :     int									=			"""\n""".join(self.mismatches       )
																	raise ValueError(
																	    """Please correct the following DeepSpeed config values that mismatch TrainingArguments"""
																	    f''' values:\n{mismatches}\nThe easiest method is to set these DeepSpeed config values to \'auto\'.'''       )




_lowercase		:					int													=		None




def 				lowerCamelCase      ( UpperCAmelCase__			:   Tuple					)     -> Any:
							# this is a special weakref global object to allow us to get to Deepspeed config from APIs
							# that don't have an easy way to get to the Deepspeed config outside of the Trainer domain.
							global _hf_deepspeed_config_weak_ref
							# will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
							lowercase_   :     List[Any]									=			weakref.ref(UpperCAmelCase__					)




def 				lowerCamelCase      ( )     -> Optional[int]:
							# useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
							global _hf_deepspeed_config_weak_ref
							lowercase_   :     List[Any]									=			None




def 				lowerCamelCase      ( )     -> Any:
							if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
														return _hf_deepspeed_config_weak_ref().is_zeroa()
							else:
														return False




def 				lowerCamelCase      ( )     -> Tuple:
							if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
														return _hf_deepspeed_config_weak_ref().config
							else:
														return None




def 				lowerCamelCase      ( UpperCAmelCase__			:   Dict , UpperCAmelCase__			:   Optional[int] , UpperCAmelCase__			:   Tuple , UpperCAmelCase__			:   Optional[Any] , UpperCAmelCase__			:   Optional[Any]					)     -> Any:
							from accelerate.utils import DummyOptim, DummyScheduler

							lowercase_   :     Any									=			hf_deepspeed_config.config

							# Optimizer + Scheduler
							# Currently supported combos:
							# 1. DS scheduler + DS optimizer: Yes
							# 2. HF scheduler + HF optimizer: Yes
							# 3. DS scheduler + HF optimizer: Yes
							# 4. HF scheduler + DS optimizer: No
							#
							# Unless Offload is enabled in which case it's:
							# 1. DS scheduler + DS optimizer: Yes
							# 2. HF scheduler + HF optimizer: Mostly*
							# 3. DS scheduler + HF optimizer: Mostly*
							# 4. HF scheduler + DS optimizer: No
							#
							# Mostly*: All non-native DeepSpeed optimizers that have both CPU and GPU implementation should work (except LAMB)

							lowercase_   :     int									=			None
							if "optimizer" in config:
														if args.adafactor:
																					raise ValueError(
																					    """--adafactor was passed, but also found `optimizer` configured in the DeepSpeed config. """
																					    """Only one optimizer can be configured."""					)
														lowercase_   :     List[Any]									=			DummyOptim(params=UpperCAmelCase__					)
							else:
														if hf_deepspeed_config.is_offload():
																					logger.info(
																					    """Detected ZeRO Offload and non-DeepSpeed optimizers: This combination should work as long as the"""
																					    """ custom optimizer has both CPU and GPU implementation (except LAMB)"""					)

														# ds supports Adam, OneBitAdam, and Lamb optimizers and can import other optimizers from torch.
														# But trainer uses AdamW by default.
														lowercase_   :     Tuple									=			trainer.create_optimizer()
														# To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
														lowercase_   :     Any									=			True

							lowercase_   :     str									=			None
							if "scheduler" in config:
														lowercase_   :     Any									=			DummyScheduler(UpperCAmelCase__					)
							else:
														if isinstance(UpperCAmelCase__ , UpperCAmelCase__					):
																					raise ValueError(
																					    """Found `optimizer` configured in the DeepSpeed config, but no `scheduler`. """
																					    """Please configure a scheduler in the DeepSpeed config."""					)
														lowercase_   :     Union[str, Any]									=			trainer.create_scheduler(num_training_steps=UpperCAmelCase__ , optimizer=UpperCAmelCase__					)

							return optimizer, lr_scheduler




def 				lowerCamelCase      ( UpperCAmelCase__			:   Union[str, Any] , UpperCAmelCase__			:   Tuple , UpperCAmelCase__			:   int=False					)     -> int:
							from deepspeed.utils import logger as ds_logger

							lowercase_   :     Dict									=			trainer.model
							lowercase_   :     Dict									=			trainer.args

							lowercase_   :     Union[str, Any]									=			trainer.accelerator.state.deepspeed_plugin.hf_ds_config

							# resume config update - some bits like `model` and `num_training_steps` only become available during train
							hf_deepspeed_config.trainer_config_finalize(UpperCAmelCase__ , UpperCAmelCase__ , UpperCAmelCase__					)

							# set the Deepspeed log level consistent with the Trainer
							ds_logger.setLevel(args.get_process_log_level()					)

							if inference:
														# only Z3 makes sense for the inference
														if not hf_deepspeed_config.is_zeroa():
																					raise ValueError("""ZeRO inference only makes sense with ZeRO Stage 3 - please adjust your config"""					)

														# in case the training config is re-used for inference
														hf_deepspeed_config.del_config_sub_tree("""optimizer"""					)
														hf_deepspeed_config.del_config_sub_tree("""lr_scheduler"""					)
														lowercase_		,		lowercase_   :     Dict									=			None, None
														lowercase_   :     List[str]									=			None
							else:
														lowercase_   :     List[Any]									=			None  # important for when deepspeed_init is used as re-init
														lowercase_   :     int									=			list(filter(lambda UpperCAmelCase__					: p.requires_grad , model.parameters()					)					)
														lowercase_		,		lowercase_   :     List[str]									=			deepspeed_optim_sched(
														    UpperCAmelCase__ , UpperCAmelCase__ , UpperCAmelCase__ , UpperCAmelCase__ , UpperCAmelCase__					)

							# keep for quick debug:
							# from pprint import pprint; pprint(config)

							return optimizer, lr_scheduler




def 				lowerCamelCase      ( UpperCAmelCase__			:   List[str] , UpperCAmelCase__			:   Tuple					)     -> List[Any]:
							# it's possible that the user is trying to resume from model_path, which doesn't necessarily
							# contain a deepspeed checkpoint. e.g. examples just check if the dir exists and assume it's
							# a resume from a checkpoint and not just a local pretrained weight. So we check here if the
							# path contains what looks like a deepspeed checkpoint
							import glob

							lowercase_   :     Optional[int]									=			sorted(glob.glob(F'''{checkpoint_path}/global_step*'''					)					)

							if len(UpperCAmelCase__					) > 0:
														logger.info(F'''Attempting to resume from {checkpoint_path}'''					)
														# this magically updates self.optimizer and self.lr_scheduler
														lowercase_		,		lowercase_   :     Any									=			deepspeed_engine.load_checkpoint(
														    UpperCAmelCase__ , load_optimizer_states=UpperCAmelCase__ , load_lr_scheduler_states=UpperCAmelCase__					)
														if load_path is None:
																					raise ValueError(F'''[deepspeed] failed to resume from checkpoint {checkpoint_path}'''					)
							else:
														raise ValueError(F'''Can\'t find a valid checkpoint at {checkpoint_path}'''					)






'''simple docstring'''




from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    convert_to_rgb,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


_lowercase		:					Optional[int]													=		logging.get_logger(__name__)


if is_vision_available():
						import PIL





class        __magic_name__		(       _UpperCAmelCase):
			UpperCamelCase__							=			['''pixel_values''']







			def __init__(      self							:    Any							,     lowercase_							:    bool = True							,     lowercase_							:    Dict[str, int] = None							,     lowercase_							:    PILImageResampling = PILImageResampling.BICUBIC							,     lowercase_							:    bool = True							,     lowercase_							:    Dict[str, int] = None							,     lowercase_							:    bool = True							,     lowercase_							:    Union[int, float] = 1 / 255							,     lowercase_							:    bool = True							,     lowercase_							:    Optional[Union[float, List[float]]] = None							,     lowercase_							:    Optional[Union[float, List[float]]] = None							,     lowercase_							:    bool = True							,     **lowercase_							:    Optional[Any]							,     ):
										super().__init__(**lowercase_       )
										lowercase_   :     List[Any]									=			size if size is not None else {"""shortest_edge""": 224}
										lowercase_   :     int									=			get_size_dict(lowercase_							,     default_to_square=lowercase_       )
										lowercase_   :     List[Any]									=			crop_size if crop_size is not None else {"""height""": 224, """width""": 224}
										lowercase_   :     int									=			get_size_dict(lowercase_       )

										lowercase_   :     Tuple									=			do_resize
										lowercase_   :     int									=			size
										lowercase_   :     List[Any]									=			resample
										lowercase_   :     str									=			do_center_crop
										lowercase_   :     Optional[int]									=			crop_size
										lowercase_   :     str									=			do_rescale
										lowercase_   :     List[Any]									=			rescale_factor
										lowercase_   :     Dict									=			do_normalize
										lowercase_   :     List[str]									=			image_mean if image_mean is not None else OPENAI_CLIP_MEAN
										lowercase_   :     Any									=			image_std if image_std is not None else OPENAI_CLIP_STD
										lowercase_   :     List[str]									=			do_convert_rgb







			def       SCREAMING_SNAKE_CASE_	(      self							:    Union[str, Any]							,     lowercase_							:    np.ndarray							,     lowercase_							:    Dict[str, int]							,     lowercase_							:    PILImageResampling = PILImageResampling.BICUBIC							,     lowercase_							:    Optional[Union[str, ChannelDimension]] = None							,     **lowercase_							:    Dict							,     ):
										lowercase_   :     Tuple									=			get_size_dict(lowercase_							,     default_to_square=lowercase_       )
										lowercase_   :     List[Any]									=			get_resize_output_image_size(
										    lowercase_							,     size=(size["""height"""], size["""width"""])							,     default_to_square=lowercase_       )
										return resize(lowercase_							,     size=lowercase_							,     resample=lowercase_							,     data_format=lowercase_							,     **lowercase_       )







			def       SCREAMING_SNAKE_CASE_	(      self							:    List[Any]							,     lowercase_							:    np.ndarray							,     lowercase_							:    Dict[str, int]							,     lowercase_							:    Optional[Union[str, ChannelDimension]] = None							,     **lowercase_							:    Any							,     ):
										lowercase_   :     int									=			get_size_dict(lowercase_       )
										return center_crop(lowercase_							,     size=(size["""height"""], size["""width"""])							,     data_format=lowercase_							,     **lowercase_       )







			def       SCREAMING_SNAKE_CASE_	(      self							:    Tuple							,     lowercase_							:    np.ndarray							,     lowercase_							:    Union[int, float]							,     lowercase_							:    Optional[Union[str, ChannelDimension]] = None							,     **lowercase_							:    List[Any]							,     ):
										return rescale(lowercase_							,     scale=lowercase_							,     data_format=lowercase_							,     **lowercase_       )







			def       SCREAMING_SNAKE_CASE_	(      self							:    int							,     lowercase_							:    np.ndarray							,     lowercase_							:    Union[float, List[float]]							,     lowercase_							:    Union[float, List[float]]							,     lowercase_							:    Optional[Union[str, ChannelDimension]] = None							,     **lowercase_							:    Optional[Any]							,     ):
										return normalize(lowercase_							,     mean=lowercase_							,     std=lowercase_							,     data_format=lowercase_							,     **lowercase_       )







			def       SCREAMING_SNAKE_CASE_	(      self							:    List[Any]							,     lowercase_							:    ImageInput							,     lowercase_							:    bool = None							,     lowercase_							:    Dict[str, int] = None							,     lowercase_							:    PILImageResampling = None							,     lowercase_							:    bool = None							,     lowercase_							:    int = None							,     lowercase_							:    bool = None							,     lowercase_							:    float = None							,     lowercase_							:    bool = None							,     lowercase_							:    Optional[Union[float, List[float]]] = None							,     lowercase_							:    Optional[Union[float, List[float]]] = None							,     lowercase_							:    bool = None							,     lowercase_							:    Optional[Union[str, TensorType]] = None							,     lowercase_							:    Optional[ChannelDimension] = ChannelDimension.FIRST							,     **lowercase_							:    Optional[int]							,     ):
										lowercase_   :     Optional[Any]									=			do_resize if do_resize is not None else self.do_resize
										lowercase_   :     Union[str, Any]									=			size if size is not None else self.size
										lowercase_   :     Optional[int]									=			get_size_dict(lowercase_							,     default_to_square=lowercase_       )
										lowercase_   :     int									=			resample if resample is not None else self.resample
										lowercase_   :     str									=			do_center_crop if do_center_crop is not None else self.do_center_crop
										lowercase_   :     int									=			crop_size if crop_size is not None else self.crop_size
										lowercase_   :     int									=			get_size_dict(lowercase_       )
										lowercase_   :     Dict									=			do_rescale if do_rescale is not None else self.do_rescale
										lowercase_   :     List[str]									=			rescale_factor if rescale_factor is not None else self.rescale_factor
										lowercase_   :     Optional[Any]									=			do_normalize if do_normalize is not None else self.do_normalize
										lowercase_   :     List[Any]									=			image_mean if image_mean is not None else self.image_mean
										lowercase_   :     str									=			image_std if image_std is not None else self.image_std
										lowercase_   :     Any									=			do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

										lowercase_   :     int									=			make_list_of_images(lowercase_       )

										if not valid_images(lowercase_       ):
																	raise ValueError(
																	    """Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, """
																	    """torch.Tensor, tf.Tensor or jax.ndarray."""       )

										if do_resize and size is None:
																	raise ValueError("""Size must be specified if do_resize is True."""       )

										if do_center_crop and crop_size is None:
																	raise ValueError("""Crop size must be specified if do_center_crop is True."""       )

										if do_rescale and rescale_factor is None:
																	raise ValueError("""Rescale factor must be specified if do_rescale is True."""       )

										if do_normalize and (image_mean is None or image_std is None):
																	raise ValueError("""Image mean and std must be specified if do_normalize is True."""       )

										# PIL RGBA images are converted to RGB
										if do_convert_rgb:
																	lowercase_   :     Dict									=			[convert_to_rgb(lowercase_       ) for image in images]

										# All transformations expect numpy arrays.
										lowercase_   :     Any									=			[to_numpy_array(lowercase_       ) for image in images]

										if do_resize:
																	lowercase_   :     Union[str, Any]									=			[self.resize(image=lowercase_							,     size=lowercase_							,     resample=lowercase_       ) for image in images]

										if do_center_crop:
																	lowercase_   :     List[str]									=			[self.center_crop(image=lowercase_							,     size=lowercase_       ) for image in images]

										if do_rescale:
																	lowercase_   :     int									=			[self.rescale(image=lowercase_							,     scale=lowercase_       ) for image in images]

										if do_normalize:
																	lowercase_   :     List[Any]									=			[self.normalize(image=lowercase_							,     mean=lowercase_							,     std=lowercase_       ) for image in images]

										lowercase_   :     List[Any]									=			[to_channel_dimension_format(lowercase_							,     lowercase_       ) for image in images]

										lowercase_   :     Any									=			{"""pixel_values""": images}
										return BatchFeature(data=lowercase_							,     tensor_type=lowercase_       )





