


'''simple docstring'''






import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__UpperCAmelCase					       =							logging.get_logger(__name__)


def       _snake_case     (  )		->					Optional[int]:
							# Get the sagemaker specific mp parameters from smp_options variable.
							lowerCAmelCase__      =			os.getenv('''SM_HP_MP_PARAMETERS'''				,					'''{}''' )
							try:
														# Parse it and check the field "partitions" is included, it is required for model parallel.
														lowerCAmelCase__      =			json.loads(A )
														if "partitions" not in smp_options:
																					return False
							except json.JSONDecodeError:
														return False

							# Get the sagemaker specific framework parameters from mpi_options variable.
							lowerCAmelCase__      =			os.getenv('''SM_FRAMEWORK_PARAMS'''				,					'''{}''' )
							try:
														# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
														lowerCAmelCase__      =			json.loads(A )
														if not mpi_options.get('''sagemaker_mpi_enabled'''				,					A ):
																					return False
							except json.JSONDecodeError:
														return False
							# Lastly, check if the `smdistributed` module is present.
							return importlib.util.find_spec('''smdistributed''' ) is not None


if is_sagemaker_model_parallel_available():
		import smdistributed.modelparallel.torch as smp

		smp.init()






@dataclass
class 	a__       (							a__						):





							'''simple docstring'''






							lowercase__    :     str					=   field(
							    default=""       ,       metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}       ,       )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Tuple:
														super().__post_init__()
														warnings.warn(
														    '''`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use '''
														    '''`TrainingArguments` instead.'''      ,					lowerCamelCase_      ,					)







							@cached_property
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							"torch.device":
														logger.info('''PyTorch: setting up devices'''     )
														if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
																					logger.warning(
																					    '''torch.distributed process group is initialized, but local_rank == -1. '''
																					    '''In order to use Torch DDP, launch your script with `python -m torch.distributed.launch'''     )
														if self.no_cuda:
																					lowerCAmelCase__      =			torch.device('''cpu'''     )
																					lowerCAmelCase__      =			0
														elif is_sagemaker_model_parallel_available():
																					lowerCAmelCase__      =			smp.local_rank()
																					lowerCAmelCase__      =			torch.device('''cuda'''      ,					lowerCamelCase_     )
																					lowerCAmelCase__      =			1
														elif is_sagemaker_dp_enabled():
																					import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

																					torch.distributed.init_process_group(backend='''smddp'''      ,					timeout=self.ddp_timeout_delta     )
																					lowerCAmelCase__      =			int(os.getenv('''SMDATAPARALLEL_LOCAL_RANK'''     )     )
																					lowerCAmelCase__      =			torch.device('''cuda'''      ,					self.local_rank     )
																					lowerCAmelCase__      =			1
														elif self.local_rank == -1:
																					# if n_gpu is > 1 we'll use nn.DataParallel.
																					# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
																					# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
																					# trigger an error that a device index is missing. Index 0 takes into account the
																					# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
																					# will use the first GPU in that env, i.e. GPU#1
																					lowerCAmelCase__      =			torch.device('''cuda:0''' if torch.cuda.is_available() else '''cpu'''     )
																					# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
																					# the default value.
																					lowerCAmelCase__      =			torch.cuda.device_count()
														else:
																					# Here, we'll use torch.distributed.
																					# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
																					if not torch.distributed.is_initialized():
																												torch.distributed.init_process_group(backend='''nccl'''      ,					timeout=self.ddp_timeout_delta     )
																					lowerCAmelCase__      =			torch.device('''cuda'''      ,					self.local_rank     )
																					lowerCAmelCase__      =			1

														if device.type == "cuda":
																					torch.cuda.set_device(lowerCamelCase_     )

														return device







							@property
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Any:
														if is_sagemaker_model_parallel_available():
																					return smp.dp_size()

														return super().world_size







							@property
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							int:
														return not is_sagemaker_model_parallel_available()







							@property
							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Tuple:
														return False



'''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")