



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


      '''simple docstring'''



      def __init__( self				)     ->     Any:
       super().__init__()
       lowerCAmelCase   =     nn.Linear(3     ,	4				)
       lowerCAmelCase   =     nn.BatchNormad(4				)
       lowerCAmelCase   =     nn.Linear(4     ,	5				)




      def     __snake_case   ( self     ,	A_				)     ->     Union[str, Any]:
       return self.lineara(self.batchnorm(self.lineara(A_				)				)				)







class __snake_case(				_lowerCAmelCase ):


      '''simple docstring'''



      def     __snake_case   ( self     ,	A_     ,	*A_     ,	**A_				)     ->     Any:
       return (args[0] + 1,) + args[1:], kwargs







class __snake_case(				_lowerCAmelCase ):


      '''simple docstring'''



      def     __snake_case   ( self     ,	A_     ,	A_				)     ->     Tuple:
       return output + 1







class __snake_case(				unittest.TestCase ):


      '''simple docstring'''



      def     __snake_case   ( self				)     ->     Dict:
       lowerCAmelCase   =     ModelForTest()
       lowerCAmelCase   =     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     __snake_case   ( self				)     ->     Tuple:
       lowerCAmelCase   =     ModelForTest()
       lowerCAmelCase   =     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     __snake_case   ( self				)     ->     Optional[Any]:
       lowerCAmelCase   =     ModelForTest()
       lowerCAmelCase   =     torch.randn(2     ,	3				)
       lowerCAmelCase   =     test_model(x + 1				)
       lowerCAmelCase   =     test_model(x + 2				)

       lowerCAmelCase   =     PreForwardHook()
       add_hook_to_module(A_     ,	A_				)
       lowerCAmelCase   =     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
       lowerCAmelCase   =     PreForwardHook()
       add_hook_to_module(A_     ,	A_				)
       lowerCAmelCase   =     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
       lowerCAmelCase   =     SequentialHook(PreForwardHook()     ,	PreForwardHook()				)
       add_hook_to_module(A_     ,	A_				)

       lowerCAmelCase   =     test_model(A_				)
       assert torch.allclose(A_     ,	A_     ,	atol=1e-5				)


      def     __snake_case   ( self				)     ->     Optional[Any]:
       lowerCAmelCase   =     ModelForTest()
       lowerCAmelCase   =     torch.randn(2     ,	3				)
       lowerCAmelCase   =     test_model(A_				)

       lowerCAmelCase   =     PostForwardHook()
       add_hook_to_module(A_     ,	A_				)
       lowerCAmelCase   =     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
       lowerCAmelCase   =     PostForwardHook()
       add_hook_to_module(A_     ,	A_				)
       lowerCAmelCase   =     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
       lowerCAmelCase   =     SequentialHook(PostForwardHook()     ,	PostForwardHook()				)
       add_hook_to_module(A_     ,	A_				)

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


      def     __snake_case   ( self				)     ->     List[str]:
       lowerCAmelCase   =     ModelForTest()
       lowerCAmelCase   =     torch.randn(2     ,	3				)
       lowerCAmelCase   =     test_model(A_				)

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

       lowerCAmelCase   =     True
       lowerCAmelCase   =     test_model(A_				)
       self.assertFalse(outputa.requires_grad				)


      @require_multi_gpu
      def     __snake_case   ( self				)     ->     Tuple:
       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(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_				)				)
       lowerCAmelCase   =     torch.randn(2     ,	3				).to(0				)
       lowerCAmelCase   =     model(A_				)
       self.assertEqual(output.device     ,	torch.device(0				)				)


      def     __snake_case   ( self				)     ->     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   =     {"""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
       lowerCAmelCase   =     torch.device(hook_kwargs["""execution_device"""]				)
       self.assertEqual(model.batchnorm.running_mean.device     ,	A_				)

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

       lowerCAmelCase   =     torch.randn(2     ,	3				)
       lowerCAmelCase   =     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     __snake_case   ( self				)     ->     int:
       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(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
       lowerCAmelCase   =     torch.device(A_				)
       self.assertEqual(model.batchnorm.running_mean.device     ,	A_				)

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

       lowerCAmelCase   =     torch.randn(2     ,	3				)
       lowerCAmelCase   =     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     __snake_case   ( self				)     ->     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(
           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
       lowerCAmelCase   =     torch.device(A_				)
       self.assertEqual(model.batchnorm.running_mean.device     ,	A_				)

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

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




'''simple docstring'''

import pytest

import datasets


# Import fixture modules as plugins
UpperCAmelCase     		=   ['tests.fixtures.files', 'tests.fixtures.hub', 'tests.fixtures.fsspec']






def 		_snake_case		( _SCREAMING_SNAKE_CASE					:						str							,		_SCREAMING_SNAKE_CASE					:						int							)      ->						Optional[int]:







	"""simple docstring"""
	# Mark tests as "unit" by default if not marked as "integration" (or already marked as "unit")
	for item in items:
		if any(marker in item.keywords for marker in ["""integration""", """unit"""]							):
			continue
		item.add_marker(pytest.mark.unit							)






def 		_snake_case		( _SCREAMING_SNAKE_CASE					:						List[str]							)      ->						Any:







	"""simple docstring"""
	config.addinivalue_line("""markers"""							,		"""torchaudio_latest: mark test to run with torchaudio>=0.12"""							)






@pytest.fixture(autouse=_SCREAMING_SNAKE_CASE							)
def 		_snake_case		( _SCREAMING_SNAKE_CASE					:						Dict							,		_SCREAMING_SNAKE_CASE					:						Dict							)      ->						str:







	"""simple docstring"""
	# test_hf_cache_home = tmp_path_factory.mktemp("cache")  # TODO: why a cache dir per test function does not work?
	lowerCAmelCase   =     tmp_path_factory.getbasetemp() / """cache"""
	lowerCAmelCase   =     test_hf_cache_home / """datasets"""
	lowerCAmelCase   =     test_hf_cache_home / """metrics"""
	lowerCAmelCase   =     test_hf_cache_home / """modules"""
	monkeypatch.setattr("""datasets.config.HF_DATASETS_CACHE"""							,		str(_SCREAMING_SNAKE_CASE							)							)
	monkeypatch.setattr("""datasets.config.HF_METRICS_CACHE"""							,		str(_SCREAMING_SNAKE_CASE							)							)
	monkeypatch.setattr("""datasets.config.HF_MODULES_CACHE"""							,		str(_SCREAMING_SNAKE_CASE							)							)
	lowerCAmelCase   =     test_hf_datasets_cache / """downloads"""
	monkeypatch.setattr("""datasets.config.DOWNLOADED_DATASETS_PATH"""							,		str(_SCREAMING_SNAKE_CASE							)							)
	lowerCAmelCase   =     test_hf_datasets_cache / """downloads""" / """extracted"""
	monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_PATH"""							,		str(_SCREAMING_SNAKE_CASE							)							)






@pytest.fixture(autouse=_SCREAMING_SNAKE_CASE							,		scope="""session"""							)
def 		_snake_case		( )      ->						Optional[Any]:







	"""simple docstring"""
	datasets.disable_progress_bar()






@pytest.fixture(autouse=_SCREAMING_SNAKE_CASE							)
def 		_snake_case		( _SCREAMING_SNAKE_CASE					:						Any							)      ->						Optional[int]:







	"""simple docstring"""
	# don't take tests into account when counting downloads
	monkeypatch.setattr("""datasets.config.HF_UPDATE_DOWNLOAD_COUNTS"""							,		_SCREAMING_SNAKE_CASE							)






@pytest.fixture
def 		_snake_case		( _SCREAMING_SNAKE_CASE					:						List[str]							)      ->						str:







	"""simple docstring"""
	# Required to suppress RemovedIn20Warning when feature(s) are not compatible with SQLAlchemy 2.0
	# To be removed once SQLAlchemy 2.0 supported
	monkeypatch.setattr("""sqlalchemy.util.deprecations.SILENCE_UBER_WARNING"""							,		_SCREAMING_SNAKE_CASE							)