


'''simple docstring'''






import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment


def       _snake_case     (  A				,					A=()				,					A=None				,					A="no"				,					A="29500" )		->					int:
							lowerCAmelCase__      =			False
							lowerCAmelCase__      =			False
							if any(key.startswith('''KAGGLE''' ) for key in os.environ.keys() ):
														lowerCAmelCase__      =			True
							elif "IPython" in sys.modules:
														lowerCAmelCase__      =			'''google.colab''' in str(sys.modules['''IPython'''].get_ipython() )

							try:
														lowerCAmelCase__      =			PrecisionType(mixed_precision.lower() )
							except ValueError:
														raise ValueError(
														    F"""Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.""" )

							if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''				,					A ) is not None):
														# TPU launch
														import torch_xla.distributed.xla_multiprocessing as xmp

														if len(AcceleratorState._shared_state ) > 0:
																					raise ValueError(
																					    '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
																					    '''your training function. Restart your notebook and make sure no cells initializes an '''
																					    '''`Accelerator`.''' )
														if num_processes is None:
																					lowerCAmelCase__      =			8

														lowerCAmelCase__      =			PrepareForLaunch(A				,					distributed_type='''TPU''' )
														print(F"""Launching a training on {num_processes} TPU cores.""" )
														xmp.spawn(A				,					args=A				,					nprocs=A				,					start_method='''fork''' )
							elif in_colab:
														# No need for a distributed launch otherwise as it's either CPU or one GPU.
														if torch.cuda.is_available():
																					print('''Launching training on one GPU.''' )
														else:
																					print('''Launching training on one CPU.''' )
														function(*A )
							else:
														if num_processes is None:
																					raise ValueError(
																					    '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.''' )

														if num_processes > 1:
																					# Multi-GPU launch
																					from torch.multiprocessing import start_processes
																					from torch.multiprocessing.spawn import ProcessRaisedException

																					if len(AcceleratorState._shared_state ) > 0:
																												raise ValueError(
																												    '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
																												    '''inside your training function. Restart your notebook and make sure no cells initializes an '''
																												    '''`Accelerator`.''' )

																					if torch.cuda.is_initialized():
																												raise ValueError(
																												    '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
																												    '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
																												    '''function.''' )

																					# torch.distributed will expect a few environment variable to be here. We set the ones common to each
																					# process here (the other ones will be set be the launcher).
																					with patch_environment(
																					    world_size=A				,					master_addr='''127.0.01'''				,					master_port=A				,					mixed_precision=A ):
																												lowerCAmelCase__      =			PrepareForLaunch(A				,					distributed_type='''MULTI_GPU''' )
																												print(F"""Launching training on {num_processes} GPUs.""" )
																												try:
																																			start_processes(A				,					args=A				,					nprocs=A				,					start_method='''fork''' )
																												except ProcessRaisedException as e:
																																			if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																																										raise RuntimeError(
																																										    '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
																																										    '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
																																										    '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
																																										    '''which one is problematic.''' ) from e

														else:
																					# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
																					if is_mps_available():
																												lowerCAmelCase__      =			'''1'''
																												print('''Launching training on MPS.''' )
																					elif torch.cuda.is_available():
																												print('''Launching training on one GPU.''' )
																					else:
																												print('''Launching training on CPU.''' )
																					function(*A )






def       _snake_case     (  A				,					A=()				,					A=2 )		->					Optional[Any]:
							from torch.multiprocessing import start_processes

							with tempfile.NamedTemporaryFile() as tmp_file:
														# torch.distributed will expect a few environment variable to be here. We set the ones common to each
														# process here (the other ones will be set be the launcher).
														with patch_environment(
														    world_size=A				,					master_addr='''127.0.01'''				,					master_port='''29500'''				,					accelerate_mixed_precision='''no'''				,					accelerate_debug_rdv_file=tmp_file.name				,					accelerate_use_cpu='''yes'''				,					):
																					lowerCAmelCase__      =			PrepareForLaunch(A				,					debug=A )
																					start_processes(A				,					args=A				,					nprocs=A				,					start_method='''fork''' )



'''simple docstring'''






from __future__ import annotations




class 	a__       :





							'''simple docstring'''






							def __init__(	self      ,					lowerCamelCase_ = 0     ) ->							List[Any]:
														lowerCAmelCase__      =			key







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_     ) ->							list[str]:
														assert isinstance(lowerCamelCase_      ,					lowerCamelCase_     ) and isinstance(lowerCamelCase_      ,					lowerCamelCase_     )

														lowerCAmelCase__      =			key or self.__key or 1

														# make sure key is an appropriate size
														key %= 2_55

														return [chr(ord(lowerCamelCase_     ) ^ key     ) for ch in content]







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_     ) ->							list[str]:
														assert isinstance(lowerCamelCase_      ,					lowerCamelCase_     ) and isinstance(lowerCamelCase_      ,					lowerCamelCase_     )

														lowerCAmelCase__      =			key or self.__key or 1

														# make sure key is an appropriate size
														key %= 2_55

														return [chr(ord(lowerCamelCase_     ) ^ key     ) for ch in content]







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_ = 0     ) ->							str:
														assert isinstance(lowerCamelCase_      ,					lowerCamelCase_     ) and isinstance(lowerCamelCase_      ,					lowerCamelCase_     )

														lowerCAmelCase__      =			key or self.__key or 1

														# make sure key can be any size
														while key > 2_55:
																					key -= 2_55

														# This will be returned
														lowerCAmelCase__      =			''''''

														for ch in content:
																					ans += chr(ord(lowerCamelCase_     ) ^ key     )

														return ans







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_ = 0     ) ->							str:
														assert isinstance(lowerCamelCase_      ,					lowerCamelCase_     ) and isinstance(lowerCamelCase_      ,					lowerCamelCase_     )

														lowerCAmelCase__      =			key or self.__key or 1

														# make sure key can be any size
														while key > 2_55:
																					key -= 2_55

														# This will be returned
														lowerCAmelCase__      =			''''''

														for ch in content:
																					ans += chr(ord(lowerCamelCase_     ) ^ key     )

														return ans







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_ = 0     ) ->							bool:
														assert isinstance(lowerCamelCase_      ,					lowerCamelCase_     ) and isinstance(lowerCamelCase_      ,					lowerCamelCase_     )

														try:
																					with open(lowerCamelCase_     ) as fin, open('''encrypt.out'''      ,					'''w+'''     ) as fout:
																												# actual encrypt-process
																												for line in fin:
																																			fout.write(self.encrypt_string(lowerCamelCase_      ,					lowerCamelCase_     )     )

														except OSError:
																					return False

														return True







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_     ) ->							bool:
														assert isinstance(lowerCamelCase_      ,					lowerCamelCase_     ) and isinstance(lowerCamelCase_      ,					lowerCamelCase_     )

														try:
																					with open(lowerCamelCase_     ) as fin, open('''decrypt.out'''      ,					'''w+'''     ) as fout:
																												# actual encrypt-process
																												for line in fin:
																																			fout.write(self.decrypt_string(lowerCamelCase_      ,					lowerCamelCase_     )     )

														except OSError:
																					return False

														return True


# Tests
# crypt = XORCipher()
# key = 67

# # test encrypt
# print(crypt.encrypt("hallo welt",key))
# # test decrypt
# print(crypt.decrypt(crypt.encrypt("hallo welt",key), key))

# # test encrypt_string
# print(crypt.encrypt_string("hallo welt",key))

# # test decrypt_string
# print(crypt.decrypt_string(crypt.encrypt_string("hallo welt",key),key))

# if (crypt.encrypt_file("test.txt",key)):
#       print("encrypt successful")
# else:
#       print("encrypt unsuccessful")

# if (crypt.decrypt_file("encrypt.out",key)):
#       print("decrypt successful")
# else:
#       print("decrypt unsuccessful")