


'''simple docstring'''




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


_a      :      Dict       				=      namedtuple("ExampleNamedTuple", "a b c")




class  _lowercase   (  unittest.TestCase       ):
   def a     (	self					:			List[Any]			)   ->       Union[str, Any]:
        __snake_case    			=	torch.randn(5					,      2			)
        __snake_case    			=	torch.device('cuda'			) if torch.cuda.is_available() else torch.device('cpu'			)

        __snake_case    			=	send_to_device(_lowerCAmelCase					,      _lowerCAmelCase			)
        self.assertTrue(torch.equal(resulta.cpu()					,      _lowerCAmelCase			)			)

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

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

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

        __snake_case    			=	send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}			)					,      _lowerCAmelCase			)
        self.assertIsInstance(_lowerCAmelCase					,      _lowerCAmelCase			)
        self.assertTrue(torch.equal(resulta['a'].cpu()					,      _lowerCAmelCase			)			)
        self.assertIsInstance(resulta['b']					,      _lowerCAmelCase			)
        self.assertTrue(torch.equal(resulta['b'][0].cpu()					,      _lowerCAmelCase			)			)
        self.assertTrue(torch.equal(resulta['b'][1].cpu()					,      _lowerCAmelCase			)			)
        self.assertEqual(resulta['c']					,      1			)
   def a     (	self					:			List[Any]			)   ->       int:
        with self.assertRaises(_lowerCAmelCase			) as cm:
             __snake_case    			=	recursively_apply(torch.tensor					,      (torch.tensor(1			), 1)					,      error_on_other_type=_lowerCAmelCase			)
        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 a     (	self					:			Union[str, Any]			)   ->       int:
        __snake_case    			=	torch.tensor([1, 2, 3, 4, 5]			)
        self.assertEqual(listify(_lowerCAmelCase			)					,      [1, 2, 3, 4, 5]			)

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

        __snake_case    			=	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(_lowerCAmelCase			)					,      [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]			)
   def a     (	self					:			List[str]			)   ->       Tuple:
        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 a     (	self					:			List[str]			)   ->       List[Any]:
        __snake_case    			=	RegressionModel()
        __snake_case    			=	model.forward
        __snake_case    			=	convert_outputs_to_fpaa(model.forward			)
        __snake_case    			=	extract_model_from_parallel(_lowerCAmelCase					,      keep_fpaa_wrapper=_lowerCAmelCase			)
        __snake_case    			=	pickle.dumps(_lowerCAmelCase			)
   @require_cuda
   def a     (	self					:			List[Any]			)   ->       Any:
        __snake_case    			=	RegressionModel()
        __snake_case    			=	model.forward
        __snake_case    			=	torch.cuda.amp.autocast(dtype=torch.floataa			)(model.forward			)
        __snake_case    			=	convert_outputs_to_fpaa(model.forward			)
        __snake_case    			=	extract_model_from_parallel(_lowerCAmelCase					,      keep_fpaa_wrapper=_lowerCAmelCase			)
        __snake_case    			=	pickle.dumps(_lowerCAmelCase			)
   @require_cuda
   @require_torch_min_version(version='2.0'			)
   def a     (	self					:			Optional[Any]			)   ->       Tuple:
        __snake_case    			=	RegressionModel()
        __snake_case    			=	model.forward
        __snake_case    			=	torch.cuda.amp.autocast(dtype=torch.floataa			)(model.forward			)
        __snake_case    			=	convert_outputs_to_fpaa(model.forward			)
        __snake_case    			=	torch.compile(model.forward					,      backend='inductor'			)
        __snake_case    			=	torch.randn(4					,      10			).cuda()
        __snake_case    			=	model(_lowerCAmelCase			)
   def a     (	self					:			List[str]			)   ->       Union[str, Any]:
        __snake_case    			=	RegressionModel()
        # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
        __snake_case    			=	torch.nn.parallel.DataParallel(_lowerCAmelCase			)
        __snake_case    			=	extract_model_from_parallel(_lowerCAmelCase			)

        self.assertEqual(_lowerCAmelCase					,      _lowerCAmelCase			)
   @require_torch_min_version(version='2.0'			)
   def a     (	self					:			Tuple			)   ->       List[str]:
        __snake_case    			=	RegressionModel()
        __snake_case    			=	torch.compile(_lowerCAmelCase			)

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

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





   def a     (	self					:			int			)   ->       Optional[Any]:
        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']			)			)








'''simple docstring'''




from functools import partial
from typing import Any, Callable, Dict, List, Type, TypeVar, Union, overload

import torch
import torch.nn as nn
import torch.types





def      _a		(lowercase__ :      torch.Tensor       ,      lowercase__ :      torch.Tensor       ,      lowercase__ :      bool	) -> torch.Tensor:






     """simple docstring"""
     # The first operation in a checkpoint can't be in-place, but it's
     # nice to have in-place addition during inference. Thus...
     if not inplace:
          __snake_case    			=	ma + ma
     else:
          ma += ma

     return ma





def      _a		(lowercase__ :      torch.Tensor       ,      lowercase__ :      List[int]	) -> torch.Tensor:






     """simple docstring"""
     __snake_case    			=	-1 * len(lowercase__	)
     __snake_case    			=	list(range(len(tensor.shape[:zero_index]	)	)	)
     return tensor.permute(first_inds + [zero_index + i for i in inds]	)





def      _a		(lowercase__ :      torch.Tensor       ,      lowercase__ :      int	) -> torch.Tensor:






     """simple docstring"""
     return t.reshape(t.shape[:-no_dims] + (-1,)	)





def      _a		(lowercase__ :      torch.Tensor       ,      lowercase__ :      torch.Tensor       ,      lowercase__ :      int       ,      lowercase__ :      float = 1e-4	) -> torch.Tensor:






     """simple docstring"""
     __snake_case    			=	mask.expand(*value.shape	)
     return torch.sum(mask * value       ,      dim=lowercase__	) / (eps + torch.sum(lowercase__       ,      dim=lowercase__	))





def      _a		(lowercase__ :      torch.Tensor       ,      lowercase__ :      torch.types.Number = 2.31_25       ,      lowercase__ :      torch.types.Number = 21.68_75       ,      lowercase__ :      int = 6_4	) -> torch.Tensor:






     """simple docstring"""
     __snake_case    			=	torch.linspace(lowercase__       ,      lowercase__       ,      no_bins - 1       ,      device=pts.device	)
     __snake_case    			=	torch.sqrt(torch.sum((pts.unsqueeze(-2	) - pts.unsqueeze(-3	)) ** 2       ,      dim=-1	)	)
     return torch.bucketize(lowercase__       ,      lowercase__	)





def      _a		(lowercase__ :      Callable[[list], Any]       ,      lowercase__ :      List[dict]	) -> dict:






     """simple docstring"""
     __snake_case    			=	dicts[0]
     __snake_case    			=	{}
     for k, v in first.items():
          __snake_case    			=	[d[k] for d in dicts]
          if isinstance(lowercase__       ,      lowercase__	):
               __snake_case    			=	dict_multimap(lowercase__       ,      lowercase__	)
          else:
               __snake_case    			=	fn(lowercase__	)

     return new_dict





def      _a		(lowercase__ :      torch.Tensor       ,      lowercase__ :      torch.Tensor	) -> torch.Tensor:






     """simple docstring"""
     __snake_case    			=	v_bins.view(((1,) * len(x.shape	)) + (len(lowercase__	),)	)
     __snake_case    			=	x[..., None] - reshaped_bins
     __snake_case    			=	torch.argmin(torch.abs(lowercase__	)       ,      dim=-1	)
     return nn.functional.one_hot(lowercase__       ,      num_classes=len(lowercase__	)	).float()





def      _a		(lowercase__ :      torch.Tensor       ,      lowercase__ :      torch.Tensor       ,      lowercase__ :      int = 0       ,      lowercase__ :      int = 0	) -> torch.Tensor:






     """simple docstring"""
     __snake_case    			=	[]
     for i, s in enumerate(data.shape[:no_batch_dims]	):
          __snake_case    			=	torch.arange(lowercase__	)
          __snake_case    			=	r.view(*(*((1,) * i), -1, *((1,) * (len(inds.shape	) - i - 1)))	)
          ranges.append(lowercase__	)

     __snake_case    			=	[slice(lowercase__	) for _ in range(len(data.shape	) - no_batch_dims	)]
     __snake_case    			=	inds
     ranges.extend(lowercase__	)
     # Matt note: Editing this to get around the behaviour of using a list as an array index changing
     # in recent Numpy versions
     return data[tuple(lowercase__	)]


_a      :      str       				=      TypeVar("T")





def      _a		(lowercase__ :      Callable[[T], Any]       ,      lowercase__ :      Dict[Any, Union[dict, list, tuple, T]]       ,      lowercase__ :      Type[T]	) -> Dict[Any, Union[dict, list, tuple, Any]]:






     """simple docstring"""
     __snake_case    			=	{}
     for k, v in dic.items():
          if isinstance(lowercase__       ,      lowercase__	):
               __snake_case    			=	dict_map(lowercase__       ,      lowercase__       ,      lowercase__	)
          else:
               __snake_case    			=	tree_map(lowercase__       ,      lowercase__       ,      lowercase__	)

     return new_dict





@overload
def      _a		(lowercase__ :      Callable[[T], Any]       ,      lowercase__ :      T       ,      lowercase__ :      Type[T]	) -> Any:






     """simple docstring"""
     ...





@overload
def      _a		(lowercase__ :      Callable[[T], Any]       ,      lowercase__ :      dict       ,      lowercase__ :      Type[T]	) -> dict:






     """simple docstring"""
     ...





@overload
def      _a		(lowercase__ :      Callable[[T], Any]       ,      lowercase__ :      list       ,      lowercase__ :      Type[T]	) -> list:






     """simple docstring"""
     ...





@overload
def      _a		(lowercase__ :      Callable[[T], Any]       ,      lowercase__ :      tuple       ,      lowercase__ :      Type[T]	) -> tuple:






     """simple docstring"""
     ...





def      _a		(lowercase__ :      str       ,      lowercase__ :      Any       ,      lowercase__ :      Dict	) -> Union[str, Any]:






     """simple docstring"""
     if isinstance(lowercase__       ,      lowercase__	):
          return dict_map(lowercase__       ,      lowercase__       ,      lowercase__	)
     elif isinstance(lowercase__       ,      lowercase__	):
          return [tree_map(lowercase__       ,      lowercase__       ,      lowercase__	) for x in tree]
     elif isinstance(lowercase__       ,      lowercase__	):
          return tuple(tree_map(lowercase__       ,      lowercase__       ,      lowercase__	) for x in tree	)
     elif isinstance(lowercase__       ,      lowercase__	):
          return fn(lowercase__	)
     else:
          print(type(lowercase__	)	)
          raise ValueError('Not supported'	)


_a      :      Union[str, Any]       				=      partial(tree_map, leaf_type=torch.Tensor)




