

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







		"""simple docstring"""




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







								super().__init__()
								SCREAMING_SNAKE_CASE :      str       =   nn.Linear(3				,				4		)
								SCREAMING_SNAKE_CASE :      Tuple       =   nn.BatchNormad(4		)
								SCREAMING_SNAKE_CASE :      str       =   nn.Linear(4				,				5		)




		def  _lowercase      (			self    : List[str]				,				UpperCAmelCase__    : Optional[Any]		)					->Any:
								"""simple docstring"""







								return self.lineara(self.batchnorm(self.lineara(_lowercase		)		)		)







class        a__       (   SCREAMING_SNAKE_CASE__       ):







		"""simple docstring"""




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







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







class        a__       (   SCREAMING_SNAKE_CASE__       ):







		"""simple docstring"""




		def  _lowercase      (			self    : Optional[Any]				,				UpperCAmelCase__    : Tuple				,				UpperCAmelCase__    : int		)					->Any:
								"""simple docstring"""







								return output + 1







class        a__       (   unittest.TestCase       ):







		"""simple docstring"""




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







								SCREAMING_SNAKE_CASE :      Tuple       =   ModelForTest()
								SCREAMING_SNAKE_CASE :      Union[str, Any]       =   ModelHook()

								add_hook_to_module(_lowercase				,				_lowercase		)
								self.assertEqual(test_model._hf_hook				,				_lowercase		)
								self.assertTrue(hasattr(_lowercase				,				"""_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(_lowercase		)
								self.assertFalse(hasattr(_lowercase				,				"""_hf_hook"""		)		)
								self.assertFalse(hasattr(_lowercase				,				"""_old_forward"""		)		)


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







								SCREAMING_SNAKE_CASE :      str       =   ModelForTest()
								SCREAMING_SNAKE_CASE :      List[Any]       =   ModelHook()

								add_hook_to_module(_lowercase				,				_lowercase		)
								add_hook_to_module(_lowercase				,				_lowercase				,				append=_lowercase		)

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


		def  _lowercase      (			self    : Tuple		)					->Any:
								"""simple docstring"""







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

								SCREAMING_SNAKE_CASE :      List[str]       =   PreForwardHook()
								add_hook_to_module(_lowercase				,				_lowercase		)
								SCREAMING_SNAKE_CASE :      Dict       =   test_model(_lowercase		)
								self.assertTrue(torch.allclose(_lowercase				,				_lowercase				,				atol=1e-5		)		)

								# Attaching a hook to a model when it already has one replaces, does not chain
								SCREAMING_SNAKE_CASE :      List[Any]       =   PreForwardHook()
								add_hook_to_module(_lowercase				,				_lowercase		)
								SCREAMING_SNAKE_CASE :      Tuple       =   test_model(_lowercase		)
								self.assertTrue(torch.allclose(_lowercase				,				_lowercase				,				atol=1e-5		)		)

								# You need to use the sequential hook to chain two or more hooks
								SCREAMING_SNAKE_CASE :      Any       =   SequentialHook(PreForwardHook()				,				PreForwardHook()		)
								add_hook_to_module(_lowercase				,				_lowercase		)

								SCREAMING_SNAKE_CASE :      List[Any]       =   test_model(_lowercase		)
								assert torch.allclose(_lowercase				,				_lowercase				,				atol=1e-5		)


		def  _lowercase      (			self    : Tuple		)					->Dict:
								"""simple docstring"""







								SCREAMING_SNAKE_CASE :      Tuple       =   ModelForTest()
								SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      Optional[Any]       =   test_model(_lowercase		)

								SCREAMING_SNAKE_CASE :      int       =   PostForwardHook()
								add_hook_to_module(_lowercase				,				_lowercase		)
								SCREAMING_SNAKE_CASE :      Tuple       =   test_model(_lowercase		)
								self.assertTrue(torch.allclose(_lowercase				,				output + 1				,				atol=1e-5		)		)

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

								# You need to use the sequential hook to chain two or more hooks
								SCREAMING_SNAKE_CASE :      List[str]       =   SequentialHook(PostForwardHook()				,				PostForwardHook()		)
								add_hook_to_module(_lowercase				,				_lowercase		)

								SCREAMING_SNAKE_CASE :      str       =   test_model(_lowercase		)
								assert torch.allclose(_lowercase				,				output + 2				,				atol=1e-5		)


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







								SCREAMING_SNAKE_CASE :      Tuple       =   ModelForTest()
								SCREAMING_SNAKE_CASE :      Dict       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      Optional[int]       =   test_model(_lowercase		)

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

								SCREAMING_SNAKE_CASE :      Optional[int]       =   True
								SCREAMING_SNAKE_CASE :      List[Any]       =   test_model(_lowercase		)
								self.assertFalse(outputa.requires_grad		)


		@require_multi_gpu
		def  _lowercase      (			self    : int		)					->List[str]:
								"""simple docstring"""







								SCREAMING_SNAKE_CASE :      Optional[int]       =   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 :      Tuple       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      Optional[Any]       =   model(_lowercase		)
								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(_lowercase				,				AlignDevicesHook(io_same_device=_lowercase		)		)
								SCREAMING_SNAKE_CASE :      List[Any]       =   torch.randn(2				,				3		).to(0		)
								SCREAMING_SNAKE_CASE :      Tuple       =   model(_lowercase		)
								self.assertEqual(output.device				,				torch.device(0		)		)


		def  _lowercase      (			self    : Optional[Any]		)					->Optional[Any]:
								"""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 :      Optional[Any]       =   {"""execution_device""": 0 if torch.cuda.is_available() else """cpu""", """offload""": True}

								add_hook_to_module(model.lineara				,				AlignDevicesHook(**_lowercase		)		)
								add_hook_to_module(model.batchnorm				,				AlignDevicesHook(**_lowercase		)		)
								add_hook_to_module(model.lineara				,				AlignDevicesHook(**_lowercase		)		)

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

								SCREAMING_SNAKE_CASE :      Optional[int]       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      Dict       =   model(_lowercase		)
								self.assertEqual(output.device				,				_lowercase		)

								# 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 :      List[str]       =   {
								    """execution_device""": 0 if torch.cuda.is_available() else """cpu""",
								    """offload""": True,
								    """offload_buffers""": True,
								}

								add_hook_to_module(model.lineara				,				AlignDevicesHook(**_lowercase		)		)
								add_hook_to_module(model.batchnorm				,				AlignDevicesHook(**_lowercase		)		)
								add_hook_to_module(model.lineara				,				AlignDevicesHook(**_lowercase		)		)

								# 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 :      Dict       =   model(_lowercase		)
								self.assertEqual(output.device				,				_lowercase		)

								# 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    : Any		)					->List[str]:
								"""simple docstring"""







								SCREAMING_SNAKE_CASE :      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
								SCREAMING_SNAKE_CASE :      Tuple       =   0 if torch.cuda.is_available() else """cpu"""
								attach_align_device_hook(_lowercase				,				execution_device=_lowercase				,				offload=_lowercase		)

								# 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 :      List[Any]       =   torch.device(_lowercase		)
								self.assertEqual(model.batchnorm.running_mean.device				,				_lowercase		)

								SCREAMING_SNAKE_CASE :      int       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      Optional[Any]       =   model(_lowercase		)
								self.assertEqual(output.device				,				_lowercase		)

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

								# 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[Any]       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      int       =   model(_lowercase		)
								self.assertEqual(output.device				,				_lowercase		)

								# Removing hooks loads back the weights in the model.
								remove_hook_from_submodules(_lowercase		)
								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    : int		)					->int:
								"""simple docstring"""







								SCREAMING_SNAKE_CASE :      Optional[int]       =   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 :      Dict       =   0 if torch.cuda.is_available() else """cpu"""
								attach_align_device_hook(
								    _lowercase				,				execution_device=_lowercase				,				offload=_lowercase				,				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 :      List[Any]       =   torch.device(_lowercase		)
								self.assertEqual(model.batchnorm.running_mean.device				,				_lowercase		)

								SCREAMING_SNAKE_CASE :      Any       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      int       =   model(_lowercase		)
								self.assertEqual(output.device				,				_lowercase		)

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

								# 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 :      Optional[int]       =   torch.randn(2				,				3		)
								SCREAMING_SNAKE_CASE :      str       =   model(_lowercase		)
								self.assertEqual(output.device				,				_lowercase		)

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







import os
import random
import sys

from . import cryptomath_module as cryptomath
from . import rabin_miller

UpperCAmelCase__ :     Optional[Any]				    =    3






def 					__lowercase      (     _A			)     ->						int:
      print("""Generating primitive root of p"""			)
      while True:
            SCREAMING_SNAKE_CASE :      int       =   random.randrange(3						,    _A			)
            if pow(_A						,    2						,    _A			) == 1:
                  continue
            if pow(_A						,    _A						,    _A			) == 1:
                  continue
            return g






def 					__lowercase      (     _A			)     ->						tuple[tuple[int, int, int, int], tuple[int, int]]:
      print("""Generating prime p..."""			)
      SCREAMING_SNAKE_CASE :      Optional[Any]       =   rabin_miller.generate_large_prime(_A			)  # select large prime number.
      SCREAMING_SNAKE_CASE :      Optional[Any]       =   primitive_root(_A			)  # one primitive root on modulo p.
      SCREAMING_SNAKE_CASE :      Optional[Any]       =   random.randrange(3						,    _A			)  # private_key -> have to be greater than 2 for safety.
      SCREAMING_SNAKE_CASE :      List[str]       =   cryptomath.find_mod_inverse(pow(_A						,    _A						,    _A			)						,    _A			)

      SCREAMING_SNAKE_CASE :      str       =   (key_size, e_a, e_a, p)
      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   (key_size, d)

      return public_key, private_key






def 					__lowercase      (     _A						,    _A			)     ->						None:
      if os.path.exists(F"{name}_pubkey.txt"			) or os.path.exists(F"{name}_privkey.txt"			):
            print("""\nWARNING:"""			)
            print(
                F"\"{name}_pubkey.txt\" or \"{name}_privkey.txt\" already exists. \n"
                """Use a different name or delete these files and re-run this program."""			)
            sys.exit()

      SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      List[str]       =   generate_key(_A			)
      print(F"\nWriting public key to file {name}_pubkey.txt..."			)
      with open(F"{name}_pubkey.txt"						,    """w"""			) as fo:
            fo.write(F"{public_key[0]},{public_key[1]},{public_key[2]},{public_key[3]}"			)

      print(F"Writing private key to file {name}_privkey.txt..."			)
      with open(F"{name}_privkey.txt"						,    """w"""			) as fo:
            fo.write(F"{private_key[0]},{private_key[1]}"			)






def 					__lowercase      (     )     ->						None:
      print("""Making key files..."""			)
      make_key_files("""elgamal"""						,    2048			)
      print("""Key files generation successful"""			)


if __name__ == "__main__":
  main()




