






import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
	from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
	import torch_xla.core.xla_model as xm





def 					lowercase_ (    __snake_case     :							int      )     -> Any:





				'''simple docstring'''


				if is_torch_version("<"	,	"2.0.0"      ) or not hasattr(__snake_case	,	"_dynamo"      ):
								return False
				return isinstance(__snake_case	,	torch._dynamo.eval_frame.OptimizedModule      )





def 					lowercase_ (    __snake_case     :							Optional[Any]	,	__snake_case     :							bool = True      )     -> Optional[int]:





				'''simple docstring'''


				snake_case__      :List[str]         =   (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

				snake_case__      :List[str]         =   is_compiled_module(__snake_case      )
				if is_compiled:
								snake_case__      :Tuple         =   model
								snake_case__      :Union[str, Any]         =   model._orig_mod

				if is_deepspeed_available():
								options += (DeepSpeedEngine,)

				while isinstance(__snake_case	,	__snake_case      ):
								snake_case__      :List[str]         =   model.module

				if not keep_fpaa_wrapper:
								snake_case__      :Optional[Any]         =   getattr(__snake_case	,	"forward"      )
								snake_case__      :Tuple         =   model.__dict__.pop("_original_forward"	,	__snake_case      )
								if original_forward is not None:
												while hasattr(__snake_case	,	"__wrapped__"      ):
																snake_case__      :Tuple         =   forward.__wrapped__
																if forward == original_forward:
																				break
												snake_case__      :Dict         =   forward
								if getattr(__snake_case	,	"_converted_to_transformer_engine"	,	__snake_case      ):
												convert_model(__snake_case	,	to_transformer_engine=__snake_case      )

				if is_compiled:
								snake_case__      :Union[str, Any]         =   model
								snake_case__      :str         =   compiled_model

				return model





def 					lowercase_ (    )     -> Optional[int]:





				'''simple docstring'''


				PartialState().wait_for_everyone()





def 					lowercase_ (    __snake_case     :							List[str]	,	__snake_case     :							Tuple      )     -> List[str]:





				'''simple docstring'''


				if PartialState().distributed_type == DistributedType.TPU:
								xm.save(__snake_case	,	__snake_case      )
				elif PartialState().local_process_index == 0:
								torch.save(__snake_case	,	__snake_case      )





@contextmanager
def 					lowercase_ (    **__snake_case     :							str      )     -> List[Any]:





				'''simple docstring'''


				for key, value in kwargs.items():
								snake_case__      :Dict         =   str(__snake_case      )

				yield

				for key in kwargs:
								if key.upper() in os.environ:
												del os.environ[key.upper()]





def 					lowercase_ (    __snake_case     :							Dict      )     -> List[str]:





				'''simple docstring'''


				if not hasattr(__snake_case	,	"__qualname__"      ) and not hasattr(__snake_case	,	"__name__"      ):
								snake_case__      :List[Any]         =   getattr(__snake_case	,	"__class__"	,	__snake_case      )
				if hasattr(__snake_case	,	"__qualname__"      ):
								return obj.__qualname__
				if hasattr(__snake_case	,	"__name__"      ):
								return obj.__name__
				return str(__snake_case      )





def 					lowercase_ (    __snake_case     :							Union[str, Any]	,	__snake_case     :							List[Any]      )     -> Union[str, Any]:





				'''simple docstring'''


				for key, value in source.items():
								if isinstance(__snake_case	,	__snake_case      ):
												snake_case__      :str         =   destination.setdefault(__snake_case	,	{}      )
												merge_dicts(__snake_case	,	__snake_case      )
								else:
												snake_case__      :Any         =   value

				return destination





def 					lowercase_ (    __snake_case     :							int = None      )     -> bool:





				'''simple docstring'''


				if port is None:
								snake_case__      :Dict         =   2_95_00
				with socket.socket(socket.AF_INET	,	socket.SOCK_STREAM      ) as s:
								return s.connect_ex(("localhost", port)      ) == 0







from __future__ import annotations





def 					lowercase_ (    __snake_case     :							list      )     -> float:





				'''simple docstring'''


				if not nums:
								raise ValueError("List is empty"      )
				return sum(__snake_case      ) / len(__snake_case      )


if __name__ == "__main__":
	import doctest

	doctest.testmod()