
import json
import os
import tempfile
from unittest.mock import patch

import torch
from torch.utils.data import DataLoader, TensorDataset

from accelerate import DistributedType, infer_auto_device_map, init_empty_weights
from accelerate.accelerator import Accelerator
from accelerate.state import GradientState, PartialState
from accelerate.test_utils import require_bnb, require_multi_gpu, slow
from accelerate.test_utils.testing import AccelerateTestCase, require_cuda
from accelerate.utils import patch_environment

def 				SCREAMING_SNAKE_CASE     (      ) ->       Union[str, Any]:

				"""simple docstring"""







				A__  :	Optional[int]         =							torch.nn.Linear(2							,						4       )
				A__  :	int         =							torch.optim.AdamW(model.parameters()							,						lr=1.0       )
				A__  :	Any         =							torch.optim.lr_scheduler.OneCycleLR(__UpperCamelCase							,						max_lr=0.0_1							,						steps_per_epoch=2							,						epochs=1       )
				A__  :	Optional[Any]         =							DataLoader(TensorDataset(torch.tensor([1, 2, 3]       )       )       )
				A__  :	Union[str, Any]         =							DataLoader(TensorDataset(torch.tensor([4, 5, 6]       )       )       )

				return model, optimizer, scheduler, train_dl, valid_dl

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Union[str, Any]       ) ->       Dict:

				"""simple docstring"""







				return (model.weight.abs().sum() + model.bias.abs().sum()).item()

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						List[str]       ) ->       List[str]:

				"""simple docstring"""







				A__  :	Union[str, Any]         =							torch.nn.Linear(*tuple(model.weight.T.shape       )       ).state_dict()
				model.load_state_dict(__UpperCamelCase       )







class  UpperCamelCase__		(    SCREAMING_SNAKE_CASE_   ):







							'''simple docstring'''





							@require_cuda
							def   __snake_case				(			self		):
											A__  :	Any         =							Accelerator()
											assert PartialState._shared_state["_cpu"] is False
											assert PartialState._shared_state["device"].type == "cuda"
											with self.assertRaises(UpperCamelCase__		):
															A__  :	Dict         =							Accelerator(cpu=UpperCamelCase__		)





							def   __snake_case				(			self		):
											A__  :	Optional[int]         =							Accelerator()
											A__  :	str         =							GradientState()
											assert state.num_steps == 1
											A__  :	Optional[Any]         =							4
											assert state.num_steps == 4

											assert state.sync_gradients is True
											A__  :	int         =							False
											assert state.sync_gradients is False
											GradientState._reset_state()





							def   __snake_case				(			self		):
											A__  :	Optional[int]         =							Accelerator()
											A__  :	Dict         =							create_components()

											(
											    A__							
											)  :	Any         =							accelerator.prepare(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)

											self.assertTrue(prepared_model in accelerator._models		)
											self.assertTrue(prepared_optimizer in accelerator._optimizers		)
											self.assertTrue(prepared_scheduler in accelerator._schedulers		)
											self.assertTrue(prepared_train_dl in accelerator._dataloaders		)
											self.assertTrue(prepared_valid_dl in accelerator._dataloaders		)





							def   __snake_case				(			self		):
											A__  :	Dict         =							Accelerator()
											A__  :	str         =							create_components()
											accelerator.prepare(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)
											accelerator.free_memory()

											self.assertTrue(len(accelerator._models		) == 0		)
											self.assertTrue(len(accelerator._optimizers		) == 0		)
											self.assertTrue(len(accelerator._schedulers		) == 0		)
											self.assertTrue(len(accelerator._dataloaders		) == 0		)





							def   __snake_case				(			self		):
											PartialState._reset_state()

											# Mock torch.cuda.set_device to avoid an exception as the device doesn't exist
											def noop(*UpperCamelCase__				,       **UpperCamelCase__		):
															pass

											with patch('''torch.cuda.set_device'''				,       UpperCamelCase__		), patch_environment(ACCELERATE_TORCH_DEVICE='''cuda:64'''		):
															A__  :	Optional[int]         =							Accelerator()
															self.assertEqual(str(accelerator.state.device		)				,       '''cuda:64'''		)





							def   __snake_case				(			self		):
											A__  :	Tuple         =							Accelerator()
											A__  :	Any         =							create_components()
											accelerator.prepare(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)

											A__  :	Optional[Any]         =							get_signature(UpperCamelCase__		)

											with tempfile.TemporaryDirectory() as tmpdirname:
															accelerator.save_state(UpperCamelCase__		)

															# make sure random weights don't match
															load_random_weights(UpperCamelCase__		)
															self.assertTrue(abs(model_signature - get_signature(UpperCamelCase__		)		) > 1e-3		)

															# make sure loaded weights match
															accelerator.load_state(UpperCamelCase__		)
															self.assertTrue(abs(model_signature - get_signature(UpperCamelCase__		)		) < 1e-3		)





							def   __snake_case				(			self		):
											A__  :	List[Any]         =							Accelerator()
											A__  :	Optional[int]         =							create_components()
											accelerator.prepare(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)

											A__  :	Union[str, Any]         =							get_signature(UpperCamelCase__		)

											# saving hook
											def save_config(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		):
															A__  :	Tuple         =							{'''class_name''': models[0].__class__.__name__}

															with open(os.path.join(UpperCamelCase__				,       '''data.json'''		)				,       '''w'''		) as f:
																			json.dump(UpperCamelCase__				,       UpperCamelCase__		)

        # loading hook
											def load_config(UpperCamelCase__				,       UpperCamelCase__		):
															with open(os.path.join(UpperCamelCase__				,       '''data.json'''		)				,       '''r'''		) as f:
																			A__  :	Optional[int]         =							json.load(UpperCamelCase__		)

															A__  :	List[Any]         =							config['''class_name''']

											A__  :	Dict         =							accelerator.register_save_state_pre_hook(UpperCamelCase__		)
											A__  :	Optional[int]         =							accelerator.register_load_state_pre_hook(UpperCamelCase__		)

											with tempfile.TemporaryDirectory() as tmpdirname:
															accelerator.save_state(UpperCamelCase__		)

															# make sure random weights don't match with hooks
															load_random_weights(UpperCamelCase__		)
															self.assertTrue(abs(model_signature - get_signature(UpperCamelCase__		)		) > 1e-3		)

															# random class name to verify correct one is loaded
															A__  :	Union[str, Any]         =							'''random'''

															# make sure loaded weights match with hooks
															accelerator.load_state(UpperCamelCase__		)
															self.assertTrue(abs(model_signature - get_signature(UpperCamelCase__		)		) < 1e-3		)

															# mode.class_name is loaded from config
															self.assertTrue(model.class_name == model.__class__.__name__		)

											# remove hooks
											save_hook.remove()
											load_hook.remove()

											with tempfile.TemporaryDirectory() as tmpdirname:
															accelerator.save_state(UpperCamelCase__		)

															# make sure random weights don't match with hooks removed
															load_random_weights(UpperCamelCase__		)
															self.assertTrue(abs(model_signature - get_signature(UpperCamelCase__		)		) > 1e-3		)

															# random class name to verify correct one is loaded
															A__  :	Union[str, Any]         =							'''random'''

															# make sure loaded weights match with hooks removed
															accelerator.load_state(UpperCamelCase__		)
															self.assertTrue(abs(model_signature - get_signature(UpperCamelCase__		)		) < 1e-3		)

															# mode.class_name is NOT loaded from config
															self.assertTrue(model.class_name != model.__class__.__name__		)





							def   __snake_case				(			self		):
											A__  :	int         =							Accelerator()
											A__  :	Tuple         =							create_components()
											A__  :	List[Any]         =							None

											# This should work
											A__  :	Tuple         =							accelerator.prepare(
											    UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)
											self.assertTrue(dummy_obj is None		)





							def   __snake_case				(			self		):
											A__  :	str         =							Accelerator()
											A__  :	int         =							create_components()
											A__  :	Any         =							[1, 2, 3]

											# This should work
											A__  :	List[Any]         =							accelerator.prepare(
											    UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)
											self.assertEqual(
											    getattr(UpperCamelCase__				,       '''_is_accelerate_prepared'''				,       UpperCamelCase__		)				,       UpperCamelCase__				,       '''Dummy object should have `_is_accelerate_prepared` set to `True`'''				,       )
											self.assertEqual(
											    getattr(UpperCamelCase__				,       '''_is_accelerate_prepared'''				,       UpperCamelCase__		)				,       UpperCamelCase__				,       '''Model is missing `_is_accelerator_prepared` or is set to `False`'''				,       )
											self.assertEqual(
											    getattr(UpperCamelCase__				,       '''_is_accelerate_prepared'''				,       UpperCamelCase__		)				,       UpperCamelCase__				,       '''Optimizer is missing `_is_accelerator_prepared` or is set to `False`'''				,       )
											self.assertEqual(
											    getattr(UpperCamelCase__				,       '''_is_accelerate_prepared'''				,       UpperCamelCase__		)				,       UpperCamelCase__				,       '''Scheduler is missing `_is_accelerator_prepared` or is set to `False`'''				,       )
											self.assertEqual(
											    getattr(UpperCamelCase__				,       '''_is_accelerate_prepared'''				,       UpperCamelCase__		)				,       UpperCamelCase__				,       '''Train Dataloader is missing `_is_accelerator_prepared` or is set to `False`'''				,       )
											self.assertEqual(
											    getattr(UpperCamelCase__				,       '''_is_accelerate_prepared'''				,       UpperCamelCase__		)				,       UpperCamelCase__				,       '''Valid Dataloader is missing `_is_accelerator_prepared` or is set to `False`'''				,       )





							@slow
							@require_bnb
							def   __snake_case				(			self		):
											from transformers import AutoModelForCausalLM

											A__  :	Tuple         =							AutoModelForCausalLM.from_pretrained(
											    '''EleutherAI/gpt-neo-125m'''				,       load_in_abit=UpperCamelCase__				,       device_map={'''''': 0}				,       )
											A__  :	Optional[int]         =							Accelerator()

											# This should work
											A__  :	Optional[Any]         =							accelerator.prepare(UpperCamelCase__		)





							@slow
							@require_bnb
							def   __snake_case				(			self		):
											from transformers import AutoModelForCausalLM

											A__  :	List[str]         =							Accelerator()

											with init_empty_weights():
															A__  :	List[Any]         =							AutoModelForCausalLM.from_pretrained(
															    '''EleutherAI/gpt-neo-125m'''				,       )
															model.tie_weights()
															A__  :	Any         =							infer_auto_device_map(UpperCamelCase__		)
															A__  :	Any         =							'''cpu'''

											A__  :	Dict         =							AutoModelForCausalLM.from_pretrained(
											    '''EleutherAI/gpt-neo-125m'''				,       device_map=UpperCamelCase__				,       load_in_abit=UpperCamelCase__				,       llm_inta_enable_fpaa_cpu_offload=UpperCamelCase__		)

											# This should not work and get value error
											with self.assertRaises(UpperCamelCase__		):
															A__  :	Union[str, Any]         =							accelerator.prepare(UpperCamelCase__		)





							@slow
							@require_bnb
							@require_multi_gpu
							def   __snake_case				(			self		):
											from transformers import AutoModelForCausalLM

											A__  :	Any         =							{'''distributed_type''': DistributedType.MULTI_GPU}

											with init_empty_weights():
															A__  :	Tuple         =							AutoModelForCausalLM.from_pretrained(
															    '''EleutherAI/gpt-neo-125m'''				,       )
															model.tie_weights()
															A__  :	Dict         =							infer_auto_device_map(UpperCamelCase__		)
															A__  :	Dict         =							1

											A__  :	Dict         =							AutoModelForCausalLM.from_pretrained(
											    '''EleutherAI/gpt-neo-125m'''				,       load_in_abit=UpperCamelCase__				,       device_map=UpperCamelCase__				,       )
											A__  :	Any         =							Accelerator()

											# This should not work and get value error
											with self.assertRaises(UpperCamelCase__		):
															A__  :	Dict         =							accelerator.prepare(UpperCamelCase__		)

											PartialState._reset_state()





							@slow
							@require_bnb
							@require_multi_gpu
							def   __snake_case				(			self		):
											from transformers import AutoModelForCausalLM

											with init_empty_weights():
															A__  :	str         =							AutoModelForCausalLM.from_pretrained(
															    '''EleutherAI/gpt-neo-125m'''				,       )
															A__  :	Tuple         =							infer_auto_device_map(UpperCamelCase__		)
															A__  :	Optional[int]         =							1

											A__  :	str         =							AutoModelForCausalLM.from_pretrained(
											    '''EleutherAI/gpt-neo-125m'''				,       load_in_abit=UpperCamelCase__				,       device_map=UpperCamelCase__				,       )
											A__  :	List[Any]         =							Accelerator()

											# This should work
											A__  :	List[Any]         =							accelerator.prepare(UpperCamelCase__		)





							@require_cuda
							def   __snake_case				(			self		):
											A__  :	int         =							torch.nn.Linear(10				,       10		)
											A__  :	int         =							torch.optim.SGD(model.parameters()				,       lr=0.0_1		)
											A__  :	str         =							Accelerator(cpu=UpperCamelCase__		)
											A__  :	Optional[Any]         =							accelerator.prepare(UpperCamelCase__		)

_SCREAMING_SNAKE_CASE    :      List[str]          =			range(2, 2_0 + 1)
_SCREAMING_SNAKE_CASE    :      Optional[Any]          =			[1_0**k for k in range(ks[-1] + 1)]
_SCREAMING_SNAKE_CASE    :      dict[int, dict[int, list[list[int]]]]          =			{}

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Any							,						__UpperCamelCase :						Dict							,						__UpperCamelCase :						Any							,						__UpperCamelCase :						Optional[Any]       ) ->       int:

				"""simple docstring"""







				A__  :	Tuple         =							sum(a_i[j] for j in range(__UpperCamelCase							,						len(__UpperCamelCase       )       )       )
				A__  :	Tuple         =							sum(a_i[j] * base[j] for j in range(min(len(__UpperCamelCase       )							,						__UpperCamelCase       )       )       )

				A__       ,							A__  :	Optional[int]         =							0, 0
				A__  :	List[Any]         =							n - i

				A__  :	Any         =							memo.get(__UpperCamelCase       )

				if sub_memo is not None:
								A__  :	Optional[int]         =							sub_memo.get(__UpperCamelCase       )

								if jumps is not None and len(__UpperCamelCase       ) > 0:
												# find and make the largest jump without going over
												A__  :	List[Any]         =							-1
												for _k in range(len(__UpperCamelCase       ) - 1							,						-1							,						-1       ):
																if jumps[_k][2] <= k and jumps[_k][1] <= max_dn:
																				A__  :	List[str]         =							_k
																				break

												if max_jump >= 0:
																A__       ,							A__       ,							A__  :	List[Any]         =							jumps[max_jump]
																# since the difference between jumps is cached, add c
																A__  :	int         =							diff + c
																for j in range(min(__UpperCamelCase							,						len(__UpperCamelCase       )       )       ):
																				A__       ,							A__  :	List[str]         =							divmod(__UpperCamelCase							,						10       )
																if new_c > 0:
																				add(__UpperCamelCase							,						__UpperCamelCase							,						__UpperCamelCase       )

								else:
												A__  :	List[Any]         =							[]
				else:
								A__  :	Optional[Any]         =							{c: []}
								A__  :	int         =							sub_memo

				if dn >= max_dn or c + diff >= base[k]:
								return diff, dn

				if k > ks[0]:
								while True:
												# keep doing smaller jumps
												A__       ,							A__  :	str         =							next_term(__UpperCamelCase							,						k - 1							,						i + dn							,						__UpperCamelCase       )
												diff += _diff
												dn += terms_jumped

												if dn >= max_dn or c + diff >= base[k]:
																break
				else:
								# would be too small a jump, just compute sequential terms instead
								A__       ,							A__  :	str         =							compute(__UpperCamelCase							,						__UpperCamelCase							,						i + dn							,						__UpperCamelCase       )
								diff += _diff
								dn += terms_jumped

				A__  :	str         =							sub_memo[c]

				# keep jumps sorted by # of terms skipped
				A__  :	List[Any]         =							0
				while j < len(__UpperCamelCase       ):
								if jumps[j][1] > dn:
												break
								j += 1

				# cache the jump for this value digitsum(b) and c
				sub_memo[c].insert(__UpperCamelCase							,						(diff, dn, k)       )
				return (diff, dn)

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						int							,						__UpperCamelCase :						Optional[int]							,						__UpperCamelCase :						List[str]							,						__UpperCamelCase :						int       ) ->       Any:

				"""simple docstring"""







				if i >= n:
								return 0, i
				if k > len(__UpperCamelCase       ):
								a_i.extend([0 for _ in range(k - len(__UpperCamelCase       )       )]       )

				# note: a_i -> b * 10^k + c
				# ds_b -> digitsum(b)
				# ds_c -> digitsum(c)
				A__  :	Optional[Any]         =							i
				A__       ,							A__       ,							A__  :	Dict         =							0, 0, 0
				for j in range(len(__UpperCamelCase       )       ):
								if j >= k:
												ds_b += a_i[j]
								else:
												ds_c += a_i[j]

				while i < n:
								i += 1
								A__  :	int         =							ds_c + ds_b
								diff += addend
								A__  :	List[Any]         =							0
								for j in range(__UpperCamelCase       ):
												A__  :	Optional[Any]         =							a_i[j] + addend
												A__       ,							A__  :	List[str]         =							divmod(__UpperCamelCase							,						10       )

												ds_c += a_i[j]

								if addend > 0:
												break

				if addend > 0:
								add(__UpperCamelCase							,						__UpperCamelCase							,						__UpperCamelCase       )
				return diff, i - start_i

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Dict							,						__UpperCamelCase :						List[Any]							,						__UpperCamelCase :						int       ) ->       Tuple:

				"""simple docstring"""







				for j in range(__UpperCamelCase							,						len(__UpperCamelCase       )       ):
								A__  :	Any         =							digits[j] + addend
								if s >= 10:
												A__       ,							A__  :	Union[str, Any]         =							divmod(__UpperCamelCase							,						10       )
												A__  :	Optional[int]         =							addend // 10 + quotient
								else:
												A__  :	Any         =							s
												A__  :	Dict         =							addend // 10

								if addend == 0:
												break

				while addend > 0:
								A__       ,							A__  :	Dict         =							divmod(__UpperCamelCase							,						10       )
								digits.append(__UpperCamelCase       )

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						int = 10**15       ) ->       int:

				"""simple docstring"""







				A__  :	List[Any]         =							[1]
				A__  :	Dict         =							1
				A__  :	Tuple         =							0
				while True:
								A__       ,							A__  :	List[str]         =							next_term(__UpperCamelCase							,						20							,						i + dn							,						__UpperCamelCase       )
								dn += terms_jumped
								if dn == n - i:
												break

				A__  :	List[str]         =							0
				for j in range(len(__UpperCamelCase       )       ):
								a_n += digits[j] * 10**j
				return a_n


if __name__ == "__main__":
					print(f"""{solution() = }""")