


'''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						)					->							str:
      super().__init__()
      _a			      =							nn.Linear(3      ,				4						)
      _a			      =							nn.BatchNormad(4						)
      _a			      =							nn.Linear(4      ,				5						)




 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							List[Any]:
      return self.lineara(self.batchnorm(self.lineara(__magic_name__						)						)						)





class  a     (     _SCREAMING_SNAKE_CASE       ):
 def 							__UpperCAmelCase		( self      ,				__magic_name__      ,				*__magic_name__      ,				**__magic_name__						)					->							Optional[int]:
      return (args[0] + 1,) + args[1:], kwargs





class  a     (     _SCREAMING_SNAKE_CASE       ):
 def 							__UpperCAmelCase		( self      ,				__magic_name__      ,				__magic_name__						)					->							Dict:
      return output + 1





class  a     (     unittest.TestCase       ):
 def 							__UpperCAmelCase		( self						)					->							List[str]:
      _a			      =							ModelForTest()
      _a			      =							ModelHook()

      add_hook_to_module(__magic_name__      ,				__magic_name__						)
      self.assertEqual(test_model._hf_hook      ,				__magic_name__						)
      self.assertTrue(hasattr(__magic_name__      ,				'_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(__magic_name__						)
      self.assertFalse(hasattr(__magic_name__      ,				'_hf_hook'						)						)
      self.assertFalse(hasattr(__magic_name__      ,				'_old_forward'						)						)
 def 							__UpperCAmelCase		( self						)					->							int:
      _a			      =							ModelForTest()
      _a			      =							ModelHook()

      add_hook_to_module(__magic_name__      ,				__magic_name__						)
      add_hook_to_module(__magic_name__      ,				__magic_name__      ,				append=__magic_name__						)

      self.assertEqual(isinstance(test_model._hf_hook      ,				__magic_name__						)      ,				__magic_name__						)
      self.assertEqual(len(test_model._hf_hook.hooks						)      ,				2						)
      self.assertTrue(hasattr(__magic_name__      ,				'_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(__magic_name__						)
      self.assertFalse(hasattr(__magic_name__      ,				'_hf_hook'						)						)
      self.assertFalse(hasattr(__magic_name__      ,				'_old_forward'						)						)
 def 							__UpperCAmelCase		( self						)					->							List[Any]:
      _a			      =							ModelForTest()
      _a			      =							torch.randn(2      ,				3						)
      _a			      =							test_model(x + 1						)
      _a			      =							test_model(x + 2						)

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

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

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

      _a			      =							test_model(__magic_name__						)
      assert torch.allclose(__magic_name__      ,				__magic_name__      ,				atol=1e-5						)
 def 							__UpperCAmelCase		( self						)					->							str:
      _a			      =							ModelForTest()
      _a			      =							torch.randn(2      ,				3						)
      _a			      =							test_model(__magic_name__						)

      _a			      =							PostForwardHook()
      add_hook_to_module(__magic_name__      ,				__magic_name__						)
      _a			      =							test_model(__magic_name__						)
      self.assertTrue(torch.allclose(__magic_name__      ,				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(__magic_name__      ,				__magic_name__						)
      _a			      =							test_model(__magic_name__						)
      self.assertTrue(torch.allclose(__magic_name__      ,				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(__magic_name__      ,				__magic_name__						)

      _a			      =							test_model(__magic_name__						)
      assert torch.allclose(__magic_name__      ,				output + 2      ,				atol=1e-5						)
 def 							__UpperCAmelCase		( self						)					->							Optional[int]:
      _a			      =							ModelForTest()
      _a			      =							torch.randn(2      ,				3						)
      _a			      =							test_model(__magic_name__						)

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

      _a			      =							True
      _a			      =							test_model(__magic_name__						)
      self.assertFalse(outputa.requires_grad						)
 @require_multi_gpu
 def 							__UpperCAmelCase		( self						)					->							Optional[Any]:
      _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(__magic_name__						)
      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(__magic_name__      ,				AlignDevicesHook(io_same_device=__magic_name__						)						)
      _a			      =							torch.randn(2      ,				3						).to(0						)
      _a			      =							model(__magic_name__						)
      self.assertEqual(output.device      ,				torch.device(0						)						)
 def 							__UpperCAmelCase		( self						)					->							int:
      _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(**__magic_name__						)						)
      add_hook_to_module(model.batchnorm      ,				AlignDevicesHook(**__magic_name__						)						)
      add_hook_to_module(model.lineara      ,				AlignDevicesHook(**__magic_name__						)						)

      # 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      ,				__magic_name__						)

      _a			      =							torch.randn(2      ,				3						)
      _a			      =							model(__magic_name__						)
      self.assertEqual(output.device      ,				__magic_name__						)

      # 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(**__magic_name__						)						)
      add_hook_to_module(model.batchnorm      ,				AlignDevicesHook(**__magic_name__						)						)
      add_hook_to_module(model.lineara      ,				AlignDevicesHook(**__magic_name__						)						)

      # 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(__magic_name__						)
      self.assertEqual(output.device      ,				__magic_name__						)

      # 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						)					->							str:
      _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(__magic_name__      ,				execution_device=__magic_name__      ,				offload=__magic_name__						)

      # 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(__magic_name__						)
      self.assertEqual(model.batchnorm.running_mean.device      ,				__magic_name__						)

      _a			      =							torch.randn(2      ,				3						)
      _a			      =							model(__magic_name__						)
      self.assertEqual(output.device      ,				__magic_name__						)

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

      # 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(__magic_name__						)
      self.assertEqual(output.device      ,				__magic_name__						)

      # Removing hooks loads back the weights in the model.
      remove_hook_from_submodules(__magic_name__						)
      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						)					->							str:
      _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(
          __magic_name__      ,				execution_device=__magic_name__      ,				offload=__magic_name__      ,				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(__magic_name__						)
      self.assertEqual(model.batchnorm.running_mean.device      ,				__magic_name__						)

      _a			      =							torch.randn(2      ,				3						)
      _a			      =							model(__magic_name__						)
      self.assertEqual(output.device      ,				__magic_name__						)

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

      # 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(__magic_name__						)
      self.assertEqual(output.device      ,				__magic_name__						)

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






def 	_A				(lowerCAmelCase__  :list		)   ->  list[Any]:




     '''simple docstring'''



     if not input_list:
          return []
     _a			      =							[input_list.count(lowerCAmelCase__		) for value in input_list]
     _a			      =							max(lowerCAmelCase__		)  # Gets the maximum count in the input list.
     # Gets values of modes
     return sorted({input_list[i] for i, value in enumerate(lowerCAmelCase__		) if value == y}		)


if __name__ == "__main__":
 import doctest

 doctest.testmod()

