


import unittest

import numpy as np

from transformers import AlbertConfig, is_flax_available
from transformers.testing_utils import require_flax, slow

from ...test_modeling_flax_common import FlaxModelTesterMixin, ids_tensor, random_attention_mask


if is_flax_available():
     import jax.numpy as jnp

     from transformers.models.albert.modeling_flax_albert import (
         FlaxAlbertForMaskedLM,
         FlaxAlbertForMultipleChoice,
         FlaxAlbertForPreTraining,
         FlaxAlbertForQuestionAnswering,
         FlaxAlbertForSequenceClassification,
         FlaxAlbertForTokenClassification,
         FlaxAlbertModel,
     )





class 			__lowerCamelCase	(   unittest.TestCase   ):




    """simple docstring"""

    def __init__(	self					:					Any       ,     SCREAMING_SNAKE_CASE__					:					List[Any]       ,     SCREAMING_SNAKE_CASE__					:					Optional[Any]=13       ,     SCREAMING_SNAKE_CASE__					:					Tuple=7       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]=True       ,     SCREAMING_SNAKE_CASE__					:					Any=True       ,     SCREAMING_SNAKE_CASE__					:					Union[str, Any]=True       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]=True       ,     SCREAMING_SNAKE_CASE__					:					List[str]=99       ,     SCREAMING_SNAKE_CASE__					:					Any=32       ,     SCREAMING_SNAKE_CASE__					:					int=5       ,     SCREAMING_SNAKE_CASE__					:					Tuple=4       ,     SCREAMING_SNAKE_CASE__					:					Optional[Any]=37       ,     SCREAMING_SNAKE_CASE__					:					Tuple="gelu"       ,     SCREAMING_SNAKE_CASE__					:					str=0.1       ,     SCREAMING_SNAKE_CASE__					:					List[str]=0.1       ,     SCREAMING_SNAKE_CASE__					:					Optional[Any]=512       ,     SCREAMING_SNAKE_CASE__					:					int=16       ,     SCREAMING_SNAKE_CASE__					:					Any=2       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]=0.02       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]=4       ,     )				->	Tuple:
         lowerCAmelCase__          =			parent
         lowerCAmelCase__          =			batch_size
         lowerCAmelCase__          =			seq_length
         lowerCAmelCase__          =			is_training
         lowerCAmelCase__          =			use_attention_mask
         lowerCAmelCase__          =			use_token_type_ids
         lowerCAmelCase__          =			use_labels
         lowerCAmelCase__          =			vocab_size
         lowerCAmelCase__          =			hidden_size
         lowerCAmelCase__          =			num_hidden_layers
         lowerCAmelCase__          =			num_attention_heads
         lowerCAmelCase__          =			intermediate_size
         lowerCAmelCase__          =			hidden_act
         lowerCAmelCase__          =			hidden_dropout_prob
         lowerCAmelCase__          =			attention_probs_dropout_prob
         lowerCAmelCase__          =			max_position_embeddings
         lowerCAmelCase__          =			type_vocab_size
         lowerCAmelCase__          =			type_sequence_label_size
         lowerCAmelCase__          =			initializer_range
         lowerCAmelCase__          =			num_choices


    def        a				(	self					:					Any       )				->	Tuple:
         lowerCAmelCase__          =			ids_tensor([self.batch_size, self.seq_length]       ,     self.vocab_size       )

         lowerCAmelCase__          =			None
         if self.use_attention_mask:
              lowerCAmelCase__          =			random_attention_mask([self.batch_size, self.seq_length]       )

         lowerCAmelCase__          =			None
         if self.use_token_type_ids:
              lowerCAmelCase__          =			ids_tensor([self.batch_size, self.seq_length]       ,     self.type_vocab_size       )

         lowerCAmelCase__          =			AlbertConfig(
             vocab_size=self.vocab_size       ,     hidden_size=self.hidden_size       ,     num_hidden_layers=self.num_hidden_layers       ,     num_attention_heads=self.num_attention_heads       ,     intermediate_size=self.intermediate_size       ,     hidden_act=self.hidden_act       ,     hidden_dropout_prob=self.hidden_dropout_prob       ,     attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,     max_position_embeddings=self.max_position_embeddings       ,     type_vocab_size=self.type_vocab_size       ,     is_decoder=SCREAMING_SNAKE_CASE__       ,     initializer_range=self.initializer_range       ,     )

         return config, input_ids, token_type_ids, attention_mask


    def        a				(	self					:					Optional[Any]       )				->	Dict:
         lowerCAmelCase__          =			self.prepare_config_and_inputs()
         lowerCAmelCase__     ,		lowerCAmelCase__     ,		lowerCAmelCase__     ,		lowerCAmelCase__          =			config_and_inputs
         lowerCAmelCase__          =			{"input_ids": input_ids, "token_type_ids": token_type_ids, "attention_mask": attention_mask}
         return config, inputs_dict






@require_flax
class 			__lowerCamelCase	(   UpperCamelCase__     ,    unittest.TestCase   ):




    """simple docstring"""

    snake_case__ =				(
        (
            FlaxAlbertModel,
            FlaxAlbertForPreTraining,
            FlaxAlbertForMaskedLM,
            FlaxAlbertForMultipleChoice,
            FlaxAlbertForQuestionAnswering,
            FlaxAlbertForSequenceClassification,
            FlaxAlbertForTokenClassification,
            FlaxAlbertForQuestionAnswering,
        )
        if is_flax_available()
        else ()
    )


    def        a				(	self					:					Dict       )				->	Union[str, Any]:
         lowerCAmelCase__          =			FlaxAlbertModelTester(self       )


    @slow
    def        a				(	self					:					int       )				->	Any:
         for model_class_name in self.all_model_classes:
              lowerCAmelCase__          =			model_class_name.from_pretrained("albert-base-v2"       )
              lowerCAmelCase__          =			model(np.ones((1, 1)       )       )
              self.assertIsNotNone(SCREAMING_SNAKE_CASE__       )






@require_flax
class 			__lowerCamelCase	(   unittest.TestCase   ):




    """simple docstring"""

    @slow
    def        a				(	self					:					Any       )				->	Tuple:
         lowerCAmelCase__          =			FlaxAlbertModel.from_pretrained("albert-base-v2"       )
         lowerCAmelCase__          =			np.array([[0, 345, 232, 328, 740, 140, 1_695, 69, 6_078, 1_588, 2]]       )
         lowerCAmelCase__          =			np.array([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]       )
         lowerCAmelCase__          =			model(SCREAMING_SNAKE_CASE__       ,     attention_mask=SCREAMING_SNAKE_CASE__       )[0]
         lowerCAmelCase__          =			(1, 11, 768)
         self.assertEqual(output.shape       ,     SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			np.array(
             [[[-0.6_513, 1.5_035, -0.2_766], [-0.6_515, 1.5_046, -0.2_780], [-0.6_512, 1.5_049, -0.2_784]]]       )

         self.assertTrue(jnp.allclose(output[:, 1:4, 1:4]       ,     SCREAMING_SNAKE_CASE__       ,     atol=1e-4       )       )






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




    """simple docstring"""

    def __init__(	self					:					Dict       )				->	Optional[int]:
         super().__init__()
         lowerCAmelCase__          =			nn.Linear(3       ,     4       )
         lowerCAmelCase__          =			nn.BatchNormad(4       )
         lowerCAmelCase__          =			nn.Linear(4       ,     5       )


    def        a				(	self					:					Union[str, Any]       ,     SCREAMING_SNAKE_CASE__					:					int       )				->	List[Any]:
         return self.lineara(self.batchnorm(self.lineara(SCREAMING_SNAKE_CASE__       )       )       )






class 			__lowerCamelCase	(   UpperCamelCase__   ):




    """simple docstring"""

    def        a				(	self					:					Any       ,     SCREAMING_SNAKE_CASE__					:					str       ,     *SCREAMING_SNAKE_CASE__					:					Tuple       ,     **SCREAMING_SNAKE_CASE__					:					Any       )				->	Union[str, Any]:
         return (args[0] + 1,) + args[1:], kwargs






class 			__lowerCamelCase	(   UpperCamelCase__   ):




    """simple docstring"""

    def        a				(	self					:					List[Any]       ,     SCREAMING_SNAKE_CASE__					:					Tuple       ,     SCREAMING_SNAKE_CASE__					:					str       )				->	Dict:
         return output + 1






class 			__lowerCamelCase	(   unittest.TestCase   ):




    """simple docstring"""

    def        a				(	self					:					List[str]       )				->	Tuple:
         lowerCAmelCase__          =			ModelForTest()
         lowerCAmelCase__          =			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        a				(	self					:					Union[str, Any]       )				->	int:
         lowerCAmelCase__          =			ModelForTest()
         lowerCAmelCase__          =			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        a				(	self					:					List[str]       )				->	Any:
         lowerCAmelCase__          =			ModelForTest()
         lowerCAmelCase__          =			torch.randn(2       ,     3       )
         lowerCAmelCase__          =			test_model(x + 1       )
         lowerCAmelCase__          =			test_model(x + 2       )

         lowerCAmelCase__          =			PreForwardHook()
         add_hook_to_module(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			PreForwardHook()
         add_hook_to_module(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			SequentialHook(PreForwardHook()       ,     PreForwardHook()       )
         add_hook_to_module(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			test_model(SCREAMING_SNAKE_CASE__       )
         assert torch.allclose(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       ,     atol=1e-5       )


    def        a				(	self					:					Any       )				->	Union[str, Any]:
         lowerCAmelCase__          =			ModelForTest()
         lowerCAmelCase__          =			torch.randn(2       ,     3       )
         lowerCAmelCase__          =			test_model(SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			PostForwardHook()
         add_hook_to_module(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			PostForwardHook()
         add_hook_to_module(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			SequentialHook(PostForwardHook()       ,     PostForwardHook()       )
         add_hook_to_module(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )

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


    def        a				(	self					:					Optional[int]       )				->	int:
         lowerCAmelCase__          =			ModelForTest()
         lowerCAmelCase__          =			torch.randn(2       ,     3       )
         lowerCAmelCase__          =			test_model(SCREAMING_SNAKE_CASE__       )

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

         lowerCAmelCase__          =			True
         lowerCAmelCase__          =			test_model(SCREAMING_SNAKE_CASE__       )
         self.assertFalse(outputa.requires_grad       )


    @require_multi_gpu
    def        a				(	self					:					Optional[Any]       )				->	List[str]:
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			torch.randn(2       ,     3       )
         lowerCAmelCase__          =			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__       )       )
         lowerCAmelCase__          =			torch.randn(2       ,     3       ).to(0       )
         lowerCAmelCase__          =			model(SCREAMING_SNAKE_CASE__       )
         self.assertEqual(output.device       ,     torch.device(0       )       )


    def        a				(	self					:					List[str]       )				->	List[str]:
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			{"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
         lowerCAmelCase__          =			torch.device(hook_kwargs["execution_device"]       )
         self.assertEqual(model.batchnorm.running_mean.device       ,     SCREAMING_SNAKE_CASE__       )

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

         lowerCAmelCase__          =			torch.randn(2       ,     3       )
         lowerCAmelCase__          =			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        a				(	self					:					Optional[int]       )				->	Union[str, Any]:
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			torch.device(SCREAMING_SNAKE_CASE__       )
         self.assertEqual(model.batchnorm.running_mean.device       ,     SCREAMING_SNAKE_CASE__       )

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

         lowerCAmelCase__          =			torch.randn(2       ,     3       )
         lowerCAmelCase__          =			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        a				(	self					:					Optional[Any]       )				->	str:
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			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
         lowerCAmelCase__          =			torch.device(SCREAMING_SNAKE_CASE__       )
         self.assertEqual(model.batchnorm.running_mean.device       ,     SCREAMING_SNAKE_CASE__       )

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

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