


"""simple docstring"""






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 					A_      (      ):






     """simple docstring"""

     _a					= torch.nn.Linear(2,   4				)
     _a					= torch.optim.AdamW(model.parameters(),   lr=1.0				)
     _a					= torch.optim.lr_scheduler.OneCycleLR(A_,   max_lr=0.0_1,   steps_per_epoch=2,   epochs=1				)
     _a					= DataLoader(TensorDataset(torch.tensor([1, 2, 3]				)				)				)
     _a					= DataLoader(TensorDataset(torch.tensor([4, 5, 6]				)				)				)

     return model, optimizer, scheduler, train_dl, valid_dl
def 					A_      (      _lowerCAmelCase   :    int				):






     """simple docstring"""

     return (model.weight.abs().sum() + model.bias.abs().sum()).item()
def 					A_      (      _lowerCAmelCase   :    Any				):






     """simple docstring"""

     _a					= torch.nn.Linear(*tuple(model.weight.T.shape				)				).state_dict()
     model.load_state_dict(A_				)


class 			__lowerCamelCase     (    a_ ):

     '''simple docstring'''


     @require_cuda
     def 							_UpperCAmelCase					(					self							)				->							str:
          _a					= Accelerator()
          assert PartialState._shared_state["_cpu"] is False
          assert PartialState._shared_state["device"].type == "cuda"
          with self.assertRaises(lowercase_							):
               _a					= Accelerator(cpu=lowercase_							)


     def 							_UpperCAmelCase					(					self							)				->							Any:
          _a					= Accelerator()
          _a					= GradientState()
          assert state.num_steps == 1
          _a					= 4
          assert state.num_steps == 4

          assert state.sync_gradients is True
          _a					= False
          assert state.sync_gradients is False
          GradientState._reset_state()


     def 							_UpperCAmelCase					(					self							)				->							Dict:
          _a					= Accelerator()
          _a					= create_components()

          (
              _a
          )					= accelerator.prepare(lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_							)

          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 							_UpperCAmelCase					(					self							)				->							List[str]:
          _a					= Accelerator()
          _a					= create_components()
          accelerator.prepare(lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_							)
          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 							_UpperCAmelCase					(					self							)				->							List[str]:
          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'''			, lowercase_							), patch_environment(ACCELERATE_TORCH_DEVICE='''cuda:64'''							):
               _a					= Accelerator()
               self.assertEqual(str(accelerator.state.device							)			, '''cuda:64'''							)


     def 							_UpperCAmelCase					(					self							)				->							Dict:
          _a					= Accelerator()
          _a					= create_components()
          accelerator.prepare(lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_							)

          _a					= get_signature(lowercase_							)

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

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

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


     def 							_UpperCAmelCase					(					self							)				->							Union[str, Any]:
          _a					= Accelerator()
          _a					= create_components()
          accelerator.prepare(lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_							)

          _a					= get_signature(lowercase_							)

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

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

        # loading hook
          def load_config(__UpperCAmelCase			, __UpperCAmelCase							):
               with open(os.path.join(lowercase_			, '''data.json'''							)			, '''r'''							) as f:
                    _a					= json.load(lowercase_							)

               _a					= config['''class_name''']

          _a					= accelerator.register_save_state_pre_hook(lowercase_							)
          _a					= accelerator.register_load_state_pre_hook(lowercase_							)

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

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

               # random class name to verify correct one is loaded
               _a					= '''random'''

               # make sure loaded weights match with hooks
               accelerator.load_state(lowercase_							)
               self.assertTrue(abs(model_signature - get_signature(lowercase_							)							) < 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(lowercase_							)

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

               # random class name to verify correct one is loaded
               _a					= '''random'''

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

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


     def 							_UpperCAmelCase					(					self							)				->							Union[str, Any]:
          _a					= Accelerator()
          _a					= create_components()
          _a					= None

          # This should work
          _a					= accelerator.prepare(
              lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_							)
          self.assertTrue(dummy_obj is None							)


     def 							_UpperCAmelCase					(					self							)				->							List[Any]:
          _a					= Accelerator()
          _a					= create_components()
          _a					= [1, 2, 3]

          # This should work
          _a					= accelerator.prepare(
              lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_			, lowercase_							)
          self.assertEqual(
              getattr(lowercase_			, '''_is_accelerate_prepared'''			, lowercase_							)			, lowercase_			, '''Dummy object should have `_is_accelerate_prepared` set to `True`'''			, )
          self.assertEqual(
              getattr(lowercase_			, '''_is_accelerate_prepared'''			, lowercase_							)			, lowercase_			, '''Model is missing `_is_accelerator_prepared` or is set to `False`'''			, )
          self.assertEqual(
              getattr(lowercase_			, '''_is_accelerate_prepared'''			, lowercase_							)			, lowercase_			, '''Optimizer is missing `_is_accelerator_prepared` or is set to `False`'''			, )
          self.assertEqual(
              getattr(lowercase_			, '''_is_accelerate_prepared'''			, lowercase_							)			, lowercase_			, '''Scheduler is missing `_is_accelerator_prepared` or is set to `False`'''			, )
          self.assertEqual(
              getattr(lowercase_			, '''_is_accelerate_prepared'''			, lowercase_							)			, lowercase_			, '''Train Dataloader is missing `_is_accelerator_prepared` or is set to `False`'''			, )
          self.assertEqual(
              getattr(lowercase_			, '''_is_accelerate_prepared'''			, lowercase_							)			, lowercase_			, '''Valid Dataloader is missing `_is_accelerator_prepared` or is set to `False`'''			, )


     @slow
     @require_bnb
     def 							_UpperCAmelCase					(					self							)				->							Union[str, Any]:
          from transformers import AutoModelForCausalLM

          _a					= AutoModelForCausalLM.from_pretrained(
              '''EleutherAI/gpt-neo-125m'''			, load_in_abit=lowercase_			, device_map={'''''': 0}			, )
          _a					= Accelerator()

          # This should work
          _a					= accelerator.prepare(lowercase_							)


     @slow
     @require_bnb
     def 							_UpperCAmelCase					(					self							)				->							Optional[int]:
          from transformers import AutoModelForCausalLM

          _a					= Accelerator()

          with init_empty_weights():
               _a					= AutoModelForCausalLM.from_pretrained(
                   '''EleutherAI/gpt-neo-125m'''			, )
               model.tie_weights()
               _a					= infer_auto_device_map(lowercase_							)
               _a					= '''cpu'''

          _a					= AutoModelForCausalLM.from_pretrained(
              '''EleutherAI/gpt-neo-125m'''			, device_map=lowercase_			, load_in_abit=lowercase_			, llm_inta_enable_fpaa_cpu_offload=lowercase_							)

          # This should not work and get value error
          with self.assertRaises(lowercase_							):
               _a					= accelerator.prepare(lowercase_							)


     @slow
     @require_bnb
     @require_multi_gpu
     def 							_UpperCAmelCase					(					self							)				->							List[Any]:
          from transformers import AutoModelForCausalLM

          _a					= {'''distributed_type''': DistributedType.MULTI_GPU}

          with init_empty_weights():
               _a					= AutoModelForCausalLM.from_pretrained(
                   '''EleutherAI/gpt-neo-125m'''			, )
               model.tie_weights()
               _a					= infer_auto_device_map(lowercase_							)
               _a					= 1

          _a					= AutoModelForCausalLM.from_pretrained(
              '''EleutherAI/gpt-neo-125m'''			, load_in_abit=lowercase_			, device_map=lowercase_			, )
          _a					= Accelerator()

          # This should not work and get value error
          with self.assertRaises(lowercase_							):
               _a					= accelerator.prepare(lowercase_							)

          PartialState._reset_state()


     @slow
     @require_bnb
     @require_multi_gpu
     def 							_UpperCAmelCase					(					self							)				->							Optional[int]:
          from transformers import AutoModelForCausalLM

          with init_empty_weights():
               _a					= AutoModelForCausalLM.from_pretrained(
                   '''EleutherAI/gpt-neo-125m'''			, )
               _a					= infer_auto_device_map(lowercase_							)
               _a					= 1

          _a					= AutoModelForCausalLM.from_pretrained(
              '''EleutherAI/gpt-neo-125m'''			, load_in_abit=lowercase_			, device_map=lowercase_			, )
          _a					= Accelerator()

          # This should work
          _a					= accelerator.prepare(lowercase_							)





     @require_cuda
     def 							_UpperCAmelCase					(					self							)				->							List[str]:
          _a					= torch.nn.Linear(10			, 10							)
          _a					= torch.optim.SGD(model.parameters()			, lr=0.01							)
          _a					= Accelerator(cpu=lowercase_							)
          _a					= accelerator.prepare(lowercase_							)


"""simple docstring"""





def   __SCREAMING_SNAKE_CASE  (      A_						,     A_ ):
	return numa ^ numa < 0


if __name__ == "__main__":
							import doctest

							doctest.testmod()




