pywiki/pyhmy/util.py

"""
Basic pyhmy utils like is_shard_active
ONE address format conversion
Chain id (str) to int conversion
"""
import json
import subprocess
import os
import sys
import datetime

from eth_utils import to_checksum_address

from .blockchain import get_latest_header

from .rpc.exceptions import (
    RPCError,
    RequestsError,
    RequestsTimeoutError,
)

from .account import is_valid_address

from .bech32.bech32 import bech32_decode, bech32_encode, convertbits

class Typgpy(str):
    """Typography constants for pretty printing.

    Note that an ENDC is needed to mark the end of a 'highlighted' text
    segment.
    """

    HEADER = "\033[95m"
    OKBLUE = "\033[94m"
    OKGREEN = "\033[92m"
    WARNING = "\033[93m"
    FAIL = "\033[91m"
    ENDC = "\033[0m"
    BOLD = "\033[1m"
    UNDERLINE = "\033[4m"


def chain_id_to_int(chain_id):
    """
    If chain_id is a string, converts it to int.
    If chain_id is an int, returns the int.

    Else raises TypeError
    """
    chain_ids = dict(
        Default=0,
        EthMainnet=1,
        Morden=2,
        Ropsten=3,
        Rinkeby=4,
        RootstockMainnet=30,
        RootstockTestnet=31,
        Kovan=42,
        EtcMainnet=61,
        EtcTestnet=62,
        Geth=1337,
        Ganache=0,
        HmyMainnet=1,
        HmyTestnet=2,
        HmyLocal=2,
        HmyPangaea=3,
    )

    # do not validate integer chainids, only known strings
    if isinstance(chain_id, str):
        assert (
            chain_id in chain_ids
        ), f"Chain {chain_id} unknown, specify an integer chainId"
        return chain_ids.get(chain_id)
    if isinstance(chain_id, int):
        return chain_id
    raise TypeError("chainId must be str or int")


def get_gopath():
    """
    :returns The go-path, assuming that go is installed.
    """
    return subprocess.check_output(["go", "env", "GOPATH"]).decode().strip()


def get_goversion():
    """
    :returns The go-version, assuming that go is installed.
    """
    return subprocess.check_output(["go", "version"]).decode().strip()


def convert_one_to_hex(addr):
    """Given a one address, convert it to hex checksum address."""
    if not is_valid_address(addr):
        return to_checksum_address(addr)
    _, data = bech32_decode(addr)
    buf = convertbits(data, 5, 8, False)
    address = "0x" + "".join(f"{x:02x}" for x in buf)
    return str(to_checksum_address(address))


def convert_hex_to_one(addr):
    """Given a hex address, convert it to a one address."""
    if is_valid_address(addr):
        return addr
    checksum_addr = str(to_checksum_address(addr))
    data = bytearray.fromhex(
        checksum_addr[2:] if checksum_addr.startswith("0x") else checksum_addr
    )
    buf = convertbits(data, 8, 5)
    return str(bech32_encode("one", buf))


def is_active_shard(endpoint, delay_tolerance=60):
    """
    :param endpoint: The endpoint of the SHARD to check
    :param delay_tolerance: The time (in seconds) that the shard timestamp can be behind
    :return: If shard is active or not
    """
    try:
        curr_time = datetime.datetime.utcnow()
        latest_header = get_latest_header(endpoint=endpoint)
        time_str = latest_header["timestamp"][:19] + ".0"  # Fit time format
        timestamp = datetime.datetime.strptime(
            time_str, "%Y-%m-%d %H:%M:%S.%f"
        ).replace(tzinfo=None)
        time_delta = curr_time - timestamp
        return abs(time_delta.seconds) < delay_tolerance
    except (RPCError, RequestsError, RequestsTimeoutError):
        return False


def get_bls_build_variables():
    """
    :returns The environment variables needed to build and/or run programs that
             use the Harmony BLS & MCL repo.
    :raises RuntimeError if openssl is not found.

    Note that this assumes that the BLS & MCL repo are in the appropriate directory
    as stated here: https://github.com/harmony-one/harmony/blob/master/README.md
    """
    variables = {}
    try:
        openssl_dir = (
            subprocess.check_output(["which", "openssl"])
            .decode()
            .strip()
            .split("\n", maxsplit=1)[0]
        )
    except (IndexError, subprocess.CalledProcessError) as exception:
        raise RuntimeError("`openssl` not found") from exception
    hmy_path = f"{get_gopath()}/src/github.com/harmony-one"
    bls_dir = f"{hmy_path}/bls"
    mcl_dir = f"{hmy_path}/mcl"
    assert os.path.exists(bls_dir), f"Harmony BLS repo not found at {bls_dir}"
    assert os.path.exists(mcl_dir), f"Harmony MCL repo not found at {mcl_dir}"
    if sys.platform.startswith("darwin"):
        variables[
            "CGO_CFLAGS"
        ] = f"-I{bls_dir}/include -I{mcl_dir}/include -I{openssl_dir}/include"
        variables["CGO_LDFLAGS"] = f"-L{bls_dir}/lib -L{openssl_dir}/lib"
        variables["LD_LIBRARY_PATH"] = f"{bls_dir}/lib:{mcl_dir}/lib:{openssl_dir}/lib"
        variables["DYLD_FALLBACK_LIBRARY_PATH"] = variables["LD_LIBRARY_PATH"]
    else:
        variables["CGO_CFLAGS"] = f"-I{bls_dir}/include -I{mcl_dir}/include"
        variables["CGO_LDFLAGS"] = f"-L{bls_dir}/lib"
        variables["LD_LIBRARY_PATH"] = f"{bls_dir}/lib:{mcl_dir}/lib"
    return variables


def json_load(string, **kwargs):
    """
    :param string: The JSON string to load
    :returns A dictionary loaded from a JSON string to a dictionary.
    :raises The exception caused by the load (if present).

    Note that this prints the failed input should an error arise.
    """
    try:
        return json.loads(string, **kwargs)
    except Exception as exception:
        print(f"{Typgpy.FAIL}Could not parse input: '{string}'{Typgpy.ENDC}")
        raise exception