

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


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





class 			__lowerCAmelCase					( unittest.TestCase							):




  def   A__					(    self )  ->			Optional[int]:
        '''simple docstring'''






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

        _lowercase     	=send_to_device(lowerCAmelCase							,       lowerCAmelCase )
        self.assertTrue(torch.equal(resulta.cpu()							,       lowerCAmelCase ) )

        _lowercase     	=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 )

        _lowercase     	=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 )

        _lowercase     	=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 )

        _lowercase     	=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]:
        '''simple docstring'''






        with self.assertRaises(lowerCAmelCase ) as cm:
              _lowercase     	=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 )  ->			Any:
        '''simple docstring'''






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

        _lowercase     	=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]] )

        _lowercase     	=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]:
        '''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   A__					(    self )  ->			Union[str, Any]:
        '''simple docstring'''






        _lowercase     	=RegressionModel()
        _lowercase     	=model.forward
        _lowercase     	=convert_outputs_to_fpaa(model.forward )
        _lowercase     	=extract_model_from_parallel(lowerCAmelCase							,       keep_fpaa_wrapper=lowerCAmelCase )
        _lowercase     	=pickle.dumps(lowerCAmelCase )




  @require_cuda
  def   A__					(    self )  ->			List[str]:
        '''simple docstring'''






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




  @require_cuda
  @require_torch_min_version(version='2.0' )
  def   A__					(    self )  ->			Any:
        '''simple docstring'''






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




  def   A__					(    self )  ->			int:
        '''simple docstring'''






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

        self.assertEqual(lowerCAmelCase							,       lowerCAmelCase )




  @require_torch_min_version(version='2.0' )
  def   A__					(    self )  ->			List[str]:
        '''simple docstring'''






        _lowercase     	=RegressionModel()
        _lowercase     	=torch.compile(lowerCAmelCase )

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

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




  def   A__					(    self )  ->			List[Any]:
        '''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'] ) )





import unittest

from cer import CER


lowercase_     =       CER()





class 			__lowerCAmelCase					( unittest.TestCase							):




  def   A__					(    self )  ->			Any:
        '''simple docstring'''






        _lowercase     	=['White House']
        _lowercase     	=['white house']
        # S = 2, D = 0, I = 0, N = 11, CER = 2 / 11
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.1818181818 ) < 1e-6 )




  def   A__					(    self )  ->			Dict:
        '''simple docstring'''






        _lowercase     	=['were wolf']
        _lowercase     	=['werewolf']
        # S = 0, D = 0, I = 1, N = 9, CER = 1 / 9
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.1111111 ) < 1e-6 )

        _lowercase     	=['werewolf']
        _lowercase     	=['weae     wolf']
        # S = 1, D = 1, I = 0, N = 8, CER = 0.25
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.25 ) < 1e-6 )

        # consecutive whitespaces case 1
        _lowercase     	=['were wolf']
        _lowercase     	=['were               wolf']
        # S = 0, D = 0, I = 0, N = 9, CER = 0
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.0 ) < 1e-6 )

        # consecutive whitespaces case 2
        _lowercase     	=['were   wolf']
        _lowercase     	=['were               wolf']
        # S = 0, D = 0, I = 0, N = 9, CER = 0
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.0 ) < 1e-6 )




  def   A__					(    self )  ->			Optional[Any]:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=['weaewolf']
        # S = 1, D = 0, I = 0, N = 8, CER = 0.125
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.125 ) < 1e-6 )




  def   A__					(    self )  ->			Union[str, Any]:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=['wereawolf']
        # S = 0, D = 1, I = 0, N = 8, CER = 0.125
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.125 ) < 1e-6 )




  def   A__					(    self )  ->			Any:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=['wereolf']
        # S = 0, D = 0, I = 1, N = 8, CER = 0.125
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.125 ) < 1e-6 )




  def   A__					(    self )  ->			Any:
        '''simple docstring'''






        _lowercase     	=['werewolf']
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertEqual(lowerCAmelCase							,       0.0 )




  def   A__					(    self )  ->			List[Any]:
        '''simple docstring'''






        _lowercase     	=['werewolf', 'I am your father']
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertEqual(lowerCAmelCase							,       0.0 )

        _lowercase     	=['werewolf', 'I am your father', 'doge']
        _lowercase     	=['werxwolf', 'I       am your father', 'doge']
        # S = 1, D = 0, I = 0, N = 28, CER = 1 / 28
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.03571428 ) < 1e-6 )




  def   A__					(    self )  ->			str:
        '''simple docstring'''






        _lowercase     	=['My hovercraft', 'is full of eels']
        _lowercase     	=['My hovercraft is full', ' of eels']
        # S = 0, D = 0, I = 2, N = 28, CER = 2 / 28
        # whitespace at the front of " of eels" will be strip during preporcessing
        # so need to insert 2 whitespaces
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase							,       concatenate_texts=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.071428 ) < 1e-6 )




  def   A__					(    self )  ->			Union[str, Any]:
        '''simple docstring'''






        _lowercase     	=['我能吞下玻璃而不伤身体']
        _lowercase     	=[' 能吞虾玻璃而 不霜身体啦']
        # S = 3, D = 2, I = 0, N = 11, CER = 5 / 11
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.4545454545 ) < 1e-6 )

        _lowercase     	=['我能吞下玻璃', '而不伤身体']
        _lowercase     	=['我    能 吞 下 玻 璃', '而不伤身体']
        # S = 0, D = 5, I = 0, N = 11, CER = 5 / 11
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertTrue(abs(char_error_rate - 0.454545454545 ) < 1e-6 )

        _lowercase     	=['我能吞下玻璃而不伤身体']
        _lowercase     	=cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )
        self.assertFalse(lowerCAmelCase							,       0.0 )




  def   A__					(    self )  ->			List[str]:
        '''simple docstring'''






        _lowercase     	=['']
        _lowercase     	=['Hypothesis']
        with self.assertRaises(lowerCAmelCase ):
              cer.compute(predictions=lowerCAmelCase							,       references=lowerCAmelCase )


if __name__ == "__main__":
  unittest.main()


