




"""simple docstring"""


import unittest
from typing import Dict, List, Optional, Union

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
     import torch

if is_vision_available():
     from PIL import Image

     from transformers import BridgeTowerImageProcessor






class  lowercase(					unittest.TestCase			):







 '''simple docstring'''





 def __init__(    self:		Dict,				a_:		List[str],				a_:		Optional[int] = True,				a_:		Optional[int] = None,				a_:		List[Any] = 32,				a_:		Dict = True,				a_:		List[str] = 1 / 255,				a_:		Tuple = True,				a_:		List[Any] = True,				a_:		str = [0.48_145_466, 0.4_578_275, 0.40_821_073],				a_:		Union[str, Any] = [0.26_862_954, 0.26_130_258, 0.27_577_711],				a_:		Optional[int] = True,				a_:		Union[str, Any]=7,				a_:		int=30,				a_:		Optional[Any]=400,				a_:		Optional[Any]=3,				):




   '''simple docstring'''



   _snake_case							:		Union[str, Any]								=  parent
   _snake_case							:		Dict								=  do_resize
   _snake_case							:		List[str]								=  size if size is not None else {"""shortest_edge""": 288}
   _snake_case							:		Optional[int]								=  size_divisor
   _snake_case							:		Optional[Any]								=  do_rescale
   _snake_case							:		Any								=  rescale_factor
   _snake_case							:		Optional[Any]								=  do_normalize
   _snake_case							:		List[Any]								=  do_center_crop
   _snake_case							:		Any								=  image_mean
   _snake_case							:		int								=  image_std
   _snake_case							:		Optional[Any]								=  do_pad
   _snake_case							:		Dict								=  batch_size
   _snake_case							:		Optional[int]								=  num_channels
   _snake_case							:		List[Any]								=  min_resolution
   _snake_case							:		Tuple								=  max_resolution


 def 			UpperCamelCase_       (    self:		Optional[Any]		):




   '''simple docstring'''



   return {
       "image_mean": self.image_mean,
       "image_std": self.image_std,
       "do_normalize": self.do_normalize,
       "do_resize": self.do_resize,
       "size": self.size,
       "size_divisor": self.size_divisor,
   }


 def 			UpperCamelCase_       (    self:		List[str],				a_:		Optional[Any],				a_:		List[Any]=False		):




   '''simple docstring'''



   if not batched:
     _snake_case							:		List[str]								=  self.size["""shortest_edge"""]
     _snake_case							:		Union[str, Any]								=  image_inputs[0]
     if isinstance(a_,				Image.Image		):
       _snake_case		, _snake_case							:		Optional[int]								=  image.size
     else:
       _snake_case		, _snake_case							:		List[str]								=  image.shape[1], image.shape[2]
     _snake_case							:		Tuple								=  size / min(a_,				a_		)
     if h < w:
       _snake_case		, _snake_case							:		Tuple								=  size, scale * w
     else:
       _snake_case		, _snake_case							:		Dict								=  scale * h, size

     _snake_case							:		str								=  int((1_333 / 800) * size		)
     if max(a_,				a_		) > max_size:
       _snake_case							:		Any								=  max_size / max(a_,				a_		)
       _snake_case							:		Optional[int]								=  newh * scale
       _snake_case							:		str								=  neww * scale

     _snake_case		, _snake_case							:		Tuple								=  int(newh + 0.5		), int(neww + 0.5		)
     _snake_case		, _snake_case							:		int								=  (
         newh // self.size_divisor * self.size_divisor,
         neww // self.size_divisor * self.size_divisor,
     )

   else:
     _snake_case							:		Any								=  []
     for image in image_inputs:
       _snake_case		, _snake_case							:		List[Any]								=  self.get_expected_values([image]		)
       expected_values.append((expected_height, expected_width)		)
     _snake_case							:		Tuple								=  max(a_,				key=lambda a_		: item[0]		)[0]
     _snake_case							:		Union[str, Any]								=  max(a_,				key=lambda a_		: item[1]		)[1]

   return expected_height, expected_width







@require_torch
@require_vision
class  lowercase(					lowerCamelCase__	,      unittest.TestCase			):







 '''simple docstring'''





 lowercase__			 = BridgeTowerImageProcessor if is_vision_available() else None


 def 			UpperCamelCase_       (    self:		List[str]		):




   '''simple docstring'''



   _snake_case							:		int								=  BridgeTowerImageProcessingTester(self		)


 @property
 def 			UpperCamelCase_       (    self:		List[Any]		):




   '''simple docstring'''



   return self.image_processor_tester.prepare_image_processor_dict()


 def 			UpperCamelCase_       (    self:		int		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  self.image_processing_class(**self.image_processor_dict		)
   self.assertTrue(hasattr(a_,				"""image_mean"""		)		)
   self.assertTrue(hasattr(a_,				"""image_std"""		)		)
   self.assertTrue(hasattr(a_,				"""do_normalize"""		)		)
   self.assertTrue(hasattr(a_,				"""do_resize"""		)		)
   self.assertTrue(hasattr(a_,				"""size"""		)		)
   self.assertTrue(hasattr(a_,				"""size_divisor"""		)		)


 def 			UpperCamelCase_       (    self:		Optional[int]		):




   '''simple docstring'''



   pass


 def 			UpperCamelCase_       (    self:		Optional[Any]		):




   '''simple docstring'''



   _snake_case							:		List[Any]								=  self.image_processing_class(**self.image_processor_dict		)
   # create random PIL images
   _snake_case							:		Dict								=  prepare_image_inputs(self.image_processor_tester,				equal_resolution=a_		)
   for image in image_inputs:
     self.assertIsInstance(a_,				Image.Image		)

   # Test not batched input
   _snake_case							:		Union[str, Any]								=  image_processing(image_inputs[0],				return_tensors="""pt"""		).pixel_values

   _snake_case		, _snake_case							:		List[str]								=  self.image_processor_tester.get_expected_values(a_		)
   self.assertEqual(
       encoded_images.shape,				(1, self.image_processor_tester.num_channels, expected_height, expected_width),				)

   # Test batched
   _snake_case							:		Union[str, Any]								=  image_processing(a_,				return_tensors="""pt"""		).pixel_values

   _snake_case		, _snake_case							:		int								=  self.image_processor_tester.get_expected_values(a_,				batched=a_		)
   self.assertEqual(
       encoded_images.shape,				(
           self.image_processor_tester.batch_size,
           self.image_processor_tester.num_channels,
           expected_height,
           expected_width,
       ),				)


 def 			UpperCamelCase_       (    self:		List[str]		):




   '''simple docstring'''



   _snake_case							:		Any								=  self.image_processing_class(**self.image_processor_dict		)
   # create random numpy tensors
   _snake_case							:		str								=  prepare_image_inputs(self.image_processor_tester,				equal_resolution=a_,				numpify=a_		)
   for image in image_inputs:
     self.assertIsInstance(a_,				np.ndarray		)

   # Test not batched input
   _snake_case							:		Tuple								=  image_processing(image_inputs[0],				return_tensors="""pt"""		).pixel_values

   _snake_case		, _snake_case							:		str								=  self.image_processor_tester.get_expected_values(a_		)
   self.assertEqual(
       encoded_images.shape,				(1, self.image_processor_tester.num_channels, expected_height, expected_width),				)

   # Test batched
   _snake_case							:		Optional[int]								=  image_processing(a_,				return_tensors="""pt"""		).pixel_values

   _snake_case		, _snake_case							:		List[Any]								=  self.image_processor_tester.get_expected_values(a_,				batched=a_		)
   self.assertEqual(
       encoded_images.shape,				(
           self.image_processor_tester.batch_size,
           self.image_processor_tester.num_channels,
           expected_height,
           expected_width,
       ),				)


 def 			UpperCamelCase_       (    self:		int		):




   '''simple docstring'''



   _snake_case							:		List[Any]								=  self.image_processing_class(**self.image_processor_dict		)
   # create random PyTorch tensors
   _snake_case							:		int								=  prepare_image_inputs(self.image_processor_tester,				equal_resolution=a_,				torchify=a_		)
   for image in image_inputs:
     self.assertIsInstance(a_,				torch.Tensor		)

   # Test not batched input
   _snake_case							:		Tuple								=  image_processing(image_inputs[0],				return_tensors="""pt"""		).pixel_values

   _snake_case		, _snake_case							:		str								=  self.image_processor_tester.get_expected_values(a_		)
   self.assertEqual(
       encoded_images.shape,				(1, self.image_processor_tester.num_channels, expected_height, expected_width),				)

   # Test batched
   _snake_case							:		Tuple								=  image_processing(a_,				return_tensors="""pt"""		).pixel_values

   _snake_case		, _snake_case							:		Optional[int]								=  self.image_processor_tester.get_expected_values(a_,				batched=a_		)
   self.assertEqual(
       encoded_images.shape,				(
           self.image_processor_tester.batch_size,
           self.image_processor_tester.num_channels,
           expected_height,
           expected_width,
       ),				)








"""simple docstring"""


import inspect
import unittest

import torch
import torch.nn as nn

from accelerate.hooks import (
    AlignDevicesHook,
    ModelHook,
    SequentialHook,
    add_hook_to_module,
    attach_align_device_hook,
    remove_hook_from_module,
    remove_hook_from_submodules,
)
from accelerate.test_utils import require_multi_gpu






class  lowercase(					nn.Module			):







 '''simple docstring'''





 def __init__(    self:		Optional[int]		):




   '''simple docstring'''



   super().__init__()
   _snake_case							:		List[str]								=  nn.Linear(3,				4		)
   _snake_case							:		int								=  nn.BatchNormad(4		)
   _snake_case							:		List[str]								=  nn.Linear(4,				5		)


 def 			UpperCamelCase_       (    self:		Any,				a_:		Union[str, Any]		):




   '''simple docstring'''



   return self.lineara(self.batchnorm(self.lineara(a_		)		)		)







class  lowercase(					__a			):







 '''simple docstring'''





 def 			UpperCamelCase_       (    self:		Any,				a_:		int,				*a_:		Dict,				**a_:		Dict		):




   '''simple docstring'''



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







class  lowercase(					__a			):







 '''simple docstring'''





 def 			UpperCamelCase_       (    self:		str,				a_:		Tuple,				a_:		Union[str, Any]		):




   '''simple docstring'''



   return output + 1







class  lowercase(					unittest.TestCase			):







 '''simple docstring'''





 def 			UpperCamelCase_       (    self:		Union[str, Any]		):




   '''simple docstring'''



   _snake_case							:		Tuple								=  ModelForTest()
   _snake_case							:		List[str]								=  ModelHook()

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


 def 			UpperCamelCase_       (    self:		Tuple		):




   '''simple docstring'''



   _snake_case							:		Dict								=  ModelForTest()
   _snake_case							:		List[Any]								=  ModelHook()

   add_hook_to_module(a_,				a_		)
   add_hook_to_module(a_,				a_,				append=a_		)

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


 def 			UpperCamelCase_       (    self:		Tuple		):




   '''simple docstring'''



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

   _snake_case							:		Any								=  PreForwardHook()
   add_hook_to_module(a_,				a_		)
   _snake_case							:		List[Any]								=  test_model(a_		)
   self.assertTrue(torch.allclose(a_,				a_,				atol=1E-5		)		)

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

   # You need to use the sequential hook to chain two or more hooks
   _snake_case							:		str								=  SequentialHook(PreForwardHook(),				PreForwardHook()		)
   add_hook_to_module(a_,				a_		)

   _snake_case							:		str								=  test_model(a_		)
   assert torch.allclose(a_,				a_,				atol=1E-5		)


 def 			UpperCamelCase_       (    self:		Optional[int]		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  ModelForTest()
   _snake_case							:		List[Any]								=  torch.randn(2,				3		)
   _snake_case							:		List[str]								=  test_model(a_		)

   _snake_case							:		List[Any]								=  PostForwardHook()
   add_hook_to_module(a_,				a_		)
   _snake_case							:		Union[str, Any]								=  test_model(a_		)
   self.assertTrue(torch.allclose(a_,				output + 1,				atol=1E-5		)		)

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

   # You need to use the sequential hook to chain two or more hooks
   _snake_case							:		Dict								=  SequentialHook(PostForwardHook(),				PostForwardHook()		)
   add_hook_to_module(a_,				a_		)

   _snake_case							:		List[str]								=  test_model(a_		)
   assert torch.allclose(a_,				output + 2,				atol=1E-5		)


 def 			UpperCamelCase_       (    self:		Optional[Any]		):




   '''simple docstring'''



   _snake_case							:		str								=  ModelForTest()
   _snake_case							:		Any								=  torch.randn(2,				3		)
   _snake_case							:		List[str]								=  test_model(a_		)

   _snake_case							:		List[Any]								=  PostForwardHook()
   add_hook_to_module(a_,				a_		)
   _snake_case							:		Dict								=  test_model(a_		)
   self.assertTrue(torch.allclose(a_,				output + 1		)		)
   self.assertTrue(outputa.requires_grad		)

   _snake_case							:		Union[str, Any]								=  True
   _snake_case							:		Dict								=  test_model(a_		)
   self.assertFalse(outputa.requires_grad		)


 @require_multi_gpu
 def 			UpperCamelCase_       (    self:		Tuple		):




   '''simple docstring'''



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


 def 			UpperCamelCase_       (    self:		Union[str, Any]		):




   '''simple docstring'''



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

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

   # 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
   _snake_case							:		Union[str, Any]								=  torch.device(hook_kwargs["""execution_device"""]		)
   self.assertEqual(model.batchnorm.running_mean.device,				a_		)

   _snake_case							:		str								=  torch.randn(2,				3		)
   _snake_case							:		Optional[int]								=  model(a_		)
   self.assertEqual(output.device,				a_		)

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

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

   # 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"""		)		)

   _snake_case							:		List[Any]								=  torch.randn(2,				3		)
   _snake_case							:		Any								=  model(a_		)
   self.assertEqual(output.device,				a_		)

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




   '''simple docstring'''



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

   # 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
   _snake_case							:		Optional[int]								=  torch.device(a_		)
   self.assertEqual(model.batchnorm.running_mean.device,				a_		)

   _snake_case							:		List[Any]								=  torch.randn(2,				3		)
   _snake_case							:		List[Any]								=  model(a_		)
   self.assertEqual(output.device,				a_		)

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

   # 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"""		)		)

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

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




   '''simple docstring'''



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

   _snake_case							:		int								=  torch.randn(2,				3		)
   _snake_case							:		str								=  model(a_		)
   self.assertEqual(output.device,				a_		)

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

   # 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"""		)		)

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

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


