
"""simple docstring"""




import os
import sys

from . import rsa_key_generator as rkg

a		:					int    	=					1_28
a		:					List[str]    	=					2_56



def 						__magic_name__		(				UpperCamelCase    :		Union[str, Any]       ,		UpperCamelCase    :		Union[str, Any] = DEFAULT_BLOCK_SIZE    )    ->	list[int]:
    a__    = message.encode('ascii'    )

    a__    = []
    for block_start in range(0       ,		len(_UpperCAmelCase    )       ,		_UpperCAmelCase    ):
        a__    = 0
        for i in range(_UpperCAmelCase       ,		min(block_start + block_size       ,		len(_UpperCAmelCase    )    )    ):
            block_int += message_bytes[i] * (BYTE_SIZE ** (i % block_size))
        block_ints.append(_UpperCAmelCase    )
    return block_ints



def 						__magic_name__		(				UpperCamelCase    :		Union[str, Any]       ,		UpperCamelCase    :		str       ,		UpperCamelCase    :		str = DEFAULT_BLOCK_SIZE    )    ->	str:
    a__    = []
    for block_int in block_ints:
        a__    = []
        for i in range(block_size - 1       ,		-1       ,		-1    ):
            if len(_UpperCAmelCase    ) + i < message_length:
                a__    = block_int // (BYTE_SIZE**i)
                a__    = block_int % (BYTE_SIZE**i)
                block_message.insert(0       ,		chr(_UpperCAmelCase    )    )
        message.extend(_UpperCAmelCase    )
    return "".join(_UpperCAmelCase    )



def 						__magic_name__		(				UpperCamelCase    :		Tuple       ,		UpperCamelCase    :		Optional[Any]       ,		UpperCamelCase    :		List[str] = DEFAULT_BLOCK_SIZE    )    ->	list[int]:
    a__    = []
    a__  ,    a__    = key

    for block in get_blocks_from_text(_UpperCAmelCase       ,		_UpperCAmelCase    ):
        encrypted_blocks.append(pow(_UpperCAmelCase       ,		_UpperCAmelCase       ,		_UpperCAmelCase    )    )
    return encrypted_blocks



def 						__magic_name__		(				UpperCamelCase    :		List[Any]       ,		UpperCamelCase    :		Any       ,		UpperCamelCase    :		List[Any]       ,		UpperCamelCase    :		Dict = DEFAULT_BLOCK_SIZE       ,		)    ->	str:
    a__    = []
    a__  ,    a__    = key
    for block in encrypted_blocks:
        decrypted_blocks.append(pow(_UpperCAmelCase       ,		_UpperCAmelCase       ,		_UpperCAmelCase    )    )
    return get_text_from_blocks(_UpperCAmelCase       ,		_UpperCAmelCase       ,		_UpperCAmelCase    )



def 						__magic_name__		(				UpperCamelCase    :		str    )    ->	tuple[int, int, int]:
    with open(_UpperCAmelCase    ) as fo:
        a__    = fo.read()
    a__  ,    a__  ,    a__    = content.split(','    )
    return (int(_UpperCAmelCase    ), int(_UpperCAmelCase    ), int(_UpperCAmelCase    ))



def 						__magic_name__		(				UpperCamelCase    :		str       ,		UpperCamelCase    :		Optional[Any]       ,		UpperCamelCase    :		Optional[int]       ,		UpperCamelCase    :		List[str] = DEFAULT_BLOCK_SIZE       ,		)    ->	str:
    a__  ,    a__  ,    a__    = read_key_file(_UpperCAmelCase    )
    if key_size < block_size * 8:
        sys.exit(
            'ERROR: Block size is {} bits and key size is {} bits. The RSA cipher '
            'requires the block size to be equal to or greater than the key size. '
            'Either decrease the block size or use different keys.'.format(
                block_size * 8       ,		_UpperCAmelCase    )    )

    a__    = [str(_UpperCAmelCase    ) for i in encrypt_message(_UpperCAmelCase       ,		(n, e)       ,		_UpperCAmelCase    )]

    a__    = ','.join(_UpperCAmelCase    )
    a__    = f'{len(_UpperCAmelCase    )}_{block_size}_{encrypted_content}'
    with open(_UpperCAmelCase       ,		'w'    ) as fo:
        fo.write(_UpperCAmelCase    )
    return encrypted_content



def 						__magic_name__		(				UpperCamelCase    :		Any       ,		UpperCamelCase    :		Union[str, Any]    )    ->	str:
    a__  ,    a__  ,    a__    = read_key_file(_UpperCAmelCase    )
    with open(_UpperCAmelCase    ) as fo:
        a__    = fo.read()
    a__  ,    a__  ,    a__    = content.split('_'    )
    a__    = int(_UpperCAmelCase    )
    a__    = int(_UpperCAmelCase    )

    if key_size < block_size * 8:
        sys.exit(
            'ERROR: Block size is {} bits and key size is {} bits. The RSA cipher '
            'requires the block size to be equal to or greater than the key size. '
            'Did you specify the correct key file and encrypted file?'.format(
                block_size * 8       ,		_UpperCAmelCase    )    )

    a__    = []
    for block in encrypted_message.split(','    ):
        encrypted_blocks.append(int(_UpperCAmelCase    )    )

    return decrypt_message(_UpperCAmelCase       ,		_UpperCAmelCase       ,		(n, d)       ,		_UpperCAmelCase    )



def 						__magic_name__		(				)    ->	None:
    a__    = 'encrypted_file.txt'
    a__    = input(r'Encrypt\Decrypt [e\d]: '    )

    if response.lower().startswith('e'    ):
        a__    = 'encrypt'
    elif response.lower().startswith('d'    ):
        a__    = 'decrypt'

    if mode == "encrypt":
        if not os.path.exists('rsa_pubkey.txt'    ):
            rkg.make_key_files('rsa'       ,		1024    )

        a__    = input('\nEnter message: '    )
        a__    = 'rsa_pubkey.txt'
        print(f'Encrypting and writing to {filename}...'    )
        a__    = encrypt_and_write_to_file(_UpperCAmelCase       ,		_UpperCAmelCase       ,		_UpperCAmelCase    )

        print('\nEncrypted text:'    )
        print(_UpperCAmelCase    )

    elif mode == "decrypt":
        a__    = 'rsa_privkey.txt'
        print(f'Reading from {filename} and decrypting...'    )
        a__    = read_from_file_and_decrypt(_UpperCAmelCase       ,		_UpperCAmelCase    )
        print('writing decryption to rsa_decryption.txt...'    )
        with open('rsa_decryption.txt'       ,		'w'    ) as dec:
            dec.write(_UpperCAmelCase    )

        print('\nDecryption:'    )
        print(_UpperCAmelCase    )


if __name__ == "__main__":
    main()








"""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		:					Tuple    	=					namedtuple('ExampleNamedTuple', 'a b c')







class 						lowercase(unittest.TestCase					):



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







         """simple docstring"""



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

         a__    = send_to_device(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta.cpu()    ,							__SCREAMING_SNAKE_CASE      )      )

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

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

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

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



     def  lowercase__     (       self      )		->    Dict:







         """simple docstring"""



         with self.assertRaises(__SCREAMING_SNAKE_CASE      ) as cm:
             a__    = recursively_apply(torch.tensor    ,							(torch.tensor(1      ), 1)    ,							error_on_other_type=__SCREAMING_SNAKE_CASE      )
         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      )		->    Any:







         """simple docstring"""



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

         a__    = torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )
         self.assertEqual(listify(__SCREAMING_SNAKE_CASE      )    ,							[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )

         a__    = torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )
         self.assertEqual(
             listify(__SCREAMING_SNAKE_CASE      )    ,							[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )



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







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE    ,							keep_fpaa_wrapper=__SCREAMING_SNAKE_CASE      )
         a__    = pickle.dumps(__SCREAMING_SNAKE_CASE      )



     @require_cuda
     def  lowercase__     (       self      )		->    Union[str, Any]:







         """simple docstring"""



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



     @require_cuda
     @require_torch_min_version(version='2.0'      )
     def  lowercase__     (       self      )		->    Optional[int]:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = torch.cuda.amp.autocast(dtype=torch.floataa      )(model.forward      )
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = torch.compile(model.forward    ,							backend='inductor'      )
         a__    = torch.randn(4    ,							1_0      ).cuda()
         a__    = model(__SCREAMING_SNAKE_CASE      )



     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



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

         self.assertEqual(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )



     @require_torch_min_version(version='2.0'      )
     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = torch.compile(__SCREAMING_SNAKE_CASE      )

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

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




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


