
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


__lowerCAmelCase   :     Dict      =						get_logger(__name__)


def       UpperCAmelCase_					(  __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase = None		,   )	->  List[str]:
    # Model states
    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						:     int		      =    opt.state_dict()
        __lowercase						:     Union[str, Any]		      =    F'{OPTIMIZER_NAME}.bin' if i == 0 else F'{OPTIMIZER_NAME}_{i}.bin'
        __lowercase						:     Optional[int]		      =    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						:     Tuple		      =    scheduler.state_dict()
        __lowercase						:     Optional[Any]		      =    F'{SCHEDULER_NAME}.bin' if i == 0 else F'{SCHEDULER_NAME}_{i}.bin'
        __lowercase						:     str		      =    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						:     Tuple		      =    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						:     Dict		      =    {}
    __lowercase						:     Optional[int]		      =    F'{RNG_STATE_NAME}_{process_index}.pkl'
    __lowercase						:     Optional[Any]		      =    random.getstate()
    __lowercase						:     Dict		      =    np.random.get_state()
    __lowercase						:     List[Any]		      =    torch.get_rng_state()
    if is_xpu_available():
        __lowercase						:     Optional[Any]		      =    torch.xpu.get_rng_state_all()
    else:
        __lowercase						:     int		      =    torch.cuda.get_rng_state_all()
    if is_tpu_available():
        __lowercase						:     Any		      =    xm.get_rng_state()
    __lowercase						:     Tuple		      =    os.path.join(__lowerCAmelCase		,   __lowerCAmelCase		)
    torch.save(__lowerCAmelCase		,   __lowerCAmelCase		)
    logger.info(F'Random states saved in {output_states_file}'		)
    return output_dir


def       UpperCAmelCase_					(  __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase=None		,   __lowerCAmelCase=None		,   **__lowerCAmelCase		,   )	->  Any:
    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						:     Any		      =    '''cpu'''
    elif map_location == "on_device":
        __lowercase						:     str		      =    PartialState().device
    # Model states
    for i, model in enumerate(__lowerCAmelCase		):
        __lowercase						:     int		      =    F'{MODEL_NAME}.bin' if i == 0 else F'{MODEL_NAME}_{i}.bin'
        __lowercase						:     Optional[int]		      =    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						:     Any		      =    F'{OPTIMIZER_NAME}.bin' if i == 0 else F'{OPTIMIZER_NAME}_{i}.bin'
        __lowercase						:     List[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						:     Any		      =    F'{SCHEDULER_NAME}.bin' if i == 0 else F'{SCHEDULER_NAME}_{i}.bin'
        __lowercase						:     str		      =    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						:     List[str]		      =    os.path.join(__lowerCAmelCase		,   __lowerCAmelCase		)
        scaler.load_state_dict(torch.load(__lowerCAmelCase		)		)
        logger.info('''GradScaler state loaded successfully'''		)

    # Random states
    try:
        __lowercase						:     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       UpperCAmelCase_					(  __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase = 0		)	->  List[Any]:
    __lowercase						:     Dict		      =    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       UpperCAmelCase_					(  __lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase = 0		)	->  Optional[Any]:
    __lowercase						:     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'''		)		)




import importlib.metadata
import sys

from transformers.testing_utils import TestCasePlus
from transformers.utils.versions import require_version, require_version_core


__lowerCAmelCase   :     Optional[Any]      =						importlib.metadata.version("numpy")
__lowerCAmelCase   :     Any      =						".".join([str(x) for x in sys.version_info[:3]])


class   __lowerCAmelCase		(		lowerCAmelCase_ ):



      """simple docstring"""







      def 						snake_case_     ( self						:    List[Any]			):
          # lt + different version strings
          require_version_core('''numpy<1000.4.5'''			)
          require_version_core('''numpy<1000.4'''			)
          require_version_core('''numpy<1000'''			)

          # le
          require_version_core('''numpy<=1000.4.5'''			)
          require_version_core(F'numpy<={numpy_ver}'			)

          # eq
          require_version_core(F'numpy=={numpy_ver}'			)

          # ne
          require_version_core('''numpy!=1000.4.5'''			)

          # ge
          require_version_core('''numpy>=1.0'''			)
          require_version_core('''numpy>=1.0.0'''			)
          require_version_core(F'numpy>={numpy_ver}'			)

          # gt
          require_version_core('''numpy>1.0.0'''			)

          # mix
          require_version_core('''numpy>1.0.0,<1000'''			)

          # requirement w/o version
          require_version_core('''numpy'''			)

          # unmet requirements due to version conflict
          for req in ["numpy==1.0.0", "numpy>=1000.0.0", F'numpy<{numpy_ver}']:
              try:
                  require_version_core(_snake_case			)
              except ImportError as e:
                  self.assertIn(F'{req} is required'			,    str(_snake_case			)			)
                  self.assertIn('''but found'''			,    str(_snake_case			)			)

        # unmet requirements due to missing module
          for req in ["numpipypie>1", "numpipypie2"]:
              try:
                  require_version_core(_snake_case			)
              except importlib.metadata.PackageNotFoundError as e:
                  self.assertIn(F'The \'{req}\' distribution was not found and is required by this application'			,    str(_snake_case			)			)
                  self.assertIn('''Try: pip install transformers -U'''			,    str(_snake_case			)			)

        # bogus requirements formats:
        # 1. whole thing
          for req in ["numpy??1.0.0", "numpy1.0.0"]:
              try:
                  require_version_core(_snake_case			)
              except ValueError as e:
                  self.assertIn('''requirement needs to be in the pip package format'''			,    str(_snake_case			)			)
        # 2. only operators
          for req in ["numpy=1.0.0", "numpy == 1.00", "numpy<>1.0.0", "numpy><1.00", "numpy>>1.0.0"]:
              try:
                  require_version_core(_snake_case			)
              except ValueError as e:
                  self.assertIn('''need one of '''			,    str(_snake_case			)			)




      def 						snake_case_     ( self						:    List[str]			):
          # matching requirement
          require_version('''python>=3.6.0'''			)

          # not matching requirements
          for req in ["python>9.9.9", "python<3.0.0"]:
              try:
                  require_version_core(_snake_case			)
              except ImportError as e:
                  self.assertIn(F'{req} is required'			,    str(_snake_case			)			)
                  self.assertIn(F'but found python=={python_ver}'			,    str(_snake_case			)			)


