

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

__A     :    Optional[int]    =		logging.get_logger(__name__)
def    __SCREAMING_SNAKE_CASE						(						):
					"""simple docstring"""
					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 			__UpperCamelCase (   _A						):


		def __init__(self :		int     ,     __SCREAMING_SNAKE_CASE :		int):
							# set global weakref object
							set_hf_deepspeed_config(self)
							dep_version_check("accelerate")
							dep_version_check("deepspeed")
							super().__init__(__SCREAMING_SNAKE_CASE)





class 			__UpperCamelCase (   _A						):


		def __init__(self :		Any     ,     __SCREAMING_SNAKE_CASE :		Optional[Any]):
							super().__init__(__SCREAMING_SNAKE_CASE)
							A			=					None
							A			=					[]


		def 					SCREAMING_SNAKE_CASE__   (self :		Union[str, 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 :		Optional[int]     ,     __SCREAMING_SNAKE_CASE :		List[Any]):
							A			=					self.get_value(__SCREAMING_SNAKE_CASE)
							if val is None:
												return False
							else:
												return val == "auto"


		def 					SCREAMING_SNAKE_CASE__   (self :		List[str]     ,     __SCREAMING_SNAKE_CASE :		Tuple     ,     __SCREAMING_SNAKE_CASE :		Optional[int]     ,     __SCREAMING_SNAKE_CASE :		Optional[Any]=None     ,     __SCREAMING_SNAKE_CASE :		Optional[Any]=True):
							A			,    A			=					self.find_config_node(__SCREAMING_SNAKE_CASE)
							if config is None:
												return

							if config.get(__SCREAMING_SNAKE_CASE) == "auto":
												A			=					hf_val
												return

							if not must_match:
												return

							A			=					config.get(__SCREAMING_SNAKE_CASE)
							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}""")

		SCREAMING_SNAKE_CASE								=    partialmethod(_A     ,		must_match=_A						)


		def 					SCREAMING_SNAKE_CASE__   (self :		int     ,     __SCREAMING_SNAKE_CASE :		Union[str, Any]):
							A			=					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"     ,     __SCREAMING_SNAKE_CASE     ,     "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:
												A			=					"apex" if args.fpaa_backend == "apex" else "amp"
							else:
												A			=					None

							if args.save_on_each_node:
												# deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
												A			=					self.config.get("checkpoint"     ,     {})
												A			=					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"):
												A			=					torch.bfloataa
							elif self.is_false("fp16.enabled"):
												A			=					torch.floataa
							else:
												A			=					torch.floataa


		def 					SCREAMING_SNAKE_CASE__   (self :		str     ,     __SCREAMING_SNAKE_CASE :		Tuple     ,     __SCREAMING_SNAKE_CASE :		Optional[int]     ,     __SCREAMING_SNAKE_CASE :		int):
							A			=					[
							    "zero_optimization.reduce_bucket_size",
							    "zero_optimization.stage3_prefetch_bucket_size",
							    "zero_optimization.stage3_param_persistence_threshold",
							]
							A			=					[x for x in hidden_size_based_keys if self.is_auto(__SCREAMING_SNAKE_CASE)]

							if len(__SCREAMING_SNAKE_CASE) > 0:
												if hasattr(model.config     ,     "hidden_size"):
																	A			=					model.config.hidden_size
												elif hasattr(model.config     ,     "hidden_sizes"):
																	# if there are many hidden sizes pick the largest one
																	A			=					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"     ,     1_0 * hidden_size)

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

							if len(self.mismatches) > 0:
												A			=					"\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'.""")

__A     :    Tuple    =		None
def    __SCREAMING_SNAKE_CASE						(						lowercase__      ):
					"""simple docstring"""
					# 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)
					A			=					weakref.ref(lowercase__      )
def    __SCREAMING_SNAKE_CASE						(						):
					"""simple docstring"""
					# useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
					global _hf_deepspeed_config_weak_ref
					A			=					None
def    __SCREAMING_SNAKE_CASE						(						):
					"""simple docstring"""
					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    __SCREAMING_SNAKE_CASE						(						):
					"""simple docstring"""
					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    __SCREAMING_SNAKE_CASE						(						lowercase__			,							lowercase__			,							lowercase__			,							lowercase__			,							lowercase__      ):
					"""simple docstring"""
					from accelerate.utils import DummyOptim, DummyScheduler

					A			=					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)

					A			=					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."      )
										A			=					DummyOptim(params=lowercase__      )
					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.
										A			=					trainer.create_optimizer()
										# To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
										A			=					True

					A			=					None
					if "scheduler" in config:
										A			=					DummyScheduler(lowercase__      )
					else:
										if isinstance(lowercase__			,							lowercase__      ):
															raise ValueError(
															    "Found `optimizer` configured in the DeepSpeed config, but no `scheduler`. "
															    "Please configure a scheduler in the DeepSpeed config."      )
										A			=					trainer.create_scheduler(num_training_steps=lowercase__			,							optimizer=lowercase__      )

					return optimizer, lr_scheduler
def    __SCREAMING_SNAKE_CASE						(						lowercase__			,							lowercase__			,							lowercase__=False      ):
					"""simple docstring"""
					from deepspeed.utils import logger as ds_logger

					A			=					trainer.model
					A			=					trainer.args

					A			=					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(lowercase__			,							lowercase__			,							lowercase__      )

					# 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"      )
										A			,    A			=					None, None
										A			=					None
					else:
										A			=					None  # important for when deepspeed_init is used as re-init
										A			=					list(filter(lambda lowercase__      : p.requires_grad			,							model.parameters()      )      )
										A			,    A			=					deepspeed_optim_sched(
										    lowercase__			,							lowercase__			,							lowercase__			,							lowercase__			,							lowercase__      )

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

					return optimizer, lr_scheduler
def    __SCREAMING_SNAKE_CASE						(						lowercase__			,							lowercase__      ):
					"""simple docstring"""
					# 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

					A			=					sorted(glob.glob(F"""{checkpoint_path}/global_step*"""      )      )

					if len(lowercase__      ) > 0:
										logger.info(F"""Attempting to resume from {checkpoint_path}"""      )
										# this magically updates self.optimizer and self.lr_scheduler
										A			,    A			=					deepspeed_engine.load_checkpoint(
										    lowercase__			,							load_optimizer_states=lowercase__			,							load_lr_scheduler_states=lowercase__      )
										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"""




import textwrap

import pyarrow as pa
import pytest

from datasets import Features, Image
from datasets.packaged_modules.text.text import Text

from ..utils import require_pil
@pytest.fixture
def    __SCREAMING_SNAKE_CASE						(						lowercase__      ):
					"""simple docstring"""
					A			=					tmp_path / "text.txt"
					A			=					textwrap.dedent(
					    "\\n        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.\n        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.\n        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n\n        Second paragraph:\n        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.\n        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.\n        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n        "      )
					with open(lowercase__			,							"w"			,							encoding="utf-8"      ) as f:
										f.write(lowercase__      )
					return str(lowercase__      )
@pytest.fixture
def    __SCREAMING_SNAKE_CASE						(						lowercase__			,							lowercase__      ):
					"""simple docstring"""
					A			=					tmp_path / "text_with_image.txt"
					with open(lowercase__			,							"w"			,							encoding="utf-8"      ) as f:
										f.write(lowercase__      )
					return str(lowercase__      )
@pytest.mark.parametrize("keep_linebreaks"			,							[True, False]      )
def    __SCREAMING_SNAKE_CASE						(						lowercase__			,							lowercase__      ):
					"""simple docstring"""
					with open(lowercase__			,							encoding="utf-8"      ) as f:
										A			=					f.read().splitlines(keepends=lowercase__      )
					A			=					Text(keep_linebreaks=lowercase__			,							encoding="utf-8"      )
					A			=					text._generate_tables([[text_file]]      )
					A			=					pa.concat_tables([table for _, table in generator]      ).to_pydict()["text"]
					assert generated_content == expected_content
@require_pil
def    __SCREAMING_SNAKE_CASE						(						lowercase__      ):
					"""simple docstring"""
					with open(lowercase__			,							encoding="utf-8"      ) as f:
										A			=					f.read().splitlines()[0]
					A			=					Text(encoding="utf-8"			,							features=Features({"image": Image()}      )      )
					A			=					text._generate_tables([[text_file_with_image]]      )
					A			=					pa.concat_tables([table for _, table in generator]      )
					assert pa_table.schema.field("image"      ).type == Image()()
					A			=					pa_table.to_pydict()["image"]
					assert generated_content == [{"path": image_file, "bytes": None}]
@pytest.mark.parametrize("sample_by"			,							["line", "paragraph", "document"]      )
def    __SCREAMING_SNAKE_CASE						(						lowercase__			,							lowercase__      ):
					"""simple docstring"""
					with open(lowercase__			,							encoding="utf-8"      ) as f:
										A			=					f.read()
					if sample_by == "line":
										A			=					expected_content.splitlines()
					elif sample_by == "paragraph":
										A			=					expected_content.split("\n\n"      )
					elif sample_by == "document":
										A			=					[expected_content]
					A			=					Text(sample_by=lowercase__			,							encoding="utf-8"			,							chunksize=100      )
					A			=					text._generate_tables([[text_file]]      )
					A			=					pa.concat_tables([table for _, table in generator]      ).to_pydict()["text"]
					assert generated_content == expected_content





