



import inspect
import logging
import os
import random
import shutil
import tempfile
import unittest

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

from accelerate import Accelerator
from accelerate.test_utils import execute_subprocess_async, require_cuda
from accelerate.utils import ProjectConfiguration, set_seed


lowercase_								=     logging.getLogger(__name__)



def __lowerCAmelCase (     SCREAMING_SNAKE_CASE_=2						,  SCREAMING_SNAKE_CASE_=3						,  SCREAMING_SNAKE_CASE_=16						,  SCREAMING_SNAKE_CASE_ = 10						,  SCREAMING_SNAKE_CASE_ = 2  ):

			def get_dataset(SCREAMING_SNAKE_CASE_  ):
						lowercase__					  =				torch.randn(batch_size * n_batches						,  1  )
						return TensorDataset(SCREAMING_SNAKE_CASE_						,  a * x + b + 0.1 * torch.randn(batch_size * n_batches						,  1  )  )

			lowercase__					  =				get_dataset(SCREAMING_SNAKE_CASE_  )
			lowercase__					  =				get_dataset(SCREAMING_SNAKE_CASE_  )
			lowercase__					  =				DataLoader(SCREAMING_SNAKE_CASE_						,  shuffle=SCREAMING_SNAKE_CASE_						,  batch_size=SCREAMING_SNAKE_CASE_						,  num_workers=4  )
			lowercase__					  =				DataLoader(SCREAMING_SNAKE_CASE_						,  shuffle=SCREAMING_SNAKE_CASE_						,  batch_size=SCREAMING_SNAKE_CASE_						,  num_workers=4  )
			return (train_dataloader, valid_dataloader)



def __lowerCAmelCase (     SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_=None  ):
			lowercase__					  =				[]
			for epoch in range(SCREAMING_SNAKE_CASE_  ):
						# Train quickly
						model.train()
						for batch in dataloader:
									lowercase__  ,       lowercase__					  =				batch
									lowercase__					  =				model(SCREAMING_SNAKE_CASE_  )
									lowercase__					  =				torch.nn.functional.mse_loss(SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  )
									accelerator.backward(SCREAMING_SNAKE_CASE_  )
									optimizer.step()
									optimizer.zero_grad()
						rands.append(random.random()  )  # Introduce some randomness
						if scheduler is not None:
									scheduler.step()
			return rands
class 				_snake_case			(					nn.Module):




				def __init__(						self					:						List[Any]						):
							super().__init__()
							lowercase__					  =				nn.Parameter(torch.randn(1						)						)
							lowercase__					  =				nn.Parameter(torch.randn(1						)						)




				def 							A__      (						self					:						Tuple,					__lowercase					:						int						):
							return x * self.a + self.b
class 				_snake_case			(					unittest.TestCase):




				def 							A__      (						self					:						Any						):
							with tempfile.TemporaryDirectory() as tmpdir:
										set_seed(42						)
										lowercase__					  =				DummyModel()
										lowercase__					  =				torch.optim.Adam(params=model.parameters(),					lr=1e-3						)
										lowercase__  ,       lowercase__					  =				dummy_dataloaders()
										lowercase__					  =				ProjectConfiguration(total_limit=1,					project_dir=__lowercase,					automatic_checkpoint_naming=__lowercase						)
										# Train baseline
										lowercase__					  =				Accelerator(project_config=__lowercase						)
										lowercase__  ,       lowercase__  ,       lowercase__  ,       lowercase__					  =				accelerator.prepare(
										    __lowercase,					__lowercase,					__lowercase,					__lowercase						)
										# Save initial
										accelerator.save_state()

										# Save second state
										accelerator.save_state()
										self.assertEqual(len(os.listdir(accelerator.project_dir						)						),					1						)




				def 							A__      (						self					:						Optional[Any]						):
							with tempfile.TemporaryDirectory() as tmpdir:
										set_seed(42						)
										lowercase__					  =				DummyModel()
										lowercase__					  =				torch.optim.Adam(params=model.parameters(),					lr=1e-3						)
										lowercase__  ,       lowercase__					  =				dummy_dataloaders()
										# Train baseline
										lowercase__					  =				Accelerator()
										lowercase__  ,       lowercase__  ,       lowercase__  ,       lowercase__					  =				accelerator.prepare(
										    __lowercase,					__lowercase,					__lowercase,					__lowercase						)
										# Save initial
										lowercase__					  =				os.path.join(__lowercase,					"initial"						)
										accelerator.save_state(__lowercase						)
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()
										lowercase__					  =				train(3,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()

										# Train partially
										set_seed(42						)
										lowercase__					  =				DummyModel()
										lowercase__					  =				torch.optim.Adam(params=model.parameters(),					lr=1e-3						)
										lowercase__  ,       lowercase__					  =				dummy_dataloaders()
										lowercase__					  =				Accelerator()
										lowercase__  ,       lowercase__  ,       lowercase__  ,       lowercase__					  =				accelerator.prepare(
										    __lowercase,					__lowercase,					__lowercase,					__lowercase						)
										accelerator.load_state(__lowercase						)
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)

										lowercase__					  =				train(2,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										# Save everything
										lowercase__					  =				os.path.join(__lowercase,					"checkpoint"						)
										accelerator.save_state(__lowercase						)

										# Load everything back in and make sure all states work
										accelerator.load_state(__lowercase						)
										test_rands += train(1,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)




				def 							A__      (						self					:						List[Any]						):
							with tempfile.TemporaryDirectory() as tmpdir:
										set_seed(42						)
										lowercase__					  =				DummyModel()
										lowercase__					  =				torch.optim.Adam(params=model.parameters(),					lr=1e-3						)
										lowercase__  ,       lowercase__					  =				dummy_dataloaders()
										lowercase__					  =				ProjectConfiguration(automatic_checkpoint_naming=__lowercase						)

										# Train baseline
										lowercase__					  =				Accelerator(project_dir=__lowercase,					project_config=__lowercase						)
										lowercase__  ,       lowercase__  ,       lowercase__  ,       lowercase__					  =				accelerator.prepare(
										    __lowercase,					__lowercase,					__lowercase,					__lowercase						)
										# Save initial
										accelerator.save_state()
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()
										lowercase__					  =				train(3,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()

										# Train partially
										set_seed(42						)
										lowercase__					  =				DummyModel()
										lowercase__					  =				torch.optim.Adam(params=model.parameters(),					lr=1e-3						)
										lowercase__  ,       lowercase__					  =				dummy_dataloaders()
										lowercase__					  =				ProjectConfiguration(iteration=1,					automatic_checkpoint_naming=__lowercase						)
										lowercase__					  =				Accelerator(project_dir=__lowercase,					project_config=__lowercase						)
										lowercase__  ,       lowercase__  ,       lowercase__  ,       lowercase__					  =				accelerator.prepare(
										    __lowercase,					__lowercase,					__lowercase,					__lowercase						)
										accelerator.load_state(os.path.join(__lowercase,					"checkpoints",					"checkpoint_0"						)						)
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)

										lowercase__					  =				train(2,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										# Save everything
										accelerator.save_state()

										# Load everything back in and make sure all states work
										accelerator.load_state(os.path.join(__lowercase,					"checkpoints",					"checkpoint_1"						)						)
										test_rands += train(1,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										((lowercase__)  ,       (lowercase__))					  =				model.a.item(), model.b.item()
										lowercase__					  =				optimizer.state_dict()
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)
										self.assertEqual(__lowercase,					__lowercase						)




				def 							A__      (						self					:						List[str]						):
							lowercase__					  =				torch.tensor([1, 2, 3]						)
							lowercase__					  =				torch.tensor([2, 3, 4]						)
							lowercase__					  =				DummyModel()
							lowercase__					  =				torch.optim.Adam(net.parameters()						)
							lowercase__					  =				Accelerator()
							with self.assertRaises(__lowercase						) as ve:
										accelerator.register_for_checkpointing(__lowercase,					__lowercase,					__lowercase,					__lowercase						)
							lowercase__					  =				str(ve.exception						)
							self.assertTrue("Item at index 0" in message						)
							self.assertTrue("Item at index 1" in message						)
							self.assertFalse("Item at index 2" in message						)
							self.assertFalse("Item at index 3" in message						)




				def 							A__      (						self					:						Union[str, Any]						):
							with tempfile.TemporaryDirectory() as tmpdir:
										set_seed(42						)
										lowercase__					  =				DummyModel()
										lowercase__					  =				torch.optim.Adam(params=model.parameters(),					lr=1e-3						)
										lowercase__					  =				torch.optim.lr_scheduler.StepLR(__lowercase,					step_size=1,					gamma=0.99						)
										lowercase__  ,       lowercase__					  =				dummy_dataloaders()
										lowercase__					  =				ProjectConfiguration(automatic_checkpoint_naming=__lowercase						)
										# Train baseline
										lowercase__					  =				Accelerator(project_dir=__lowercase,					project_config=__lowercase						)
										lowercase__  ,       lowercase__  ,       lowercase__  ,       lowercase__  ,       lowercase__					  =				accelerator.prepare(
										    __lowercase,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										# Save initial
										accelerator.save_state()
										lowercase__					  =				scheduler.state_dict()
										train(3,					__lowercase,					__lowercase,					__lowercase,					__lowercase,					__lowercase						)
										self.assertNotEqual(__lowercase,					scheduler.state_dict()						)

										# Load everything back in and make sure all states work
										accelerator.load_state(os.path.join(__lowercase,					"checkpoints",					"checkpoint_0"						)						)
										self.assertEqual(__lowercase,					scheduler.state_dict()						)




				def 							A__      (						self					:						Dict						):
							with tempfile.TemporaryDirectory() as tmpdir:
										set_seed(42						)
										lowercase__					  =				DummyModel()
										lowercase__					  =				ProjectConfiguration(automatic_checkpoint_naming=__lowercase,					total_limit=2						)
										# Train baseline
										lowercase__					  =				Accelerator(project_dir=__lowercase,					project_config=__lowercase						)
										lowercase__					  =				accelerator.prepare(__lowercase						)
										# Save 3 states:
										for _ in range(11						):
													accelerator.save_state()
										self.assertTrue(not os.path.exists(os.path.join(__lowercase,					"checkpoints",					"checkpoint_0"						)						)						)
										self.assertTrue(os.path.exists(os.path.join(__lowercase,					"checkpoints",					"checkpoint_9"						)						)						)
										self.assertTrue(os.path.exists(os.path.join(__lowercase,					"checkpoints",					"checkpoint_10"						)						)						)




				@require_cuda
				def 							A__      (						self					:						Tuple						):
							lowercase__					  =				["torchrun", F'''--nproc_per_node={torch.cuda.device_count()}''', inspect.getfile(self.__class__						)]
							execute_subprocess_async(__lowercase,					env=os.environ.copy()						)


if __name__ == "__main__":
							lowercase_								=     """/tmp/accelerate/state_checkpointing"""
							lowercase_								=     DummyModel()
							lowercase_								=     torch.optim.Adam(params=model.parameters(), lr=1e-3)
							lowercase_								=     torch.optim.lr_scheduler.StepLR(optimizer, step_size=1, gamma=0.99)
							lowercase_				,    lowercase_								=     dummy_dataloaders()
							lowercase_								=     ProjectConfiguration(automatic_checkpoint_naming=True)
							# Train baseline
							lowercase_								=     Accelerator(project_dir=savedir, project_config=project_config, mixed_precision="""no""")
							if accelerator.process_index == 0:
														if os.path.exists(savedir):
																					shutil.rmtree(savedir)
														os.makedirs(savedir)
							lowercase_				,    lowercase_				,    lowercase_				,    lowercase_				,    lowercase_								=     accelerator.prepare(
							    model, optimizer, train_dataloader, valid_dataloader, scheduler
							)
							lowercase_				,    lowercase_								=     accelerator.prepare(model, optimizer)
							train(3, model, train_dataloader, optimizer, accelerator, scheduler)
							# Check that the intial optimizer is loaded on the GPU
							for group in optimizer.param_groups:
														lowercase_								=     group["""params"""][0].device
														break
							assert param_device.type == accelerator.device.type
							lowercase_								=     model.cpu()
							accelerator.wait_for_everyone()
							accelerator.save_state()
							accelerator.wait_for_everyone()

							# Check CPU state
							accelerator.load_state(os.path.join(savedir, """checkpoints""", """checkpoint_0"""), map_location="""cpu""")
							for group in optimizer.param_groups:
														lowercase_								=     group["""params"""][0].device
														break
							assert (
							    param_device.type == torch.device("""cpu""").type
							), F"Loaded optimizer states did not match, expected to be loaded on the CPU but got {param_device}"

							# Check device state
							model.to(accelerator.device)
							accelerator.load_state(os.path.join(savedir, """checkpoints""", """checkpoint_0"""), map_location="""on_device""")
							for group in optimizer.param_groups:
														lowercase_								=     group["""params"""][0].device
														break
							assert (
							    param_device.type == accelerator.device.type
							), F"Loaded optimizer states did not match, expected to be loaded on {accelerator.device} but got {param_device}"

							# Check error
							with pytest.raises(TypeError, match="""Unsupported optimizer map location passed"""):
														accelerator.load_state(os.path.join(savedir, """checkpoints""", """checkpoint_0"""), map_location="""invalid""")
							accelerator.wait_for_everyone()
							if accelerator.process_index == 0:
														shutil.rmtree(savedir)
							accelerator.wait_for_everyone()









import math



def __lowerCAmelCase (     SCREAMING_SNAKE_CASE_						,  SCREAMING_SNAKE_CASE_  ):
			if 0 not in (x, y):
						# We use the relation x^y = y*log10(x), where 10 is the base.
						return y * math.logaa(SCREAMING_SNAKE_CASE_  )
			else:
						if x == 0:  # 0 raised to any number is 0
									return 0
						elif y == 0:
									return 1  # any number raised to 0 is 1
			raise AssertionError("This should never happen"  )


if __name__ == "__main__":  # Main function
							# Read two numbers from input and typecast them to int using map function.
							# Here x is the base and y is the power.
							lowercase_								=     """Enter the base and the power separated by a comma: """
							lowercase_				,    lowercase_								=     map(int, input(prompt).split(""","""))
							lowercase_				,    lowercase_								=     map(int, input(prompt).split(""","""))

							# We find the log of each number, using the function res(), which takes two
							# arguments.
							lowercase_								=     res(xa, ya)
							lowercase_								=     res(xa, ya)

							# We check for the largest number
							if resa > resa:
														print("""Largest number is""", xa, """^""", ya)
							elif resa > resa:
														print("""Largest number is""", xa, """^""", ya)
							else:
														print("""Both are equal""")




