




# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING, Dict, Optional

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.logging import get_logger
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
					import jax
					import jaxlib

UpperCamelCase__    			=			get_logger()

UpperCamelCase__    			=			None
class 			a__    (					TensorFormatter[Mapping, """jax.Array""", Mapping]      ):






						def __init__(   self		,			_A=None		,			_A=None		,			**_A			):




											"""simple docstring"""



											super().__init__(features=_A			)
											import jax
											from jaxlib.xla_client import Device

											if isinstance(_A		,			_A			):
																raise ValueError(
																    f"""Expected {device} to be a `str` not {type(_A			)}, as `jaxlib.xla_extension.Device` """
																    "is not serializable neither with `pickle` nor with `dill`. Instead you can surround "
																    "the device with `str()` to get its string identifier that will be internally mapped "
																    "to the actual `jaxlib.xla_extension.Device`."			)
											__lowerCAmelCase				   =       device if isinstance(_A		,			_A			) else str(jax.devices()[0]			)
											# using global variable since `jaxlib.xla_extension.Device` is not serializable neither
											# with `pickle` nor with `dill`, so we need to use a global variable instead
											global DEVICE_MAPPING
											if DEVICE_MAPPING is None:
																__lowerCAmelCase				   =       self._map_devices_to_str()
											if self.device not in list(DEVICE_MAPPING.keys()			):
																logger.warning(
																    f"""Device with string identifier {self.device} not listed among the available """
																    f"""devices: {list(DEVICE_MAPPING.keys()			)}, so falling back to the default """
																    f"""device: {str(jax.devices()[0]			)}."""			)
																__lowerCAmelCase				   =       str(jax.devices()[0]			)
											__lowerCAmelCase				   =       jnp_array_kwargs






						@staticmethod
						def 				__SCREAMING_SNAKE_CASE(   ):




											"""simple docstring"""



											import jax

											return {str(_A			): device for device in jax.devices()}






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A			):




											"""simple docstring"""



											import jax
											import jax.numpy as jnp

											if isinstance(_A		,			_A			) and column:
																if all(
																    isinstance(_A		,			jax.Array			) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column			):
																					return jnp.stack(_A		,			axis=0			)
											return column






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A			):




											"""simple docstring"""



											import jax
											import jax.numpy as jnp

											if isinstance(_A		,			(str, bytes, type(_A			))			):
																return value
											elif isinstance(_A		,			(np.character, np.ndarray)			) and np.issubdtype(value.dtype		,			np.character			):
																return value.tolist()

											__lowerCAmelCase				   =       {}

											if isinstance(_A		,			(np.number, np.ndarray)			) and np.issubdtype(value.dtype		,			np.integer			):
																# the default int precision depends on the jax config
																# see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
																if jax.config.jax_enable_xaa:
																					__lowerCAmelCase				   =       {"dtype": jnp.intaa}
																else:
																					__lowerCAmelCase				   =       {"dtype": jnp.intaa}
											elif isinstance(_A		,			(np.number, np.ndarray)			) and np.issubdtype(value.dtype		,			np.floating			):
																__lowerCAmelCase				   =       {"dtype": jnp.floataa}
											elif config.PIL_AVAILABLE and "PIL" in sys.modules:
																import PIL.Image

																if isinstance(_A		,			PIL.Image.Image			):
																					__lowerCAmelCase				   =       np.asarray(_A			)

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
											global DEVICE_MAPPING
											if DEVICE_MAPPING is None:
																__lowerCAmelCase				   =       self._map_devices_to_str()

											with jax.default_device(DEVICE_MAPPING[self.device]			):
																# calling jnp.array on a np.ndarray does copy the data
																# see https://github.com/google/jax/issues/4486
																return jnp.array(_A		,			**{**default_dtype, **self.jnp_array_kwargs}			)






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A			):




											"""simple docstring"""



											import jax

											# support for torch, tf, jax etc.
											if config.TORCH_AVAILABLE and "torch" in sys.modules:
																import torch

																if isinstance(_A		,			torch.Tensor			):
																					return self._tensorize(data_struct.detach().cpu().numpy()[()]			)
											if hasattr(_A		,			"__array__"			) and not isinstance(_A		,			jax.Array			):
																__lowerCAmelCase				   =       data_struct.__array__()
											# support for nested types like struct of list of struct
											if isinstance(_A		,			np.ndarray			):
																if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
																					return self._consolidate([self.recursive_tensorize(_A			) for substruct in data_struct]			)
											elif isinstance(_A		,			(list, tuple)			):
																return self._consolidate([self.recursive_tensorize(_A			) for substruct in data_struct]			)
											return self._tensorize(_A			)






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A			):




											"""simple docstring"""



											return map_nested(self._recursive_tensorize		,			_A		,			map_list=_A			)






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A			):




											"""simple docstring"""



											__lowerCAmelCase				   =       self.numpy_arrow_extractor().extract_row(_A			)
											__lowerCAmelCase				   =       self.python_features_decoder.decode_row(_A			)
											return self.recursive_tensorize(_A			)






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A			):




											"""simple docstring"""



											__lowerCAmelCase				   =       self.numpy_arrow_extractor().extract_column(_A			)
											__lowerCAmelCase				   =       self.python_features_decoder.decode_column(_A		,			pa_table.column_names[0]			)
											__lowerCAmelCase				   =       self.recursive_tensorize(_A			)
											__lowerCAmelCase				   =       self._consolidate(_A			)
											return column






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A			):




											"""simple docstring"""



											__lowerCAmelCase				   =       self.numpy_arrow_extractor().extract_batch(_A			)
											__lowerCAmelCase				   =       self.python_features_decoder.decode_batch(_A			)
											__lowerCAmelCase				   =       self.recursive_tensorize(_A			)
											for column_name in batch:
																__lowerCAmelCase				   =       self._consolidate(batch[column_name]			)
											return batch










import os

import torch

from ..logging import get_logger
from .constants import FSDP_PYTORCH_VERSION, MODEL_NAME, OPTIMIZER_NAME
from .versions import is_torch_version


if is_torch_version(""">=""", FSDP_PYTORCH_VERSION):
					import torch.distributed.checkpoint as dist_cp
					from torch.distributed.checkpoint.default_planner import DefaultLoadPlanner, DefaultSavePlanner
					from torch.distributed.checkpoint.optimizer import load_sharded_optimizer_state_dict
					from torch.distributed.fsdp.fully_sharded_data_parallel import FullyShardedDataParallel as FSDP
					from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType


UpperCamelCase__    			=			get_logger(__name__)







def 	_a      (						SCREAMING_SNAKE_CASE_							:				List[str]      ,      SCREAMING_SNAKE_CASE_							:				Tuple      ,      SCREAMING_SNAKE_CASE_							:				Tuple      ,      SCREAMING_SNAKE_CASE_							:				Optional[Any]      ,      SCREAMING_SNAKE_CASE_							:				str=0     ):
					os.makedirs(SCREAMING_SNAKE_CASE_      ,      exist_ok=SCREAMING_SNAKE_CASE_     )
					with FSDP.state_dict_type(
					    SCREAMING_SNAKE_CASE_      ,      fsdp_plugin.state_dict_type      ,      fsdp_plugin.state_dict_config      ,      fsdp_plugin.optim_state_dict_config     ):
										__lowerCAmelCase				   =       model.state_dict()
										if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
															__lowerCAmelCase				   =       F"""{MODEL_NAME}.bin""" if model_index == 0 else F"""{MODEL_NAME}_{model_index}.bin"""
															__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
															if accelerator.process_index == 0:
																				logger.info(F"""Saving model to {output_model_file}"""     )
																				torch.save(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
																				logger.info(F"""Model saved to {output_model_file}"""     )
										elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
															__lowerCAmelCase				   =       (
															    F"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
															    if model_index == 0
															    else F"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
															)
															__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Saving model to {output_model_file}"""     )
															torch.save(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Model saved to {output_model_file}"""     )
										elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
															__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      F"""{MODEL_NAME}_{model_index}"""     )
															os.makedirs(SCREAMING_SNAKE_CASE_      ,      exist_ok=SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Saving model to {ckpt_dir}"""     )
															__lowerCAmelCase				   =       {"model": state_dict}

															dist_cp.save_state_dict(
															    state_dict=SCREAMING_SNAKE_CASE_      ,      storage_writer=dist_cp.FileSystemWriter(SCREAMING_SNAKE_CASE_     )      ,      planner=DefaultSavePlanner()      ,      )
															logger.info(F"""Model saved to {ckpt_dir}"""     )







def 	_a      (						SCREAMING_SNAKE_CASE_							:				Dict      ,      SCREAMING_SNAKE_CASE_							:				int      ,      SCREAMING_SNAKE_CASE_							:				Optional[int]      ,      SCREAMING_SNAKE_CASE_							:				List[str]      ,      SCREAMING_SNAKE_CASE_							:				Any=0     ):
					accelerator.wait_for_everyone()
					with FSDP.state_dict_type(
					    SCREAMING_SNAKE_CASE_      ,      fsdp_plugin.state_dict_type      ,      fsdp_plugin.state_dict_config      ,      fsdp_plugin.optim_state_dict_config     ):
										if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
															if type(SCREAMING_SNAKE_CASE_     ) != FSDP and accelerator.process_index != 0:
																				if not fsdp_plugin.sync_module_states:
																									raise ValueError(
																									    "Set the `sync_module_states` flag to `True` so that model states are synced across processes when "
																									    "initializing FSDP object"     )
																				return
															__lowerCAmelCase				   =       F"""{MODEL_NAME}.bin""" if model_index == 0 else F"""{MODEL_NAME}_{model_index}.bin"""
															__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Loading model from {input_model_file}"""     )
															__lowerCAmelCase				   =       torch.load(SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Model loaded from {input_model_file}"""     )
										elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
															__lowerCAmelCase				   =       (
															    F"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
															    if model_index == 0
															    else F"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
															)
															__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Loading model from {input_model_file}"""     )
															__lowerCAmelCase				   =       torch.load(SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Model loaded from {input_model_file}"""     )
										elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
															__lowerCAmelCase				   =       (
															    os.path.join(SCREAMING_SNAKE_CASE_      ,      F"""{MODEL_NAME}_{model_index}"""     )
															    if F"""{MODEL_NAME}""" not in input_dir
															    else input_dir
															)
															logger.info(F"""Loading model from {ckpt_dir}"""     )
															__lowerCAmelCase				   =       {"model": model.state_dict()}
															dist_cp.load_state_dict(
															    state_dict=SCREAMING_SNAKE_CASE_      ,      storage_reader=dist_cp.FileSystemReader(SCREAMING_SNAKE_CASE_     )      ,      planner=DefaultLoadPlanner()      ,      )
															__lowerCAmelCase				   =       state_dict["model"]
															logger.info(F"""Model loaded from {ckpt_dir}"""     )
										model.load_state_dict(SCREAMING_SNAKE_CASE_     )







def 	_a      (						SCREAMING_SNAKE_CASE_							:				Dict      ,      SCREAMING_SNAKE_CASE_							:				int      ,      SCREAMING_SNAKE_CASE_							:				List[str]      ,      SCREAMING_SNAKE_CASE_							:				Optional[Any]      ,      SCREAMING_SNAKE_CASE_							:				Optional[int]      ,      SCREAMING_SNAKE_CASE_							:				str=0     ):
					os.makedirs(SCREAMING_SNAKE_CASE_      ,      exist_ok=SCREAMING_SNAKE_CASE_     )
					with FSDP.state_dict_type(
					    SCREAMING_SNAKE_CASE_      ,      fsdp_plugin.state_dict_type      ,      fsdp_plugin.state_dict_config      ,      fsdp_plugin.optim_state_dict_config     ):
										__lowerCAmelCase				   =       FSDP.optim_state_dict(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
										if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
															if accelerator.process_index == 0:
																				__lowerCAmelCase				   =       (
																				    F"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else F"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
																				)
																				__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
																				logger.info(F"""Saving Optimizer state to {output_optimizer_file}"""     )
																				torch.save(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
																				logger.info(F"""Optimizer state saved in {output_optimizer_file}"""     )
										else:
															__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      F"""{OPTIMIZER_NAME}_{optimizer_index}"""     )
															os.makedirs(SCREAMING_SNAKE_CASE_      ,      exist_ok=SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Saving Optimizer state to {ckpt_dir}"""     )
															dist_cp.save_state_dict(
															    state_dict={"optimizer": optim_state}      ,      storage_writer=dist_cp.FileSystemWriter(SCREAMING_SNAKE_CASE_     )      ,      planner=DefaultSavePlanner()      ,      )
															logger.info(F"""Optimizer state saved in {ckpt_dir}"""     )







def 	_a      (						SCREAMING_SNAKE_CASE_							:				Dict      ,      SCREAMING_SNAKE_CASE_							:				Dict      ,      SCREAMING_SNAKE_CASE_							:				Tuple      ,      SCREAMING_SNAKE_CASE_							:				List[str]      ,      SCREAMING_SNAKE_CASE_							:				str      ,      SCREAMING_SNAKE_CASE_							:				Dict=0     ):
					accelerator.wait_for_everyone()
					with FSDP.state_dict_type(
					    SCREAMING_SNAKE_CASE_      ,      fsdp_plugin.state_dict_type      ,      fsdp_plugin.state_dict_config      ,      fsdp_plugin.optim_state_dict_config     ):
										if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
															__lowerCAmelCase				   =       None
															# below check should work but currently it isn't working (mostly opytorch issue),
															# in the meantime disabling it at the cost of excess memory usage
															# if accelerator.process_index == 0 or not fsdp_plugin.optim_state_dict_config.rank0_only:
															__lowerCAmelCase				   =       (
															    F"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else F"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
															)
															__lowerCAmelCase				   =       os.path.join(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Loading Optimizer state from {input_optimizer_file}"""     )
															__lowerCAmelCase				   =       torch.load(SCREAMING_SNAKE_CASE_     )
															logger.info(F"""Optimizer state loaded from {input_optimizer_file}"""     )
										else:
															__lowerCAmelCase				   =       (
															    os.path.join(SCREAMING_SNAKE_CASE_      ,      F"""{OPTIMIZER_NAME}_{optimizer_index}"""     )
															    if F"""{OPTIMIZER_NAME}""" not in input_dir
															    else input_dir
															)
															logger.info(F"""Loading Optimizer from {ckpt_dir}"""     )
															__lowerCAmelCase				   =       load_sharded_optimizer_state_dict(
															    model_state_dict=model.state_dict()      ,      optimizer_key="optimizer"      ,      storage_reader=dist_cp.FileSystemReader(SCREAMING_SNAKE_CASE_     )      ,      )
															__lowerCAmelCase				   =       optim_state["optimizer"]
															logger.info(F"""Optimizer loaded from {ckpt_dir}"""     )
										__lowerCAmelCase				   =       FSDP.optim_state_dict_to_load(SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_      ,      SCREAMING_SNAKE_CASE_     )
										optimizer.load_state_dict(SCREAMING_SNAKE_CASE_     )




