



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







    def __init__(      self :  Any	):
      super().__init__()
      a						:					Tuple        =     nn.Linear(3						,			4	)
      a						:					Optional[Any]        =     nn.BatchNormad(4	)
      a						:					Dict        =     nn.Linear(4						,			5	)







    def 			lowercase_       (      self :  Optional[int]						,			__snake_case :  List[Any]	):
      return self.lineara(self.batchnorm(self.lineara(__snake_case	)	)	)

class 							a__(					lowerCamelCase__	):







    def 			lowercase_       (      self :  Optional[Any]						,			__snake_case :  Tuple						,			*__snake_case :  Tuple						,			**__snake_case :  str	):
      return (args[0] + 1,) + args[1:], kwargs

class 							a__(					lowerCamelCase__	):







    def 			lowercase_       (      self :  List[str]						,			__snake_case :  List[Any]						,			__snake_case :  Union[str, Any]	):
      return output + 1

class 							a__(					unittest.TestCase	):







    def 			lowercase_       (      self :  List[str]	):
      a						:					List[str]        =     ModelForTest()
      a						:					Optional[int]        =     ModelHook()

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







    def 			lowercase_       (      self :  List[Any]	):
      a						:					Optional[int]        =     ModelForTest()
      a						:					Optional[int]        =     ModelHook()

      add_hook_to_module(__snake_case						,			__snake_case	)
      add_hook_to_module(__snake_case						,			__snake_case						,			append=__snake_case	)

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







    def 			lowercase_       (      self :  List[Any]	):
      a						:					Tuple        =     ModelForTest()
      a						:					Optional[Any]        =     torch.randn(2						,			3	)
      a						:					Optional[int]        =     test_model(x + 1	)
      a						:					str        =     test_model(x + 2	)

      a						:					List[Any]        =     PreForwardHook()
      add_hook_to_module(__snake_case						,			__snake_case	)
      a						:					Tuple        =     test_model(__snake_case	)
      self.assertTrue(torch.allclose(__snake_case						,			__snake_case						,			atol=1e-5	)	)

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

      # You need to use the sequential hook to chain two or more hooks
      a						:					Dict        =     SequentialHook(PreForwardHook()						,			PreForwardHook()	)
      add_hook_to_module(__snake_case						,			__snake_case	)

      a						:					int        =     test_model(__snake_case	)
      assert torch.allclose(__snake_case						,			__snake_case						,			atol=1e-5	)







    def 			lowercase_       (      self :  Optional[Any]	):
      a						:					Optional[int]        =     ModelForTest()
      a						:					Dict        =     torch.randn(2						,			3	)
      a						:					Union[str, Any]        =     test_model(__snake_case	)

      a						:					Any        =     PostForwardHook()
      add_hook_to_module(__snake_case						,			__snake_case	)
      a						:					str        =     test_model(__snake_case	)
      self.assertTrue(torch.allclose(__snake_case						,			output + 1						,			atol=1e-5	)	)

      # Attaching a hook to a model when it already has one replaces, does not chain
      a						:					List[Any]        =     PostForwardHook()
      add_hook_to_module(__snake_case						,			__snake_case	)
      a						:					Dict        =     test_model(__snake_case	)
      self.assertTrue(torch.allclose(__snake_case						,			output + 1						,			atol=1e-5	)	)

      # You need to use the sequential hook to chain two or more hooks
      a						:					Optional[int]        =     SequentialHook(PostForwardHook()						,			PostForwardHook()	)
      add_hook_to_module(__snake_case						,			__snake_case	)

      a						:					Union[str, Any]        =     test_model(__snake_case	)
      assert torch.allclose(__snake_case						,			output + 2						,			atol=1e-5	)







    def 			lowercase_       (      self :  List[str]	):
      a						:					Optional[Any]        =     ModelForTest()
      a						:					Optional[int]        =     torch.randn(2						,			3	)
      a						:					Union[str, Any]        =     test_model(__snake_case	)

      a						:					Optional[Any]        =     PostForwardHook()
      add_hook_to_module(__snake_case						,			__snake_case	)
      a						:					int        =     test_model(__snake_case	)
      self.assertTrue(torch.allclose(__snake_case						,			output + 1	)	)
      self.assertTrue(outputa.requires_grad	)

      a						:					Union[str, Any]        =     True
      a						:					Union[str, Any]        =     test_model(__snake_case	)
      self.assertFalse(outputa.requires_grad	)







    @require_multi_gpu
    def 			lowercase_       (      self :  List[str]	):
      a						:					List[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
      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						:					Union[str, Any]        =     torch.randn(2						,			3	)
      a						:					List[str]        =     model(__snake_case	)
      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(__snake_case						,			AlignDevicesHook(io_same_device=__snake_case	)	)
      a						:					Dict        =     torch.randn(2						,			3	).to(0	)
      a						:					Tuple        =     model(__snake_case	)
      self.assertEqual(output.device						,			torch.device(0	)	)







    def 			lowercase_       (      self :  str	):
      a						:					List[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
      a						:					Dict        =     {'''execution_device''': 0 if torch.cuda.is_available() else '''cpu''', '''offload''': True}

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

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

      a						:					Optional[Any]        =     torch.randn(2						,			3	)
      a						:					Dict        =     model(__snake_case	)
      self.assertEqual(output.device						,			__snake_case	)

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

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

      # 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						:					Optional[int]        =     torch.randn(2						,			3	)
      a						:					Optional[int]        =     model(__snake_case	)
      self.assertEqual(output.device						,			__snake_case	)

      # 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 			lowercase_       (      self :  List[Any]	):
      a						:					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
      a						:					Any        =     0 if torch.cuda.is_available() else '''cpu'''
      attach_align_device_hook(__snake_case						,			execution_device=__snake_case						,			offload=__snake_case	)

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

      a						:					str        =     torch.randn(2						,			3	)
      a						:					Tuple        =     model(__snake_case	)
      self.assertEqual(output.device						,			__snake_case	)

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

      # 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						:					str        =     torch.randn(2						,			3	)
      a						:					Tuple        =     model(__snake_case	)
      self.assertEqual(output.device						,			__snake_case	)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(__snake_case	)
      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 			lowercase_       (      self :  List[str]	):
      a						:					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
      a						:					Optional[Any]        =     0 if torch.cuda.is_available() else '''cpu'''
      attach_align_device_hook(
          __snake_case						,			execution_device=__snake_case						,			offload=__snake_case						,			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						:					Any        =     torch.device(__snake_case	)
      self.assertEqual(model.batchnorm.running_mean.device						,			__snake_case	)

      a						:					Any        =     torch.randn(2						,			3	)
      a						:					Optional[int]        =     model(__snake_case	)
      self.assertEqual(output.device						,			__snake_case	)

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

      # 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						:					Tuple        =     torch.randn(2						,			3	)
      a						:					Dict        =     model(__snake_case	)
      self.assertEqual(output.device						,			__snake_case	)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(__snake_case	)
      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 ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCAmelCase    :int     =					logging.get_logger(__name__)

_lowerCAmelCase    :Union[str, Any]     =					{
    'alibaba-damo/mgp-str-base': 'https://huggingface.co/alibaba-damo/mgp-str-base/resolve/main/config.json',
}
class 							_UpperCAmelCase							( a      ):





     '''simple docstring'''



     a__									='''mgp-str'''





     def __init__(					self       ,	A=[3_2, 1_2_8]       ,	A=4       ,	A=3       ,	A=2_7       ,	A=3_8       ,	A=5_0_2_5_7       ,	A=3_0_5_2_2       ,	A=7_6_8       ,	A=1_2       ,	A=1_2       ,	A=4.0       ,	A=True       ,	A=False       ,	A=1E-5       ,	A=0.0       ,	A=0.0       ,	A=0.0       ,	A=False       ,	A=0.02       ,	**A       ,	)							->					Union[str, Any]:
         super().__init__(**A		)

         _UpperCAmelCase							:  Any											=    image_size
         _UpperCAmelCase							:  str											=    patch_size
         _UpperCAmelCase							:  Dict											=    num_channels
         _UpperCAmelCase							:  Dict											=    max_token_length
         _UpperCAmelCase							:  Optional[Any]											=    num_character_labels
         _UpperCAmelCase							:  int											=    num_bpe_labels
         _UpperCAmelCase							:  List[str]											=    num_wordpiece_labels
         _UpperCAmelCase							:  Optional[int]											=    hidden_size
         _UpperCAmelCase							:  Any											=    num_hidden_layers
         _UpperCAmelCase							:  List[Any]											=    num_attention_heads
         _UpperCAmelCase							:  List[Any]											=    mlp_ratio
         _UpperCAmelCase							:  List[str]											=    distilled
         _UpperCAmelCase							:  Optional[int]											=    layer_norm_eps
         _UpperCAmelCase							:  str											=    drop_rate
         _UpperCAmelCase							:  List[Any]											=    qkv_bias
         _UpperCAmelCase							:  List[str]											=    attn_drop_rate
         _UpperCAmelCase							:  Dict											=    drop_path_rate
         _UpperCAmelCase							:  Union[str, Any]											=    output_aa_attentions
         _UpperCAmelCase							:  List[str]											=    initializer_range


