

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_					  =   get_logger(__name__)



def   UpperCamelCase(   lowercase_     ,   lowercase_     ,   lowercase_     ,   lowercase_     ,   lowercase_     ,   lowercase_ = None     ,   )      ->							Dict:





  '''simple docstring'''





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



def   UpperCamelCase(   lowercase_     ,   lowercase_     ,   lowercase_     ,   lowercase_     ,   lowercase_     ,   lowercase_=None     ,   lowercase_=None     ,   **lowercase_     ,   )      ->							List[Any]:





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

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

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

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

  # Random states
  try:
    snake_case_					= torch.load(os.path.join(a__     ,   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(   lowercase_     ,   lowercase_     ,   lowercase_ = 0				)      ->							Any:





  '''simple docstring'''





  snake_case_					= Path(a__				) / f'''custom_checkpoint_{index}.pkl'''
  logger.info(f'''Saving the state of {get_pretty_name(a__				)} to {save_location}'''				)
  torch.save(obj.state_dict()     ,   a__				)



def   UpperCamelCase(   lowercase_     ,   lowercase_     ,   lowercase_ = 0				)      ->							Optional[Any]:





  '''simple docstring'''





  snake_case_					= f'''{path}/custom_checkpoint_{index}.pkl'''
  logger.info(f'''Loading the state of {get_pretty_name(a__				)} from {load_location}'''				)
  obj.load_state_dict(torch.load(a__     ,   map_location="""cpu"""				)				)
'''simple docstring'''

import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


lowercase_  		=			namedtuple('''ExampleNamedTuple''', '''a b c''')
class 	SCREAMING_SNAKE_CASE      (    unittest.TestCase):







		"""simple docstring"""






		def 		UpperCamelCase__							( self					)			->  int:
								_lowerCAmelCase					    =torch.randn(5		,    2					)
								_lowerCAmelCase					    =torch.device('cuda'					) if torch.cuda.is_available() else torch.device('cpu'					)

								_lowerCAmelCase					    =send_to_device(__A		,    __A					)
								self.assertTrue(torch.equal(resulta.cpu()		,    __A					)					)

								_lowerCAmelCase					    =send_to_device((tensor, [tensor, tensor], 1)		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta[0].cpu()		,    __A					)					)
								self.assertIsInstance(resulta[1]		,    __A					)
								self.assertTrue(torch.equal(resulta[1][0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta[1][1].cpu()		,    __A					)					)
								self.assertEqual(resulta[2]		,    1					)

								_lowerCAmelCase					    =send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta['a'].cpu()		,    __A					)					)
								self.assertIsInstance(resulta['b']		,    __A					)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta['b'][1].cpu()		,    __A					)					)
								self.assertEqual(resulta['c']		,    1					)

								_lowerCAmelCase					    =send_to_device(ExampleNamedTuple(a=__A		,    b=[tensor, tensor]		,    c=1					)		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta.a.cpu()		,    __A					)					)
								self.assertIsInstance(resulta.b		,    __A					)
								self.assertTrue(torch.equal(resulta.b[0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta.b[1].cpu()		,    __A					)					)
								self.assertEqual(resulta.c		,    1					)

								_lowerCAmelCase					    =send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}					)		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta['a'].cpu()		,    __A					)					)
								self.assertIsInstance(resulta['b']		,    __A					)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta['b'][1].cpu()		,    __A					)					)
								self.assertEqual(resulta['c']		,    1					)






		def 		UpperCamelCase__							( self					)			->  Optional[int]:
								with self.assertRaises(__A					) as cm:
														_lowerCAmelCase					    =recursively_apply(torch.tensor		,    (torch.tensor(1					), 1)		,    error_on_other_type=__A					)
								self.assertEqual(
								    str(cm.exception					)		,    'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'		,    )






		def 		UpperCamelCase__							( self					)			->  int:
								_lowerCAmelCase					    =torch.tensor([1, 2, 3, 4, 5]					)
								self.assertEqual(listify(__A					)		,    [1, 2, 3, 4, 5]					)

								_lowerCAmelCase					    =torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)
								self.assertEqual(listify(__A					)		,    [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)

								_lowerCAmelCase					    =torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]					)
								self.assertEqual(
								    listify(__A					)		,    [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]					)






		def 		UpperCamelCase__							( self					)			->  List[Any]:
								with patch_environment(aa=1		,    BB=2					):
														self.assertEqual(os.environ.get('AA'					)		,    '1'					)
														self.assertEqual(os.environ.get('BB'					)		,    '2'					)

								self.assertNotIn('AA'		,    os.environ					)
								self.assertNotIn('BB'		,    os.environ					)






		def 		UpperCamelCase__							( self					)			->  Union[str, Any]:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =model.forward
								_lowerCAmelCase					    =convert_outputs_to_fpaa(model.forward					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A		,    keep_fpaa_wrapper=__A					)
								_lowerCAmelCase					    =pickle.dumps(__A					)






		@require_cuda
		def 		UpperCamelCase__							( self					)			->  Dict:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =model.forward
								_lowerCAmelCase					    =torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
								_lowerCAmelCase					    =convert_outputs_to_fpaa(model.forward					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A		,    keep_fpaa_wrapper=__A					)
								_lowerCAmelCase					    =pickle.dumps(__A					)






		@require_cuda
		@require_torch_min_version(version='2.0'					)
		def 		UpperCamelCase__							( self					)			->  int:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =model.forward
								_lowerCAmelCase					    =torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
								_lowerCAmelCase					    =convert_outputs_to_fpaa(model.forward					)
								_lowerCAmelCase					    =torch.compile(model.forward		,    backend='inductor'					)
								_lowerCAmelCase					    =torch.randn(4		,    10					).cuda()
								_lowerCAmelCase					    =model(__A					)






		def 		UpperCamelCase__							( self					)			->  Optional[int]:
								_lowerCAmelCase					    =RegressionModel()
								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								_lowerCAmelCase					    =torch.nn.parallel.DataParallel(__A					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A					)

								self.assertEqual(__A		,    __A					)






		@require_torch_min_version(version='2.0'					)
		def 		UpperCamelCase__							( self					)			->  str:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =torch.compile(__A					)

								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								_lowerCAmelCase					    =torch.nn.parallel.DataParallel(__A					)
								_lowerCAmelCase					    =torch.compile(__A					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A					)

								self.assertEqual(compiled_model._orig_mod		,    compiled_model_unwrapped._orig_mod					)






		def 		UpperCamelCase__							( self					)			->  Tuple:
								self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3]					)]					)		,    torch.device('cpu'					)					)
								self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3]					)}					)		,    torch.device('cpu'					)					)
								self.assertIsNone(find_device([1, 'a']					)					)





