"""simple docstring"""




from .constants import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SAFE_WEIGHTS_INDEX_NAME,
    SAFE_WEIGHTS_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    TORCH_LAUNCH_PARAMS,
    WEIGHTS_INDEX_NAME,
    WEIGHTS_NAME,
)
from .dataclasses import (
    BnbQuantizationConfig,
    ComputeEnvironment,
    CustomDtype,
    DeepSpeedPlugin,
    DistributedDataParallelKwargs,
    DistributedType,
    DynamoBackend,
    FPaRecipeKwargs,
    FullyShardedDataParallelPlugin,
    GradientAccumulationPlugin,
    GradScalerKwargs,
    InitProcessGroupKwargs,
    KwargsHandler,
    LoggerType,
    MegatronLMPlugin,
    PrecisionType,
    ProjectConfiguration,
    RNGType,
    SageMakerDistributedType,
    TensorInformation,
    TorchDynamoPlugin,
)
from .environment import get_int_from_env, parse_choice_from_env, parse_flag_from_env
from .imports import (
    get_ccl_version,
    is_abit_bnb_available,
    is_abit_bnb_available,
    is_aim_available,
    is_bfaa_available,
    is_bnb_available,
    is_botoa_available,
    is_ccl_available,
    is_comet_ml_available,
    is_datasets_available,
    is_deepspeed_available,
    is_fpa_available,
    is_ipex_available,
    is_megatron_lm_available,
    is_mlflow_available,
    is_mps_available,
    is_npu_available,
    is_rich_available,
    is_safetensors_available,
    is_sagemaker_available,
    is_tensorboard_available,
    is_tpu_available,
    is_transformers_available,
    is_wandb_available,
    is_xpu_available,
)
from .modeling import (
    check_device_map,
    check_tied_parameters_in_config,
    check_tied_parameters_on_same_device,
    compute_module_sizes,
    convert_file_size_to_int,
    dtype_byte_size,
    find_tied_parameters,
    get_balanced_memory,
    get_max_layer_size,
    get_max_memory,
    get_mixed_precision_context_manager,
    id_tensor_storage,
    infer_auto_device_map,
    load_checkpoint_in_model,
    load_offloaded_weights,
    load_state_dict,
    named_module_tensors,
    retie_parameters,
    set_module_tensor_to_device,
    shard_checkpoint,
)
from .offload import (
    OffloadedWeightsLoader,
    PrefixedDataset,
    extract_submodules_state_dict,
    load_offloaded_weight,
    offload_state_dict,
    offload_weight,
    save_offload_index,
)
from .operations import (
    broadcast,
    broadcast_object_list,
    concatenate,
    convert_outputs_to_fpaa,
    convert_to_fpaa,
    find_batch_size,
    find_device,
    gather,
    gather_object,
    get_data_structure,
    honor_type,
    initialize_tensors,
    is_namedtuple,
    is_tensor_information,
    is_torch_tensor,
    listify,
    pad_across_processes,
    recursively_apply,
    reduce,
    send_to_device,
    slice_tensors,
)
from .versions import compare_versions, is_torch_version


if is_deepspeed_available():
     from .deepspeed import (
         DeepSpeedEngineWrapper,
         DeepSpeedOptimizerWrapper,
         DeepSpeedSchedulerWrapper,
         DummyOptim,
         DummyScheduler,
         HfDeepSpeedConfig,
     )

from .bnb import has_abit_bnb_layers, load_and_quantize_model
from .fsdp_utils import load_fsdp_model, load_fsdp_optimizer, save_fsdp_model, save_fsdp_optimizer
from .launch import (
    PrepareForLaunch,
    _filter_args,
    prepare_deepspeed_cmd_env,
    prepare_multi_gpu_env,
    prepare_sagemager_args_inputs,
    prepare_simple_launcher_cmd_env,
    prepare_tpu,
)
from .megatron_lm import (
    AbstractTrainStep,
    BertTrainStep,
    GPTTrainStep,
    MegatronEngine,
    MegatronLMDummyDataLoader,
    MegatronLMDummyScheduler,
    MegatronLMOptimizerWrapper,
    MegatronLMSchedulerWrapper,
    TaTrainStep,
    avg_losses_across_data_parallel_group,
    gather_across_data_parallel_groups,
)
from .megatron_lm import initialize as megatron_lm_initialize
from .megatron_lm import prepare_data_loader as megatron_lm_prepare_data_loader
from .megatron_lm import prepare_model as megatron_lm_prepare_model
from .megatron_lm import prepare_optimizer as megatron_lm_prepare_optimizer
from .megatron_lm import prepare_scheduler as megatron_lm_prepare_scheduler
from .memory import find_executable_batch_size, release_memory
from .other import (
    extract_model_from_parallel,
    get_pretty_name,
    is_port_in_use,
    merge_dicts,
    patch_environment,
    save,
    wait_for_everyone,
    write_basic_config,
)
from .random import set_seed, synchronize_rng_state, synchronize_rng_states
from .torch_xla import install_xla
from .tqdm import tqdm
from .transformer_engine import convert_model, has_transformer_engine_layers





"""simple docstring"""




import json
import os
import tempfile
from unittest.mock import patch

import torch
from torch.utils.data import DataLoader, TensorDataset

from accelerate import DistributedType, infer_auto_device_map, init_empty_weights
from accelerate.accelerator import Accelerator
from accelerate.state import GradientState, PartialState
from accelerate.test_utils import require_bnb, require_multi_gpu, slow
from accelerate.test_utils.testing import AccelerateTestCase, require_cuda
from accelerate.utils import patch_environment





def 							SCREAMING_SNAKE_CASE					(      )	->				Tuple:
  _lowerCAmelCase						:	Tuple        =					torch.nn.Linear(2      ,4     )
  _lowerCAmelCase						:	Union[str, Any]        =					torch.optim.AdamW(model.parameters()      ,lr=1.0     )
  _lowerCAmelCase						:	Tuple        =					torch.optim.lr_scheduler.OneCycleLR(_lowerCamelCase      ,max_lr=0.01      ,steps_per_epoch=2      ,epochs=1     )
  _lowerCAmelCase						:	Tuple        =					DataLoader(TensorDataset(torch.tensor([1, 2, 3]     )     )     )
  _lowerCAmelCase						:	List[Any]        =					DataLoader(TensorDataset(torch.tensor([4, 5, 6]     )     )     )

  return model, optimizer, scheduler, train_dl, valid_dl





def 							SCREAMING_SNAKE_CASE					(      _lowerCamelCase						:	Any     )	->				int:
  return (model.weight.abs().sum() + model.bias.abs().sum()).item()





def 							SCREAMING_SNAKE_CASE					(      _lowerCamelCase						:	List[str]     )	->				Any:
  _lowerCAmelCase						:	List[str]        =					torch.nn.Linear(*tuple(model.weight.T.shape     )     ).state_dict()
  model.load_state_dict(_lowerCamelCase     )






class 						__A							(			SCREAMING_SNAKE_CASE_		):






     @require_cuda
     def    __A  (		self  ):
       _lowerCAmelCase						:	Union[str, Any]        =					Accelerator()
       assert PartialState._shared_state["_cpu"] is False
       assert PartialState._shared_state["device"].type == "cuda"
       with self.assertRaises(a__  ):
         _lowerCAmelCase						:	Tuple        =					Accelerator(cpu=a__  )






     def    __A  (		self  ):
       _lowerCAmelCase						:	Dict        =					Accelerator()
       _lowerCAmelCase						:	Any        =					GradientState()
       assert state.num_steps == 1
       _lowerCAmelCase						:	Optional[int]        =					4
       assert state.num_steps == 4

       assert state.sync_gradients is True
       _lowerCAmelCase						:	Dict        =					False
       assert state.sync_gradients is False
       GradientState._reset_state()






     def    __A  (		self  ):
       _lowerCAmelCase						:	Optional[int]        =					Accelerator()
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	Any        =					create_components()

       (
           (
           _lowerCAmelCase
       )      ,     (
           _lowerCAmelCase
       )      ,     (
           _lowerCAmelCase
       )      ,     (
           _lowerCAmelCase
       )      ,     (
           _lowerCAmelCase
       )      ,     
       )						:	int        =					accelerator.prepare(a__    ,					a__    ,					a__    ,					a__    ,					a__  )

       self.assertTrue(prepared_model in accelerator._models  )
       self.assertTrue(prepared_optimizer in accelerator._optimizers  )
       self.assertTrue(prepared_scheduler in accelerator._schedulers  )
       self.assertTrue(prepared_train_dl in accelerator._dataloaders  )
       self.assertTrue(prepared_valid_dl in accelerator._dataloaders  )






     def    __A  (		self  ):
       _lowerCAmelCase						:	Optional[Any]        =					Accelerator()
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	List[Any]        =					create_components()
       accelerator.prepare(a__    ,					a__    ,					a__    ,					a__    ,					a__  )
       accelerator.free_memory()

       self.assertTrue(len(accelerator._models  ) == 0  )
       self.assertTrue(len(accelerator._optimizers  ) == 0  )
       self.assertTrue(len(accelerator._schedulers  ) == 0  )
       self.assertTrue(len(accelerator._dataloaders  ) == 0  )






     def    __A  (		self  ):
       PartialState._reset_state()

       # Mock torch.cuda.set_device to avoid an exception as the device doesn't exist
       def noop(*a__    ,					**a__  ):
         pass

       with patch("""torch.cuda.set_device"""    ,					a__  ), patch_environment(ACCELERATE_TORCH_DEVICE="""cuda:64"""  ):
         _lowerCAmelCase						:	Dict        =					Accelerator()
         self.assertEqual(str(accelerator.state.device  )    ,					"""cuda:64"""  )






     def    __A  (		self  ):
       _lowerCAmelCase						:	Any        =					Accelerator()
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	List[str]        =					create_components()
       accelerator.prepare(a__    ,					a__    ,					a__    ,					a__    ,					a__  )

       _lowerCAmelCase						:	List[Any]        =					get_signature(a__  )

       with tempfile.TemporaryDirectory() as tmpdirname:
         accelerator.save_state(a__  )

         # make sure random weights don't match
         load_random_weights(a__  )
         self.assertTrue(abs(model_signature - get_signature(a__  )  ) > 1e-3  )

         # make sure loaded weights match
         accelerator.load_state(a__  )
         self.assertTrue(abs(model_signature - get_signature(a__  )  ) < 1e-3  )






     def    __A  (		self  ):
       _lowerCAmelCase						:	str        =					Accelerator()
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	str        =					create_components()
       accelerator.prepare(a__    ,					a__    ,					a__    ,					a__    ,					a__  )

       _lowerCAmelCase						:	Optional[Any]        =					get_signature(a__  )

       # saving hook
       def save_config(a__    ,					a__    ,					a__  ):
         _lowerCAmelCase						:	Dict        =					{"""class_name""": models[0].__class__.__name__}

         with open(os.path.join(a__    ,					"""data.json"""  )    ,					"""w"""  ) as f:
           json.dump(a__    ,					a__  )

        # loading hook
       def load_config(a__    ,					a__  ):
         with open(os.path.join(a__    ,					"""data.json"""  )    ,					"""r"""  ) as f:
           _lowerCAmelCase						:	int        =					json.load(a__  )

         _lowerCAmelCase						:	str        =					config["""class_name"""]

       _lowerCAmelCase						:	Union[str, Any]        =					accelerator.register_save_state_pre_hook(a__  )
       _lowerCAmelCase						:	int        =					accelerator.register_load_state_pre_hook(a__  )

       with tempfile.TemporaryDirectory() as tmpdirname:
         accelerator.save_state(a__  )

         # make sure random weights don't match with hooks
         load_random_weights(a__  )
         self.assertTrue(abs(model_signature - get_signature(a__  )  ) > 1e-3  )

         # random class name to verify correct one is loaded
         _lowerCAmelCase						:	Dict        =					"""random"""

         # make sure loaded weights match with hooks
         accelerator.load_state(a__  )
         self.assertTrue(abs(model_signature - get_signature(a__  )  ) < 1e-3  )

         # mode.class_name is loaded from config
         self.assertTrue(model.class_name == model.__class__.__name__  )

       # remove hooks
       save_hook.remove()
       load_hook.remove()

       with tempfile.TemporaryDirectory() as tmpdirname:
         accelerator.save_state(a__  )

         # make sure random weights don't match with hooks removed
         load_random_weights(a__  )
         self.assertTrue(abs(model_signature - get_signature(a__  )  ) > 1e-3  )

         # random class name to verify correct one is loaded
         _lowerCAmelCase						:	Optional[Any]        =					"""random"""

         # make sure loaded weights match with hooks removed
         accelerator.load_state(a__  )
         self.assertTrue(abs(model_signature - get_signature(a__  )  ) < 1e-3  )

         # mode.class_name is NOT loaded from config
         self.assertTrue(model.class_name != model.__class__.__name__  )






     def    __A  (		self  ):
       _lowerCAmelCase						:	Optional[Any]        =					Accelerator()
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	Any        =					create_components()
       _lowerCAmelCase						:	Any        =					None

       # This should work
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	int        =					accelerator.prepare(
           a__    ,					a__    ,					a__    ,					a__    ,					a__    ,					a__  )
       self.assertTrue(dummy_obj is None  )






     def    __A  (		self  ):
       _lowerCAmelCase						:	str        =					Accelerator()
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	Any        =					create_components()
       _lowerCAmelCase						:	Optional[int]        =					[1, 2, 3]

       # This should work
       _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase						:	str        =					accelerator.prepare(
           a__    ,					a__    ,					a__    ,					a__    ,					a__    ,					a__  )
       self.assertEqual(
           getattr(a__    ,					"""_is_accelerate_prepared"""    ,					a__  )    ,					a__    ,					"""Dummy object should have `_is_accelerate_prepared` set to `True`"""    ,					)
       self.assertEqual(
           getattr(a__    ,					"""_is_accelerate_prepared"""    ,					a__  )    ,					a__    ,					"""Model is missing `_is_accelerator_prepared` or is set to `False`"""    ,					)
       self.assertEqual(
           getattr(a__    ,					"""_is_accelerate_prepared"""    ,					a__  )    ,					a__    ,					"""Optimizer is missing `_is_accelerator_prepared` or is set to `False`"""    ,					)
       self.assertEqual(
           getattr(a__    ,					"""_is_accelerate_prepared"""    ,					a__  )    ,					a__    ,					"""Scheduler is missing `_is_accelerator_prepared` or is set to `False`"""    ,					)
       self.assertEqual(
           getattr(a__    ,					"""_is_accelerate_prepared"""    ,					a__  )    ,					a__    ,					"""Train Dataloader is missing `_is_accelerator_prepared` or is set to `False`"""    ,					)
       self.assertEqual(
           getattr(a__    ,					"""_is_accelerate_prepared"""    ,					a__  )    ,					a__    ,					"""Valid Dataloader is missing `_is_accelerator_prepared` or is set to `False`"""    ,					)






     @slow
     @require_bnb
     def    __A  (		self  ):
       from transformers import AutoModelForCausalLM

       _lowerCAmelCase						:	List[str]        =					AutoModelForCausalLM.from_pretrained(
           """EleutherAI/gpt-neo-125m"""    ,					load_in_abit=a__    ,					device_map={"""""": 0}    ,					)
       _lowerCAmelCase						:	List[str]        =					Accelerator()

       # This should work
       _lowerCAmelCase						:	List[Any]        =					accelerator.prepare(a__  )






     @slow
     @require_bnb
     def    __A  (		self  ):
       from transformers import AutoModelForCausalLM

       _lowerCAmelCase						:	Any        =					Accelerator()

       with init_empty_weights():
         _lowerCAmelCase						:	Dict        =					AutoModelForCausalLM.from_pretrained(
             """EleutherAI/gpt-neo-125m"""    ,					)
         model.tie_weights()
         _lowerCAmelCase						:	int        =					infer_auto_device_map(a__  )
         _lowerCAmelCase						:	Optional[Any]        =					"""cpu"""

       _lowerCAmelCase						:	Dict        =					AutoModelForCausalLM.from_pretrained(
           """EleutherAI/gpt-neo-125m"""    ,					device_map=a__    ,					load_in_abit=a__    ,					llm_inta_enable_fpaa_cpu_offload=a__  )

       # This should not work and get value error
       with self.assertRaises(a__  ):
         _lowerCAmelCase						:	List[str]        =					accelerator.prepare(a__  )






     @slow
     @require_bnb
     @require_multi_gpu
     def    __A  (		self  ):
       from transformers import AutoModelForCausalLM

       _lowerCAmelCase						:	Dict        =					{"""distributed_type""": DistributedType.MULTI_GPU}

       with init_empty_weights():
         _lowerCAmelCase						:	Optional[Any]        =					AutoModelForCausalLM.from_pretrained(
             """EleutherAI/gpt-neo-125m"""    ,					)
         model.tie_weights()
         _lowerCAmelCase						:	List[str]        =					infer_auto_device_map(a__  )
         _lowerCAmelCase						:	Union[str, Any]        =					1

       _lowerCAmelCase						:	int        =					AutoModelForCausalLM.from_pretrained(
           """EleutherAI/gpt-neo-125m"""    ,					load_in_abit=a__    ,					device_map=a__    ,					)
       _lowerCAmelCase						:	Tuple        =					Accelerator()

       # This should not work and get value error
       with self.assertRaises(a__  ):
         _lowerCAmelCase						:	Optional[int]        =					accelerator.prepare(a__  )

       PartialState._reset_state()






     @slow
     @require_bnb
     @require_multi_gpu
     def    __A  (		self  ):
       from transformers import AutoModelForCausalLM

       with init_empty_weights():
         _lowerCAmelCase						:	Dict        =					AutoModelForCausalLM.from_pretrained(
             """EleutherAI/gpt-neo-125m"""    ,					)
         _lowerCAmelCase						:	int        =					infer_auto_device_map(a__  )
         _lowerCAmelCase						:	List[Any]        =					1

       _lowerCAmelCase						:	Optional[Any]        =					AutoModelForCausalLM.from_pretrained(
           """EleutherAI/gpt-neo-125m"""    ,					load_in_abit=a__    ,					device_map=a__    ,					)
       _lowerCAmelCase						:	str        =					Accelerator()

       # This should work
       _lowerCAmelCase						:	str        =					accelerator.prepare(a__  )






     @require_cuda
     def    __A  (		self  ):
       _lowerCAmelCase						:	Union[str, Any]        =					torch.nn.Linear(10    ,					10  )
       _lowerCAmelCase						:	Any        =					torch.optim.SGD(model.parameters()    ,					lr=0.0_1  )
       _lowerCAmelCase						:	List[str]        =					Accelerator(cpu=a__  )
       _lowerCAmelCase						:	Tuple        =					accelerator.prepare(a__  )




