
"""simple docstring"""






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)

__SCREAMING_SNAKE_CASE							=     'bert-base-cased'
__SCREAMING_SNAKE_CASE							=     'fp16'
__SCREAMING_SNAKE_CASE							=     'bf16'
__SCREAMING_SNAKE_CASE							=     [FPaa, BFaa]





@require_fsdp
@require_cuda
class 				a__					(						_UpperCamelCase       ):



     def  lowerCamelCase_						(							self :Tuple   ):







      '''simple docstring'''



      super().setUp()

      UpperCamelCase_     :					List[str]   	=dict(
          ACCELERATE_USE_FSDP='true'      ,  MASTER_ADDR='localhost'      ,  MASTER_PORT='10999'      ,  RANK='0'      ,  LOCAL_RANK='0'      ,  WORLD_SIZE='1'      ,  )



     def  lowerCamelCase_						(							self :Union[str, Any]   ):







      '''simple docstring'''



      from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

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



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

      for i, prefetch_policy in enumerate(lowercase__   ):
       UpperCamelCase_     :					int   	=self.dist_env.copy()
       UpperCamelCase_     :					Tuple   	=prefetch_policy
       with mockenv_context(**lowercase__   ):
        UpperCamelCase_     :					Tuple   	=FullyShardedDataParallelPlugin()
        if prefetch_policy == "NO_PREFETCH":
         self.assertIsNone(fsdp_plugin.backward_prefetch   )
        else:
         self.assertEqual(fsdp_plugin.backward_prefetch      ,  BackwardPrefetch(i + 1   )   )



     def  lowerCamelCase_						(							self :int   ):







      '''simple docstring'''



      from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

      for i, state_dict_type in enumerate(lowercase__   ):
       UpperCamelCase_     :					Optional[Any]   	=self.dist_env.copy()
       UpperCamelCase_     :					List[str]   	=state_dict_type
       with mockenv_context(**lowercase__   ):
        UpperCamelCase_     :					List[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  lowerCamelCase_						(							self :Optional[Any]   ):







      '''simple docstring'''



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

      UpperCamelCase_     :					Dict   	=self.dist_env.copy()
      UpperCamelCase_     :					Optional[Any]   	='''TRANSFORMER_BASED_WRAP'''
      UpperCamelCase_     :					Tuple   	='''T5Layer'''
      with mockenv_context(**lowercase__   ):
       UpperCamelCase_     :					Any   	=FullyShardedDataParallelPlugin()
       with self.assertRaises(lowercase__   ) as cm:
        fsdp_plugin.set_auto_wrap_policy(lowercase__   )
       self.assertTrue('Could not find the transformer layer class to wrap in the model.' in str(cm.exception   )   )

      UpperCamelCase_     :					Dict   	=self.dist_env.copy()
      UpperCamelCase_     :					Optional[Any]   	='''SIZE_BASED_WRAP'''
      UpperCamelCase_     :					int   	='''0'''
      with mockenv_context(**lowercase__   ):
       UpperCamelCase_     :					List[Any]   	=FullyShardedDataParallelPlugin()
       fsdp_plugin.set_auto_wrap_policy(lowercase__   )
       self.assertIsNone(fsdp_plugin.auto_wrap_policy   )



     def  lowerCamelCase_						(							self :Tuple   ):







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




     def  lowerCamelCase_						(							self :List[str]   ):







      '''simple docstring'''



      from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

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





@require_fsdp
@require_multi_gpu
@slow
class 				a__					(						_UpperCamelCase       ):



     def  lowerCamelCase_						(							self :str   ):







      '''simple docstring'''



      super().setUp()
      UpperCamelCase_     :					List[Any]   	=0.82
      UpperCamelCase_     :					int   	=[
          '''fsdp_shard_grad_op_transformer_based_wrap''',
          '''fsdp_full_shard_transformer_based_wrap''',
      ]
      UpperCamelCase_     :					Any   	={
          '''multi_gpu_fp16''': 3_200,
          '''fsdp_shard_grad_op_transformer_based_wrap_fp16''': 2_000,
          '''fsdp_full_shard_transformer_based_wrap_fp16''': 1_900,
          # 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
      }
      UpperCamelCase_     :					Tuple   	=160
      UpperCamelCase_     :					int   	=160

      UpperCamelCase_     :					str   	=inspect.getfile(accelerate.test_utils   )
      UpperCamelCase_     :					int   	=os.path.sep.join(mod_file.split(os.path.sep   )[:-1] + ['scripts', 'external_deps']   )



     def  lowerCamelCase_						(							self :Optional[Any]   ):







      '''simple docstring'''



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



     def  lowerCamelCase_						(							self :List[Any]   ):







      '''simple docstring'''



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

        UpperCamelCase_     :					str   	=cmd_config[:-1]
        UpperCamelCase_     :					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(lowercase__      ,  env=os.environ.copy()   )




     def  lowerCamelCase_						(							self :Tuple   ):







      '''simple docstring'''



      UpperCamelCase_     :					Union[str, Any]   	=os.path.join(self.test_scripts_folder      ,  'test_peak_memory_usage.py'   )
      UpperCamelCase_     :					int   	=[
          '''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():
       UpperCamelCase_     :					List[str]   	=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(lowercase__   ):
         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(lowercase__      ,  env=os.environ.copy()   )




import importlib

import torch
import yaml
from omegaconf import OmegaConf
from taming.models.vqgan import VQModel




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_=False      )		->     Optional[Any]:




						'''simple docstring'''







						__UpperCAmelCase			:							Optional[Any]	    =	OmegaConf.load(lowercase_      )
						if display:
												print(yaml.dump(OmegaConf.to_container(lowercase_      )      )      )
						return config




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_=None      ,       lowercase_=None      )		->     Union[str, Any]:




						'''simple docstring'''







						if conf_path is None:
												__UpperCAmelCase			:							Optional[int]	    =	'''./model_checkpoints/vqgan_only.yaml'''
						__UpperCAmelCase			:							Dict	    =	load_config(lowercase_      ,       display=lowercase_      )
						__UpperCAmelCase			:							Tuple	    =	VQModel(**config.model.params      )
						if ckpt_path is None:
												__UpperCAmelCase			:							Dict	    =	'''./model_checkpoints/vqgan_only.pt'''
						__UpperCAmelCase			:							int	    =	torch.load(lowercase_      ,       map_location=lowercase_      )
						if ".ckpt" in ckpt_path:
												__UpperCAmelCase			:							Union[str, Any]	    =	sd['''state_dict''']
						model.load_state_dict(lowercase_      ,       strict=lowercase_      )
						model.to(lowercase_      )
						del sd
						return model




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_      )		->     List[Any]:




						'''simple docstring'''







						__UpperCAmelCase    ,      __UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[Any]	    =	model.encode(lowercase_      )
						print(f"VQGAN --- {model.__class__.__name__}: latent shape: {z.shape[2:]}"      )
						__UpperCAmelCase			:							List[str]	    =	model.decode(lowercase_      )
						return xrec




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_=False      )		->     int:




						'''simple docstring'''







						__UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[int]	    =	string.rsplit('''.'''      ,       1      )
						if reload:
												__UpperCAmelCase			:							Optional[int]	    =	importlib.import_module(lowercase_      )
												importlib.reload(lowercase_      )
						return getattr(importlib.import_module(lowercase_      ,       package=lowercase_      )      ,       cls      )




def     __SCREAMING_SNAKE_CASE							(       lowercase_      )		->     Union[str, Any]:




						'''simple docstring'''







						if "target" not in config:
												raise KeyError('''Expected key `target` to instantiate.'''      )
						return get_obj_from_str(config['''target''']      )(**config.get('''params'''      ,       {}      )      )




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_      ,       lowercase_=True      ,       lowercase_=True      )		->     int:




						'''simple docstring'''







						__UpperCAmelCase			:							Dict	    =	instantiate_from_config(lowercase_      )
						if sd is not None:
												model.load_state_dict(lowercase_      )
						if gpu:
												model.cuda()
						if eval_mode:
												model.eval()
						return {"model": model}





def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_      ,       lowercase_      ,       lowercase_      )		->     Optional[Any]:




						'''simple docstring'''







						if ckpt:
												__UpperCAmelCase			:							Optional[int]	    =	torch.load(lowercase_      ,       map_location='''cpu'''      )
												__UpperCAmelCase			:							Union[str, Any]	    =	pl_sd['''global_step''']
												print(f"loaded model from global step {global_step}."      )
						else:
												__UpperCAmelCase			:							Union[str, Any]	    =	{'''state_dict''': None}
												__UpperCAmelCase			:							Optional[int]	    =	None
						__UpperCAmelCase			:							str	    =	load_model_from_config(config.model      ,       pl_sd['''state_dict''']      ,       gpu=lowercase_      ,       eval_mode=lowercase_      )['''model''']
						return model, global_step

