




"""simple docstring"""


import argparse
from collections import OrderedDict
from pathlib import Path

import torch
from huggingface_hub import hf_hub_download
from PIL import Image
from torchvision.transforms import functional as F

from transformers import DetrImageProcessor, TableTransformerConfig, TableTransformerForObjectDetection
from transformers.utils import logging


logging.set_verbosity_info()
A_      =    logging.get_logger(__name__)

# here we list all keys to be renamed (original name on the left, our name on the right)
A_      =    []
for i in range(6):
     # encoder layers: output projection, 2 feedforward neural networks and 2 layernorms
     rename_keys.append(
         (F'''transformer.encoder.layers.{i}.self_attn.out_proj.weight''', F'''encoder.layers.{i}.self_attn.out_proj.weight''')
     )
     rename_keys.append(
         (F'''transformer.encoder.layers.{i}.self_attn.out_proj.bias''', F'''encoder.layers.{i}.self_attn.out_proj.bias''')
     )
     rename_keys.append((F'''transformer.encoder.layers.{i}.linear1.weight''', F'''encoder.layers.{i}.fc1.weight'''))
     rename_keys.append((F'''transformer.encoder.layers.{i}.linear1.bias''', F'''encoder.layers.{i}.fc1.bias'''))
     rename_keys.append((F'''transformer.encoder.layers.{i}.linear2.weight''', F'''encoder.layers.{i}.fc2.weight'''))
     rename_keys.append((F'''transformer.encoder.layers.{i}.linear2.bias''', F'''encoder.layers.{i}.fc2.bias'''))
     rename_keys.append(
         (F'''transformer.encoder.layers.{i}.norm1.weight''', F'''encoder.layers.{i}.self_attn_layer_norm.weight''')
     )
     rename_keys.append((F'''transformer.encoder.layers.{i}.norm1.bias''', F'''encoder.layers.{i}.self_attn_layer_norm.bias'''))
     rename_keys.append((F'''transformer.encoder.layers.{i}.norm2.weight''', F'''encoder.layers.{i}.final_layer_norm.weight'''))
     rename_keys.append((F'''transformer.encoder.layers.{i}.norm2.bias''', F'''encoder.layers.{i}.final_layer_norm.bias'''))
     # decoder layers: 2 times output projection, 2 feedforward neural networks and 3 layernorms
     rename_keys.append(
         (F'''transformer.decoder.layers.{i}.self_attn.out_proj.weight''', F'''decoder.layers.{i}.self_attn.out_proj.weight''')
     )
     rename_keys.append(
         (F'''transformer.decoder.layers.{i}.self_attn.out_proj.bias''', F'''decoder.layers.{i}.self_attn.out_proj.bias''')
     )
     rename_keys.append(
         (
             F'''transformer.decoder.layers.{i}.multihead_attn.out_proj.weight''',
             F'''decoder.layers.{i}.encoder_attn.out_proj.weight''',
         )
     )
     rename_keys.append(
         (
             F'''transformer.decoder.layers.{i}.multihead_attn.out_proj.bias''',
             F'''decoder.layers.{i}.encoder_attn.out_proj.bias''',
         )
     )
     rename_keys.append((F'''transformer.decoder.layers.{i}.linear1.weight''', F'''decoder.layers.{i}.fc1.weight'''))
     rename_keys.append((F'''transformer.decoder.layers.{i}.linear1.bias''', F'''decoder.layers.{i}.fc1.bias'''))
     rename_keys.append((F'''transformer.decoder.layers.{i}.linear2.weight''', F'''decoder.layers.{i}.fc2.weight'''))
     rename_keys.append((F'''transformer.decoder.layers.{i}.linear2.bias''', F'''decoder.layers.{i}.fc2.bias'''))
     rename_keys.append(
         (F'''transformer.decoder.layers.{i}.norm1.weight''', F'''decoder.layers.{i}.self_attn_layer_norm.weight''')
     )
     rename_keys.append((F'''transformer.decoder.layers.{i}.norm1.bias''', F'''decoder.layers.{i}.self_attn_layer_norm.bias'''))
     rename_keys.append(
         (F'''transformer.decoder.layers.{i}.norm2.weight''', F'''decoder.layers.{i}.encoder_attn_layer_norm.weight''')
     )
     rename_keys.append(
         (F'''transformer.decoder.layers.{i}.norm2.bias''', F'''decoder.layers.{i}.encoder_attn_layer_norm.bias''')
     )
     rename_keys.append((F'''transformer.decoder.layers.{i}.norm3.weight''', F'''decoder.layers.{i}.final_layer_norm.weight'''))
     rename_keys.append((F'''transformer.decoder.layers.{i}.norm3.bias''', F'''decoder.layers.{i}.final_layer_norm.bias'''))

# convolutional projection + query embeddings + layernorm of encoder + layernorm of decoder + class and bounding box heads
rename_keys.extend(
    [
        ('''input_proj.weight''', '''input_projection.weight'''),
        ('''input_proj.bias''', '''input_projection.bias'''),
        ('''query_embed.weight''', '''query_position_embeddings.weight'''),
        ('''transformer.encoder.norm.weight''', '''encoder.layernorm.weight'''),
        ('''transformer.encoder.norm.bias''', '''encoder.layernorm.bias'''),
        ('''transformer.decoder.norm.weight''', '''decoder.layernorm.weight'''),
        ('''transformer.decoder.norm.bias''', '''decoder.layernorm.bias'''),
        ('''class_embed.weight''', '''class_labels_classifier.weight'''),
        ('''class_embed.bias''', '''class_labels_classifier.bias'''),
        ('''bbox_embed.layers.0.weight''', '''bbox_predictor.layers.0.weight'''),
        ('''bbox_embed.layers.0.bias''', '''bbox_predictor.layers.0.bias'''),
        ('''bbox_embed.layers.1.weight''', '''bbox_predictor.layers.1.weight'''),
        ('''bbox_embed.layers.1.bias''', '''bbox_predictor.layers.1.bias'''),
        ('''bbox_embed.layers.2.weight''', '''bbox_predictor.layers.2.weight'''),
        ('''bbox_embed.layers.2.bias''', '''bbox_predictor.layers.2.bias'''),
    ]
)


def  UpperCAmelCase__	(snake_case__							: Optional[int]		,       snake_case__							: Optional[int]		,       snake_case__							: Any   ):
  """simple docstring"""

  _snake_case							:		List[Any]								=  state_dict.pop(lowercase__   )
  _snake_case							:		List[Any]								=  val


def  UpperCAmelCase__	(snake_case__							: Dict   ):
  """simple docstring"""

  _snake_case							:		Any								=  OrderedDict()
  for key, value in state_dict.items():
    if "backbone.0.body" in key:
      _snake_case							:		Optional[Any]								=  key.replace("""backbone.0.body"""		,       """backbone.conv_encoder.model"""   )
      _snake_case							:		int								=  value
    else:
      _snake_case							:		Tuple								=  value

  return new_state_dict


def  UpperCAmelCase__	(snake_case__							: List[str]   ):
  """simple docstring"""

  _snake_case							:		List[str]								=  """"""

  # first: transformer encoder
  for i in range(6   ):
    # read in weights + bias of input projection layer (in PyTorch's MultiHeadAttention, this is a single matrix + bias)
    _snake_case							:		Dict								=  state_dict.pop(F"{prefix}transformer.encoder.layers.{i}.self_attn.in_proj_weight"   )
    _snake_case							:		List[str]								=  state_dict.pop(F"{prefix}transformer.encoder.layers.{i}.self_attn.in_proj_bias"   )
    # next, add query, keys and values (in that order) to the state dict
    _snake_case							:		str								=  in_proj_weight[:2_56, :]
    _snake_case							:		Optional[int]								=  in_proj_bias[:2_56]
    _snake_case							:		Dict								=  in_proj_weight[2_56:5_12, :]
    _snake_case							:		int								=  in_proj_bias[2_56:5_12]
    _snake_case							:		Optional[Any]								=  in_proj_weight[-2_56:, :]
    _snake_case							:		Any								=  in_proj_bias[-2_56:]
  # next: transformer decoder (which is a bit more complex because it also includes cross-attention)
  for i in range(6   ):
    # read in weights + bias of input projection layer of self-attention
    _snake_case							:		Union[str, Any]								=  state_dict.pop(F"{prefix}transformer.decoder.layers.{i}.self_attn.in_proj_weight"   )
    _snake_case							:		List[Any]								=  state_dict.pop(F"{prefix}transformer.decoder.layers.{i}.self_attn.in_proj_bias"   )
    # next, add query, keys and values (in that order) to the state dict
    _snake_case							:		int								=  in_proj_weight[:2_56, :]
    _snake_case							:		Dict								=  in_proj_bias[:2_56]
    _snake_case							:		int								=  in_proj_weight[2_56:5_12, :]
    _snake_case							:		Optional[Any]								=  in_proj_bias[2_56:5_12]
    _snake_case							:		Dict								=  in_proj_weight[-2_56:, :]
    _snake_case							:		Any								=  in_proj_bias[-2_56:]
    # read in weights + bias of input projection layer of cross-attention
    _snake_case							:		List[Any]								=  state_dict.pop(
        F"{prefix}transformer.decoder.layers.{i}.multihead_attn.in_proj_weight"   )
    _snake_case							:		List[Any]								=  state_dict.pop(F"{prefix}transformer.decoder.layers.{i}.multihead_attn.in_proj_bias"   )
    # next, add query, keys and values (in that order) of cross-attention to the state dict
    _snake_case							:		Dict								=  in_proj_weight_cross_attn[:2_56, :]
    _snake_case							:		List[str]								=  in_proj_bias_cross_attn[:2_56]
    _snake_case							:		Tuple								=  in_proj_weight_cross_attn[2_56:5_12, :]
    _snake_case							:		Tuple								=  in_proj_bias_cross_attn[2_56:5_12]
    _snake_case							:		Optional[int]								=  in_proj_weight_cross_attn[-2_56:, :]
    _snake_case							:		List[str]								=  in_proj_bias_cross_attn[-2_56:]


def  UpperCAmelCase__	(snake_case__							: Tuple		,       snake_case__							: List[Any]   ):
  """simple docstring"""

  _snake_case							:		Tuple								=  image.size
  _snake_case							:		List[Any]								=  max(lowercase__		,       lowercase__   )
  _snake_case							:		Optional[Any]								=  8_00 if """detection""" in checkpoint_url else 10_00
  _snake_case							:		Dict								=  target_max_size / current_max_size
  _snake_case							:		Tuple								=  image.resize((int(round(scale * width   )   ), int(round(scale * height   )   ))   )

  return resized_image


def  UpperCAmelCase__	(snake_case__							: Optional[Any]   ):
  """simple docstring"""

  _snake_case							:		Optional[int]								=  F.to_tensor(lowercase__   )
  _snake_case							:		Optional[int]								=  F.normalize(lowercase__		,       mean=[0.4_85, 0.4_56, 0.4_06]		,       std=[0.2_29, 0.2_24, 0.2_25]   )
  return image


@torch.no_grad()
def  UpperCAmelCase__	(snake_case__							: List[Any]		,       snake_case__							: Optional[int]		,       snake_case__							: List[Any]   ):
  """simple docstring"""

  logger.info("""Converting model..."""   )

  # load original state dict
  _snake_case							:		Optional[int]								=  torch.hub.load_state_dict_from_url(lowercase__		,       map_location="""cpu"""   )
  # rename keys
  for src, dest in rename_keys:
    rename_key(lowercase__		,       lowercase__		,       lowercase__   )
  _snake_case							:		List[str]								=  rename_backbone_keys(lowercase__   )
  # query, key and value matrices need special treatment
  read_in_q_k_v(lowercase__   )
  # important: we need to prepend a prefix to each of the base model keys as the head models use different attributes for them
  _snake_case							:		Any								=  """model."""
  for key in state_dict.copy().keys():
    if not key.startswith("""class_labels_classifier"""   ) and not key.startswith("""bbox_predictor"""   ):
      _snake_case							:		Tuple								=  state_dict.pop(lowercase__   )
      _snake_case							:		Tuple								=  val
    # create HuggingFace model and load state dict
  _snake_case							:		List[str]								=  TableTransformerConfig(
      backbone="""resnet18"""		,       mask_loss_coefficient=1		,       dice_loss_coefficient=1		,       ce_loss_coefficient=1		,       bbox_loss_coefficient=5		,       giou_loss_coefficient=2		,       eos_coefficient=0.4		,       class_cost=1		,       bbox_cost=5		,       giou_cost=2		,       )

  if "detection" in checkpoint_url:
    _snake_case							:		List[Any]								=  15
    _snake_case							:		int								=  2
    _snake_case							:		str								=  {0: """table""", 1: """table rotated"""}
    _snake_case							:		Tuple								=  idalabel
    _snake_case							:		Optional[Any]								=  {v: k for k, v in idalabel.items()}
  else:
    _snake_case							:		Union[str, Any]								=  1_25
    _snake_case							:		int								=  6
    _snake_case							:		List[Any]								=  {
        0: """table""",
        1: """table column""",
        2: """table row""",
        3: """table column header""",
        4: """table projected row header""",
        5: """table spanning cell""",
    }
    _snake_case							:		List[str]								=  idalabel
    _snake_case							:		Union[str, Any]								=  {v: k for k, v in idalabel.items()}

  _snake_case							:		Tuple								=  DetrImageProcessor(
      format="""coco_detection"""		,       max_size=8_00 if """detection""" in checkpoint_url else 10_00   )
  _snake_case							:		List[Any]								=  TableTransformerForObjectDetection(lowercase__   )
  model.load_state_dict(lowercase__   )
  model.eval()

  # verify our conversion
  _snake_case							:		Optional[Any]								=  """example_pdf.png""" if """detection""" in checkpoint_url else """example_table.png"""
  _snake_case							:		int								=  hf_hub_download(repo_id="""nielsr/example-pdf"""		,       repo_type="""dataset"""		,       filename=lowercase__   )
  _snake_case							:		Any								=  Image.open(lowercase__   ).convert("""RGB"""   )
  _snake_case							:		Optional[int]								=  normalize(resize(lowercase__		,       lowercase__   )   ).unsqueeze(0   )

  _snake_case							:		Any								=  model(lowercase__   )

  if "detection" in checkpoint_url:
    _snake_case							:		Optional[int]								=  (1, 15, 3)
    _snake_case							:		Tuple								=  torch.tensor(
        [[-6.78_97, -16.99_85, 6.79_37], [-8.01_86, -22.21_92, 6.96_77], [-7.31_17, -21.07_08, 7.40_55]]   )
    _snake_case							:		Any								=  torch.tensor([[0.48_67, 0.17_67, 0.67_32], [0.67_18, 0.44_79, 0.38_30], [0.47_16, 0.17_60, 0.63_64]]   )

  else:
    _snake_case							:		Any								=  (1, 1_25, 7)
    _snake_case							:		Dict								=  torch.tensor(
        [[-18.14_30, -8.32_14, 4.82_74], [-18.46_85, -7.13_61, -4.26_67], [-26.36_93, -9.34_29, -4.99_62]]   )
    _snake_case							:		str								=  torch.tensor([[0.49_83, 0.55_95, 0.94_40], [0.49_16, 0.63_15, 0.59_54], [0.61_08, 0.86_37, 0.11_35]]   )

  assert outputs.logits.shape == expected_shape
  assert torch.allclose(outputs.logits[0, :3, :3]		,       lowercase__		,       atol=1e-4   )
  assert torch.allclose(outputs.pred_boxes[0, :3, :3]		,       lowercase__		,       atol=1e-4   )
  print("""Looks ok!"""   )

  if pytorch_dump_folder_path is not None:
    # Save model and image processor
    logger.info(F"Saving PyTorch model and image processor to {pytorch_dump_folder_path}..."   )
    Path(lowercase__   ).mkdir(exist_ok=lowercase__   )
    model.save_pretrained(lowercase__   )
    image_processor.save_pretrained(lowercase__   )

  if push_to_hub:
    # Push model to HF hub
    logger.info("""Pushing model to the hub..."""   )
    _snake_case							:		int								=  (
        """microsoft/table-transformer-detection"""
        if """detection""" in checkpoint_url
        else """microsoft/table-transformer-structure-recognition"""
    )
    model.push_to_hub(lowercase__   )
    image_processor.push_to_hub(lowercase__   )


if __name__ == "__main__":
     A_      =    argparse.ArgumentParser()

     parser.add_argument(
         '''--checkpoint_url''',
         default='''https://pubtables1m.blob.core.windows.net/model/pubtables1m_detection_detr_r18.pth''',
         type=str,
         choices=[
             '''https://pubtables1m.blob.core.windows.net/model/pubtables1m_detection_detr_r18.pth''',
             '''https://pubtables1m.blob.core.windows.net/model/pubtables1m_structure_detr_r18.pth''',
         ],
         help='''URL of the Table Transformer checkpoint you\'d like to convert.''',
     )
     parser.add_argument(
         '''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the folder to output PyTorch model.'''
     )
     parser.add_argument(
         '''--push_to_hub''', action='''store_true''', help='''Whether or not to push the converted model to the 🤗 hub.'''
     )
     A_      =    parser.parse_args()
     convert_table_transformer_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path, args.push_to_hub)








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


