
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							=       """bert-base-cased"""
__lowerCamelCase							=       """fp16"""
__lowerCamelCase							=       """bf16"""
__lowerCamelCase							=       [FPaa, BFaa]






@require_fsdp
@require_cuda
class   UpperCAmelCase		(						A_ ):



   def 				_SCREAMING_SNAKE_CASE   (self		:					Tuple )		->	int:


          '''simple docstring'''


          super().setUp()

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



   def 				_SCREAMING_SNAKE_CASE   (self		:					int )		->	Optional[int]:


          '''simple docstring'''


          from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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



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


          '''simple docstring'''


          from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

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



   def 				_SCREAMING_SNAKE_CASE   (self		:					List[str] )		->	Tuple:


          '''simple docstring'''


          from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

          for i, state_dict_type in enumerate(snake_case__ ):
                 snake_case       :   int							=      self.dist_env.copy()
                 snake_case       :   Any							=      state_dict_type
                 with mockenv_context(**snake_case__ ):
                        snake_case       :   Union[str, Any]							=      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 				_SCREAMING_SNAKE_CASE   (self		:					str )		->	List[str]:


          '''simple docstring'''


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

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

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



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[int] )		->	Optional[Any]:


          '''simple docstring'''


          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       :   List[str]							=      self.dist_env.copy()
                 snake_case       :   Optional[Any]							=      mp_dtype
                 with mockenv_context(**snake_case__ ):
                        snake_case       :   int							=      Accelerator()
                        if mp_dtype == "fp16":
                               snake_case       :   int							=      torch.floataa
                        elif mp_dtype == "bf16":
                               snake_case       :   Tuple							=      torch.bfloataa
                        snake_case       :   Dict							=      MixedPrecision(param_dtype=snake_case__    ,						reduce_dtype=snake_case__    ,						buffer_dtype=snake_case__ )
                        self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy    ,						snake_case__ )
                        if mp_dtype == FPaa:
                               self.assertTrue(isinstance(accelerator.scaler    ,						snake_case__ ) )
                        elif mp_dtype == BFaa:
                               self.assertIsNone(accelerator.scaler )
                        AcceleratorState._reset_state(snake_case__ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Any )		->	int:


          '''simple docstring'''


          from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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






@require_fsdp
@require_multi_gpu
@slow
class   UpperCAmelCase		(						A_ ):



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


          '''simple docstring'''


          super().setUp()
          snake_case       :   int							=      0.82
          snake_case       :   Optional[int]							=      [
              "fsdp_shard_grad_op_transformer_based_wrap",
              "fsdp_full_shard_transformer_based_wrap",
          ]
          snake_case       :   List[str]							=      {
              "multi_gpu_fp16": 32_00,
              "fsdp_shard_grad_op_transformer_based_wrap_fp16": 20_00,
              "fsdp_full_shard_transformer_based_wrap_fp16": 19_00,
              # 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       :   Tuple							=      1_60
          snake_case       :   List[str]							=      1_60

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



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


          '''simple docstring'''


          snake_case       :   Tuple							=      os.path.join(self.test_scripts_folder    ,						"test_performance.py" )
          snake_case       :   Optional[Any]							=      ["accelerate", "launch", "--num_processes=2", "--num_machines=1", "--machine_rank=0", "--use_fsdp"]
          for config in self.performance_configs:
                 snake_case       :   Optional[int]							=      cmd.copy()
                 for i, strategy in enumerate(snake_case__ ):
                        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(snake_case__    ,						env=os.environ.copy() )



   def 				_SCREAMING_SNAKE_CASE   (self		:					str )		->	Optional[Any]:


          '''simple docstring'''


          snake_case       :   Union[str, Any]							=      os.path.join(self.test_scripts_folder    ,						"test_checkpointing.py" )
          snake_case       :   str							=      [
              "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(snake_case__ ):
                 snake_case       :   List[str]							=      cmd.copy()
                 cmd_config.append(f"""--fsdp_sharding_strategy={i+1}""" )
                 if strategy != "FULL_SHARD":
                        continue
                 snake_case       :   List[Any]							=      len(snake_case__ )
                 for state_dict_type in FSDP_STATE_DICT_TYPE:
                        snake_case       :   Optional[int]							=      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(snake_case__    ,						env=os.environ.copy() )

                        snake_case       :   Optional[Any]							=      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(snake_case__    ,						env=os.environ.copy() )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[Any] )		->	str:


          '''simple docstring'''


          snake_case       :   Optional[int]							=      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       :   Optional[int]							=      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(snake_case__ ):
                               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(snake_case__    ,						env=os.environ.copy() )







def 			UpperCamelCase					(    __lowerCamelCase     : str				,					__lowerCamelCase     : int    ):
       snake_case       :   list[list[str]]							=      [[] for _ in range(__lowerCamelCase    )]
       snake_case       :   int							=      key - 1

       if key <= 0:
              raise ValueError("Height of grid can't be 0 or negative"    )
       if key == 1 or len(__lowerCamelCase    ) <= key:
              return input_string

       for position, character in enumerate(__lowerCamelCase    ):
              snake_case       :   Any							=      position % (lowest * 2)  # puts it in bounds
              snake_case       :   Optional[int]							=      min(__lowerCamelCase				,					lowest * 2 - num    )  # creates zigzag pattern
              temp_grid[num].append(__lowerCamelCase    )
       snake_case       :   List[str]							=      ["".join(__lowerCamelCase    ) for row in temp_grid]
       snake_case       :   Tuple							=      "".join(__lowerCamelCase    )

       return output_string


def 			UpperCamelCase					(    __lowerCamelCase     : str				,					__lowerCamelCase     : int    ):
       snake_case       :   Dict							=      []
       snake_case       :   Union[str, Any]							=      key - 1

       if key <= 0:
              raise ValueError("Height of grid can't be 0 or negative"    )
       if key == 1:
              return input_string

       snake_case       :   list[list[str]]							=      [[] for _ in range(__lowerCamelCase    )]  # generates template
       for position in range(len(__lowerCamelCase    )    ):
              snake_case       :   List[str]							=      position % (lowest * 2)  # puts it in bounds
              snake_case       :   Optional[int]							=      min(__lowerCamelCase				,					lowest * 2 - num    )  # creates zigzag pattern
              temp_grid[num].append("*"    )

       snake_case       :   Tuple							=      0
       for row in temp_grid:  # fills in the characters
              snake_case       :   Union[str, Any]							=      input_string[counter : counter + len(__lowerCamelCase    )]
              grid.append(list(__lowerCamelCase    )    )
              counter += len(__lowerCamelCase    )

       snake_case       :   str							=      ""  # reads as zigzag
       for position in range(len(__lowerCamelCase    )    ):
              snake_case       :   Optional[int]							=      position % (lowest * 2)  # puts it in bounds
              snake_case       :   Tuple							=      min(__lowerCamelCase				,					lowest * 2 - num    )  # creates zigzag pattern
              output_string += grid[num][0]
              grid[num].pop(0    )
       return output_string


def 			UpperCamelCase					(    __lowerCamelCase     : str    ):
       snake_case       :   Tuple							=      {}
       for key_guess in range(1				,					len(__lowerCamelCase    )    ):  # tries every key
              snake_case       :   Any							=      decrypt(__lowerCamelCase				,					__lowerCamelCase    )
       return results


if __name__ == "__main__":
   import doctest

   doctest.testmod()





