



'''simple docstring'''





import inspect
import unittest

import torch
import torch.nn as nn

from accelerate.hooks import (
    AlignDevicesHook,
    ModelHook,
    SequentialHook,
    add_hook_to_module,
    attach_align_device_hook,
    remove_hook_from_module,
    remove_hook_from_submodules,
)
from accelerate.test_utils import require_multi_gpu






class __A     (    nn.Module		):



 '''simple docstring'''







 def __init__(self			)       ->	Dict:


      """simple docstring"""



      super().__init__()
      _a           =    nn.Linear(3					,  4			)
      _a           =    nn.BatchNormad(4			)
      _a           =    nn.Linear(4					,  5			)



 def    a__    (self					,  A			)       ->	Dict:


      """simple docstring"""



      return self.lineara(self.batchnorm(self.lineara(A			)			)			)






class __A     (    A		):



 '''simple docstring'''







 def    a__    (self					,  A					,  *A					,  **A			)       ->	Optional[Any]:


      """simple docstring"""



      return (args[0] + 1,) + args[1:], kwargs






class __A     (    A		):



 '''simple docstring'''







 def    a__    (self					,  A					,  A			)       ->	int:


      """simple docstring"""



      return output + 1






class __A     (    unittest.TestCase		):



 '''simple docstring'''







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


      """simple docstring"""



      _a           =    ModelForTest()
      _a           =    ModelHook()

      add_hook_to_module(A					,  A			)
      self.assertEqual(test_model._hf_hook					,  A			)
      self.assertTrue(hasattr(A					,  '''_old_forward'''			)			)

      # Check adding the hook did not change the name or the signature
      self.assertEqual(test_model.forward.__name__					,  '''forward'''			)
      self.assertListEqual(list(inspect.signature(test_model.forward			).parameters			)					,  ['''x''']			)

      remove_hook_from_module(A			)
      self.assertFalse(hasattr(A					,  '''_hf_hook'''			)			)
      self.assertFalse(hasattr(A					,  '''_old_forward'''			)			)



 def    a__    (self			)       ->	Any:


      """simple docstring"""



      _a           =    ModelForTest()
      _a           =    ModelHook()

      add_hook_to_module(A					,  A			)
      add_hook_to_module(A					,  A					,  append=A			)

      self.assertEqual(isinstance(test_model._hf_hook					,  A			)					,  A			)
      self.assertEqual(len(test_model._hf_hook.hooks			)					,  2			)
      self.assertTrue(hasattr(A					,  '''_old_forward'''			)			)

      # Check adding the hook did not change the name or the signature
      self.assertEqual(test_model.forward.__name__					,  '''forward'''			)
      self.assertListEqual(list(inspect.signature(test_model.forward			).parameters			)					,  ['''x''']			)

      remove_hook_from_module(A			)
      self.assertFalse(hasattr(A					,  '''_hf_hook'''			)			)
      self.assertFalse(hasattr(A					,  '''_old_forward'''			)			)



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


      """simple docstring"""



      _a           =    ModelForTest()
      _a           =    torch.randn(2					,  3			)
      _a           =    test_model(x + 1			)
      _a           =    test_model(x + 2			)

      _a           =    PreForwardHook()
      add_hook_to_module(A					,  A			)
      _a           =    test_model(A			)
      self.assertTrue(torch.allclose(A					,  A					,  atol=1E-5			)			)

      # Attaching a hook to a model when it already has one replaces, does not chain
      _a           =    PreForwardHook()
      add_hook_to_module(A					,  A			)
      _a           =    test_model(A			)
      self.assertTrue(torch.allclose(A					,  A					,  atol=1E-5			)			)

      # You need to use the sequential hook to chain two or more hooks
      _a           =    SequentialHook(PreForwardHook()					,  PreForwardHook()			)
      add_hook_to_module(A					,  A			)

      _a           =    test_model(A			)
      assert torch.allclose(A					,  A					,  atol=1E-5			)



 def    a__    (self			)       ->	str:


      """simple docstring"""



      _a           =    ModelForTest()
      _a           =    torch.randn(2					,  3			)
      _a           =    test_model(A			)

      _a           =    PostForwardHook()
      add_hook_to_module(A					,  A			)
      _a           =    test_model(A			)
      self.assertTrue(torch.allclose(A					,  output + 1					,  atol=1E-5			)			)

      # Attaching a hook to a model when it already has one replaces, does not chain
      _a           =    PostForwardHook()
      add_hook_to_module(A					,  A			)
      _a           =    test_model(A			)
      self.assertTrue(torch.allclose(A					,  output + 1					,  atol=1E-5			)			)

      # You need to use the sequential hook to chain two or more hooks
      _a           =    SequentialHook(PostForwardHook()					,  PostForwardHook()			)
      add_hook_to_module(A					,  A			)

      _a           =    test_model(A			)
      assert torch.allclose(A					,  output + 2					,  atol=1E-5			)



 def    a__    (self			)       ->	List[str]:


      """simple docstring"""



      _a           =    ModelForTest()
      _a           =    torch.randn(2					,  3			)
      _a           =    test_model(A			)

      _a           =    PostForwardHook()
      add_hook_to_module(A					,  A			)
      _a           =    test_model(A			)
      self.assertTrue(torch.allclose(A					,  output + 1			)			)
      self.assertTrue(outputa.requires_grad			)

      _a           =    True
      _a           =    test_model(A			)
      self.assertFalse(outputa.requires_grad			)



 @require_multi_gpu
 def    a__    (self			)       ->	List[Any]:


      """simple docstring"""



      _a           =    ModelForTest()
      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)

      # This will move each submodule on different devices
      add_hook_to_module(model.lineara					,  AlignDevicesHook(execution_device=0			)			)
      add_hook_to_module(model.batchnorm					,  AlignDevicesHook(execution_device=0			)			)
      add_hook_to_module(model.lineara					,  AlignDevicesHook(execution_device=1			)			)

      self.assertEqual(model.lineara.weight.device					,  torch.device(0			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device(0			)			)
      self.assertEqual(model.batchnorm.running_mean.device					,  torch.device(0			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device(1			)			)

      # We can still make a forward pass. The input does not need to be on any particular device
      _a           =    torch.randn(2					,  3			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  torch.device(1			)			)

      # We can add a general hook to put back output on same device as input.
      add_hook_to_module(A					,  AlignDevicesHook(io_same_device=A			)			)
      _a           =    torch.randn(2					,  3			).to(0			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  torch.device(0			)			)



 def    a__    (self			)       ->	List[str]:


      """simple docstring"""



      _a           =    ModelForTest()

      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)

      # This will move each submodule on different devices
      _a           =    {'''execution_device''': 0 if torch.cuda.is_available() else '''cpu''', '''offload''': True}

      add_hook_to_module(model.lineara					,  AlignDevicesHook(**A			)			)
      add_hook_to_module(model.batchnorm					,  AlignDevicesHook(**A			)			)
      add_hook_to_module(model.lineara					,  AlignDevicesHook(**A			)			)

      # Parameters have been offloaded, so on the meta device
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      # Buffers are not included in the offload by default, so are on the execution device
      _a           =    torch.device(hook_kwargs['''execution_device''']			)
      self.assertEqual(model.batchnorm.running_mean.device					,  A			)

      _a           =    torch.randn(2					,  3			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  A			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_module(model.lineara			)
      remove_hook_from_module(model.batchnorm			)
      remove_hook_from_module(model.lineara			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)

      # Now test with buffers included in the offload
      _a           =    {
          '''execution_device''': 0 if torch.cuda.is_available() else '''cpu''',
          '''offload''': True,
          '''offload_buffers''': True,
      }

      add_hook_to_module(model.lineara					,  AlignDevicesHook(**A			)			)
      add_hook_to_module(model.batchnorm					,  AlignDevicesHook(**A			)			)
      add_hook_to_module(model.lineara					,  AlignDevicesHook(**A			)			)

      # Parameters have been offloaded, so on the meta device, buffers included
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.running_mean.device					,  torch.device('''meta'''			)			)

      _a           =    torch.randn(2					,  3			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  A			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_module(model.lineara			)
      remove_hook_from_module(model.batchnorm			)
      remove_hook_from_module(model.lineara			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)



 def    a__    (self			)       ->	Optional[int]:


      """simple docstring"""



      _a           =    ModelForTest()

      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)

      # This will move each submodule on different devices
      _a           =    0 if torch.cuda.is_available() else '''cpu'''
      attach_align_device_hook(A					,  execution_device=A					,  offload=A			)

      # Parameters have been offloaded, so on the meta device
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      # Buffers are not included in the offload by default, so are on the execution device
      _a           =    torch.device(A			)
      self.assertEqual(model.batchnorm.running_mean.device					,  A			)

      _a           =    torch.randn(2					,  3			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  A			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(A			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)

      # Now test with buffers included in the offload
      attach_align_device_hook(A					,  execution_device=A					,  offload=A					,  offload_buffers=A			)

      # Parameters have been offloaded, so on the meta device, buffers included
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.running_mean.device					,  torch.device('''meta'''			)			)

      _a           =    torch.randn(2					,  3			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  A			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(A			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)



 def    a__    (self			)       ->	Any:


      """simple docstring"""



      _a           =    ModelForTest()

      # Everything is on CPU
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)

      # This will move each submodule on different devices
      _a           =    0 if torch.cuda.is_available() else '''cpu'''
      attach_align_device_hook(
          A					,  execution_device=A					,  offload=A					,  weights_map=model.state_dict()			)

      # Parameters have been offloaded, so on the meta device
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      # Buffers are not included in the offload by default, so are on the execution device
      _a           =    torch.device(A			)
      self.assertEqual(model.batchnorm.running_mean.device					,  A			)

      _a           =    torch.randn(2					,  3			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  A			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(A			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)

      # Now test with buffers included in the offload
      attach_align_device_hook(
          A					,  execution_device=A					,  offload=A					,  weights_map=model.state_dict()					,  offload_buffers=A					,  )

      # Parameters have been offloaded, so on the meta device, buffers included
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''meta'''			)			)
      self.assertEqual(model.batchnorm.running_mean.device					,  torch.device('''meta'''			)			)

      _a           =    torch.randn(2					,  3			)
      _a           =    model(A			)
      self.assertEqual(output.device					,  A			)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(A			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.batchnorm.weight.device					,  torch.device('''cpu'''			)			)
      self.assertEqual(model.lineara.weight.device					,  torch.device('''cpu'''			)			)






def 						lowerCamelCase__      ( __A    :int	,__A    :float	,__A    :float		):


       """simple docstring"""




       return round(float(moles / volume		) * nfactor		)







def 						lowerCamelCase__      ( __A    :float	,__A    :float	,__A    :float		):


       """simple docstring"""




       return round(float((moles * 0.0_821 * temperature) / (volume)		)		)







def 						lowerCamelCase__      ( __A    :float	,__A    :float	,__A    :float		):


       """simple docstring"""




       return round(float((moles * 0.0_821 * temperature) / (pressure)		)		)







def 						lowerCamelCase__      ( __A    :float	,__A    :float	,__A    :float		):


       """simple docstring"""




       return round(float((pressure * volume) / (0.0_821 * moles)		)		)


if __name__ == "__main__":
    import doctest

    doctest.testmod()

