

import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
  import torch_xla.core.xla_model as xm  # noqa: F401

  lowercase_     =       True
except ImportError:
  lowercase_     =       False


# Cache this result has it's a C FFI call which can be pretty time-consuming
lowercase_     =       torch.distributed.is_available()
def 					a				(      A__     :       str					)   ->				Tuple:





      """simple docstring"""



      _lowercase     	=importlib.util.find_spec(A__					) is not None
      if package_exists:
            try:
                  _lowercase     	=importlib.metadata.metadata(A__					)
                  return True
            except importlib.metadata.PackageNotFoundError:
                  return False
def 					a				(      )   ->				bool:





      """simple docstring"""



      return _torch_distributed_available
def 					a				(      )   ->				Optional[int]:





      """simple docstring"""



      try:
            pass
      except ImportError:
            print(
                'Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not'
                ' detected. If you see \"ValueError: Invalid backend: \'ccl\'\" error, please install Intel(R) oneCCL'
                ' Bindings for PyTorch*.'					)
      return (
          importlib.util.find_spec('torch_ccl'					) is not None
          or importlib.util.find_spec('oneccl_bindings_for_pytorch'					) is not None
      )
def 					a				(      )   ->				List[Any]:





      """simple docstring"""



      return importlib.metadata.version('oneccl_bind_pt'					)
def 					a				(      )   ->				List[str]:





      """simple docstring"""



      return _is_package_available('transformer_engine'					)
@lru_cache
def 					a				(      A__     :       Optional[int]=True					)   ->				Any:





      """simple docstring"""



      if torch.cuda.is_available():
            return False
      if _tpu_available and check_device:
            try:
                  # Will raise a RuntimeError if no XLA configuration is found
                  _lowercase     	=xm.xla_device()
                  return True
            except RuntimeError:
                  return False
      return _tpu_available
def 					a				(      )   ->				Tuple:





      """simple docstring"""



      return _is_package_available('deepspeed'					)
def 					a				(      A__     :       str=False					)   ->				int:





      """simple docstring"""



      if is_tpu_available():
            return not ignore_tpu
      if torch.cuda.is_available():
            return torch.cuda.is_bfaa_supported()
      if is_npu_available():
            return False
      return True
def 					a				(      )   ->				Optional[Any]:





      """simple docstring"""



      _lowercase     	=_is_package_available('bitsandbytes'					)
      if package_exists:
            _lowercase     	=version.parse(importlib.metadata.version('bitsandbytes'					)					)
            return compare_versions(A__				,       '>='				,       '0.39.0'					)
      return False
def 					a				(      )   ->				str:





      """simple docstring"""



      _lowercase     	=_is_package_available('bitsandbytes'					)
      if package_exists:
            _lowercase     	=version.parse(importlib.metadata.version('bitsandbytes'					)					)
            return compare_versions(A__				,       '>='				,       '0.37.2'					)
      return False
def 					a				(      )   ->				Any:





      """simple docstring"""



      return _is_package_available('bitsandbytes'					)
def 					a				(      )   ->				int:





      """simple docstring"""



      if strtobool(os.environ.get('ACCELERATE_USE_MEGATRON_LM'				,       'False'					)					) == 1:
            _lowercase     	=importlib.util.find_spec('megatron'					) is not None
            if package_exists:
                  try:
                        _lowercase     	=parse(importlib.metadata.version('megatron-lm'					)					)
                        return compare_versions(A__				,       '>='				,       '2.2.0'					)
                  except Exception as e:
                        warnings.warn(F'''Parse Megatron version failed. Exception:{e}'''					)
                        return False
def 					a				(      )   ->				List[Any]:





      """simple docstring"""



      return _is_package_available('safetensors'					)
def 					a				(      )   ->				int:





      """simple docstring"""



      return _is_package_available('transformers'					)
def 					a				(      )   ->				Optional[int]:





      """simple docstring"""



      return _is_package_available('datasets'					)
def 					a				(      )   ->				List[str]:





      """simple docstring"""



      return _is_package_available('aim'					)
def 					a				(      )   ->				Optional[Any]:





      """simple docstring"""



      return _is_package_available('tensorboard'					) or _is_package_available('tensorboardX'					)
def 					a				(      )   ->				str:





      """simple docstring"""



      return _is_package_available('wandb'					)
def 					a				(      )   ->				Dict:





      """simple docstring"""



      return _is_package_available('comet_ml'					)
def 					a				(      )   ->				Optional[Any]:





      """simple docstring"""



      return _is_package_available('boto3'					)
def 					a				(      )   ->				Optional[int]:





      """simple docstring"""



      if _is_package_available('rich'					):
            if "ACCELERATE_DISABLE_RICH" in os.environ:
                  warnings.warn(
                      '`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`.'					)
                  return not parse_flag_from_env('ACCELERATE_DISABLE_RICH'				,       A__					)
            return parse_flag_from_env('ACCELERATE_ENABLE_RICH'				,       A__					)
      return False
def 					a				(      )   ->				Any:





      """simple docstring"""



      return _is_package_available('sagemaker'					)
def 					a				(      )   ->				List[str]:





      """simple docstring"""



      return _is_package_available('tqdm'					)
def 					a				(      )   ->				Optional[Any]:





      """simple docstring"""



      return _is_package_available('mlflow'					)
def 					a				(      )   ->				Dict:





      """simple docstring"""



      return is_torch_version('>='				,       '1.12'					) and torch.backends.mps.is_available() and torch.backends.mps.is_built()
def 					a				(      )   ->				Optional[int]:





      """simple docstring"""



      def get_major_and_minor_from_version(A__     :       Union[str, Any]					):
            return str(version.parse(A__					).major					) + "." + str(version.parse(A__					).minor					)

      _lowercase     	=importlib.metadata.version('torch'					)
      if importlib.util.find_spec('intel_extension_for_pytorch'					) is None:
            return False
      _lowercase     	='N/A'
      try:
            _lowercase     	=importlib.metadata.version('intel_extension_for_pytorch'					)
      except importlib.metadata.PackageNotFoundError:
            return False
      _lowercase     	=get_major_and_minor_from_version(_torch_version					)
      _lowercase     	=get_major_and_minor_from_version(_ipex_version					)
      if torch_major_and_minor != ipex_major_and_minor:
            warnings.warn(
                F'''Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'''
                F''' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'''					)
            return False
      return True
@lru_cache
def 					a				(      A__     :       Any=False					)   ->				Optional[Any]:





      """simple docstring"""



      if importlib.util.find_spec('torch'					) is None or importlib.util.find_spec('torch_npu'					) is None:
            return False

      import torch
      import torch_npu  # noqa: F401

      if check_device:
            try:
                  # Will raise a RuntimeError if no NPU is found
                  _lowercase     	=torch.npu.device_count()
                  return torch.npu.is_available()
            except RuntimeError:
                  return False
      return hasattr(A__				,       'npu'					) and torch.npu.is_available()







@lru_cache
def 					a				(      A__     :       Optional[int]=False					)   ->				List[Any]:





      """simple docstring"""



      if not parse_flag_from_env('ACCELERATE_USE_XPU'				,       default=A__					):
            return False
      if is_ipex_available():
            import torch

            if is_torch_version('<='				,       '1.12'					):
                  return False
      else:
            return False

      import intel_extension_for_pytorch  # noqa: F401

      if check_device:
            try:
                  # Will raise a RuntimeError if no XPU  is found
                  _lowercase     	=torch.xpu.device_count()
                  return torch.xpu.is_available()
            except RuntimeError:
                  return False
      return hasattr(A__				,       'xpu'					) and torch.xpu.is_available()





import os
import sys

from . import rsa_key_generator as rkg

lowercase_     =       1_2_8
lowercase_     =       2_5_6
def 					a				(      A__     :       str				,       A__     :       int = DEFAULT_BLOCK_SIZE					)   ->				list[int]:





      """simple docstring"""



      _lowercase     	=message.encode('ascii'					)

      _lowercase     	=[]
      for block_start in range(0				,       len(A__					)				,       A__					):
            _lowercase     	=0
            for i in range(A__				,       min(block_start + block_size				,       len(A__					)					)					):
                  block_int += message_bytes[i] * (BYTE_SIZE ** (i % block_size))
            block_ints.append(A__					)
      return block_ints
def 					a				(      A__     :       list[int]				,       A__     :       int				,       A__     :       int = DEFAULT_BLOCK_SIZE					)   ->				str:





      """simple docstring"""



      _lowercase     	=[]
      for block_int in block_ints:
            _lowercase     	=[]
            for i in range(block_size - 1				,       -1				,       -1					):
                  if len(A__					) + i < message_length:
                        _lowercase     	=block_int // (BYTE_SIZE**i)
                        _lowercase     	=block_int % (BYTE_SIZE**i)
                        block_message.insert(0				,       chr(A__					)					)
            message.extend(A__					)
      return "".join(A__					)
def 					a				(      A__     :       str				,       A__     :       tuple[int, int]				,       A__     :       int = DEFAULT_BLOCK_SIZE					)   ->				list[int]:





      """simple docstring"""



      _lowercase     	=[]
      _lowercase	,	_lowercase     	=key

      for block in get_blocks_from_text(A__				,       A__					):
            encrypted_blocks.append(pow(A__				,       A__				,       A__					)					)
      return encrypted_blocks
def 					a				(      A__     :       list[int]				,       A__     :       int				,       A__     :       tuple[int, int]				,       A__     :       int = DEFAULT_BLOCK_SIZE				,       )   ->				str:





      """simple docstring"""



      _lowercase     	=[]
      _lowercase	,	_lowercase     	=key
      for block in encrypted_blocks:
            decrypted_blocks.append(pow(A__				,       A__				,       A__					)					)
      return get_text_from_blocks(A__				,       A__				,       A__					)
def 					a				(      A__     :       str					)   ->				tuple[int, int, int]:





      """simple docstring"""



      with open(A__					) as fo:
            _lowercase     	=fo.read()
      _lowercase	,	_lowercase	,	_lowercase     	=content.split(','					)
      return (int(A__					), int(A__					), int(A__					))
def 					a				(      A__     :       str				,       A__     :       str				,       A__     :       str				,       A__     :       int = DEFAULT_BLOCK_SIZE				,       )   ->				str:





      """simple docstring"""



      _lowercase	,	_lowercase	,	_lowercase     	=read_key_file(A__					)
      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				,       A__					)					)

      _lowercase     	=[str(A__					) for i in encrypt_message(A__				,       (n, e)				,       A__					)]

      _lowercase     	=','.join(A__					)
      _lowercase     	=F'''{len(A__					)}_{block_size}_{encrypted_content}'''
      with open(A__				,       'w'					) as fo:
            fo.write(A__					)
      return encrypted_content
def 					a				(      A__     :       str				,       A__     :       str					)   ->				str:





      """simple docstring"""



      _lowercase	,	_lowercase	,	_lowercase     	=read_key_file(A__					)
      with open(A__					) as fo:
            _lowercase     	=fo.read()
      _lowercase	,	_lowercase	,	_lowercase     	=content.split('_'					)
      _lowercase     	=int(A__					)
      _lowercase     	=int(A__					)

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

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

      return decrypt_message(A__				,       A__				,       (n, d)				,       A__					)
def 					a				(      )   ->				None:





      """simple docstring"""



      _lowercase     	='encrypted_file.txt'
      _lowercase     	=input(r'Encrypt\Decrypt [e\d]: '					)

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

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

            _lowercase     	=input('\nEnter message: '					)
            _lowercase     	='rsa_pubkey.txt'
            print(F'''Encrypting and writing to {filename}...'''					)
            _lowercase     	=encrypt_and_write_to_file(A__				,       A__				,       A__					)

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

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

            print('\nDecryption:'					)
            print(A__					)


if __name__ == "__main__":
  main()


