

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 _UpperCamelCase							(							nn.Module   ):
			'''simple docstring'''







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

				"""simple docstring"""

				super().__init__()
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       nn.Linear(3      ,      4	)
				SCREAMING_SNAKE_CASE   :      str				     =       nn.BatchNormad(4	)
				SCREAMING_SNAKE_CASE   :      str				     =       nn.Linear(4      ,      5	)






			def 							__UpperCamelCase (			self							: str      ,      a							: Optional[Any]	)							->       Any:

				"""simple docstring"""

				return self.lineara(self.batchnorm(self.lineara(__a	)	)	)






class _UpperCamelCase							(							UpperCamelCase_   ):
			'''simple docstring'''







			def 							__UpperCamelCase (			self							: Dict      ,      a							: List[Any]      ,      *a							: List[str]      ,      **a							: int	)							->       Tuple:

				"""simple docstring"""

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






class _UpperCamelCase							(							UpperCamelCase_   ):
			'''simple docstring'''







			def 							__UpperCamelCase (			self							: int      ,      a							: Optional[Any]      ,      a							: str	)							->       int:

				"""simple docstring"""

				return output + 1






class _UpperCamelCase							(							unittest.TestCase   ):
			'''simple docstring'''







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

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      Dict				     =       ModelForTest()
				SCREAMING_SNAKE_CASE   :      Dict				     =       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 							__UpperCamelCase (			self							: Optional[Any]	)							->       Any:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      int				     =       ModelForTest()
				SCREAMING_SNAKE_CASE   :      List[Any]				     =       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 							__UpperCamelCase (			self							: List[str]	)							->       List[str]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      List[Any]				     =       ModelForTest()
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      Dict				     =       test_model(x + 1	)
				SCREAMING_SNAKE_CASE   :      int				     =       test_model(x + 2	)

				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       PreForwardHook()
				add_hook_to_module(__a      ,      __a	)
				SCREAMING_SNAKE_CASE   :      int				     =       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
				SCREAMING_SNAKE_CASE   :      Dict				     =       PreForwardHook()
				add_hook_to_module(__a      ,      __a	)
				SCREAMING_SNAKE_CASE   :      str				     =       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
				SCREAMING_SNAKE_CASE   :      Dict				     =       SequentialHook(PreForwardHook()      ,      PreForwardHook()	)
				add_hook_to_module(__a      ,      __a	)

				SCREAMING_SNAKE_CASE   :      Tuple				     =       test_model(__a	)
				assert torch.allclose(__a      ,      __a      ,      atol=1e-5	)
			def 							__UpperCamelCase (			self							: Any	)							->       int:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       ModelForTest()
				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      int				     =       test_model(__a	)

				SCREAMING_SNAKE_CASE   :      Any				     =       PostForwardHook()
				add_hook_to_module(__a      ,      __a	)
				SCREAMING_SNAKE_CASE   :      List[str]				     =       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
				SCREAMING_SNAKE_CASE   :      Tuple				     =       PostForwardHook()
				add_hook_to_module(__a      ,      __a	)
				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       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
				SCREAMING_SNAKE_CASE   :      List[Any]				     =       SequentialHook(PostForwardHook()      ,      PostForwardHook()	)
				add_hook_to_module(__a      ,      __a	)

				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       test_model(__a	)
				assert torch.allclose(__a      ,      output + 2      ,      atol=1e-5	)
			def 							__UpperCamelCase (			self							: Dict	)							->       List[Any]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      List[Any]				     =       ModelForTest()
				SCREAMING_SNAKE_CASE   :      Dict				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      List[str]				     =       test_model(__a	)

				SCREAMING_SNAKE_CASE   :      List[str]				     =       PostForwardHook()
				add_hook_to_module(__a      ,      __a	)
				SCREAMING_SNAKE_CASE   :      int				     =       test_model(__a	)
				self.assertTrue(torch.allclose(__a      ,      output + 1	)	)
				self.assertTrue(outputa.requires_grad	)

				SCREAMING_SNAKE_CASE   :      List[str]				     =       True
				SCREAMING_SNAKE_CASE   :      Any				     =       test_model(__a	)
				self.assertFalse(outputa.requires_grad	)
			@require_multi_gpu
			def 							__UpperCamelCase (			self							: str	)							->       Union[str, Any]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      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
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      str				     =       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	)	)
				SCREAMING_SNAKE_CASE   :      List[Any]				     =       torch.randn(2      ,      3	).to(0	)
				SCREAMING_SNAKE_CASE   :      str				     =       model(__a	)
				self.assertEqual(output.device      ,      torch.device(0	)	)
			def 							__UpperCamelCase (			self							: List[Any]	)							->       int:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      Optional[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
				SCREAMING_SNAKE_CASE   :      str				     =       {'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
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       torch.device(hook_kwargs["execution_device"]	)
				self.assertEqual(model.batchnorm.running_mean.device      ,      __a	)

				SCREAMING_SNAKE_CASE   :      List[str]				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      int				     =       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
				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       {
				    '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"	)	)

				SCREAMING_SNAKE_CASE   :      Dict				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       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 							__UpperCamelCase (			self							: Any	)							->       Union[str, Any]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      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
				SCREAMING_SNAKE_CASE   :      Any				     =       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
				SCREAMING_SNAKE_CASE   :      Dict				     =       torch.device(__a	)
				self.assertEqual(model.batchnorm.running_mean.device      ,      __a	)

				SCREAMING_SNAKE_CASE   :      Dict				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      List[Any]				     =       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"	)	)

				SCREAMING_SNAKE_CASE   :      List[str]				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      int				     =       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 							__UpperCamelCase (			self							: Union[str, Any]	)							->       Optional[int]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      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
				SCREAMING_SNAKE_CASE   :      str				     =       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
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       torch.device(__a	)
				self.assertEqual(model.batchnorm.running_mean.device      ,      __a	)

				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      Dict				     =       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"	)	)

				SCREAMING_SNAKE_CASE   :      int				     =       torch.randn(2      ,      3	)
				SCREAMING_SNAKE_CASE   :      Tuple				     =       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"	)	)



from __future__ import annotations

def  lowerCamelCase__ (					_a							,			_a):
	if b == 0:
		return (1, 0)
	((SCREAMING_SNAKE_CASE)     ,(SCREAMING_SNAKE_CASE))   :      Tuple				     =       extended_euclid(_a							,			a % b)
	SCREAMING_SNAKE_CASE   :      Dict				     =       a // b
	return (y, x - k * y)

def  lowerCamelCase__ (					_a							,			_a							,			_a							,			_a):
	((SCREAMING_SNAKE_CASE)     ,(SCREAMING_SNAKE_CASE))   :      str				     =       extended_euclid(_a							,			_a)
	SCREAMING_SNAKE_CASE   :      str				     =       na * na
	SCREAMING_SNAKE_CASE   :      int				     =       ra * x * na + ra * y * na
	return (n % m + m) % m

def  lowerCamelCase__ (					_a							,			_a):
	((SCREAMING_SNAKE_CASE)     ,(SCREAMING_SNAKE_CASE))   :      List[str]				     =       extended_euclid(_a							,			_a)
	if b < 0:
		SCREAMING_SNAKE_CASE   :      int				     =       (b % n + n) % n
	return b

def  lowerCamelCase__ (					_a							,			_a							,			_a							,			_a):
	SCREAMING_SNAKE_CASE     ,SCREAMING_SNAKE_CASE   :      Dict				     =       invert_modulo(_a							,			_a), invert_modulo(_a							,			_a)
	SCREAMING_SNAKE_CASE   :      Optional[int]				     =       na * na
	SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       ra * x * na + ra * y * na
	return (n % m + m) % m


if __name__ == "__main__":
			from doctest import testmod

			testmod(name='chinese_remainder_theorem', verbose=True)
			testmod(name='chinese_remainder_theorem2', verbose=True)
			testmod(name='invert_modulo', verbose=True)
			testmod(name='extended_euclid', verbose=True)