





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     lowercase__       (	nn.Module				):

   '''simple docstring'''


   def __init__(    self							) ->      Optional[Any]:






       """simple docstring"""
       super().__init__()
       UpperCamelCase__							:				int          =      nn.Linear(3,		4							)
       UpperCamelCase__							:				List[str]          =      nn.BatchNormad(4							)
       UpperCamelCase__							:				str          =      nn.Linear(4,		5							)




   def       UpperCamelCase__  (    self,		__magic_name__							) ->      Tuple:






       """simple docstring"""
       return self.lineara(self.batchnorm(self.lineara(__lowerCAmelCase							)							)							)






class     lowercase__       (	a__				):

   '''simple docstring'''


   def       UpperCamelCase__  (    self,		__magic_name__,		*__magic_name__,		**__magic_name__							) ->      List[str]:






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






class     lowercase__       (	a__				):

   '''simple docstring'''


   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__							) ->      Tuple:






       """simple docstring"""
       return output + 1






class     lowercase__       (	unittest.TestCase				):

   '''simple docstring'''


   def       UpperCamelCase__  (    self							) ->      Dict:






       """simple docstring"""
       UpperCamelCase__							:				List[str]          =      ModelForTest()
       UpperCamelCase__							:				str          =      ModelHook()

       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)
       self.assertEqual(test_model._hf_hook,		__lowerCAmelCase							)
       self.assertTrue(hasattr(__lowerCAmelCase,		'''_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(__lowerCAmelCase							)
       self.assertFalse(hasattr(__lowerCAmelCase,		'''_hf_hook'''							)							)
       self.assertFalse(hasattr(__lowerCAmelCase,		'''_old_forward'''							)							)




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






       """simple docstring"""
       UpperCamelCase__							:				Optional[Any]          =      ModelForTest()
       UpperCamelCase__							:				List[str]          =      ModelHook()

       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase,		append=__lowerCAmelCase							)

       self.assertEqual(isinstance(test_model._hf_hook,		__lowerCAmelCase							),		__lowerCAmelCase							)
       self.assertEqual(len(test_model._hf_hook.hooks							),		2							)
       self.assertTrue(hasattr(__lowerCAmelCase,		'''_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(__lowerCAmelCase							)
       self.assertFalse(hasattr(__lowerCAmelCase,		'''_hf_hook'''							)							)
       self.assertFalse(hasattr(__lowerCAmelCase,		'''_old_forward'''							)							)




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






       """simple docstring"""
       UpperCamelCase__							:				List[str]          =      ModelForTest()
       UpperCamelCase__							:				str          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Union[str, Any]          =      test_model(x + 1							)
       UpperCamelCase__							:				Optional[Any]          =      test_model(x + 2							)

       UpperCamelCase__							:				str          =      PreForwardHook()
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)
       UpperCamelCase__							:				List[Any]          =      test_model(__lowerCAmelCase							)
       self.assertTrue(torch.allclose(__lowerCAmelCase,		__lowerCAmelCase,		atol=1E-5							)							)

       # Attaching a hook to a model when it already has one replaces, does not chain
       UpperCamelCase__							:				int          =      PreForwardHook()
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)
       UpperCamelCase__							:				Optional[int]          =      test_model(__lowerCAmelCase							)
       self.assertTrue(torch.allclose(__lowerCAmelCase,		__lowerCAmelCase,		atol=1E-5							)							)

       # You need to use the sequential hook to chain two or more hooks
       UpperCamelCase__							:				List[Any]          =      SequentialHook(PreForwardHook(),		PreForwardHook()							)
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)

       UpperCamelCase__							:				Dict          =      test_model(__lowerCAmelCase							)
       assert torch.allclose(__lowerCAmelCase,		__lowerCAmelCase,		atol=1E-5							)




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






       """simple docstring"""
       UpperCamelCase__							:				Union[str, Any]          =      ModelForTest()
       UpperCamelCase__							:				List[Any]          =      torch.randn(2,		3							)
       UpperCamelCase__							:				List[str]          =      test_model(__lowerCAmelCase							)

       UpperCamelCase__							:				Optional[int]          =      PostForwardHook()
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)
       UpperCamelCase__							:				List[Any]          =      test_model(__lowerCAmelCase							)
       self.assertTrue(torch.allclose(__lowerCAmelCase,		output + 1,		atol=1E-5							)							)

       # Attaching a hook to a model when it already has one replaces, does not chain
       UpperCamelCase__							:				int          =      PostForwardHook()
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)
       UpperCamelCase__							:				Optional[int]          =      test_model(__lowerCAmelCase							)
       self.assertTrue(torch.allclose(__lowerCAmelCase,		output + 1,		atol=1E-5							)							)

       # You need to use the sequential hook to chain two or more hooks
       UpperCamelCase__							:				Tuple          =      SequentialHook(PostForwardHook(),		PostForwardHook()							)
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)

       UpperCamelCase__							:				List[Any]          =      test_model(__lowerCAmelCase							)
       assert torch.allclose(__lowerCAmelCase,		output + 2,		atol=1E-5							)




   def       UpperCamelCase__  (    self							) ->      Tuple:






       """simple docstring"""
       UpperCamelCase__							:				Optional[int]          =      ModelForTest()
       UpperCamelCase__							:				int          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Any          =      test_model(__lowerCAmelCase							)

       UpperCamelCase__							:				Tuple          =      PostForwardHook()
       add_hook_to_module(__lowerCAmelCase,		__lowerCAmelCase							)
       UpperCamelCase__							:				int          =      test_model(__lowerCAmelCase							)
       self.assertTrue(torch.allclose(__lowerCAmelCase,		output + 1							)							)
       self.assertTrue(outputa.requires_grad							)

       UpperCamelCase__							:				str          =      True
       UpperCamelCase__							:				List[str]          =      test_model(__lowerCAmelCase							)
       self.assertFalse(outputa.requires_grad							)




   @require_multi_gpu
   def       UpperCamelCase__  (    self							) ->      int:






       """simple docstring"""
       UpperCamelCase__							:				List[str]          =      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
       UpperCamelCase__							:				Optional[Any]          =      torch.randn(2,		3							)
       UpperCamelCase__							:				int          =      model(__lowerCAmelCase							)
       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(__lowerCAmelCase,		AlignDevicesHook(io_same_device=__lowerCAmelCase							)							)
       UpperCamelCase__							:				Optional[Any]          =      torch.randn(2,		3							).to(0							)
       UpperCamelCase__							:				int          =      model(__lowerCAmelCase							)
       self.assertEqual(output.device,		torch.device(0							)							)




   def       UpperCamelCase__  (    self							) ->      int:






       """simple docstring"""
       UpperCamelCase__							:				str          =      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
       UpperCamelCase__							:				Any          =      {'''execution_device''': 0 if torch.cuda.is_available() else '''cpu''', '''offload''': True}

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

       # 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
       UpperCamelCase__							:				Any          =      torch.device(hook_kwargs['''execution_device''']							)
       self.assertEqual(model.batchnorm.running_mean.device,		__lowerCAmelCase							)

       UpperCamelCase__							:				str          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Any          =      model(__lowerCAmelCase							)
       self.assertEqual(output.device,		__lowerCAmelCase							)

       # 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
       UpperCamelCase__							:				Union[str, Any]          =      {
           '''execution_device''': 0 if torch.cuda.is_available() else '''cpu''',
           '''offload''': True,
           '''offload_buffers''': True,
       }

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

       # 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'''							)							)

       UpperCamelCase__							:				str          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Union[str, Any]          =      model(__lowerCAmelCase							)
       self.assertEqual(output.device,		__lowerCAmelCase							)

       # 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       UpperCamelCase__  (    self							) ->      int:






       """simple docstring"""
       UpperCamelCase__							:				Union[str, Any]          =      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
       UpperCamelCase__							:				int          =      0 if torch.cuda.is_available() else '''cpu'''
       attach_align_device_hook(__lowerCAmelCase,		execution_device=__lowerCAmelCase,		offload=__lowerCAmelCase							)

       # 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
       UpperCamelCase__							:				Dict          =      torch.device(__lowerCAmelCase							)
       self.assertEqual(model.batchnorm.running_mean.device,		__lowerCAmelCase							)

       UpperCamelCase__							:				Any          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Union[str, Any]          =      model(__lowerCAmelCase							)
       self.assertEqual(output.device,		__lowerCAmelCase							)

       # Removing hooks loads back the weights in the model.
       remove_hook_from_submodules(__lowerCAmelCase							)
       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(__lowerCAmelCase,		execution_device=__lowerCAmelCase,		offload=__lowerCAmelCase,		offload_buffers=__lowerCAmelCase							)

       # 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'''							)							)

       UpperCamelCase__							:				Any          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Any          =      model(__lowerCAmelCase							)
       self.assertEqual(output.device,		__lowerCAmelCase							)

       # Removing hooks loads back the weights in the model.
       remove_hook_from_submodules(__lowerCAmelCase							)
       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       UpperCamelCase__  (    self							) ->      int:






       """simple docstring"""
       UpperCamelCase__							:				Optional[int]          =      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
       UpperCamelCase__							:				Dict          =      0 if torch.cuda.is_available() else '''cpu'''
       attach_align_device_hook(
           __lowerCAmelCase,		execution_device=__lowerCAmelCase,		offload=__lowerCAmelCase,		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
       UpperCamelCase__							:				Optional[int]          =      torch.device(__lowerCAmelCase							)
       self.assertEqual(model.batchnorm.running_mean.device,		__lowerCAmelCase							)

       UpperCamelCase__							:				List[Any]          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Tuple          =      model(__lowerCAmelCase							)
       self.assertEqual(output.device,		__lowerCAmelCase							)

       # Removing hooks loads back the weights in the model.
       remove_hook_from_submodules(__lowerCAmelCase							)
       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(
           __lowerCAmelCase,		execution_device=__lowerCAmelCase,		offload=__lowerCAmelCase,		weights_map=model.state_dict(),		offload_buffers=__lowerCAmelCase,		)

       # 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'''							)							)

       UpperCamelCase__							:				Dict          =      torch.randn(2,		3							)
       UpperCamelCase__							:				Tuple          =      model(__lowerCAmelCase							)
       self.assertEqual(output.device,		__lowerCAmelCase							)

       # Removing hooks loads back the weights in the model.
       remove_hook_from_submodules(__lowerCAmelCase							)
       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'''							)							)


'''simple docstring'''

from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


__a: Dict          =   {
    """configuration_ctrl""": ["""CTRL_PRETRAINED_CONFIG_ARCHIVE_MAP""", """CTRLConfig"""],
    """tokenization_ctrl""": ["""CTRLTokenizer"""],
}

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  __a: int          =   [
      """CTRL_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """CTRLForSequenceClassification""",
      """CTRLLMHeadModel""",
      """CTRLModel""",
      """CTRLPreTrainedModel""",
  ]

try:
  if not is_tf_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  __a: int          =   [
      """TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """TFCTRLForSequenceClassification""",
      """TFCTRLLMHeadModel""",
      """TFCTRLModel""",
      """TFCTRLPreTrainedModel""",
  ]


if TYPE_CHECKING:
  from .configuration_ctrl import CTRL_PRETRAINED_CONFIG_ARCHIVE_MAP, CTRLConfig
  from .tokenization_ctrl import CTRLTokenizer

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_ctrl import (
        CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
        CTRLForSequenceClassification,
        CTRLLMHeadModel,
        CTRLModel,
        CTRLPreTrainedModel,
    )

  try:
    if not is_tf_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_tf_ctrl import (
        TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
        TFCTRLForSequenceClassification,
        TFCTRLLMHeadModel,
        TFCTRLModel,
        TFCTRLPreTrainedModel,
    )

else:
  import sys

  __a: Dict          =   _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)

