
"""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


UpperCAmelCase__				:   List[Any]   							=			get_logger(__name__)

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

def lowercase_  (       _snake_case       ,_snake_case       ,_snake_case       ,_snake_case       ,_snake_case       ,_snake_case=None       ,_snake_case=None       ,**_snake_case       ,):
							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:
														SCREAMING_SNAKE_CASE__						:	str									=       """cpu"""
							elif map_location == "on_device":
														SCREAMING_SNAKE_CASE__						:	Any									=       PartialState().device
							# Model states
							for i, model in enumerate(_snake_case						):
														SCREAMING_SNAKE_CASE__						:	Any									=       f'''{MODEL_NAME}.bin''' if i == 0 else f'''{MODEL_NAME}_{i}.bin'''
														SCREAMING_SNAKE_CASE__						:	Tuple									=       os.path.join(_snake_case       ,_snake_case						)
														models[i].load_state_dict(torch.load(_snake_case       ,map_location=_snake_case						)       ,**_snake_case						)
							logger.info("""All model weights loaded successfully"""						)

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

							# Scheduler states
							for i, scheduler in enumerate(_snake_case						):
														SCREAMING_SNAKE_CASE__						:	Dict									=       f'''{SCHEDULER_NAME}.bin''' if i == 0 else f'''{SCHEDULER_NAME}_{i}.bin'''
														SCREAMING_SNAKE_CASE__						:	str									=       os.path.join(_snake_case       ,_snake_case						)
														scheduler.load_state_dict(torch.load(_snake_case						)						)
							logger.info("""All scheduler states loaded successfully"""						)

							# GradScaler state
							if scaler is not None:
														SCREAMING_SNAKE_CASE__						:	Tuple									=       os.path.join(_snake_case       ,_snake_case						)
														scaler.load_state_dict(torch.load(_snake_case						)						)
														logger.info("""GradScaler state loaded successfully"""						)

							# Random states
							try:
														SCREAMING_SNAKE_CASE__						:	int									=       torch.load(os.path.join(_snake_case       ,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_  (       _snake_case       ,_snake_case       ,_snake_case = 0						):
							SCREAMING_SNAKE_CASE__						:	Tuple									=       Path(_snake_case						) / f'''custom_checkpoint_{index}.pkl'''
							logger.info(f'''Saving the state of {get_pretty_name(_snake_case						)} to {save_location}'''						)
							torch.save(obj.state_dict()       ,_snake_case						)

def lowercase_  (       _snake_case       ,_snake_case       ,_snake_case = 0						):
							SCREAMING_SNAKE_CASE__						:	Union[str, Any]									=       f'''{path}/custom_checkpoint_{index}.pkl'''
							logger.info(f'''Loading the state of {get_pretty_name(_snake_case						)} from {load_location}'''						)
							obj.load_state_dict(torch.load(_snake_case       ,map_location="""cpu"""						)						)






"""simple docstring"""




import re

def lowercase_  (       _snake_case						):
							def docstring_decorator(_snake_case						):
														SCREAMING_SNAKE_CASE__						:	Any									=       fn.__doc__
														SCREAMING_SNAKE_CASE__						:	Dict									=       func_doc.split("""\n"""						)
														SCREAMING_SNAKE_CASE__						:	Any									=       0
														while i < len(_snake_case						) and re.search(R"""^\s*Examples?:\s*$"""       ,lines[i]						) is None:
																					i += 1
														if i < len(_snake_case						):
																					SCREAMING_SNAKE_CASE__						:	List[str]									=       example_docstring
																					SCREAMING_SNAKE_CASE__						:	List[Any]									=       """\n""".join(_snake_case						)
														else:
																					raise ValueError(
																					    f'''The function {fn} should have an empty \'Examples:\' in its docstring as placeholder, '''
																					    f'''current docstring is:\n{func_doc}'''						)
														SCREAMING_SNAKE_CASE__						:	int									=       func_doc
														return fn

							return docstring_decorator




