






# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("""ignore""", category=UserWarning, module="""torch.optim.lr_scheduler""")



class 							SCREAMING_SNAKE_CASE__						:





					"""simple docstring"""






					def __init__( self					:  Tuple							,		__A					:  List[Any]							,		__A					:  List[Any]							,		__A					:  bool = True							,		__A					:  bool = False	):
						snake_case__			:				Any       =  scheduler
						snake_case__			:				int       =  optimizers if isinstance(__A							,		(list, tuple)	) else [optimizers]
						snake_case__			:				Optional[int]       =  split_batches
						snake_case__			:				Union[str, Any]       =  step_with_optimizer
						snake_case__			:				Dict       =  GradientState()



					def 	_lowercase					( self					:  List[Any]							,		*__A					:  List[str]							,		**__A					:  int	):
						if not self.step_with_optimizer:
							# No link between scheduler and optimizer -> just step
							self.scheduler.step(*__A							,		**__A	)
							return

						# Otherwise, first make sure the optimizer was stepped.
						if not self.gradient_state.sync_gradients:
							if self.gradient_state.adjust_scheduler:
								self.scheduler._step_count += 1
							return

						for opt in self.optimizers:
							if opt.step_was_skipped:
								return
						if self.split_batches:
							# Split batches -> the training dataloader batch size is not changed so one step per training step
							self.scheduler.step(*__A							,		**__A	)
						else:
							# Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
							# num_processes steps per training step
							snake_case__			:				Optional[Any]       =  AcceleratorState().num_processes
							for _ in range(__A	):
								# Special case when using OneCycle and `drop_last` was not used
								if hasattr(self.scheduler							,		"total_steps"	):
									if self.scheduler._step_count <= self.scheduler.total_steps:
										self.scheduler.step(*__A							,		**__A	)
								else:
									self.scheduler.step(*__A							,		**__A	)



					def 	_lowercase					( self					:  Union[str, Any]	):
						return self.scheduler.get_last_lr()



					def 	_lowercase					( self					:  str	):
						return self.scheduler.state_dict()



					def 	_lowercase					( self					:  List[str]							,		__A					:  str	):
						self.scheduler.load_state_dict(__A	)



					def 	_lowercase					( self					:  Union[str, Any]	):
						return self.scheduler.get_lr()






					def 	_lowercase					( self					:  List[str]							,		*__A					:  int							,		**__A					:  str	):
						return self.scheduler.print_lr(*__A							,		**__A	)











from tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch



@require_torch
@require_tf
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[Any]	):
						snake_case__			:				Optional[Any]       =  SMALL_MODEL_IDENTIFIER
						snake_case__			:				Any       =  "pt"
						snake_case__			:				Any       =  "tf"



					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  List[Any]	):
						snake_case__			:				int       =  AutoModel.from_pretrained(self.test_model	)
						model_pt.save_pretrained(__A	)



					def 	_lowercase					( self					:  Optional[int]							,		__A					:  Tuple	):
						snake_case__			:				List[Any]       =  TFAutoModel.from_pretrained(self.test_model							,		from_pt=__A	)
						model_tf.save_pretrained(__A	)



					def 	_lowercase					( self					:  str	):
						snake_case__			:				Optional[Any]       =  "mock_framework"

						# Framework provided - return whatever the user provides
						snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model							,		__A	)
						self.assertEqual(__A							,		__A	)

						# Local checkpoint and framework provided - return provided framework
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				Optional[int]       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				int       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)



					def 	_lowercase					( self					:  Dict	):
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				List[str]       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_pt	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				Tuple       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Invalid local checkpoint
						with TemporaryDirectory() as local_invalid_ckpt:
							with self.assertRaises(__A	):
								snake_case__			:				int       =  FeaturesManager.determine_framework(__A	)






					def 	_lowercase					( self					:  Dict	):
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	):
							snake_case__			:				List[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# PyTorch not in environment -> use TensorFlow
						snake_case__			:				Tuple       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				int       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Both in environment -> use PyTorch
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[int]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# Both not in environment -> raise error
						snake_case__			:				List[str]       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[Any]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							with self.assertRaises(__A	):
								snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)



