






'''simple docstring'''

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 	_UpperCAmelCase    (  a		:	str							)			->			str:





      """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 	_UpperCAmelCase    (  a		:	Union[str, Any]						,			a		:	bool = True							)			->			List[str]:





      """simple docstring"""
      lowercase_      :   Optional[Any]        =			(torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

      lowercase_      :   int        =			is_compiled_module(snake_case__							)
      if is_compiled:
            lowercase_      :   str        =			model
            lowercase_      :   Optional[int]        =			model._orig_mod

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

      while isinstance(snake_case__						,			snake_case__							):
            lowercase_      :   Optional[Any]        =			model.module

      if not keep_fpaa_wrapper:
            lowercase_      :   int        =			getattr(snake_case__						,			'forward'							)
            lowercase_      :   int        =			model.__dict__.pop('_original_forward'						,			snake_case__							)
            if original_forward is not None:
                  while hasattr(snake_case__						,			'__wrapped__'							):
                        lowercase_      :   List[str]        =			forward.__wrapped__
                        if forward == original_forward:
                              break
                  lowercase_      :   Optional[int]        =			forward
            if getattr(snake_case__						,			'_converted_to_transformer_engine'						,			snake_case__							):
                  convert_model(snake_case__						,			to_transformer_engine=snake_case__							)

      if is_compiled:
            lowercase_      :   Dict        =			model
            lowercase_      :   Optional[int]        =			compiled_model

      return model






def 	_UpperCAmelCase    (  )			->			List[Any]:





      """simple docstring"""
      PartialState().wait_for_everyone()






def 	_UpperCAmelCase    (  a		:	Optional[int]						,			a		:	Optional[int]							)			->			Any:





      """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 	_UpperCAmelCase    (  **a		:	Tuple							)			->			str:





      """simple docstring"""
      for key, value in kwargs.items():
            lowercase_      :   List[Any]        =			str(snake_case__							)

      yield

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






def 	_UpperCAmelCase    (  a		:	Any							)			->			List[str]:





      """simple docstring"""
      if not hasattr(snake_case__						,			'__qualname__'							) and not hasattr(snake_case__						,			'__name__'							):
            lowercase_      :   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 	_UpperCAmelCase    (  a		:	List[str]						,			a		:	Optional[int]							)			->			List[Any]:





      """simple docstring"""
      for key, value in source.items():
            if isinstance(snake_case__						,			snake_case__							):
                  lowercase_      :   str        =			destination.setdefault(snake_case__						,			{}							)
                  merge_dicts(snake_case__						,			snake_case__							)
            else:
                  lowercase_      :   int        =			value

      return destination






def 	_UpperCAmelCase    (  a		:	int = None							)			->			bool:





      """simple docstring"""
      if port is None:
            lowercase_      :   Optional[Any]        =			2_9_5_0_0
      with socket.socket(socket.AF_INET						,			socket.SOCK_STREAM							) as s:
            return s.connect_ex(('localhost', port)							) == 0








'''simple docstring'''

import os
from distutils.util import strtobool






def 	_UpperCAmelCase    (  a		:	Any						,			a		:	int							)			->			Any:





						"""simple docstring"""
						for e in env_keys:
												lowercase_      :   Optional[Any]        =			int(os.environ.get(a						,			-1							)							)
												if val >= 0:
																		return val
						return default






def 	_UpperCAmelCase    (  a		:	List[Any]						,			a		:	Dict=False							)			->			Optional[Any]:





						"""simple docstring"""
						lowercase_      :   Optional[int]        =			os.environ.get(a						,			str(a							)							)
						return strtobool(a							) == 1  # As its name indicates `strtobool` actually returns an int...






def 	_UpperCAmelCase    (  a		:	List[Any]						,			a		:	Dict="no"							)			->			str:





						"""simple docstring"""
						lowercase_      :   List[Any]        =			os.environ.get(a						,			str(a							)							)
						return value
