




"""simple docstring"""


import os
import re
import warnings
from shutil import copyfile
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer


if TYPE_CHECKING:
     from ...tokenization_utils_base import TextInput
from ...utils import logging


A_      =    logging.get_logger(__name__)

A_      =    {'''vocab_file''': '''spiece.model'''}

A_      =    {
    '''vocab_file''': {
        '''t5-small''': '''https://huggingface.co/t5-small/resolve/main/spiece.model''',
        '''t5-base''': '''https://huggingface.co/t5-base/resolve/main/spiece.model''',
        '''t5-large''': '''https://huggingface.co/t5-large/resolve/main/spiece.model''',
        '''t5-3b''': '''https://huggingface.co/t5-3b/resolve/main/spiece.model''',
        '''t5-11b''': '''https://huggingface.co/t5-11b/resolve/main/spiece.model''',
    }
}


# TODO(PVP) - this should be removed in Transformers v5
A_      =    {
    '''t5-small''': 5_12,
    '''t5-base''': 5_12,
    '''t5-large''': 5_12,
    '''t5-3b''': 5_12,
    '''t5-11b''': 5_12,
}

A_      =    '''▁'''






class  lowercase(					__a			):







 '''simple docstring'''





 lowercase__			 = VOCAB_FILES_NAMES
 lowercase__			 = PRETRAINED_VOCAB_FILES_MAP
 lowercase__			 = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 lowercase__			 = ["input_ids", "attention_mask"]


 def __init__(    self:		Optional[Any],				a_:		Any,				a_:		List[Any]="</s>",				a_:		Tuple="<unk>",				a_:		int="<pad>",				a_:		List[str]=100,				a_:		int=None,				a_:		Optional[Dict[str, Any]] = None,				a_:		Optional[Any]=True,				**a_:		str,				):




   '''simple docstring'''



   if extra_ids > 0 and additional_special_tokens is None:
     _snake_case							:		Tuple								=  [f"<extra_id_{i}>" for i in range(a_		)]
   elif extra_ids > 0 and additional_special_tokens is not None:
     # Check that we have the right number of extra_id special tokens
     _snake_case							:		List[Any]								=  len(set(filter(lambda a_		: bool("""extra_id""" in str(a_		)		),				a_		)		)		)
     if extra_tokens != extra_ids:
       raise ValueError(
           f"Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are"
           """ provided to T5Tokenizer. In this case the additional_special_tokens must include the extra_ids"""
           """ tokens"""		)
   if legacy:
     logger.warning_once(
         f"You are using the legacy behaviour of the {self.__class__}. This means that tokens that come after special tokens will not be properly handled. We recommend you to"
         """ read the related pull request available at https://github.com/huggingface/transformers/pull/24565"""		)

   _snake_case							:		Union[str, Any]								=  legacy
   _snake_case							:		Optional[Any]								=  {} if sp_model_kwargs is None else sp_model_kwargs

   super().__init__(
       eos_token=a_,				unk_token=a_,				pad_token=a_,				extra_ids=a_,				additional_special_tokens=a_,				sp_model_kwargs=self.sp_model_kwargs,				legacy=a_,				**a_,				)

   _snake_case							:		Tuple								=  vocab_file
   _snake_case							:		Any								=  extra_ids

   _snake_case							:		Tuple								=  spm.SentencePieceProcessor(**self.sp_model_kwargs		)
   self.sp_model.Load(a_		)


 @staticmethod
 def 			UpperCamelCase_       (    a_:		Union[str, Any],				a_:		int,				a_:		int		):




   '''simple docstring'''



   if pretrained_model_name_or_path in TaTokenizer.max_model_input_sizes:
     _snake_case							:		Tuple								=  TaTokenizer.max_model_input_sizes[pretrained_model_name_or_path]
     if init_max_model_length is not None and init_max_model_length != max_model_length:
       return init_max_model_length
     elif init_max_model_length is None:
       warnings.warn(
           """This tokenizer was incorrectly instantiated with a model max length of"""
           f" {deprecated_max_model_length} which will be corrected in Transformers v5.\nFor now, this"
           """ behavior is kept to avoid breaking backwards compatibility when padding/encoding with"""
           """ `truncation is True`.\n- Be aware that you SHOULD NOT rely on"""
           f" {pretrained_model_name_or_path} automatically truncating your input to"
           f" {deprecated_max_model_length} when padding/encoding.\n- If you want to encode/pad to sequences"
           f" longer than {deprecated_max_model_length} you can either instantiate this tokenizer with"
           """ `model_max_length` or pass `max_length` when encoding/padding.\n- To avoid this warning, please"""
           """ instantiate this tokenizer with `model_max_length` set to your preferred value.""",				a_,				)

   return max_model_length


 @property
 def 			UpperCamelCase_       (    self:		str		):




   '''simple docstring'''



   return self.sp_model.get_piece_size() + self._extra_ids


 def 			UpperCamelCase_       (    self:		List[Any]		):




   '''simple docstring'''



   _snake_case							:		List[str]								=  {self.convert_ids_to_tokens(a_		): i for i in range(self.vocab_size		)}
   vocab.update(self.added_tokens_encoder		)
   return vocab


 def 			UpperCamelCase_       (    self:		List[str],				a_:		List[int],				a_:		Optional[List[int]] = None,				a_:		bool = False		):




   '''simple docstring'''



   if already_has_special_tokens:
     return super().get_special_tokens_mask(
         token_ids_a=a_,				token_ids_a=a_,				already_has_special_tokens=a_		)

   # normal case: some special tokens
   if token_ids_a is None:
     return ([0] * len(a_		)) + [1]
   return ([0] * len(a_		)) + [1] + ([0] * len(a_		)) + [1]


 def 			UpperCamelCase_       (    self:		List[Any]		):




   '''simple docstring'''



   return list(
       set(filter(lambda a_		: bool(re.search(r"""<extra_id_\d+>""",				a_		)		) is not None,				self.additional_special_tokens		)		)		)


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




   '''simple docstring'''



   return [self._convert_token_to_id(a_		) for token in self.get_sentinel_tokens()]


 def 			UpperCamelCase_       (    self:		str,				a_:		List[int]		):




   '''simple docstring'''



   if len(a_		) > 0 and token_ids[-1] == self.eos_token_id:
     warnings.warn(
         f"This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated"
         """ eos tokens being added."""		)
     return token_ids
   else:
     return token_ids + [self.eos_token_id]


 def 			UpperCamelCase_       (    self:		List[Any],				a_:		List[int],				a_:		Optional[List[int]] = None		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  [self.eos_token_id]

   if token_ids_a is None:
     return len(token_ids_a + eos		) * [0]
   return len(token_ids_a + eos + token_ids_a + eos		) * [0]


 def 			UpperCamelCase_       (    self:		List[str],				a_:		List[int],				a_:		Optional[List[int]] = None		):




   '''simple docstring'''



   _snake_case							:		List[str]								=  self._add_eos_if_not_present(a_		)
   if token_ids_a is None:
     return token_ids_a
   else:
     _snake_case							:		Tuple								=  self._add_eos_if_not_present(a_		)
     return token_ids_a + token_ids_a


 def __getstate__(    self:		str		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  self.__dict__.copy()
   _snake_case							:		Optional[int]								=  None
   return state


 def __setstate__(    self:		Optional[int],				a_:		Dict		):




   '''simple docstring'''



   _snake_case							:		Optional[Any]								=  d

   # for backward compatibility
   if not hasattr(self,				"""sp_model_kwargs"""		):
     _snake_case							:		Tuple								=  {}

   _snake_case							:		int								=  spm.SentencePieceProcessor(**self.sp_model_kwargs		)
   self.sp_model.Load(self.vocab_file		)


 def 			UpperCamelCase_       (    self:		List[Any],				a_:		"TextInput",				**a_:		Optional[Any]		):




   '''simple docstring'''



   if not self.legacy:
     _snake_case							:		int								=  SPIECE_UNDERLINE + text.replace(a_,				""" """		)
   return super().tokenize(a_,				**a_		)


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




   '''simple docstring'''



   if not self.legacy:
     _snake_case							:		Tuple								=  text.startswith(a_		)
     if is_first:
       _snake_case							:		Optional[int]								=  text[1:]

   _snake_case							:		Optional[Any]								=  self.sp_model.encode(a_,				out_type=a_		)

   if not self.legacy and not is_first and not text.startswith(""" """		) and tokens[0].startswith(a_		):
     _snake_case							:		int								=  ([tokens[0][1:]] if len(tokens[0]		) > 1 else []) + tokens[1:]
   return tokens


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




   '''simple docstring'''



   if token.startswith("""<extra_id_"""		):
     _snake_case							:		Dict								=  re.match(r"""<extra_id_(\d+)>""",				a_		)
     _snake_case							:		int								=  int(match.group(1		)		)
     return self.vocab_size - num - 1
   return self.sp_model.piece_to_id(a_		)


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




   '''simple docstring'''



   if index < self.sp_model.get_piece_size():
     _snake_case							:		Union[str, Any]								=  self.sp_model.IdToPiece(a_		)
   else:
     _snake_case							:		int								=  f"<extra_id_{self.vocab_size - 1 - index}>"
   return token


 def 			UpperCamelCase_       (    self:		Optional[int],				a_:		Optional[int]		):




   '''simple docstring'''



   _snake_case							:		Dict								=  []
   _snake_case							:		Optional[Any]								=  """"""
   _snake_case							:		str								=  False
   for token in tokens:
     # make sure that special tokens are not decoded using sentencepiece model
     if token in self.all_special_tokens:
       if not prev_is_special:
         out_string += " "
       out_string += self.sp_model.decode(a_		) + token
       _snake_case							:		List[Any]								=  True
       _snake_case							:		str								=  []
     else:
       current_sub_tokens.append(a_		)
       _snake_case							:		Union[str, Any]								=  False
   out_string += self.sp_model.decode(a_		)
   return out_string.strip()


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




   '''simple docstring'''



   if not os.path.isdir(a_		):
     logger.error(f"Vocabulary path ({save_directory}) should be a directory"		)
     return
   _snake_case							:		str								=  os.path.join(
       a_,				(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)

   if os.path.abspath(self.vocab_file		) != os.path.abspath(a_		) and os.path.isfile(self.vocab_file		):
     copyfile(self.vocab_file,				a_		)
   elif not os.path.isfile(self.vocab_file		):
     with open(a_,				"""wb"""		) as fi:
       _snake_case							:		Dict								=  self.sp_model.serialized_model_proto()
       fi.write(a_		)

   return (out_vocab_file,)








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


