






"""simple docstring"""







import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


a			:    List[Any]				 =		WebClient(token=os.environ["""CI_SLACK_BOT_TOKEN"""])





def 						lowercase__(A						)    ->Dict:




						"""simple docstring"""



						lowercase__    :				List[Any]=	test_results.split(" "						)

						lowercase__    :				Union[str, Any]=	0
						lowercase__    :				Optional[int]=	0

						# When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
						# When it is too long, those signs are not present.
						lowercase__    :				Any=	expressions[-2] if "=" in expressions[-1] else expressions[-1]

						for i, expression in enumerate(A						):
												if "failed" in expression:
																		failed += int(expressions[i - 1]						)
												if "passed" in expression:
																		success += int(expressions[i - 1]						)

						return failed, success, time_spent





def 						lowercase__(A						)    ->Optional[int]:




						"""simple docstring"""



						lowercase__    :				List[Any]=	{}
						lowercase__    :				Any=	None
						lowercase__    :				Any=	False
						for line in failures_short_lines.split("\n"						):
												if re.search(R"_ \[doctest\]" ,		A						):
																		lowercase__    :				List[Any]=	True
																		lowercase__    :				List[str]=	line.split(" "						)[2]
												elif in_error and not line.split(" "						)[0].isdigit():
																		lowercase__    :				List[str]=	line
																		lowercase__    :				Optional[Any]=	False

						return failures


class    __UpperCAmelCase:




						"""simple docstring"""




						def __init__(       self	,  snake_case__	,  snake_case__      ):
												'''simple docstring'''
												lowercase__    :				Union[str, Any]=	title

												lowercase__    :				List[Any]=	doc_test_results["time_spent"].split(","      )[0]
												lowercase__    :				Optional[Any]=	doc_test_results["success"]
												lowercase__    :				Any=	doc_test_results["failures"]
												lowercase__    :				List[Any]=	self.n_success + self.n_failures

												# Failures and success of the modeling tests
												lowercase__    :				Optional[Any]=	doc_test_results




						@property
						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				Optional[Any]=	[self._time_spent]
												lowercase__    :				Tuple=	0

												for time in time_spent:
																		lowercase__    :				str=	time.split(":"      )

																		# Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
																		if len(snake_case__      ) == 1:
																								lowercase__    :				List[Any]=	[0, 0, time_parts[0]]

																		lowercase__,     lowercase__,     lowercase__    :				Dict=	int(time_parts[0]      ), int(time_parts[1]      ), float(time_parts[2]      )
																		total_secs += hours * 3600 + minutes * 60 + seconds

												lowercase__,     lowercase__,     lowercase__    :				Union[str, Any]=	total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
												return F'''{int(snake_case__      )}h{int(snake_case__      )}m{int(snake_case__      )}s'''




						@property
						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												return {"type": "header", "text": {"type": "plain_text", "text": self.title}}




						@property
						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												return {
												    "type": "section",
												    "text": {
												        "type": "plain_text",
												        "text": F'''🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.''',
												        "emoji": True,
												    },
												    "accessory": {
												        "type": "button",
												        "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
												        "url": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
												    },
												}




						@property
						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												return {
												    "type": "section",
												    "text": {
												        "type": "plain_text",
												        "text": (
												            F'''There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in'''
												            F''' {self.time}.'''
												        ),
												        "emoji": True,
												    },
												    "accessory": {
												        "type": "button",
												        "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
												        "url": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
												    },
												}




						@property
						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				Dict=	40
												lowercase__    :				Union[str, Any]=	{k: v["failed"] for k, v in doc_test_results.items() if isinstance(snake_case__	,  snake_case__      )}

												lowercase__    :				List[Any]=	""
												for category, failures in category_failures.items():
																		if len(snake_case__      ) == 0:
																								continue

																		if report != "":
																								report += "\n\n"

																		report += F'''*{category} failures*:'''.ljust(line_length // 2      ).rjust(line_length // 2      ) + "\n"
																		report += "`"
																		report += "`\n`".join(snake_case__      )
																		report += "`"

												return {
												    "type": "section",
												    "text": {
												        "type": "mrkdwn",
												        "text": F'''The following examples had failures:\n\n\n{report}\n''',
												    },
												}




						@property
						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				str=	[self.header]

												if self.n_failures > 0:
																		blocks.append(self.failures      )

												if self.n_failures > 0:
																		blocks.extend([self.category_failures]      )

												if self.n_failures == 0:
																		blocks.append(self.no_failures      )

												return json.dumps(snake_case__      )




						@staticmethod
						def 		UpperCAmelCase_	(       ):
												'''simple docstring'''
												lowercase__    :				Tuple=	[
												    {
												        "type": "section",
												        "text": {
												            "type": "plain_text",
												            "text": "There was an issue running the tests.",
												        },
												        "accessory": {
												            "type": "button",
												            "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
												            "url": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
												        },
												    }
												]

												print("Sending the following payload"      )
												print(json.dumps({"blocks": json.loads(snake_case__      )}      )      )

												client.chat_postMessage(
												    channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"]	,  text="There was an issue running the tests."	,  blocks=snake_case__	,  )




						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												print("Sending the following payload"      )
												print(json.dumps({"blocks": json.loads(self.payload      )}      )      )

												lowercase__    :				Tuple=	F'''{self.n_failures} failures out of {self.n_tests} tests,''' if self.n_failures else "All tests passed."

												lowercase__    :				List[Any]=	client.chat_postMessage(
												    channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"]	,  blocks=self.payload	,  text=snake_case__	,  )




						def 		UpperCAmelCase_	(       self	,  snake_case__	,  snake_case__	,  snake_case__	,  snake_case__      ):
												'''simple docstring'''
												lowercase__    :				List[Any]=	""
												for key, value in failures.items():
																		lowercase__    :				Union[str, Any]=	value[:200] + " [Truncated]" if len(snake_case__      ) > 250 else value
																		failures_text += F'''*{key}*\n_{value}_\n\n'''

												lowercase__    :				Optional[Any]=	job_name
												lowercase__    :				List[Any]=	{"type": "section", "text": {"type": "mrkdwn", "text": text}}

												if job_link is not None:
																		lowercase__    :				List[str]=	{
																		    "type": "button",
																		    "text": {"type": "plain_text", "text": "GitHub Action job", "emoji": True},
																		    "url": job_link,
																		}

												return [
												    {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
												    content,
												    {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
												]




						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												if self.thread_ts is None:
																		raise ValueError("Can only post reply if a post has been made."      )

												lowercase__    :				Any=	self.doc_test_results.pop("job_link"      )
												self.doc_test_results.pop("failures"      )
												self.doc_test_results.pop("success"      )
												self.doc_test_results.pop("time_spent"      )

												lowercase__    :				int=	sorted(self.doc_test_results.items()	,  key=lambda snake_case__      : t[0]      )
												for job, job_result in sorted_dict:
																		if len(job_result["failures"]      ):
																								lowercase__    :				Any=	F'''*Num failures* :{len(job_result['failed']      )} \n'''
																								lowercase__    :				List[Any]=	job_result["failures"]
																								lowercase__    :				List[str]=	self.get_reply_blocks(snake_case__	,  snake_case__	,  snake_case__	,  text=snake_case__      )

																								print("Sending the following reply"      )
																								print(json.dumps({"blocks": blocks}      )      )

																								client.chat_postMessage(
																								    channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"]	,  text=F'''Results for {job}'''	,  blocks=snake_case__	,  thread_ts=self.thread_ts["ts"]	,  )

																								time.sleep(1      )





def 						lowercase__()    ->Any:




						"""simple docstring"""



						lowercase__    :				Any=	os.environ["GITHUB_RUN_ID"]
						lowercase__    :				Union[str, Any]=	f'''https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100'''
						lowercase__    :				Tuple=	requests.get(A						).json()
						lowercase__    :				str=	{}

						try:
												jobs.update({job["name"]: job["html_url"] for job in result["jobs"]}						)
												lowercase__    :				str=	math.ceil((result["total_count"] - 100) / 100						)

												for i in range(A						):
																		lowercase__    :				List[str]=	requests.get(url + f'''&page={i + 2}'''						).json()
																		jobs.update({job["name"]: job["html_url"] for job in result["jobs"]}						)

												return jobs
						except Exception as e:
												print("Unknown error, could not fetch links." ,		A						)

						return {}





def 						lowercase__(A						)    ->str:




						"""simple docstring"""



						lowercase__    :				List[str]=	{}

						if os.path.exists(A						):
												lowercase__    :				List[str]=	os.listdir(A						)
												for file in files:
																		try:
																								with open(os.path.join(A ,		A						) ,		encoding="utf-8"						) as f:
																														lowercase__    :				Optional[Any]=	f.read()
																		except UnicodeDecodeError as e:
																								raise ValueError(f'''Could not open {os.path.join(A ,		A						)}.'''						) from e

						return _artifact





def 						lowercase__()    ->Optional[int]:




						"""simple docstring"""



						class    __UpperCAmelCase:




												"""simple docstring"""




												def __init__(       self	,  snake_case__      ):
																		'''simple docstring'''
																		lowercase__    :				int=	name
																		lowercase__    :				Tuple=	[]




												def __str__(       self      ):
																		'''simple docstring'''
																		return self.name




												def 		UpperCAmelCase_	(       self	,  snake_case__      ):
																		'''simple docstring'''
																		self.paths.append({"name": self.name, "path": path}      )


						lowercase__    :				Dict[str, Artifact]=	{}

						lowercase__    :				int=	filter(os.path.isdir ,		os.listdir()						)
						for directory in directories:
												lowercase__    :				Optional[int]=	directory
												if artifact_name not in _available_artifacts:
																		lowercase__    :				List[Any]=	Artifact(A						)

																		_available_artifacts[artifact_name].add_path(A						)

						return _available_artifacts


if __name__ == "__main__":
			a			:    Union[str, Any]				 =		get_job_links()
			a			:    Optional[Any]				 =		retrieve_available_artifacts()

			a			:    List[Any]				 =		collections.OrderedDict(
			    [
			        ("""*.py""", """API Examples"""),
			        ("""*.md""", """MD Examples"""),
			    ]
			)

			# This dict will contain all the information relative to each doc test category:
			# - failed: list of failed tests
			# - failures: dict in the format 'test': 'error_message'
			a			:    Union[str, Any]				 =		{
			    v: {
			        """failed""": [],
			        """failures""": {},
			    }
			    for v in docs.values()
			}

			# Link to the GitHub Action job
			a			:    str				 =		github_actions_job_links.get("""run_doctests""")

			a			:    str				 =		available_artifacts["""doc_tests_gpu_test_reports"""].paths[0]
			a			:    Optional[int]				 =		retrieve_artifact(artifact_path["""name"""])
			if "stats" in artifact:
						a ,		a ,		a			:    Tuple				 =		handle_test_results(artifact["""stats"""])
						a			:    Optional[int]				 =		failed
						a			:    str				 =		success
						a			:    int				 =		time_spent[1:-1] + """, """

						a			:    Optional[int]				 =		extract_first_line_failure(artifact["""failures_short"""])
						for line in artifact["summary_short"].split("""\n"""):
									if re.search("""FAILED""", line):
												a			:    Optional[int]				 =		line.replace("""FAILED """, """""")
												a			:    List[str]				 =		line.split()[0].replace("""\n""", """""")

												if "::" in line:
															a ,		a			:    str				 =		line.split("""::""")
												else:
															a ,		a			:    str				 =		line, line

												for file_regex in docs.keys():
															if fnmatch(file_path, file_regex):
																		a			:    Dict				 =		docs[file_regex]
																		doc_test_results[category]["failed"].append(test)

																		a			:    Dict				 =		all_failures[test] if test in all_failures else """N/A"""
																		a			:    Optional[int]				 =		failure
																		break

			a			:    Optional[int]				 =		Message("""🤗 Results of the doc tests.""", doc_test_results)
			message.post()
			message.post_reply()














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




						"""simple docstring"""




						def __init__(       self      ):
												'''simple docstring'''
												super().__init__()
												lowercase__    :				Any=	nn.Linear(3	,  4      )
												lowercase__    :				Tuple=	nn.BatchNormad(4      )
												lowercase__    :				Dict=	nn.Linear(4	,  5      )




						def 		UpperCAmelCase_	(       self	,  snake_case__      ):
												'''simple docstring'''
												return self.lineara(self.batchnorm(self.lineara(snake_case__      )      )      )


class    __UpperCAmelCase(       SCREAMING_SNAKE_CASE__							):




						"""simple docstring"""




						def 		UpperCAmelCase_	(       self	,  snake_case__	,  *snake_case__	,  **snake_case__      ):
												'''simple docstring'''
												return (args[0] + 1,) + args[1:], kwargs


class    __UpperCAmelCase(       SCREAMING_SNAKE_CASE__							):




						"""simple docstring"""




						def 		UpperCAmelCase_	(       self	,  snake_case__	,  snake_case__      ):
												'''simple docstring'''
												return output + 1


class    __UpperCAmelCase(       unittest.TestCase							):




						"""simple docstring"""




						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				str=	ModelForTest()
												lowercase__    :				str=	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 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				int=	ModelForTest()
												lowercase__    :				int=	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 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				Dict=	ModelForTest()
												lowercase__    :				int=	torch.randn(2	,  3      )
												lowercase__    :				Optional[Any]=	test_model(x + 1      )
												lowercase__    :				Tuple=	test_model(x + 2      )

												lowercase__    :				str=	PreForwardHook()
												add_hook_to_module(snake_case__	,  snake_case__      )
												lowercase__    :				Tuple=	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
												lowercase__    :				Tuple=	PreForwardHook()
												add_hook_to_module(snake_case__	,  snake_case__      )
												lowercase__    :				Optional[Any]=	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
												lowercase__    :				List[str]=	SequentialHook(PreForwardHook()	,  PreForwardHook()      )
												add_hook_to_module(snake_case__	,  snake_case__      )

												lowercase__    :				Dict=	test_model(snake_case__      )
												assert torch.allclose(snake_case__	,  snake_case__	,  atol=1e-5      )




						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				Tuple=	ModelForTest()
												lowercase__    :				Optional[int]=	torch.randn(2	,  3      )
												lowercase__    :				Optional[int]=	test_model(snake_case__      )

												lowercase__    :				str=	PostForwardHook()
												add_hook_to_module(snake_case__	,  snake_case__      )
												lowercase__    :				Optional[int]=	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
												lowercase__    :				Tuple=	PostForwardHook()
												add_hook_to_module(snake_case__	,  snake_case__      )
												lowercase__    :				Dict=	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
												lowercase__    :				Optional[Any]=	SequentialHook(PostForwardHook()	,  PostForwardHook()      )
												add_hook_to_module(snake_case__	,  snake_case__      )

												lowercase__    :				List[str]=	test_model(snake_case__      )
												assert torch.allclose(snake_case__	,  output + 2	,  atol=1e-5      )




						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				int=	ModelForTest()
												lowercase__    :				Optional[Any]=	torch.randn(2	,  3      )
												lowercase__    :				int=	test_model(snake_case__      )

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

												lowercase__    :				Any=	True
												lowercase__    :				Optional[int]=	test_model(snake_case__      )
												self.assertFalse(outputa.requires_grad      )




						@require_multi_gpu
						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				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
												lowercase__    :				int=	torch.randn(2	,  3      )
												lowercase__    :				List[str]=	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__      )      )
												lowercase__    :				Tuple=	torch.randn(2	,  3      ).to(0      )
												lowercase__    :				Optional[Any]=	model(snake_case__      )
												self.assertEqual(output.device	,  torch.device(0      )      )




						def 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				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
												lowercase__    :				Optional[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
												lowercase__    :				Optional[int]=	torch.device(hook_kwargs["execution_device"]      )
												self.assertEqual(model.batchnorm.running_mean.device	,  snake_case__      )

												lowercase__    :				List[Any]=	torch.randn(2	,  3      )
												lowercase__    :				str=	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
												lowercase__    :				Optional[int]=	{
												    "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"      )      )

												lowercase__    :				str=	torch.randn(2	,  3      )
												lowercase__    :				str=	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 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				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
												lowercase__    :				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
												lowercase__    :				Dict=	torch.device(snake_case__      )
												self.assertEqual(model.batchnorm.running_mean.device	,  snake_case__      )

												lowercase__    :				Optional[Any]=	torch.randn(2	,  3      )
												lowercase__    :				List[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__	,  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"      )      )

												lowercase__    :				List[str]=	torch.randn(2	,  3      )
												lowercase__    :				List[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 		UpperCAmelCase_	(       self      ):
												'''simple docstring'''
												lowercase__    :				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
												lowercase__    :				Optional[Any]=	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
												lowercase__    :				Tuple=	torch.device(snake_case__      )
												self.assertEqual(model.batchnorm.running_mean.device	,  snake_case__      )

												lowercase__    :				str=	torch.randn(2	,  3      )
												lowercase__    :				List[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"      )      )

												lowercase__    :				Dict=	torch.randn(2	,  3      )
												lowercase__    :				List[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"      )      )






