





"""simple docstring"""




import numpy as np



def A_     (		A__		,				A__		,				A__ = 1E-12		,				A__ = 100		,				)  ->		str:
       assert np.shape(A__    )[0] == np.shape(A__    )[1]
       # Ensure proper dimensionality.
       assert np.shape(A__    )[0] == np.shape(A__    )[0]
       # Ensure inputs are either both complex or both real
       assert np.iscomplexobj(A__    ) == np.iscomplexobj(A__    )
       a__						:     List[Any]			=    np.iscomplexobj(A__    )
       if is_complex:
              # Ensure complex input_matrix is Hermitian
              assert np.array_equal(A__		,				input_matrix.conj().T    )

       # Set convergence to False. Will define convergence when we exceed max_iterations
       # or when we have small changes from one iteration to next.

       a__						:     Optional[Any]			=    False
       a__						:     List[str]			=    0
       a__						:     Optional[int]			=    0
       a__						:     Optional[int]			=    1E12

       while not convergence:
              # Multiple matrix by the vector.
              a__						:     List[str]			=    np.dot(A__		,				A__    )
              # Normalize the resulting output vector.
              a__						:     str			=    w / np.linalg.norm(A__    )
              # Find rayleigh quotient
              # (faster than usual b/c we know vector is normalized already)
              a__						:     Optional[Any]			=    vector.conj().T if is_complex else vector.T
              a__						:     List[str]			=    np.dot(A__		,				np.dot(A__		,				A__    )    )

              # Check convergence.
              a__						:     Any			=    np.abs(lambda_ - lambda_previous    ) / lambda_
              iterations += 1

              if error <= error_tol or iterations >= max_iterations:
                     a__						:     Any			=    True

              a__						:     List[Any]			=    lambda_

       if is_complex:
              a__						:     Dict			=    np.real(lambda_    )

       return lambda_, vector



def A_     (		)  ->		List[str]:
       a__						:     List[str]			=    np.array([[41, 4, 20], [4, 26, 30], [20, 30, 50]]    )
       a__						:     Dict			=    np.array([41, 4, 20]    )
       a__						:     Any			=    real_input_matrix.astype(np.complexaaa    )
       a__						:     Union[str, Any]			=    np.triu(1J * complex_input_matrix		,				1    )
       complex_input_matrix += imag_matrix
       complex_input_matrix += -1 * imag_matrix.T
       a__						:     int			=    np.array([41, 4, 20]    ).astype(np.complexaaa    )

       for problem_type in ["real", "complex"]:
              if problem_type == "real":
                     a__						:     Optional[Any]			=    real_input_matrix
                     a__						:     Optional[Any]			=    real_vector
              elif problem_type == "complex":
                     a__						:     Tuple			=    complex_input_matrix
                     a__						:     int			=    complex_vector

              # Our implementation.
              a__			,							a__						:     Optional[int]			=    power_iteration(A__		,				A__    )

              # Numpy implementation.

              # Get eigenvalues and eigenvectors using built-in numpy
              # eigh (eigh used for symmetric or hermetian matrices).
              a__			,							a__						:     str			=    np.linalg.eigh(A__    )
              # Last eigenvalue is the maximum one.
              a__						:     Tuple			=    eigen_values[-1]
              # Last column in this matrix is eigenvector corresponding to largest eigenvalue.
              a__						:     List[Any]			=    eigen_vectors[:, -1]

              # Check our implementation and numpy gives close answers.
              assert np.abs(eigen_value - eigen_value_max    ) <= 1E-6
              # Take absolute values element wise of each eigenvector.
              # as they are only unique to a minus sign.
              assert np.linalg.norm(np.abs(A__    ) - np.abs(A__    )    ) <= 1E-6


if __name__ == "__main__":
    import doctest

    doctest.testmod()
    test_power_iteration()









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 A_     (		A__    )  ->		Optional[int]:
       if is_torch_version('<'		,				'2.0.0'    ) or not hasattr(A__		,				'_dynamo'    ):
              return False
       return isinstance(A__		,				torch._dynamo.eval_frame.OptimizedModule    )



def A_     (		A__		,				A__ = True    )  ->		int:
       a__						:     Optional[Any]			=    (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

       a__						:     Union[str, Any]			=    is_compiled_module(A__    )
       if is_compiled:
              a__						:     List[str]			=    model
              a__						:     Dict			=    model._orig_mod

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

       while isinstance(A__		,				A__    ):
              a__						:     str			=    model.module

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

       if is_compiled:
              a__						:     List[str]			=    model
              a__						:     Any			=    compiled_model

       return model



def A_     (		)  ->		int:
       PartialState().wait_for_everyone()



def A_     (		A__		,				A__    )  ->		Dict:
       if PartialState().distributed_type == DistributedType.TPU:
              xm.save(A__		,				A__    )
       elif PartialState().local_process_index == 0:
              torch.save(A__		,				A__    )



@contextmanager
def A_     (		**A__    )  ->		Any:
       for key, value in kwargs.items():
              a__						:     Optional[int]			=    str(A__    )

       yield

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



def A_     (		A__    )  ->		List[str]:
       if not hasattr(A__		,				'__qualname__'    ) and not hasattr(A__		,				'__name__'    ):
              a__						:     Dict			=    getattr(A__		,				'__class__'		,				A__    )
       if hasattr(A__		,				'__qualname__'    ):
              return obj.__qualname__
       if hasattr(A__		,				'__name__'    ):
              return obj.__name__
       return str(A__    )



def A_     (		A__		,				A__    )  ->		Dict:
       for key, value in source.items():
              if isinstance(A__		,				A__    ):
                     a__						:     Optional[Any]			=    destination.setdefault(A__		,				{}    )
                     merge_dicts(A__		,				A__    )
              else:
                     a__						:     Optional[int]			=    value

       return destination



def A_     (		A__ = None    )  ->		bool:
       if port is None:
              a__						:     List[Any]			=    2_9500
       with socket.socket(socket.AF_INET		,				socket.SOCK_STREAM    ) as s:
              return s.connect_ex(('localhost', port)    ) == 0


