


'''simple docstring'''
import os

import torch

from ..logging import get_logger
from .constants import FSDP_PYTORCH_VERSION, MODEL_NAME, OPTIMIZER_NAME
from .versions import is_torch_version


if is_torch_version('>=', FSDP_PYTORCH_VERSION):
 import torch.distributed.checkpoint as dist_cp
 from torch.distributed.checkpoint.default_planner import DefaultLoadPlanner, DefaultSavePlanner
 from torch.distributed.checkpoint.optimizer import load_sharded_optimizer_state_dict
 from torch.distributed.fsdp.fully_sharded_data_parallel import FullyShardedDataParallel as FSDP
 from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType


UpperCAmelCase_			:  Optional[int]										=   get_logger(__name__)


def       _lowercase							(					UpperCamelCase__							:       int,					UpperCamelCase__							:       Optional[int],					UpperCamelCase__							:       Optional[Any],					UpperCamelCase__							:       Dict,					UpperCamelCase__							:       List[str]=0      ):
 os.makedirs(UpperCamelCase__,					exist_ok=UpperCamelCase__      )
 with FSDP.state_dict_type(
     UpperCamelCase__,					fsdp_plugin.state_dict_type,					fsdp_plugin.state_dict_config,					fsdp_plugin.optim_state_dict_config      ):
  __A	:		Any	  =      model.state_dict()
  if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
   __A	:		Dict	  =      f"""{MODEL_NAME}.bin""" if model_index == 0 else f"""{MODEL_NAME}_{model_index}.bin"""
   __A	:		str	  =      os.path.join(UpperCamelCase__,					UpperCamelCase__      )
   if accelerator.process_index == 0:
    logger.info(f"""Saving model to {output_model_file}"""      )
    torch.save(UpperCamelCase__,					UpperCamelCase__      )
    logger.info(f"""Model saved to {output_model_file}"""      )
  elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
   __A	:		Optional[Any]	  =      (
       f"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
       if model_index == 0
       else f"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
   )
   __A	:		Tuple	  =      os.path.join(UpperCamelCase__,					UpperCamelCase__      )
   logger.info(f"""Saving model to {output_model_file}"""      )
   torch.save(UpperCamelCase__,					UpperCamelCase__      )
   logger.info(f"""Model saved to {output_model_file}"""      )
  elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
   __A	:		Optional[Any]	  =      os.path.join(UpperCamelCase__,					f"""{MODEL_NAME}_{model_index}"""      )
   os.makedirs(UpperCamelCase__,					exist_ok=UpperCamelCase__      )
   logger.info(f"""Saving model to {ckpt_dir}"""      )
   __A	:		Any	  =      {'model': state_dict}

   dist_cp.save_state_dict(
       state_dict=UpperCamelCase__,					storage_writer=dist_cp.FileSystemWriter(UpperCamelCase__      ),					planner=DefaultSavePlanner(),					)
   logger.info(f"""Model saved to {ckpt_dir}"""      )


def       _lowercase							(					UpperCamelCase__							:       int,					UpperCamelCase__							:       List[str],					UpperCamelCase__							:       Union[str, Any],					UpperCamelCase__							:       Dict,					UpperCamelCase__							:       Dict=0      ):
 accelerator.wait_for_everyone()
 with FSDP.state_dict_type(
     UpperCamelCase__,					fsdp_plugin.state_dict_type,					fsdp_plugin.state_dict_config,					fsdp_plugin.optim_state_dict_config      ):
  if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
   if type(UpperCamelCase__      ) != FSDP and accelerator.process_index != 0:
    if not fsdp_plugin.sync_module_states:
     raise ValueError(
         'Set the `sync_module_states` flag to `True` so that model states are synced across processes when '
         'initializing FSDP object'      )
    return
   __A	:		Dict	  =      f"""{MODEL_NAME}.bin""" if model_index == 0 else f"""{MODEL_NAME}_{model_index}.bin"""
   __A	:		Any	  =      os.path.join(UpperCamelCase__,					UpperCamelCase__      )
   logger.info(f"""Loading model from {input_model_file}"""      )
   __A	:		List[str]	  =      torch.load(UpperCamelCase__      )
   logger.info(f"""Model loaded from {input_model_file}"""      )
  elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
   __A	:		List[str]	  =      (
       f"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
       if model_index == 0
       else f"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
   )
   __A	:		List[str]	  =      os.path.join(UpperCamelCase__,					UpperCamelCase__      )
   logger.info(f"""Loading model from {input_model_file}"""      )
   __A	:		str	  =      torch.load(UpperCamelCase__      )
   logger.info(f"""Model loaded from {input_model_file}"""      )
  elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
   __A	:		Optional[int]	  =      (
       os.path.join(UpperCamelCase__,					f"""{MODEL_NAME}_{model_index}"""      )
       if f"""{MODEL_NAME}""" not in input_dir
       else input_dir
   )
   logger.info(f"""Loading model from {ckpt_dir}"""      )
   __A	:		Union[str, Any]	  =      {'model': model.state_dict()}
   dist_cp.load_state_dict(
       state_dict=UpperCamelCase__,					storage_reader=dist_cp.FileSystemReader(UpperCamelCase__      ),					planner=DefaultLoadPlanner(),					)
   __A	:		Tuple	  =      state_dict['model']
   logger.info(f"""Model loaded from {ckpt_dir}"""      )
  model.load_state_dict(UpperCamelCase__      )


def       _lowercase							(					UpperCamelCase__							:       Optional[Any],					UpperCamelCase__							:       Optional[int],					UpperCamelCase__							:       Dict,					UpperCamelCase__							:       Optional[int],					UpperCamelCase__							:       List[str],					UpperCamelCase__							:       str=0      ):
 os.makedirs(UpperCamelCase__,					exist_ok=UpperCamelCase__      )
 with FSDP.state_dict_type(
     UpperCamelCase__,					fsdp_plugin.state_dict_type,					fsdp_plugin.state_dict_config,					fsdp_plugin.optim_state_dict_config      ):
  __A	:		Tuple	  =      FSDP.optim_state_dict(UpperCamelCase__,					UpperCamelCase__      )
  if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
   if accelerator.process_index == 0:
    __A	:		Any	  =      (
        f"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else f"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
    )
    __A	:		Union[str, Any]	  =      os.path.join(UpperCamelCase__,					UpperCamelCase__      )
    logger.info(f"""Saving Optimizer state to {output_optimizer_file}"""      )
    torch.save(UpperCamelCase__,					UpperCamelCase__      )
    logger.info(f"""Optimizer state saved in {output_optimizer_file}"""      )
  else:
   __A	:		Dict	  =      os.path.join(UpperCamelCase__,					f"""{OPTIMIZER_NAME}_{optimizer_index}"""      )
   os.makedirs(UpperCamelCase__,					exist_ok=UpperCamelCase__      )
   logger.info(f"""Saving Optimizer state to {ckpt_dir}"""      )
   dist_cp.save_state_dict(
       state_dict={'optimizer': optim_state},					storage_writer=dist_cp.FileSystemWriter(UpperCamelCase__      ),					planner=DefaultSavePlanner(),					)
   logger.info(f"""Optimizer state saved in {ckpt_dir}"""      )


def       _lowercase							(					UpperCamelCase__							:       int,					UpperCamelCase__							:       Optional[Any],					UpperCamelCase__							:       Any,					UpperCamelCase__							:       Tuple,					UpperCamelCase__							:       Optional[Any],					UpperCamelCase__							:       Tuple=0      ):
 accelerator.wait_for_everyone()
 with FSDP.state_dict_type(
     UpperCamelCase__,					fsdp_plugin.state_dict_type,					fsdp_plugin.state_dict_config,					fsdp_plugin.optim_state_dict_config      ):
  if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
   __A	:		Any	  =      None
   # below check should work but currently it isn't working (mostly opytorch issue),
   # in the meantime disabling it at the cost of excess memory usage
   # if accelerator.process_index == 0 or not fsdp_plugin.optim_state_dict_config.rank0_only:
   __A	:		Dict	  =      (
       f"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else f"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
   )
   __A	:		List[str]	  =      os.path.join(UpperCamelCase__,					UpperCamelCase__      )
   logger.info(f"""Loading Optimizer state from {input_optimizer_file}"""      )
   __A	:		List[str]	  =      torch.load(UpperCamelCase__      )
   logger.info(f"""Optimizer state loaded from {input_optimizer_file}"""      )
  else:
   __A	:		Optional[Any]	  =      (
       os.path.join(UpperCamelCase__,					f"""{OPTIMIZER_NAME}_{optimizer_index}"""      )
       if f"""{OPTIMIZER_NAME}""" not in input_dir
       else input_dir
   )
   logger.info(f"""Loading Optimizer from {ckpt_dir}"""      )
   __A	:		str	  =      load_sharded_optimizer_state_dict(
       model_state_dict=model.state_dict(),					optimizer_key='optimizer',					storage_reader=dist_cp.FileSystemReader(UpperCamelCase__      ),					)
   __A	:		Union[str, Any]	  =      optim_state['optimizer']
   logger.info(f"""Optimizer loaded from {ckpt_dir}"""      )
  __A	:		Optional[int]	  =      FSDP.optim_state_dict_to_load(UpperCamelCase__,					UpperCamelCase__,					UpperCamelCase__      )
  optimizer.load_state_dict(UpperCamelCase__      )










'''simple docstring'''
import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch

@require_torch
class     _lowerCamelCase       :




 '''simple docstring'''


 @property
 def 			snake_case__ (   self    ):

  """simple docstring"""







  return self.get_dummy_input()


 @property
 def 			snake_case__ (   self    ):

  """simple docstring"""







  if self.block_type == "down":
   return (4, 32, 16, 16)
  elif self.block_type == "mid":
   return (4, 32, 32, 32)
  elif self.block_type == "up":
   return (4, 32, 64, 64)

  raise ValueError(F"""'{self.block_type}' is not a supported block_type. Set it to 'up', 'mid', or 'down'."""    )


 def 			snake_case__ (   self   ,   __lowercase=True   ,   __lowercase=False   ,   __lowercase=False   ,   __lowercase=False   ,   ):

  """simple docstring"""







  __A	:		List[str]	  =      4
  __A	:		List[Any]	  =      32
  __A	:		Optional[Any]	  =      (32, 32)

  __A	:		List[str]	  =      torch.manual_seed(0    )
  __A	:		Union[str, Any]	  =      torch.device(__lowercase    )
  __A	:		Optional[int]	  =      (batch_size, num_channels) + sizes
  __A	:		str	  =      randn_tensor(__lowercase   ,   generator=__lowercase   ,   device=__lowercase    )
  __A	:		int	  =      {'hidden_states': hidden_states}

  if include_temb:
   __A	:		Optional[Any]	  =      128
   __A	:		Any	  =      randn_tensor((batch_size, temb_channels)   ,   generator=__lowercase   ,   device=__lowercase    )

  if include_res_hidden_states_tuple:
   __A	:		Any	  =      torch.manual_seed(1    )
   __A	:		List[Any]	  =      (randn_tensor(__lowercase   ,   generator=__lowercase   ,   device=__lowercase    ),)

  if include_encoder_hidden_states:
   __A	:		str	  =      floats_tensor((batch_size, 32, 32)    ).to(__lowercase    )

  if include_skip_sample:
   __A	:		List[Any]	  =      randn_tensor(((batch_size, 3) + sizes)   ,   generator=__lowercase   ,   device=__lowercase    )

  return dummy_input


 def 			snake_case__ (   self    ):

  """simple docstring"""







  __A	:		Tuple	  =      {
      'in_channels': 32,
      'out_channels': 32,
      'temb_channels': 128,
  }
  if self.block_type == "up":
   __A	:		int	  =      32

  if self.block_type == "mid":
   init_dict.pop('out_channels'    )

  __A	:		Optional[int]	  =      self.dummy_input
  return init_dict, inputs_dict


 def 			snake_case__ (   self   ,   __lowercase    ):

  """simple docstring"""







  __A  ,__A	:		List[Any]	  =      self.prepare_init_args_and_inputs_for_common()
  __A	:		int	  =      self.block_class(**__lowercase    )
  unet_block.to(__lowercase    )
  unet_block.eval()

  with torch.no_grad():
   __A	:		List[Any]	  =      unet_block(**__lowercase    )

  if isinstance(__lowercase   ,   __lowercase    ):
   __A	:		Tuple	  =      output[0]

  self.assertEqual(output.shape   ,   self.output_shape    )

  __A	:		Any	  =      output[0, -1, -3:, -3:]
  __A	:		Dict	  =      torch.tensor(__lowercase    ).to(__lowercase    )
  assert torch_all_close(output_slice.flatten()   ,   __lowercase   ,   atol=5E-3    )







 @unittest.skipIf(torch_device == 'mps'   ,   'Training is not supported in mps'    )
 def 			snake_case__ (   self    ):

  """simple docstring"""







  __A  ,__A	:		str	  =      self.prepare_init_args_and_inputs_for_common()
  __A	:		Optional[Any]	  =      self.block_class(**__lowercase    )
  model.to(__lowercase    )
  model.train()
  __A	:		Any	  =      model(**__lowercase    )

  if isinstance(__lowercase   ,   __lowercase    ):
   __A	:		Any	  =      output[0]

  __A	:		int	  =      torch.device(__lowercase    )
  __A	:		Dict	  =      randn_tensor(output.shape   ,   device=__lowercase    )
  __A	:		Dict	  =      torch.nn.functional.mse_loss(__lowercase   ,   __lowercase    )
  loss.backward()






