

import inspect
from typing import Callable, List, Optional, Set, Tuple, Union

import torch
from packaging import version
from safetensors.torch import storage_ptr, storage_size
from torch import nn

from .utils import logging


SCREAMING_SNAKE_CASE						:Dict      =			[nn.LayerNorm]

SCREAMING_SNAKE_CASE						:List[Any]      =			logging.get_logger(__name__)

SCREAMING_SNAKE_CASE						:int      =			version.parse(version.parse(torch.__version__).base_version)

SCREAMING_SNAKE_CASE						:List[str]      =			parsed_torch_version_base >= version.parse("""2.0""")
SCREAMING_SNAKE_CASE						:Any      =			parsed_torch_version_base >= version.parse("""1.12""")
SCREAMING_SNAKE_CASE						:Tuple      =			parsed_torch_version_base >= version.parse("""1.11""")
SCREAMING_SNAKE_CASE						:Optional[int]      =			parsed_torch_version_base < version.parse("""1.11""")




def        lowerCAmelCase(	SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		self       )->		Dict:



    """simple docstring"""






    from torch import _softmax_backward_data

    if is_torch_less_than_1_11:
        return _softmax_backward_data(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		parent.dim       ,		self       )
    else:
        return _softmax_backward_data(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		parent.dim       ,		self.dtype       )




def        lowerCAmelCase(	SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_ = 0       )->		nn.Linear:



    """simple docstring"""






    UpperCamelCase_          =		index.to(layer.weight.device       )
    UpperCamelCase_          =		layer.weight.index_select(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ).clone().detach()
    if layer.bias is not None:
        if dim == 1:
            UpperCamelCase_          =		layer.bias.clone().detach()
        else:
            UpperCamelCase_          =		layer.bias[index].clone().detach()
    UpperCamelCase_          =		list(layer.weight.size()       )
    UpperCamelCase_          =		len(SCREAMING_SNAKE_CASE_       )
    UpperCamelCase_          =		nn.Linear(new_size[1]       ,		new_size[0]       ,		bias=layer.bias is not None       ).to(layer.weight.device       )
    UpperCamelCase_          =		False
    new_layer.weight.copy_(W.contiguous()       )
    UpperCamelCase_          =		True
    if layer.bias is not None:
        UpperCamelCase_          =		False
        new_layer.bias.copy_(b.contiguous()       )
        UpperCamelCase_          =		True
    return new_layer






class 		__magic_name__ (				nn.Module							):






     def __init__(							self						,		_lowercase						,		_lowercase     )->			List[str]:
         super().__init__()
         UpperCamelCase_          =		nf
         UpperCamelCase_          =		nn.Parameter(torch.empty(_lowercase						,		_lowercase     )     )
         UpperCamelCase_          =		nn.Parameter(torch.zeros(_lowercase     )     )
         nn.init.normal_(self.weight						,		std=0.02     )






     def      UpperCAmelCase_							(							self						,		_lowercase     )->			List[str]:
         UpperCamelCase_          =		x.size()[:-1] + (self.nf,)
         UpperCamelCase_          =		torch.addmm(self.bias						,		x.view(-1						,		x.size(-1     )     )						,		self.weight     )
         UpperCamelCase_          =		x.view(_lowercase     )
         return x




def        lowerCAmelCase(	SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_ = 1       )->		ConvaD:



    """simple docstring"""






    UpperCamelCase_          =		index.to(layer.weight.device       )
    UpperCamelCase_          =		layer.weight.index_select(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ).clone().detach()
    if dim == 0:
        UpperCamelCase_          =		layer.bias.clone().detach()
    else:
        UpperCamelCase_          =		layer.bias[index].clone().detach()
    UpperCamelCase_          =		list(layer.weight.size()       )
    UpperCamelCase_          =		len(SCREAMING_SNAKE_CASE_       )
    UpperCamelCase_          =		ConvaD(new_size[1]       ,		new_size[0]       ).to(layer.weight.device       )
    UpperCamelCase_          =		False
    new_layer.weight.copy_(W.contiguous()       )
    UpperCamelCase_          =		True
    UpperCamelCase_          =		False
    new_layer.bias.copy_(b.contiguous()       )
    UpperCamelCase_          =		True
    return new_layer




def        lowerCAmelCase(	SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_ = None       )->		Union[nn.Linear, ConvaD]:



    """simple docstring"""






    if isinstance(SCREAMING_SNAKE_CASE_       ,		nn.Linear       ):
        return prune_linear_layer(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		dim=0 if dim is None else dim       )
    elif isinstance(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ):
        return prune_convad_layer(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		dim=1 if dim is None else dim       )
    else:
        raise ValueError(f"Can't prune layer of class {layer.__class__}"       )




def        lowerCAmelCase(	SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		*SCREAMING_SNAKE_CASE_       )->		torch.Tensor:



    """simple docstring"""






    assert len(SCREAMING_SNAKE_CASE_       ) > 0, f"{input_tensors} has to be a tuple/list of tensors"

    # inspect.signature exist since python 3.5 and is a python method -> no problem with backward compatibility
    UpperCamelCase_          =		len(inspect.signature(SCREAMING_SNAKE_CASE_       ).parameters       )
    if num_args_in_forward_chunk_fn != len(SCREAMING_SNAKE_CASE_       ):
        raise ValueError(
            f"forward_chunk_fn expects {num_args_in_forward_chunk_fn} arguments, but only {len(SCREAMING_SNAKE_CASE_       )} input "
            "tensors are given"       )

    if chunk_size > 0:
        UpperCamelCase_          =		input_tensors[0].shape[chunk_dim]
        for input_tensor in input_tensors:
            if input_tensor.shape[chunk_dim] != tensor_shape:
                raise ValueError(
                    f"All input tenors have to be of the same shape: {tensor_shape}, "
                    f"found shape {input_tensor.shape[chunk_dim]}"       )

        if input_tensors[0].shape[chunk_dim] % chunk_size != 0:
            raise ValueError(
                f"The dimension to be chunked {input_tensors[0].shape[chunk_dim]} has to be a multiple of the chunk "
                f"size {chunk_size}"       )

        UpperCamelCase_          =		input_tensors[0].shape[chunk_dim] // chunk_size

        # chunk input tensor into tuples
        UpperCamelCase_          =		tuple(input_tensor.chunk(SCREAMING_SNAKE_CASE_       ,		dim=SCREAMING_SNAKE_CASE_       ) for input_tensor in input_tensors       )
        # apply forward fn to every tuple
        UpperCamelCase_          =		tuple(forward_fn(*SCREAMING_SNAKE_CASE_       ) for input_tensors_chunk in zip(*SCREAMING_SNAKE_CASE_       )       )
        # concatenate output at same dimension
        return torch.cat(SCREAMING_SNAKE_CASE_       ,		dim=SCREAMING_SNAKE_CASE_       )

    return forward_fn(*SCREAMING_SNAKE_CASE_       )




def        lowerCAmelCase(	SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       )->		Tuple[Set[int], torch.LongTensor]:



    """simple docstring"""






    UpperCamelCase_          =		torch.ones(SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_       )
    UpperCamelCase_          =		set(SCREAMING_SNAKE_CASE_       ) - already_pruned_heads  # Convert to set and remove already pruned heads
    for head in heads:
        # Compute how many pruned heads are before the head and move the index accordingly
        UpperCamelCase_          =		head - sum(1 if h < head else 0 for h in already_pruned_heads       )
        UpperCamelCase_          =		0
    UpperCamelCase_          =		mask.view(-1       ).contiguous().eq(1       )
    UpperCamelCase_          =		torch.arange(len(SCREAMING_SNAKE_CASE_       )       )[mask].long()
    return heads, index




def        lowerCAmelCase(	*SCREAMING_SNAKE_CASE_       ,		SCREAMING_SNAKE_CASE_ = None       )->		Tuple[torch.Tensor, ...]:



    """simple docstring"""






    return torch.meshgrid(*SCREAMING_SNAKE_CASE_       ,		indexing=SCREAMING_SNAKE_CASE_       )




def        lowerCAmelCase(	SCREAMING_SNAKE_CASE_       )->		Tuple[torch.device, int, int]:



    """simple docstring"""






    return tensor.device, storage_ptr(SCREAMING_SNAKE_CASE_       ), storage_size(SCREAMING_SNAKE_CASE_       )



import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


_UpperCamelCase			       =			namedtuple('''ExampleNamedTuple''', '''a b c''')



class     lowercase						(       unittest.TestCase					):



 '''simple docstring'''


 def   UpperCamelCase__							(self ) ->					Tuple:






  """simple docstring"""





  UpperCAmelCase__							=				torch.randn(5       ,       2 )
  UpperCAmelCase__							=				torch.device('cuda' ) if torch.cuda.is_available() else torch.device('cpu' )

  UpperCAmelCase__							=				send_to_device(__a       ,       __a )
  self.assertTrue(torch.equal(resulta.cpu()       ,       __a ) )

  UpperCAmelCase__							=				send_to_device((tensor, [tensor, tensor], 1)       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta[0].cpu()       ,       __a ) )
  self.assertIsInstance(resulta[1]       ,       __a )
  self.assertTrue(torch.equal(resulta[1][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta[1][1].cpu()       ,       __a ) )
  self.assertEqual(resulta[2]       ,       1 )

  UpperCAmelCase__							=				send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta['a'].cpu()       ,       __a ) )
  self.assertIsInstance(resulta['b']       ,       __a )
  self.assertTrue(torch.equal(resulta['b'][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta['b'][1].cpu()       ,       __a ) )
  self.assertEqual(resulta['c']       ,       1 )

  UpperCAmelCase__							=				send_to_device(ExampleNamedTuple(a=__a       ,       b=[tensor, tensor]       ,       c=1 )       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta.a.cpu()       ,       __a ) )
  self.assertIsInstance(resulta.b       ,       __a )
  self.assertTrue(torch.equal(resulta.b[0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta.b[1].cpu()       ,       __a ) )
  self.assertEqual(resulta.c       ,       1 )

  UpperCAmelCase__							=				send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1} )       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta['a'].cpu()       ,       __a ) )
  self.assertIsInstance(resulta['b']       ,       __a )
  self.assertTrue(torch.equal(resulta['b'][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta['b'][1].cpu()       ,       __a ) )
  self.assertEqual(resulta['c']       ,       1 )




 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  with self.assertRaises(__a ) as cm:
   UpperCAmelCase__							=				recursively_apply(torch.tensor       ,       (torch.tensor(1 ), 1)       ,       error_on_other_type=__a )
  self.assertEqual(
      str(cm.exception )       ,       'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'       ,       )




 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  UpperCAmelCase__							=				torch.tensor([1, 2, 3, 4, 5] )
  self.assertEqual(listify(__a )       ,       [1, 2, 3, 4, 5] )

  UpperCAmelCase__							=				torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]] )
  self.assertEqual(listify(__a )       ,       [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]] )

  UpperCAmelCase__							=				torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]] )
  self.assertEqual(
      listify(__a )       ,       [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]] )




 def   UpperCamelCase__							(self ) ->					int:






  """simple docstring"""





  with patch_environment(aa=1       ,       BB=2 ):
   self.assertEqual(os.environ.get('AA' )       ,       '1' )
   self.assertEqual(os.environ.get('BB' )       ,       '2' )

  self.assertNotIn('AA'       ,       os.environ )
  self.assertNotIn('BB'       ,       os.environ )




 def   UpperCamelCase__							(self ) ->					Dict:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				extract_model_from_parallel(__a       ,       keep_fpaa_wrapper=__a )
  UpperCAmelCase__							=				pickle.dumps(__a )




 @require_cuda
 def   UpperCamelCase__							(self ) ->					Tuple:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				torch.cuda.amp.autocast(dtype=torch.floataa )(model.forward )
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				extract_model_from_parallel(__a       ,       keep_fpaa_wrapper=__a )
  UpperCAmelCase__							=				pickle.dumps(__a )




 @require_cuda
 @require_torch_min_version(version='2.0' )
 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				torch.cuda.amp.autocast(dtype=torch.floataa )(model.forward )
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				torch.compile(model.forward       ,       backend='inductor' )
  UpperCAmelCase__							=				torch.randn(4       ,       10 ).cuda()
  UpperCAmelCase__							=				model(__a )




 def   UpperCamelCase__							(self ) ->					Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
  UpperCAmelCase__							=				torch.nn.parallel.DataParallel(__a )
  UpperCAmelCase__							=				extract_model_from_parallel(__a )

  self.assertEqual(__a       ,       __a )




 @require_torch_min_version(version='2.0' )
 def   UpperCamelCase__							(self ) ->					Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				torch.compile(__a )

  # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
  UpperCAmelCase__							=				torch.nn.parallel.DataParallel(__a )
  UpperCAmelCase__							=				torch.compile(__a )
  UpperCAmelCase__							=				extract_model_from_parallel(__a )

  self.assertEqual(compiled_model._orig_mod       ,       compiled_model_unwrapped._orig_mod )




 def   UpperCamelCase__							(self ) ->					str:






  """simple docstring"""





  self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3] )] )       ,       torch.device('cpu' ) )
  self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3] )} )       ,       torch.device('cpu' ) )
  self.assertIsNone(find_device([1, 'a'] ) )



