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



     def __init__(							self   )		->     Any:







         """simple docstring"""



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




     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     Optional[int]:







         """simple docstring"""



         return self.lineara(self.batchnorm(self.lineara(__SCREAMING_SNAKE_CASE   )   )   )







class     lowercase(_lowercase       ):



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE     ,				*__SCREAMING_SNAKE_CASE     ,				**__SCREAMING_SNAKE_CASE   )		->     str:







         """simple docstring"""



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







class     lowercase(_lowercase       ):



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )		->     int:







         """simple docstring"""



         return output + 1







class     lowercase(unittest.TestCase       ):



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







         """simple docstring"""



         a__		=			ModelForTest()
         a__		=			ModelHook()

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



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







         """simple docstring"""



         a__		=			ModelForTest()
         a__		=			ModelHook()

         add_hook_to_module(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )
         add_hook_to_module(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				append=__SCREAMING_SNAKE_CASE   )

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



     def 			lowercase__  (							self   )		->     Any:







         """simple docstring"""



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

         a__		=			PreForwardHook()
         add_hook_to_module(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )
         a__		=			test_model(__SCREAMING_SNAKE_CASE   )
         self.assertTrue(torch.allclose(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				atol=1e-5   )   )

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

         # You need to use the sequential hook to chain two or more hooks
         a__		=			SequentialHook(PreForwardHook()     ,				PreForwardHook()   )
         add_hook_to_module(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )

         a__		=			test_model(__SCREAMING_SNAKE_CASE   )
         assert torch.allclose(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				atol=1e-5   )



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







         """simple docstring"""



         a__		=			ModelForTest()
         a__		=			torch.randn(2     ,				3   )
         a__		=			test_model(__SCREAMING_SNAKE_CASE   )

         a__		=			PostForwardHook()
         add_hook_to_module(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )
         a__		=			test_model(__SCREAMING_SNAKE_CASE   )
         self.assertTrue(torch.allclose(__SCREAMING_SNAKE_CASE     ,				output + 1     ,				atol=1e-5   )   )

         # Attaching a hook to a model when it already has one replaces, does not chain
         a__		=			PostForwardHook()
         add_hook_to_module(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )
         a__		=			test_model(__SCREAMING_SNAKE_CASE   )
         self.assertTrue(torch.allclose(__SCREAMING_SNAKE_CASE     ,				output + 1     ,				atol=1e-5   )   )

         # You need to use the sequential hook to chain two or more hooks
         a__		=			SequentialHook(PostForwardHook()     ,				PostForwardHook()   )
         add_hook_to_module(__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )

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



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







         """simple docstring"""



         a__		=			ModelForTest()
         a__		=			torch.randn(2     ,				3   )
         a__		=			test_model(__SCREAMING_SNAKE_CASE   )

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

         a__		=			True
         a__		=			test_model(__SCREAMING_SNAKE_CASE   )
         self.assertFalse(outputa.requires_grad   )



     @require_multi_gpu
     def 			lowercase__  (							self   )		->     Dict:







         """simple docstring"""



         a__		=			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__		=			torch.randn(2     ,				3   )
         a__		=			model(__SCREAMING_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(__SCREAMING_SNAKE_CASE     ,				AlignDevicesHook(io_same_device=__SCREAMING_SNAKE_CASE   )   )
         a__		=			torch.randn(2     ,				3   ).to(0   )
         a__		=			model(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(output.device     ,				torch.device(0   )   )



     def 			lowercase__  (							self   )		->     Any:







         """simple docstring"""



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

         add_hook_to_module(model.lineara     ,				AlignDevicesHook(**__SCREAMING_SNAKE_CASE   )   )
         add_hook_to_module(model.batchnorm     ,				AlignDevicesHook(**__SCREAMING_SNAKE_CASE   )   )
         add_hook_to_module(model.lineara     ,				AlignDevicesHook(**__SCREAMING_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__		=			torch.device(hook_kwargs['execution_device']   )
         self.assertEqual(model.batchnorm.running_mean.device     ,				__SCREAMING_SNAKE_CASE   )

         a__		=			torch.randn(2     ,				3   )
         a__		=			model(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(output.device     ,				__SCREAMING_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__		=			{
             'execution_device': 0 if torch.cuda.is_available() else 'cpu',
             'offload': True,
             'offload_buffers': True,
         }

         add_hook_to_module(model.lineara     ,				AlignDevicesHook(**__SCREAMING_SNAKE_CASE   )   )
         add_hook_to_module(model.batchnorm     ,				AlignDevicesHook(**__SCREAMING_SNAKE_CASE   )   )
         add_hook_to_module(model.lineara     ,				AlignDevicesHook(**__SCREAMING_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__		=			torch.randn(2     ,				3   )
         a__		=			model(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(output.device     ,				__SCREAMING_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   )		->     Union[str, Any]:







         """simple docstring"""



         a__		=			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__		=			0 if torch.cuda.is_available() else 'cpu'
         attach_align_device_hook(__SCREAMING_SNAKE_CASE     ,				execution_device=__SCREAMING_SNAKE_CASE     ,				offload=__SCREAMING_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__		=			torch.device(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(model.batchnorm.running_mean.device     ,				__SCREAMING_SNAKE_CASE   )

         a__		=			torch.randn(2     ,				3   )
         a__		=			model(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(output.device     ,				__SCREAMING_SNAKE_CASE   )

         # Removing hooks loads back the weights in the model.
         remove_hook_from_submodules(__SCREAMING_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(__SCREAMING_SNAKE_CASE     ,				execution_device=__SCREAMING_SNAKE_CASE     ,				offload=__SCREAMING_SNAKE_CASE     ,				offload_buffers=__SCREAMING_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__		=			torch.randn(2     ,				3   )
         a__		=			model(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(output.device     ,				__SCREAMING_SNAKE_CASE   )

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







         """simple docstring"""



         a__		=			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__		=			0 if torch.cuda.is_available() else 'cpu'
         attach_align_device_hook(
             __SCREAMING_SNAKE_CASE     ,				execution_device=__SCREAMING_SNAKE_CASE     ,				offload=__SCREAMING_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__		=			torch.device(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(model.batchnorm.running_mean.device     ,				__SCREAMING_SNAKE_CASE   )

         a__		=			torch.randn(2     ,				3   )
         a__		=			model(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(output.device     ,				__SCREAMING_SNAKE_CASE   )

         # Removing hooks loads back the weights in the model.
         remove_hook_from_submodules(__SCREAMING_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(
             __SCREAMING_SNAKE_CASE     ,				execution_device=__SCREAMING_SNAKE_CASE     ,				offload=__SCREAMING_SNAKE_CASE     ,				weights_map=model.state_dict()     ,				offload_buffers=__SCREAMING_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__		=			torch.randn(2     ,				3   )
         a__		=			model(__SCREAMING_SNAKE_CASE   )
         self.assertEqual(output.device     ,				__SCREAMING_SNAKE_CASE   )

         # Removing hooks loads back the weights in the model.
         remove_hook_from_submodules(__SCREAMING_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 __future__ import annotations

a		:       str     						=       {
    'A': ['B', 'C', 'E'],
    'B': ['A', 'D', 'E'],
    'C': ['A', 'F', 'G'],
    'D': ['B'],
    'E': ['A', 'B', 'D'],
    'F': ['C'],
    'G': ['C'],
}







class     lowercase:



     def __init__(							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )		->     None:







         """simple docstring"""



         a__		=			graph
         # mapping node to its parent in resulting breadth first tree
         a__		=			{}
         a__		=			source_vertex



     def 			lowercase__  (							self   )		->     None:







         """simple docstring"""



         a__		=			{self.source_vertex}
         a__		=			None
         a__		=			[self.source_vertex]  # first in first out queue

         while queue:
             a__		=			queue.pop(0   )
             for adjacent_vertex in self.graph[vertex]:
                 if adjacent_vertex not in visited:
                     visited.add(__SCREAMING_SNAKE_CASE   )
                     a__		=			vertex
                     queue.append(__SCREAMING_SNAKE_CASE   )




     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     str:







         """simple docstring"""



         if target_vertex == self.source_vertex:
             return self.source_vertex

         a__		=			self.parent.get(__SCREAMING_SNAKE_CASE   )
         if target_vertex_parent is None:
             a__		=			(
                 f'No path from vertex: {self.source_vertex} to vertex: {target_vertex}'
             )
             raise ValueError(__SCREAMING_SNAKE_CASE   )

         return self.shortest_path(__SCREAMING_SNAKE_CASE   ) + f'->{target_vertex}'


if __name__ == "__main__":
    a		:       Optional[Any]     						=       Graph(graph, 'G')
    g.breath_first_search()
    print(g.shortest_path('D'))
    print(g.shortest_path('G'))
    print(g.shortest_path('Foo'))


