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


_snake_case		  =   namedtuple("ExampleNamedTuple", "a b c")


class 			UpperCamelCase_				(  unittest.TestCase     ):




      '''simple docstring'''







      def lowercase__		( self):
            lowerCAmelCase_						=      torch.randn(5					,					2)
            lowerCAmelCase_						=      torch.device('''cuda''') if torch.cuda.is_available() else torch.device('''cpu''')

            lowerCAmelCase_						=      send_to_device(_UpperCAmelCase					,					_UpperCAmelCase)
            self.assertTrue(torch.equal(resulta.cpu()					,					_UpperCAmelCase))

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

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

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

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







      def lowercase__		( self):
            with self.assertRaises(_UpperCAmelCase) as cm:
                  lowerCAmelCase_						=      recursively_apply(torch.tensor					,					(torch.tensor(1), 1)					,					error_on_other_type=_UpperCAmelCase)
            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 lowercase__		( self):
            lowerCAmelCase_						=      torch.tensor([1, 2, 3, 4, 5])
            self.assertEqual(listify(_UpperCAmelCase)					,					[1, 2, 3, 4, 5])

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

            lowerCAmelCase_						=      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(_UpperCAmelCase)					,					[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]])







      def lowercase__		( self):
            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 lowercase__		( self):
            lowerCAmelCase_						=      RegressionModel()
            lowerCAmelCase_						=      model.forward
            lowerCAmelCase_						=      convert_outputs_to_fpaa(model.forward)
            lowerCAmelCase_						=      extract_model_from_parallel(_UpperCAmelCase					,					keep_fpaa_wrapper=_UpperCAmelCase)
            lowerCAmelCase_						=      pickle.dumps(_UpperCAmelCase)







      @require_cuda
      def lowercase__		( self):
            lowerCAmelCase_						=      RegressionModel()
            lowerCAmelCase_						=      model.forward
            lowerCAmelCase_						=      torch.cuda.amp.autocast(dtype=torch.floataa)(model.forward)
            lowerCAmelCase_						=      convert_outputs_to_fpaa(model.forward)
            lowerCAmelCase_						=      extract_model_from_parallel(_UpperCAmelCase					,					keep_fpaa_wrapper=_UpperCAmelCase)
            lowerCAmelCase_						=      pickle.dumps(_UpperCAmelCase)







      @require_cuda
      @require_torch_min_version(version='''2.0''')
      def lowercase__		( self):
            lowerCAmelCase_						=      RegressionModel()
            lowerCAmelCase_						=      model.forward
            lowerCAmelCase_						=      torch.cuda.amp.autocast(dtype=torch.floataa)(model.forward)
            lowerCAmelCase_						=      convert_outputs_to_fpaa(model.forward)
            lowerCAmelCase_						=      torch.compile(model.forward					,					backend='''inductor''')
            lowerCAmelCase_						=      torch.randn(4					,					10).cuda()
            lowerCAmelCase_						=      model(_UpperCAmelCase)







      def lowercase__		( self):
            lowerCAmelCase_						=      RegressionModel()
            # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
            lowerCAmelCase_						=      torch.nn.parallel.DataParallel(_UpperCAmelCase)
            lowerCAmelCase_						=      extract_model_from_parallel(_UpperCAmelCase)

            self.assertEqual(_UpperCAmelCase					,					_UpperCAmelCase)







      @require_torch_min_version(version='''2.0''')
      def lowercase__		( self):
            lowerCAmelCase_						=      RegressionModel()
            lowerCAmelCase_						=      torch.compile(_UpperCAmelCase)

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

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







      def lowercase__		( self):
            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''']))





from __future__ import annotations
def lowerCamelCase_       (			A			:      list[int]						):




      """simple docstring"""
      if not list_of_ints:
            return []

      if min(A						) < 0:
            raise ValueError('''All numbers must be positive'''						)

      lowerCAmelCase_						=      max(len(bin(A						)[2:]						) for x in list_of_ints						)
      return _msd_radix_sort(A		,					A						)
def lowerCamelCase_       (			A			:      list[int]		,					A			:      int						):




      """simple docstring"""
      if bit_position == 0 or len(A						) in [0, 1]:
            return list_of_ints

      lowerCAmelCase_						=      []
      lowerCAmelCase_						=      []
      # Split numbers based on bit at bit_position from the right
      for number in list_of_ints:
            if (number >> (bit_position - 1)) & 1:
                  # number has a one at bit bit_position
                  ones.append(A						)
            else:
                  # number has a zero at bit bit_position
                  zeros.append(A						)

    # recursively split both lists further
      lowerCAmelCase_						=      _msd_radix_sort(A		,					bit_position - 1						)
      lowerCAmelCase_						=      _msd_radix_sort(A		,					bit_position - 1						)

      # recombine lists
      lowerCAmelCase_						=      zeros
      res.extend(A						)

      return res
def lowerCamelCase_       (			A			:      list[int]						):




      """simple docstring"""
      lowerCAmelCase_						=      len(A						)
      if not list_of_ints or length == 1:
            return

      if min(A						) < 0:
            raise ValueError('''All numbers must be positive'''						)

      lowerCAmelCase_						=      max(len(bin(A						)[2:]						) for x in list_of_ints						)
      _msd_radix_sort_inplace(A		,					A		,					0		,					A						)
def lowerCamelCase_       (			A			:      list[int]		,					A			:      int		,					A			:      int		,					A			:      int						):




      """simple docstring"""
      if bit_position == 0 or end_index - begin_index <= 1:
            return

      bit_position -= 1

      lowerCAmelCase_						=      begin_index
      lowerCAmelCase_						=      end_index - 1
      while i <= j:
            lowerCAmelCase_						=      False
            if not (list_of_ints[i] >> bit_position) & 1:
                  # found zero at the beginning
                  i += 1
                  lowerCAmelCase_						=      True
            if (list_of_ints[j] >> bit_position) & 1:
                  # found one at the end
                  j -= 1
                  lowerCAmelCase_						=      True

            if changed:
                  continue

            lowerCAmelCase_			,		lowerCAmelCase_						=      list_of_ints[j], list_of_ints[i]
            j -= 1
            if j != i:
                  i += 1

      _msd_radix_sort_inplace(A		,					A		,					A		,					A						)
      _msd_radix_sort_inplace(A		,					A		,					A		,					A						)


if __name__ == "__main__":
 import doctest

 doctest.testmod()



