'''simple docstring'''


import os
import random
from pathlib import Path
from typing import List

import numpy as np
import torch
from torch.cuda.amp import GradScaler

from .utils import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    get_pretty_name,
    is_tpu_available,
    is_xpu_available,
    save,
)


if is_tpu_available(check_device=False):
 import torch_xla.core.xla_model as xm

from .logging import get_logger
from .state import PartialState


snake_case__	:     List[str]      =    get_logger(__name__)


def    _lowerCamelCase       ( lowerCamelCase_   :				str      ,	lowerCamelCase_   :				List[dict]      ,	lowerCamelCase_   :				list      ,	lowerCamelCase_   :				list      ,	lowerCamelCase_   :				int      ,	lowerCamelCase_   :				GradScaler = None      ,	):



     """simple docstring"""





     for i, state in enumerate(lowerCamelCase_  ):
          UpperCAmelCase_				:  Tuple  = F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
          UpperCAmelCase_				:  int  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          save(lowerCamelCase_      ,	lowerCamelCase_  )
          logger.info(F'''Model weights saved in {output_model_file}'''  )
     # Optimizer states
     for i, opt in enumerate(lowerCamelCase_  ):
          UpperCAmelCase_				:  str  = opt.state_dict()
          UpperCAmelCase_				:  Optional[Any]  = F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
          UpperCAmelCase_				:  List[str]  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          save(lowerCamelCase_      ,	lowerCamelCase_  )
          logger.info(F'''Optimizer state saved in {output_optimizer_file}'''  )
     # Scheduler states
     for i, scheduler in enumerate(lowerCamelCase_  ):
          UpperCAmelCase_				:  Tuple  = scheduler.state_dict()
          UpperCAmelCase_				:  Union[str, Any]  = F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
          UpperCAmelCase_				:  int  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          save(lowerCamelCase_      ,	lowerCamelCase_  )
          logger.info(F'''Scheduler state saved in {output_scheduler_file}'''  )
     # GradScaler state
     if scaler is not None:
          UpperCAmelCase_				:  Dict  = scaler.state_dict()
          UpperCAmelCase_				:  Dict  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          torch.save(lowerCamelCase_      ,	lowerCamelCase_  )
          logger.info(F'''Gradient scaler state saved in {output_scaler_file}'''  )
     # Random number generator states
     UpperCAmelCase_				:  Dict  = {}
     UpperCAmelCase_				:  int  = F'''{RNG_STATE_NAME}_{process_index}.pkl'''
     UpperCAmelCase_				:  Dict  = random.getstate()
     UpperCAmelCase_				:  List[str]  = np.random.get_state()
     UpperCAmelCase_				:  Any  = torch.get_rng_state()
     if is_xpu_available():
          UpperCAmelCase_				:  Optional[int]  = torch.xpu.get_rng_state_all()
     else:
          UpperCAmelCase_				:  Dict  = torch.cuda.get_rng_state_all()
     if is_tpu_available():
          UpperCAmelCase_				:  int  = xm.get_rng_state()
     UpperCAmelCase_				:  Any  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
     torch.save(lowerCamelCase_      ,	lowerCamelCase_  )
     logger.info(F'''Random states saved in {output_states_file}'''  )
     return output_dir


def    _lowerCamelCase       ( lowerCamelCase_   :				List[Any]      ,	lowerCamelCase_   :				List[Any]      ,	lowerCamelCase_   :				Dict      ,	lowerCamelCase_   :				Dict      ,	lowerCamelCase_   :				Dict      ,	lowerCamelCase_   :				Any=None      ,	lowerCamelCase_   :				int=None      ,	**lowerCamelCase_   :				Dict      ,	):



     """simple docstring"""





     if map_location not in [None, "cpu", "on_device"]:
          raise TypeError(
              'Unsupported optimizer map location passed, please choose one of `None`, `\'cpu\'`, or `\'on_device\'`'  )
     if map_location is None:
          UpperCAmelCase_				:  str  = 'cpu'
     elif map_location == "on_device":
          UpperCAmelCase_				:  Optional[int]  = PartialState().device
     # Model states
     for i, model in enumerate(lowerCamelCase_  ):
          UpperCAmelCase_				:  Union[str, Any]  = F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
          UpperCAmelCase_				:  Any  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          models[i].load_state_dict(torch.load(lowerCamelCase_      ,	map_location=lowerCamelCase_  )      ,	**lowerCamelCase_  )
     logger.info('All model weights loaded successfully'  )

     # Optimizer states
     for i, opt in enumerate(lowerCamelCase_  ):
          UpperCAmelCase_				:  List[Any]  = F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
          UpperCAmelCase_				:  List[Any]  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          UpperCAmelCase_				:  str  = torch.load(lowerCamelCase_      ,	map_location=lowerCamelCase_  )
          optimizers[i].load_state_dict(lowerCamelCase_  )
     logger.info('All optimizer states loaded successfully'  )

     # Scheduler states
     for i, scheduler in enumerate(lowerCamelCase_  ):
          UpperCAmelCase_				:  Any  = F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
          UpperCAmelCase_				:  List[Any]  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          scheduler.load_state_dict(torch.load(lowerCamelCase_  )  )
     logger.info('All scheduler states loaded successfully'  )

     # GradScaler state
     if scaler is not None:
          UpperCAmelCase_				:  Dict  = os.path.join(lowerCamelCase_      ,	lowerCamelCase_  )
          scaler.load_state_dict(torch.load(lowerCamelCase_  )  )
          logger.info('GradScaler state loaded successfully'  )

     # Random states
     try:
          UpperCAmelCase_				:  Union[str, Any]  = torch.load(os.path.join(lowerCamelCase_      ,	F'''{RNG_STATE_NAME}_{process_index}.pkl'''  )  )
          random.setstate(states['random_state']  )
          np.random.set_state(states['numpy_random_seed']  )
          torch.set_rng_state(states['torch_manual_seed']  )
          if is_xpu_available():
               torch.xpu.set_rng_state_all(states['torch_xpu_manual_seed']  )
          else:
               torch.cuda.set_rng_state_all(states['torch_cuda_manual_seed']  )
          if is_tpu_available():
               xm.set_rng_state(states['xm_seed']  )
          logger.info('All random states loaded successfully'  )
     except Exception:
          logger.info('Could not load random states'  )


def    _lowerCamelCase       ( lowerCamelCase_   :				str      ,	lowerCamelCase_   :				str      ,	lowerCamelCase_   :				int = 0  ):



     """simple docstring"""





     UpperCAmelCase_				:  Any  = Path(lowerCamelCase_  ) / F'''custom_checkpoint_{index}.pkl'''
     logger.info(F'''Saving the state of {get_pretty_name(lowerCamelCase_  )} to {save_location}'''  )
     torch.save(obj.state_dict()      ,	lowerCamelCase_  )





def    _lowerCamelCase       ( lowerCamelCase_   :				List[str]      ,	lowerCamelCase_   :				Dict      ,	lowerCamelCase_   :				int = 0  ):



     """simple docstring"""





     UpperCAmelCase_				:  Tuple  = F'''{path}/custom_checkpoint_{index}.pkl'''
     logger.info(F'''Loading the state of {get_pretty_name(lowerCamelCase_  )} from {load_location}'''  )
     obj.load_state_dict(torch.load(lowerCamelCase_      ,	map_location='cpu'  )  )








"""simple docstring"""






from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
				import torch

__UpperCAmelCase 					=    logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
				from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):






				def 		_snake_case			(      cls						:    Tuple						)    ->							List[Any]:


							'''simple docstring'''





							return cls






def 		_snake_case			(      lowercase__						:    Union[Tuple, List] ,	lowercase__						:    Optional[Union[List["torch.Generator"], "torch.Generator"]] = None ,	lowercase__						:    Optional["torch.device"] = None ,	lowercase__						:    Optional["torch.dtype"] = None ,	lowercase__						:    Optional["torch.layout"] = None ,	)    ->							Optional[Any]:


			'''simple docstring'''





			lowerCAmelCase_       :Any     	=    device
			lowerCAmelCase_       :Any     	=    shape[0]

			lowerCAmelCase_       :Dict     	=    layout or torch.strided
			lowerCAmelCase_       :List[str]     	=    device or torch.device("""cpu"""						)

			if generator is not None:
						lowerCAmelCase_       :Optional[Any]     	=    generator.device.type if not isinstance(lowercase__ ,	lowercase__						) else generator[0].device.type
						if gen_device_type != device.type and gen_device_type == "cpu":
									lowerCAmelCase_       :Optional[int]     	=    """cpu"""
									if device != "mps":
												logger.info(
												    f"""The passed generator was created on 'cpu' even though a tensor on {device} was expected."""
												    f""" Tensors will be created on 'cpu' and then moved to {device}. Note that one can probably"""
												    f""" slighly speed up this function by passing a generator that was created on the {device} device."""						)
						elif gen_device_type != device.type and gen_device_type == "cuda":
									raise ValueError(f"""Cannot generate a {device} tensor from a generator of type {gen_device_type}."""						)

			if isinstance(lowercase__ ,	lowercase__						):
						lowerCAmelCase_       :List[str]     	=    (1,) + shape[1:]
						lowerCAmelCase_       :Optional[Any]     	=    [
						    torch.randn(lowercase__ ,	generator=generator[i] ,	device=lowercase__ ,	dtype=lowercase__ ,	layout=lowercase__						)
						    for i in range(lowercase__						)
						]
						lowerCAmelCase_       :str     	=    torch.cat(lowercase__ ,	dim=0						).to(lowercase__						)
			else:
						lowerCAmelCase_       :Dict     	=    torch.randn(lowercase__ ,	generator=lowercase__ ,	device=lowercase__ ,	dtype=lowercase__ ,	layout=lowercase__						).to(lowercase__						)

			return latents






def 		_snake_case			(      lowercase__						:    Union[str, Any]						)    ->							int:


			'''simple docstring'''





			if is_torch_version("""<""" ,	"""2.0.0"""						) or not hasattr(lowercase__ ,	"""_dynamo"""						):
						return False
			return isinstance(lowercase__ ,	torch._dynamo.eval_frame.OptimizedModule						)






