






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







	def __init__(	self   :				str      )		->			Dict:
		"""simple docstring"""







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







	def 			_a (	self   :				int							,					_lowerCAmelCase   :				List[Any]      )		->			str:
		"""simple docstring"""







		return self.lineara(self.batchnorm(self.lineara(__lowerCAmelCase      )      )      )


class 							__UpperCamelCase     (  lowerCAmelCase__ ):







	def 			_a (	self   :				Optional[int]							,					_lowerCAmelCase   :				Tuple							,					*_lowerCAmelCase   :				Tuple							,					**_lowerCAmelCase   :				List[str]      )		->			int:
		"""simple docstring"""







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


class 							__UpperCamelCase     (  lowerCAmelCase__ ):







	def 			_a (	self   :				Any							,					_lowerCAmelCase   :				Union[str, Any]							,					_lowerCAmelCase   :				Tuple      )		->			str:
		"""simple docstring"""







		return output + 1


class 							__UpperCamelCase     (  unittest.TestCase ):







	def 			_a (	self   :				int      )		->			Optional[int]:
		"""simple docstring"""







		__lowercase   							=	ModelForTest()
		__lowercase   							=	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 			_a (	self   :				str      )		->			List[str]:
		"""simple docstring"""







		__lowercase   							=	ModelForTest()
		__lowercase   							=	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 			_a (	self   :				Union[str, Any]      )		->			List[Any]:
		"""simple docstring"""







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

		__lowercase   							=	PreForwardHook()
		add_hook_to_module(__lowerCAmelCase							,					__lowerCAmelCase      )
		__lowercase   							=	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
		__lowercase   							=	PreForwardHook()
		add_hook_to_module(__lowerCAmelCase							,					__lowerCAmelCase      )
		__lowercase   							=	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
		__lowercase   							=	SequentialHook(PreForwardHook()							,					PreForwardHook()      )
		add_hook_to_module(__lowerCAmelCase							,					__lowerCAmelCase      )

		__lowercase   							=	test_model(__lowerCAmelCase      )
		assert torch.allclose(__lowerCAmelCase							,					__lowerCAmelCase							,					atol=1e-5      )







	def 			_a (	self   :				List[Any]      )		->			Union[str, Any]:
		"""simple docstring"""







		__lowercase   							=	ModelForTest()
		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	test_model(__lowerCAmelCase      )

		__lowercase   							=	PostForwardHook()
		add_hook_to_module(__lowerCAmelCase							,					__lowerCAmelCase      )
		__lowercase   							=	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
		__lowercase   							=	PostForwardHook()
		add_hook_to_module(__lowerCAmelCase							,					__lowerCAmelCase      )
		__lowercase   							=	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
		__lowercase   							=	SequentialHook(PostForwardHook()							,					PostForwardHook()      )
		add_hook_to_module(__lowerCAmelCase							,					__lowerCAmelCase      )

		__lowercase   							=	test_model(__lowerCAmelCase      )
		assert torch.allclose(__lowerCAmelCase							,					output + 2							,					atol=1e-5      )







	def 			_a (	self   :				List[str]      )		->			Tuple:
		"""simple docstring"""







		__lowercase   							=	ModelForTest()
		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	test_model(__lowerCAmelCase      )

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

		__lowercase   							=	True
		__lowercase   							=	test_model(__lowerCAmelCase      )
		self.assertFalse(outputa.requires_grad      )







	@require_multi_gpu
	def 			_a (	self   :				Any      )		->			Optional[Any]:
		"""simple docstring"""







		__lowercase   							=	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
		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	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      )      )
		__lowercase   							=	torch.randn(2							,					3      ).to(0      )
		__lowercase   							=	model(__lowerCAmelCase      )
		self.assertEqual(output.device							,					torch.device(0      )      )







	def 			_a (	self   :				int      )		->			Tuple:
		"""simple docstring"""







		__lowercase   							=	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
		__lowercase   							=	{"""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
		__lowercase   							=	torch.device(hook_kwargs["""execution_device"""]      )
		self.assertEqual(model.batchnorm.running_mean.device							,					__lowerCAmelCase      )

		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	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
		__lowercase   							=	{
		    """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"""      )      )

		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	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 			_a (	self   :				Union[str, Any]      )		->			int:
		"""simple docstring"""







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

		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	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"""      )      )

		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	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 			_a (	self   :				Optional[int]      )		->			int:
		"""simple docstring"""







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

		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	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"""      )      )

		__lowercase   							=	torch.randn(2							,					3      )
		__lowercase   							=	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'''




# A Bipartite Graph is a graph whose vertices can be divided into two independent sets,
# U and V such that every edge (u, v) either connects a vertex from U to V or a vertex
# from V to U. In other words, for every edge (u, v), either u belongs to U and v to V,
# or u belongs to V and v to U. We can also say that there is no edge that connects
# vertices of same set.






def 							snake_case__				(     _A:   Dict					)       ->      str:

		'''simple docstring'''


		lowerCAmelCase     =    [False] * len(_A					)
		lowerCAmelCase     =    [-1] * len(_A					)

		def dfs(_A:   Dict ,  _A:   Tuple					):
				lowerCAmelCase     =    True
				lowerCAmelCase     =    c
				for u in graph[v]:
						if not visited[u]:
								dfs(_A ,  1 - c					)

		for i in range(len(_A					)					):
				if not visited[i]:
						dfs(_A ,  0					)

		for i in range(len(_A					)					):
				for j in graph[i]:
						if color[i] == color[j]:
								return False

		return True


# Adjacency list of graph
__lowercase        =    {0: [1, 3], 1: [0, 2], 2: [1, 3], 3: [0, 2], 4: []}




print(check_bipartite_dfs(graph))




