






import inspect
import os

import torch
from transformers import AutoModel
from transformers.testing_utils import mockenv_context
from transformers.trainer_utils import set_seed

import accelerate
from accelerate.accelerator import Accelerator
from accelerate.state import AcceleratorState
from accelerate.test_utils.testing import (
    AccelerateTestCase,
    TempDirTestCase,
    execute_subprocess_async,
    require_cuda,
    require_fsdp,
    require_multi_gpu,
    slow,
)
from accelerate.utils.constants import (
    FSDP_AUTO_WRAP_POLICY,
    FSDP_BACKWARD_PREFETCH,
    FSDP_SHARDING_STRATEGY,
    FSDP_STATE_DICT_TYPE,
)
from accelerate.utils.dataclasses import FullyShardedDataParallelPlugin
from accelerate.utils.other import patch_environment


set_seed(42)

__lowerCamelCase :      Union[str, Any]										=				"""bert-base-cased"""
__lowerCamelCase :      Optional[Any]										=				"""fp16"""
__lowerCamelCase :      Any										=				"""bf16"""
__lowerCamelCase :      Tuple										=				[FPaa, BFaa]



@require_fsdp
@require_cuda
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[int]	):
						super().setUp()

						snake_case__			:				Optional[int]       =  dict(
						    ACCELERATE_USE_FSDP="true"							,		MASTER_ADDR="localhost"							,		MASTER_PORT="10999"							,		RANK="0"							,		LOCAL_RANK="0"							,		WORLD_SIZE="1"							,		)



					def 	_lowercase					( self					:  Union[str, Any]	):
						from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

						for i, strategy in enumerate(__A	):
							snake_case__			:				Any       =  self.dist_env.copy()
							snake_case__			:				Union[str, Any]       =  f'''{i + 1}'''
							snake_case__			:				List[str]       =  strategy
							with mockenv_context(**__A	):
								snake_case__			:				Optional[int]       =  FullyShardedDataParallelPlugin()
								self.assertEqual(fsdp_plugin.sharding_strategy							,		ShardingStrategy(i + 1	)	)



					def 	_lowercase					( self					:  Any	):
						from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

						for i, prefetch_policy in enumerate(__A	):
							snake_case__			:				str       =  self.dist_env.copy()
							snake_case__			:				str       =  prefetch_policy
							with mockenv_context(**__A	):
								snake_case__			:				Any       =  FullyShardedDataParallelPlugin()
								if prefetch_policy == "NO_PREFETCH":
									self.assertIsNone(fsdp_plugin.backward_prefetch	)
								else:
									self.assertEqual(fsdp_plugin.backward_prefetch							,		BackwardPrefetch(i + 1	)	)



					def 	_lowercase					( self					:  Union[str, Any]	):
						from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

						for i, state_dict_type in enumerate(__A	):
							snake_case__			:				Any       =  self.dist_env.copy()
							snake_case__			:				str       =  state_dict_type
							with mockenv_context(**__A	):
								snake_case__			:				List[str]       =  FullyShardedDataParallelPlugin()
								self.assertEqual(fsdp_plugin.state_dict_type							,		StateDictType(i + 1	)	)
								if state_dict_type == "FULL_STATE_DICT":
									self.assertTrue(fsdp_plugin.state_dict_config.offload_to_cpu	)
									self.assertTrue(fsdp_plugin.state_dict_config.ranka_only	)



					def 	_lowercase					( self					:  Optional[int]	):
						snake_case__			:				Tuple       =  AutoModel.from_pretrained(__A	)
						for policy in FSDP_AUTO_WRAP_POLICY:
							snake_case__			:				Dict       =  self.dist_env.copy()
							snake_case__			:				int       =  policy
							if policy == "TRANSFORMER_BASED_WRAP":
								snake_case__			:				Tuple       =  "BertLayer"
							elif policy == "SIZE_BASED_WRAP":
								snake_case__			:				Any       =  "2000"
							with mockenv_context(**__A	):
								snake_case__			:				Dict       =  FullyShardedDataParallelPlugin()
								fsdp_plugin.set_auto_wrap_policy(__A	)
								if policy == "NO_WRAP":
									self.assertIsNone(fsdp_plugin.auto_wrap_policy	)
								else:
									self.assertIsNotNone(fsdp_plugin.auto_wrap_policy	)

						snake_case__			:				str       =  self.dist_env.copy()
						snake_case__			:				Optional[int]       =  "TRANSFORMER_BASED_WRAP"
						snake_case__			:				Union[str, Any]       =  "T5Layer"
						with mockenv_context(**__A	):
							snake_case__			:				Optional[int]       =  FullyShardedDataParallelPlugin()
							with self.assertRaises(__A	) as cm:
								fsdp_plugin.set_auto_wrap_policy(__A	)
							self.assertTrue("Could not find the transformer layer class to wrap in the model." in str(cm.exception	)	)

						snake_case__			:				Optional[Any]       =  self.dist_env.copy()
						snake_case__			:				Dict       =  "SIZE_BASED_WRAP"
						snake_case__			:				List[Any]       =  "0"
						with mockenv_context(**__A	):
							snake_case__			:				Dict       =  FullyShardedDataParallelPlugin()
							fsdp_plugin.set_auto_wrap_policy(__A	)
							self.assertIsNone(fsdp_plugin.auto_wrap_policy	)



					def 	_lowercase					( self					:  int	):
						from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
						from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

						for mp_dtype in dtypes:
							snake_case__			:				str       =  self.dist_env.copy()
							snake_case__			:				List[str]       =  mp_dtype
							with mockenv_context(**__A	):
								snake_case__			:				Optional[int]       =  Accelerator()
								if mp_dtype == "fp16":
									snake_case__			:				Tuple       =  torch.floataa
								elif mp_dtype == "bf16":
									snake_case__			:				str       =  torch.bfloataa
								snake_case__			:				int       =  MixedPrecision(param_dtype=__A							,		reduce_dtype=__A							,		buffer_dtype=__A	)
								self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy							,		__A	)
								if mp_dtype == FPaa:
									self.assertTrue(isinstance(accelerator.scaler							,		__A	)	)
								elif mp_dtype == BFaa:
									self.assertIsNone(accelerator.scaler	)
								AcceleratorState._reset_state(__A	)






					def 	_lowercase					( self					:  List[str]	):
						from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

						for flag in [True, False]:
							snake_case__			:				Union[str, Any]       =  self.dist_env.copy()
							snake_case__			:				Tuple       =  str(__A	).lower()
							with mockenv_context(**__A	):
								snake_case__			:				Any       =  FullyShardedDataParallelPlugin()
								self.assertEqual(fsdp_plugin.cpu_offload							,		CPUOffload(offload_params=__A	)	)



@require_fsdp
@require_multi_gpu
@slow
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					def 	_lowercase					( self					:  List[Any]	):
						super().setUp()
						snake_case__			:				Tuple       =  0.8_2
						snake_case__			:				int       =  [
						    "fsdp_shard_grad_op_transformer_based_wrap",
						    "fsdp_full_shard_transformer_based_wrap",
						]
						snake_case__			:				Optional[Any]       =  {
						    "multi_gpu_fp16": 3_2_0_0,
						    "fsdp_shard_grad_op_transformer_based_wrap_fp16": 2_0_0_0,
						    "fsdp_full_shard_transformer_based_wrap_fp16": 1_9_0_0,
						    # Disabling below test as it overwhelms the RAM memory usage
						    # on CI self-hosted runner leading to tests getting killed.
						    # "fsdp_full_shard_cpu_offload_transformer_based_wrap_fp32": 1500,  # fp16 was leading to indefinite hang
						}
						snake_case__			:				Optional[Any]       =  1_6_0
						snake_case__			:				Optional[Any]       =  1_6_0

						snake_case__			:				Tuple       =  inspect.getfile(accelerate.test_utils	)
						snake_case__			:				Optional[int]       =  os.path.sep.join(mod_file.split(os.path.sep	)[:-1] + ["scripts", "external_deps"]	)



					def 	_lowercase					( self					:  Dict	):
						snake_case__			:				Tuple       =  os.path.join(self.test_scripts_folder							,		"test_performance.py"	)
						snake_case__			:				Dict       =  ["accelerate", "launch", "--num_processes=2", "--num_machines=1", "--machine_rank=0", "--use_fsdp"]
						for config in self.performance_configs:
							snake_case__			:				str       =  cmd.copy()
							for i, strategy in enumerate(__A	):
								if strategy.lower() in config:
									cmd_config.append(f'''--fsdp_sharding_strategy={i+1}'''	)
									break

							if "fp32" in config:
								cmd_config.append("--mixed_precision=no"	)
							else:
								cmd_config.append("--mixed_precision=fp16"	)

							if "cpu_offload" in config:
								cmd_config.append("--fsdp_offload_params=True"	)

							for policy in FSDP_AUTO_WRAP_POLICY:
								if policy.lower() in config:
									cmd_config.append(f'''--fsdp_auto_wrap_policy={policy}'''	)
									break

							if policy == "TRANSFORMER_BASED_WRAP":
								cmd_config.append("--fsdp_transformer_layer_cls_to_wrap=BertLayer"	)
							elif policy == "SIZE_BASED_WRAP":
								cmd_config.append("--fsdp_min_num_params=2000"	)

							cmd_config.extend(
							    [
							        self.test_file_path,
							        f'''--output_dir={self.tmpdir}''',
							        f'''--performance_lower_bound={self.performance_lower_bound}''',
							    ]	)
							with patch_environment(omp_num_threads=1	):
								execute_subprocess_async(__A							,		env=os.environ.copy()	)



					def 	_lowercase					( self					:  List[Any]	):
						snake_case__			:				Optional[int]       =  os.path.join(self.test_scripts_folder							,		"test_checkpointing.py"	)
						snake_case__			:				List[Any]       =  [
						    "accelerate",
						    "launch",
						    "--num_processes=2",
						    "--num_machines=1",
						    "--machine_rank=0",
						    "--use_fsdp",
						    "--mixed_precision=fp16",
						    "--fsdp_transformer_layer_cls_to_wrap=BertLayer",
						]

						for i, strategy in enumerate(__A	):
							snake_case__			:				Union[str, Any]       =  cmd.copy()
							cmd_config.append(f'''--fsdp_sharding_strategy={i+1}'''	)
							if strategy != "FULL_SHARD":
								continue
							snake_case__			:				Any       =  len(__A	)
							for state_dict_type in FSDP_STATE_DICT_TYPE:
								snake_case__			:				str       =  cmd_config[:state_dict_config_index]
								cmd_config.append(f'''--fsdp_state_dict_type={state_dict_type}'''	)
								cmd_config.extend(
								    [
								        self.test_file_path,
								        f'''--output_dir={self.tmpdir}''',
								        "--partial_train_epoch=1",
								    ]	)
								with patch_environment(omp_num_threads=1	):
									execute_subprocess_async(__A							,		env=os.environ.copy()	)

								snake_case__			:				Tuple       =  cmd_config[:-1]
								snake_case__			:				Optional[Any]       =  os.path.join(self.tmpdir							,		"epoch_0"	)
								cmd_config.extend(
								    [
								        f'''--resume_from_checkpoint={resume_from_checkpoint}''',
								    ]	)
								with patch_environment(omp_num_threads=1	):
									execute_subprocess_async(__A							,		env=os.environ.copy()	)






					def 	_lowercase					( self					:  Any	):
						snake_case__			:				str       =  os.path.join(self.test_scripts_folder							,		"test_peak_memory_usage.py"	)
						snake_case__			:				Optional[Any]       =  [
						    "accelerate",
						    "launch",
						    "--num_processes=2",
						    "--num_machines=1",
						    "--machine_rank=0",
						]
						for spec, peak_mem_upper_bound in self.peak_memory_usage_upper_bound.items():
							snake_case__			:				Tuple       =  cmd.copy()
							if "fp16" in spec:
								cmd_config.extend(["--mixed_precision=fp16"]	)
							else:
								cmd_config.extend(["--mixed_precision=no"]	)

							if "multi_gpu" in spec:
								continue
							else:
								cmd_config.extend(["--use_fsdp"]	)
								for i, strategy in enumerate(__A	):
									if strategy.lower() in spec:
										cmd_config.append(f'''--fsdp_sharding_strategy={i+1}'''	)
										break

								if "cpu_offload" in spec:
									cmd_config.append("--fsdp_offload_params=True"	)

								for policy in FSDP_AUTO_WRAP_POLICY:
									if policy.lower() in spec:
										cmd_config.append(f'''--fsdp_auto_wrap_policy={policy}'''	)
										break

								if policy == "TRANSFORMER_BASED_WRAP":
									cmd_config.append("--fsdp_transformer_layer_cls_to_wrap=BertLayer"	)
								elif policy == "SIZE_BASED_WRAP":
									cmd_config.append("--fsdp_min_num_params=2000"	)

							cmd_config.extend(
							    [
							        self.test_file_path,
							        f'''--output_dir={self.tmpdir}''',
							        f'''--peak_memory_upper_bound={peak_mem_upper_bound}''',
							        f'''--n_train={self.n_train}''',
							        f'''--n_val={self.n_val}''',
							    ]	)
							with patch_environment(omp_num_threads=1	):
								execute_subprocess_async(__A							,		env=os.environ.copy()	)











from __future__ import annotations







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							float					,  snake_case_ :							float					,  snake_case_ :							float       ):
	if days_between_payments <= 0:
		raise ValueError("days_between_payments must be > 0"       )
	if daily_interest_rate < 0:
		raise ValueError("daily_interest_rate must be >= 0"       )
	if principal <= 0:
		raise ValueError("principal must be > 0"       )
	return principal * daily_interest_rate * days_between_payments







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							float					,  snake_case_ :							float					,  snake_case_ :							float					,  ):
	if number_of_compounding_periods <= 0:
		raise ValueError("number_of_compounding_periods must be > 0"       )
	if nominal_annual_interest_rate_percentage < 0:
		raise ValueError("nominal_annual_interest_rate_percentage must be >= 0"       )
	if principal <= 0:
		raise ValueError("principal must be > 0"       )

	return principal * (
	    (1 + nominal_annual_interest_rate_percentage) ** number_of_compounding_periods
	    - 1
	)







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							float					,  snake_case_ :							float					,  snake_case_ :							float					,  ):
	if number_of_years <= 0:
		raise ValueError("number_of_years must be > 0"       )
	if nominal_annual_percentage_rate < 0:
		raise ValueError("nominal_annual_percentage_rate must be >= 0"       )
	if principal <= 0:
		raise ValueError("principal must be > 0"       )

	return compound_interest(
	    snake_case_					,  nominal_annual_percentage_rate / 365					,  number_of_years * 365       )


if __name__ == "__main__":
	import doctest

	doctest.testmod()



