




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

										super().__init__()
										__magic_name__						:						Tuple										=nn.Linear(3     ,			4			)
										__magic_name__						:						Union[str, Any]										=nn.BatchNormad(4			)
										__magic_name__						:						List[str]										=nn.Linear(4     ,			5			)


						def      A__    (		self :Dict     ,			__snake_case :Tuple			):
										'''simple docstring'''

										return self.lineara(self.batchnorm(self.lineara(__snake_case			)			)			)




class 		__A   (      UpperCamelCase__      ):

						def      A__    (		self :Any     ,			__snake_case :Optional[Any]     ,			*__snake_case :List[Any]     ,			**__snake_case :Any			):
										'''simple docstring'''

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




class 		__A   (      UpperCamelCase__      ):

						def      A__    (		self :List[str]     ,			__snake_case :Tuple     ,			__snake_case :Union[str, Any]			):
										'''simple docstring'''

										return output + 1




class 		__A   (      unittest.TestCase      ):

						def      A__    (		self :List[str]			):
										'''simple docstring'''

										__magic_name__						:						int										=ModelForTest()
										__magic_name__						:						Tuple										=ModelHook()

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

						def      A__    (		self :Tuple			):
										'''simple docstring'''

										__magic_name__						:						int										=ModelForTest()
										__magic_name__						:						List[str]										=ModelHook()

										add_hook_to_module(__snake_case     ,			__snake_case			)
										add_hook_to_module(__snake_case     ,			__snake_case     ,			append=__snake_case			)

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

						def      A__    (		self :List[str]			):
										'''simple docstring'''

										__magic_name__						:						Any										=ModelForTest()
										__magic_name__						:						Any										=torch.randn(2     ,			3			)
										__magic_name__						:						Any										=test_model(x + 1			)
										__magic_name__						:						Optional[Any]										=test_model(x + 2			)

										__magic_name__						:						int										=PreForwardHook()
										add_hook_to_module(__snake_case     ,			__snake_case			)
										__magic_name__						:						int										=test_model(__snake_case			)
										self.assertTrue(torch.allclose(__snake_case     ,			__snake_case     ,			atol=1E-5			)			)

										# Attaching a hook to a model when it already has one replaces, does not chain
										__magic_name__						:						str										=PreForwardHook()
										add_hook_to_module(__snake_case     ,			__snake_case			)
										__magic_name__						:						List[str]										=test_model(__snake_case			)
										self.assertTrue(torch.allclose(__snake_case     ,			__snake_case     ,			atol=1E-5			)			)

										# You need to use the sequential hook to chain two or more hooks
										__magic_name__						:						Optional[Any]										=SequentialHook(PreForwardHook()     ,			PreForwardHook()			)
										add_hook_to_module(__snake_case     ,			__snake_case			)

										__magic_name__						:						Any										=test_model(__snake_case			)
										assert torch.allclose(__snake_case     ,			__snake_case     ,			atol=1E-5			)

						def      A__    (		self :Any			):
										'''simple docstring'''

										__magic_name__						:						Optional[Any]										=ModelForTest()
										__magic_name__						:						Dict										=torch.randn(2     ,			3			)
										__magic_name__						:						Any										=test_model(__snake_case			)

										__magic_name__						:						Dict										=PostForwardHook()
										add_hook_to_module(__snake_case     ,			__snake_case			)
										__magic_name__						:						Any										=test_model(__snake_case			)
										self.assertTrue(torch.allclose(__snake_case     ,			output + 1     ,			atol=1E-5			)			)

										# Attaching a hook to a model when it already has one replaces, does not chain
										__magic_name__						:						Union[str, Any]										=PostForwardHook()
										add_hook_to_module(__snake_case     ,			__snake_case			)
										__magic_name__						:						Optional[int]										=test_model(__snake_case			)
										self.assertTrue(torch.allclose(__snake_case     ,			output + 1     ,			atol=1E-5			)			)

										# You need to use the sequential hook to chain two or more hooks
										__magic_name__						:						Union[str, Any]										=SequentialHook(PostForwardHook()     ,			PostForwardHook()			)
										add_hook_to_module(__snake_case     ,			__snake_case			)

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

						def      A__    (		self :Tuple			):
										'''simple docstring'''

										__magic_name__						:						Tuple										=ModelForTest()
										__magic_name__						:						int										=torch.randn(2     ,			3			)
										__magic_name__						:						Union[str, Any]										=test_model(__snake_case			)

										__magic_name__						:						Union[str, Any]										=PostForwardHook()
										add_hook_to_module(__snake_case     ,			__snake_case			)
										__magic_name__						:						Dict										=test_model(__snake_case			)
										self.assertTrue(torch.allclose(__snake_case     ,			output + 1			)			)
										self.assertTrue(outputa.requires_grad			)

										__magic_name__						:						Any										=True
										__magic_name__						:						Any										=test_model(__snake_case			)
										self.assertFalse(outputa.requires_grad			)

						@require_multi_gpu
						def      A__    (		self :List[str]			):
										'''simple docstring'''

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

						def      A__    (		self :List[Any]			):
										'''simple docstring'''

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

										add_hook_to_module(model.lineara     ,			AlignDevicesHook(**__snake_case			)			)
										add_hook_to_module(model.batchnorm     ,			AlignDevicesHook(**__snake_case			)			)
										add_hook_to_module(model.lineara     ,			AlignDevicesHook(**__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
										__magic_name__						:						Optional[int]										=torch.device(hook_kwargs["""execution_device"""]			)
										self.assertEqual(model.batchnorm.running_mean.device     ,			__snake_case			)

										__magic_name__						:						Union[str, Any]										=torch.randn(2     ,			3			)
										__magic_name__						:						Optional[int]										=model(__snake_case			)
										self.assertEqual(output.device     ,			__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
										__magic_name__						:						Tuple										={
										    """execution_device""": 0 if torch.cuda.is_available() else """cpu""",
										    """offload""": True,
										    """offload_buffers""": True,
										}

										add_hook_to_module(model.lineara     ,			AlignDevicesHook(**__snake_case			)			)
										add_hook_to_module(model.batchnorm     ,			AlignDevicesHook(**__snake_case			)			)
										add_hook_to_module(model.lineara     ,			AlignDevicesHook(**__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"""			)			)

										__magic_name__						:						Tuple										=torch.randn(2     ,			3			)
										__magic_name__						:						int										=model(__snake_case			)
										self.assertEqual(output.device     ,			__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      A__    (		self :List[Any]			):
										'''simple docstring'''

										__magic_name__						:						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
										__magic_name__						:						str										=0 if torch.cuda.is_available() else """cpu"""
										attach_align_device_hook(__snake_case     ,			execution_device=__snake_case     ,			offload=__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
										__magic_name__						:						Optional[Any]										=torch.device(__snake_case			)
										self.assertEqual(model.batchnorm.running_mean.device     ,			__snake_case			)

										__magic_name__						:						List[Any]										=torch.randn(2     ,			3			)
										__magic_name__						:						str										=model(__snake_case			)
										self.assertEqual(output.device     ,			__snake_case			)

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

										__magic_name__						:						Optional[int]										=torch.randn(2     ,			3			)
										__magic_name__						:						Union[str, Any]										=model(__snake_case			)
										self.assertEqual(output.device     ,			__snake_case			)

										# Removing hooks loads back the weights in the model.
										remove_hook_from_submodules(__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      A__    (		self :List[str]			):
										'''simple docstring'''

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

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

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

										__magic_name__						:						List[Any]										=torch.randn(2     ,			3			)
										__magic_name__						:						str										=model(__snake_case			)
										self.assertEqual(output.device     ,			__snake_case			)

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

import numpy as np
import torch

from . import residue_constants as rc
from .tensor_utils import tensor_tree_map, tree_map



def        a__					(							snake_case__							)		->   Dict[str, torch.Tensor]:
			lowerCamelCase      =   []
			lowerCamelCase      =   []
			lowerCamelCase      =   []

			for rt in rc.restypes:
						lowerCamelCase      =   rc.restype_name_to_atomaa_names[rc.restype_atoa[rt]]
						restype_atomaa_to_atomaa_list.append([(rc.atom_order[name] if name else 0) for name in atom_names]							)
						lowerCamelCase      =   {name: i for i, name in enumerate(snake_case__							)}
						restype_atomaa_to_atomaa_list.append(
						    [(atom_name_to_idxaa[name] if name in atom_name_to_idxaa else 0) for name in rc.atom_types]							)

						restype_atomaa_mask_list.append([(1.0 if name else 0.0) for name in atom_names]							)

			# Add dummy mapping for restype 'UNK'
			restype_atomaa_to_atomaa_list.append([0] * 14							)
			restype_atomaa_to_atomaa_list.append([0] * 37							)
			restype_atomaa_mask_list.append([0.0] * 14							)

			lowerCamelCase      =   torch.tensor(
			    snake_case__			,					dtype=torch.intaa			,					device=protein["""aatype"""].device			,					)
			lowerCamelCase      =   torch.tensor(
			    snake_case__			,					dtype=torch.intaa			,					device=protein["""aatype"""].device			,					)
			lowerCamelCase      =   torch.tensor(
			    snake_case__			,					dtype=torch.floataa			,					device=protein["""aatype"""].device			,					)
			lowerCamelCase      =   protein["""aatype"""].to(torch.long							)

			# create the mapping for (residx, atom14) --> atom37, i.e. an array
			# with shape (num_res, 14) containing the atom37 indices for this protein
			lowerCamelCase      =   restype_atomaa_to_atomaa[protein_aatype]
			lowerCamelCase      =   restype_atomaa_mask[protein_aatype]

			lowerCamelCase      =   residx_atomaa_mask
			lowerCamelCase      =   residx_atomaa_to_atomaa.long()

			# create the gather indices for mapping back
			lowerCamelCase      =   restype_atomaa_to_atomaa[protein_aatype]
			lowerCamelCase      =   residx_atomaa_to_atomaa.long()

			# create the corresponding mask
			lowerCamelCase      =   torch.zeros([21, 37]			,					dtype=torch.floataa			,					device=protein["""aatype"""].device							)
			for restype, restype_letter in enumerate(rc.restypes							):
						lowerCamelCase      =   rc.restype_atoa[restype_letter]
						lowerCamelCase      =   rc.residue_atoms[restype_name]
						for atom_name in atom_names:
									lowerCamelCase      =   rc.atom_order[atom_name]
									lowerCamelCase      =   1

			lowerCamelCase      =   restype_atomaa_mask[protein_aatype]
			lowerCamelCase      =   residx_atomaa_mask

			return protein



def        a__					(							snake_case__							)		->   Dict[str, np.ndarray]:
			lowerCamelCase      =   tree_map(lambda snake_case__							: torch.tensor(snake_case__			,					device=batch["""aatype"""].device							)			,					snake_case__			,					np.ndarray							)
			lowerCamelCase      =   tensor_tree_map(lambda snake_case__							: np.array(snake_case__							)			,					make_atomaa_masks(snake_case__							)							)
			return out



