pywiki/pyhmy/staking.py

from .rpc.request import (
    rpc_request
)

_default_endpoint = 'http://localhost:9500'
_default_timeout = 30


##################
# Validator RPCs #
##################
def get_all_validator_addresses(endpoint=_default_endpoint, timeout=_default_timeout) -> list:
    """
    Get list of all created validator addresses on chain

    Parameters
    ----------
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    list
        List of one addresses for all validators on chain
    """
    return rpc_request('hmy_getAllValidatorAddresses', endpoint=endpoint, timeout=timeout)['result']


def get_validator_information(validator_addr, endpoint=_default_endpoint, timeout=_default_timeout) -> dict:
    """
    Get validator information for validator address

    Parameters
    ----------
    validator_addr: str
        One address of the validator to get information for
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    dict
        # TODO: Add link to reference RPC documentation
    """
    params = [
        validator_addr
    ]
    return rpc_request('hmy_getValidatorInformation', params=params, endpoint=endpoint, timeout=timeout)['result']


def get_validator_information_by_block(validator_addr, block_num, endpoint=_default_endpoint, timeout=_default_timeout):
    """
    Get validator information for validator address at a block

    Parameters
    ----------
    validator_addr: str
        One address of the validator to get information for
    block_num: int
        Block number to query validator information at
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    list
        # TODO: Add link to reference RPC documentation
    """
    params = [
        validator_addr,
        str(hex(block_num))
    ]
    return rpc_request('hmy_getValidatorInformationByBlockNumber', params=params, endpoint=endpoint, timeout=timeout)['result']


def get_all_validator_information(page=-1, endpoint=_default_endpoint, timeout=_default_timeout) -> list:
    """
    Get validator information for all validators on chain

    Parameters
    ----------
    page: :obj:`int`, optional
        Page to request (-1 for all validators)
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    list
        # TODO: Add link to reference RPC documentation
    """
    params = [
        page
    ]
    return rpc_request('hmy_getAllValidatorInformation', params=params, endpoint=endpoint, timeout=timeout)['result']


def get_all_validator_information_by_block(block_num, page=-1, endpoint=_default_endpoint, timeout=_default_timeout) -> list:
    """
    Get validator information at block number for all validators on chain

    Parameters
    ----------
    block_num: int
        Block number to get validator information for
    page: :obj:`int`, optional
        Page to request (-1 for all validators)
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    list
        # TODO: Add link to reference RPC documentation
    """
    params = [
        page,
        str(hex(block_num))
    ]
    return rpc_request('hmy_getAllValidatorInformationByBlockNumber', params=params, endpoint=endpoint, timeout=timeout)['result']


###################
# Delegation RPCs #
###################
def get_delegations_by_delegator(delegator_addr, endpoint=_default_endpoint, timeout=_default_timeout) -> list:
    """
    Get list of delegations by a delegator

    Parameters
    ----------
    delegator_addr: str
        Delegator address to get list of delegations for
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    list
        # TODO: Add link to reference RPC documentation
    """
    params = [
        delegator_addr
    ]
    return rpc_request('hmy_getDelegationsByDelegator', params=params, endpoint=endpoint, timeout=timeout)['result']


def get_delegations_by_delegator_by_block(delegator_addr, block_num, endpoint=_default_endpoint, timeout=_default_timeout) -> list:
    """
    Get list of delegations by a delegator at a specific block

    Parameters
    ----------
    delegator_addr: str
        Delegator address to get list of delegations for
    block_num: int
        Block number to query delgator information at
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    list
        # TODO: Add link to reference RPC documentation
    """
    params = [
        delegator_addr,
        str(hex(block_num))
    ]
    return rpc_request('hmy_getDelegationsByDelegatorByBlockNumber', params=params, endpoint=endpoint, timeout=timeout)['result']


def get_delegations_by_validator(validator_addr, endpoint=_default_endpoint, timeout=_default_timeout) -> list:
    """
    Get list of delegations to a validator

    Parameters
    ----------
    validator_addr: str
        Validator address to get list of delegations for
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    list
        # TODO: Add link to reference RPC documentation
    """
    params = [
        validator_addr
    ]
    return rpc_request('hmy_getDelegationsByValidator', params=params, endpoint=endpoint, timeout=timeout)['result']


########################
# Staking Network RPCs #
########################
def get_current_utility_metrics(endpoint=_default_endpoint, timeout=_default_timeout) -> dict:
    """
    Get current utility metrics of network

    Parameters
    ----------
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    dict
        # TODO: Add link to reference RPC documentation
    """
    return rpc_request('hmy_getCurrentUtilityMetrics', endpoint=endpoint, timeout=timeout)['result']


def get_staking_network_info(endpoint=_default_endpoint, timeout=_default_timeout) -> dict:
    """
    Get staking network information

    Parameters
    ----------
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    dict
        # TODO: Add link to reference RPC documentation
    """
    return rpc_request('hmy_getStakingNetworkInfo', endpoint=endpoint, timeout=timeout)['result']


def get_super_committees(endpoint=_default_endpoint, timeout=_default_timeout) -> dict:
    """
    Get voting committees for current & previous epoch

    Parameters
    ----------
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    dict
        # TODO: Add link to reference RPC documentation
    """
    return rpc_request('hmy_getSuperCommittees', endpoint=endpoint, timeout=timeout)['result']


def get_raw_median_stake_snapshot(endpoint=_default_endpoint, timeout=_default_timeout) -> dict:
    """
    Get median stake & additional committee data of the current epoch

    Parameters
    ----------
    endpoint: :obj:`str`, optional
        Endpoint to send request to
    timeout: :obj:`int`, optional
        Timeout in seconds

    Returns
    -------
    dict
        # TODO: Add link to reference RPC documentation
    """
    return rpc_request('hmy_getMedianRawStakeSnapshot', endpoint=endpoint, timeout=timeout)['result']