code: 
  
  
  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
  
  
  __a     :   Optional[Any]   				=      get_logger(__name__)
  
  
  
  
  def 			__lowercase	(						_UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase = None,    )    ->Tuple:
  
  
  
  
  
  
  
        """simple docstring"""
  
        for i, state in enumerate(lowerCamelCase__     ):
              lowercase			:   Tuple					   =		f"""{MODEL_NAME}.bin""" if i == 0 else f"""{MODEL_NAME}_{i}.bin"""
              lowercase			:   Union[str, Any]					   =		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__     ):
              lowercase			:   Tuple					   =		opt.state_dict()
              lowercase			:   List[Any]					   =		f"""{OPTIMIZER_NAME}.bin""" if i == 0 else f"""{OPTIMIZER_NAME}_{i}.bin"""
              lowercase			:   Optional[Any]					   =		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__     ):
              lowercase			:   List[Any]					   =		scheduler.state_dict()
              lowercase			:   List[str]					   =		f"""{SCHEDULER_NAME}.bin""" if i == 0 else f"""{SCHEDULER_NAME}_{i}.bin"""
              lowercase			:   List[Any]					   =		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:
              lowercase			:   str					   =		scaler.state_dict()
              lowercase			:   List[str]					   =		os.path.join(lowerCamelCase__,    lowerCamelCase__     )
              torch.save(lowerCamelCase__,    lowerCamelCase__     )
              logger.info(f"""Gradient scaler state saved in {output_scaler_file}"""     )
        # Random number generator states
        lowercase			:   Optional[int]					   =		{}
        lowercase			:   Any					   =		f"""{RNG_STATE_NAME}_{process_index}.pkl"""
        lowercase			:   Union[str, Any]					   =		random.getstate()
        lowercase			:   Any					   =		np.random.get_state()
        lowercase			:   List[str]					   =		torch.get_rng_state()
        if is_xpu_available():
              lowercase			:   List[Any]					   =		torch.xpu.get_rng_state_all()
        else:
              lowercase			:   Optional[int]					   =		torch.cuda.get_rng_state_all()
        if is_tpu_available():
              lowercase			:   Any					   =		xm.get_rng_state()
        lowercase			:   Dict					   =		os.path.join(lowerCamelCase__,    lowerCamelCase__     )
        torch.save(lowerCamelCase__,    lowerCamelCase__     )
        logger.info(f"""Random states saved in {output_states_file}"""     )
        return output_dir
  
  
  
  
  def 			__lowercase	(						_UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase=None,    _UpperCamelCase=None,    **_UpperCamelCase,    )    ->List[str]:
  
  
  
  
  
  
  
        """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:
              lowercase			:   str					   =		'''cpu'''
        elif map_location == "on_device":
              lowercase			:   List[str]					   =		PartialState().device
        # Model states
        for i, model in enumerate(lowerCamelCase__     ):
              lowercase			:   Tuple					   =		f"""{MODEL_NAME}.bin""" if i == 0 else f"""{MODEL_NAME}_{i}.bin"""
              lowercase			:   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__     ):
              lowercase			:   List[Any]					   =		f"""{OPTIMIZER_NAME}.bin""" if i == 0 else f"""{OPTIMIZER_NAME}_{i}.bin"""
              lowercase			:   str					   =		os.path.join(lowerCamelCase__,    lowerCamelCase__     )
              lowercase			:   Union[str, Any]					   =		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__     ):
              lowercase			:   List[Any]					   =		f"""{SCHEDULER_NAME}.bin""" if i == 0 else f"""{SCHEDULER_NAME}_{i}.bin"""
              lowercase			:   Dict					   =		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:
              lowercase			:   Tuple					   =		os.path.join(lowerCamelCase__,    lowerCamelCase__     )
              scaler.load_state_dict(torch.load(lowerCamelCase__     )     )
              logger.info('''GradScaler state loaded successfully'''     )
  
        # Random states
        try:
              lowercase			:   Tuple					   =		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 			__lowercase	(						_UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase = 0     )    ->Optional[int]:
  
  
  
  
  
  
  
        """simple docstring"""
  
        lowercase			:   Union[str, 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 			__lowercase	(						_UpperCamelCase,    _UpperCamelCase,    _UpperCamelCase = 0     )    ->List[str]:
  
  
  
  
  
  
  
        """simple docstring"""
  
        lowercase			:   Union[str, Any]					   =		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'''     )     )
  
  
style_context: 
  
  
  from ...utils import (
      OptionalDependencyNotAvailable,
      is_torch_available,
      is_transformers_available,
      is_transformers_version,
  )
  
  
  try:
    if not (is_transformers_available() and is_torch_available() and is_transformers_version('''>=''', '''4.27.0''')):
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    from ...utils.dummy_torch_and_transformers_objects import (
        AudioLDMPipeline,
    )
  else:
    from .pipeline_audioldm import AudioLDMPipeline
  
  