

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




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




	def        lowerCAmelCase_       (     self						:     str					,      __A						:     Any							):
		return self.lineara(self.batchnorm(self.lineara(__A							)							)							)






class      lowerCamelCase_      (	_lowercase	):




	def        lowerCAmelCase_       (     self						:     Union[str, Any]					,      __A						:     int					,      *__A						:     List[Any]					,      **__A						:     Any							):
		return (args[0] + 1,) + args[1:], kwargs






class      lowerCamelCase_      (	_lowercase	):




	def        lowerCAmelCase_       (     self						:     Optional[int]					,      __A						:     List[str]					,      __A						:     str							):
		return output + 1






class      lowerCamelCase_      (	unittest.TestCase	):




	def        lowerCAmelCase_       (     self						:     List[str]							):
		__A :			Union[str, Any]		    =			ModelForTest()
		__A :			Union[str, Any]		    =			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        lowerCAmelCase_       (     self						:     Dict							):
		__A :			str		    =			ModelForTest()
		__A :			Tuple		    =			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        lowerCAmelCase_       (     self						:     int							):
		__A :			int		    =			ModelForTest()
		__A :			List[str]		    =			torch.randn(2					,      3							)
		__A :			Tuple		    =			test_model(x + 1							)
		__A :			List[str]		    =			test_model(x + 2							)

		__A :			int		    =			PreForwardHook()
		add_hook_to_module(__A					,      __A							)
		__A :			Union[str, Any]		    =			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 :			Dict		    =			PreForwardHook()
		add_hook_to_module(__A					,      __A							)
		__A :			Union[str, Any]		    =			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 :			List[Any]		    =			SequentialHook(PreForwardHook()					,      PreForwardHook()							)
		add_hook_to_module(__A					,      __A							)

		__A :			List[str]		    =			test_model(__A							)
		assert torch.allclose(__A					,      __A					,      atol=1e-5							)




	def        lowerCAmelCase_       (     self						:     Dict							):
		__A :			Any		    =			ModelForTest()
		__A :			List[str]		    =			torch.randn(2					,      3							)
		__A :			Union[str, Any]		    =			test_model(__A							)

		__A :			Dict		    =			PostForwardHook()
		add_hook_to_module(__A					,      __A							)
		__A :			int		    =			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 :			Optional[int]		    =			PostForwardHook()
		add_hook_to_module(__A					,      __A							)
		__A :			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
		__A :			Dict		    =			SequentialHook(PostForwardHook()					,      PostForwardHook()							)
		add_hook_to_module(__A					,      __A							)

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




	def        lowerCAmelCase_       (     self						:     Tuple							):
		__A :			Optional[Any]		    =			ModelForTest()
		__A :			List[Any]		    =			torch.randn(2					,      3							)
		__A :			int		    =			test_model(__A							)

		__A :			Tuple		    =			PostForwardHook()
		add_hook_to_module(__A					,      __A							)
		__A :			List[Any]		    =			test_model(__A							)
		self.assertTrue(torch.allclose(__A					,      output + 1							)							)
		self.assertTrue(outputa.requires_grad							)

		__A :			int		    =			True
		__A :			Any		    =			test_model(__A							)
		self.assertFalse(outputa.requires_grad							)




	@require_multi_gpu
	def        lowerCAmelCase_       (     self						:     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
		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 :			Tuple		    =			torch.randn(2					,      3							)
		__A :			Tuple		    =			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 :			List[str]		    =			torch.randn(2					,      3							).to(0							)
		__A :			Tuple		    =			model(__A							)
		self.assertEqual(output.device					,      torch.device(0							)							)




	def        lowerCAmelCase_       (     self						:     Any							):
		__A :			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 :			Optional[int]		    =			{"""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 :			List[str]		    =			torch.device(hook_kwargs["""execution_device"""]							)
		self.assertEqual(model.batchnorm.running_mean.device					,      __A							)

		__A :			Optional[Any]		    =			torch.randn(2					,      3							)
		__A :			Optional[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
		__A :			str		    =			{
		    """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 :			int		    =			torch.randn(2					,      3							)
		__A :			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        lowerCAmelCase_       (     self						:     List[str]							):
		__A :			Dict		    =			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 :			str		    =			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 :			List[Any]		    =			torch.device(__A							)
		self.assertEqual(model.batchnorm.running_mean.device					,      __A							)

		__A :			List[str]		    =			torch.randn(2					,      3							)
		__A :			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					,      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 :			Dict		    =			torch.randn(2					,      3							)
		__A :			List[str]		    =			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_       (     self						:     Optional[int]							):
		__A :			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 :			int		    =			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 :			Optional[int]		    =			torch.device(__A							)
		self.assertEqual(model.batchnorm.running_mean.device					,      __A							)

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








class      lowerCamelCase_      :




	def __init__(     self						:     Dict					,      __A						:     Tuple					,      __A						:     Optional[int]					,      __A						:     int							):
		__A :			List[str]		    =			name
		__A :			Optional[int]		    =			value
		__A :			Optional[Any]		    =			weight




	def __repr__(     self						:     Any							):
		return F"""{self.__class__.__name__}({self.name}, {self.value}, {self.weight})"""




	def        lowerCAmelCase_       (     self						:     Union[str, Any]							):
		return self.value




	def        lowerCAmelCase_       (     self						:     str							):
		return self.name




	def        lowerCAmelCase_       (     self						:     str							):
		return self.weight




	def        lowerCAmelCase_       (     self						:     Dict							):
		return self.value / self.weight






def 							__SCREAMING_SNAKE_CASE     (   a__    :   str			,a__    :   Optional[int]			,a__    :   Union[str, Any]       )					->				int:
	__A :			Tuple		    =			[]
	for i in range(len(a__       )       ):
		menu.append(Things(name[i]			,value[i]			,weight[i]       )       )
	return menu
def 							__SCREAMING_SNAKE_CASE     (   a__    :   Tuple			,a__    :   Any			,a__    :   Optional[int]       )					->				Tuple:
	__A :			Optional[int]		    =			sorted(a__			,key=a__			,reverse=a__       )
	__A :			Optional[Any]		    =			[]
	__A    ,				__A :			Tuple		    =			0.0, 0.0
	for i in range(len(a__       )       ):
		if (total_cost + items_copy[i].get_weight()) <= max_cost:
			result.append(items_copy[i]       )
			total_cost += items_copy[i].get_weight()
			total_value += items_copy[i].get_value()
	return (result, total_value)
def 							__SCREAMING_SNAKE_CASE     (   )					->				List[Any]:
	pass


if __name__ == "__main__":
			import doctest

			doctest.testmod()





