Source code for eth_defi.utils

"""Bunch of random utilities."""
import calendar
import datetime
import logging
import os
import random
import socket
import time
from typing import Optional, Tuple
from urllib.parse import urlparse

import psutil

logger = logging.getLogger(__name__)

#: Ethereum 0x0000000000000000000000000000000000000000 address as a string
ZERO_ADDRESS_STR = "0x0000000000000000000000000000000000000000"

[docs]def sanitise_string(s: str) -> str: """Remove null characters.""" # return s.replace("\x00", "\U0000FFFD")
[docs]def is_localhost_port_listening(port: int, host="localhost") -> bool: """Check if a localhost is running a server already. See :return: True if there is a process occupying the port """ a_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) location = (host, port) result_of_check = a_socket.connect_ex(location) return result_of_check == 0
[docs]def find_free_port(min_port: int = 20_000, max_port: int = 40_000, max_attempt: int = 20) -> int: """Find a free localhost port to bind. Does by random. .. note :: Subject to race condition, but should be rareish. :param min_port: Minimum port range :param max_port: Maximum port range :param max_attempt: Give up and die with an exception if no port found after this many attempts. :return: Free port number """ assert type(min_port) == int assert type(max_port) == int assert type(max_attempt) == int for attempt in range(0, max_attempt): random_port = random.randrange(start=min_port, stop=max_port)"Attempting to allocate port %d to Anvil", random_port) if not is_localhost_port_listening(random_port, ""): return random_port raise RuntimeError(f"Could not open a port with a spec: {min_port} - {max_port}, {max_attempt} attempts")
[docs]def shutdown_hard( process: psutil.Popen, log_level: Optional[int] = None, block=True, block_timeout=30, check_port: Optional[int] = None, ) -> Tuple[bytes, bytes]: """Kill Psutil process. - Straight out OS `SIGKILL` a process - Log output if necessary - Use port listening to check that the process goes down and frees its ports :param process: Process to kill :param block: Block the execution until the process has terminated. You must give `check_port` option to ensure we enforce the shutdown. :param block_timeout: How long we give for process to clean up after itself :param log_level: If set, dump anything in Anvil stdout to the Python logging using level `INFO`. :param check_port: Check that TCP/IP localhost port is freed after shutdown :return: stdout, stderr as string """ stdout = b"" stderr = b"" if process.poll() is None: # Still alive, we need to kill to read the output process.kill() for line in process.stdout.readlines(): stdout += line if log_level is not None: logger._log(log_level, "stdout: %s", line.decode("utf-8").strip()) for line in process.stderr.readlines(): stderr += line if log_level is not None: logger._log(log_level, "stderr: %s", line.decode("utf-8").strip()) if block: assert check_port is not None, "Give check_port to block the execution" deadline = time.time() + 30 while time.time() < deadline: if not is_localhost_port_listening(check_port): # Port released, assume Anvil/Ganache is gone return stdout, stderr raise AssertionError(f"Could not terminate Anvil in {block_timeout} seconds, stdout is %d bytes, stderr is %d bytes", len(stdout), len(stderr)) return stdout, stderr
[docs]def to_unix_timestamp(dt: datetime.datetime) -> float: """Convert Python UTC datetime to UNIX seconds since epoch. Example: .. code-block:: python import datetime from eth_defi.utils import to_unix_timestamp dt = datetime.datetime(1970, 1, 1) unix_time = to_unix_timestamp(dt) assert unix_time == 0 :param dt: Python datetime to convert :return: Datetime as seconds since 1970-1-1 """ # return calendar.timegm(dt.utctimetuple())
[docs]def get_url_domain(url: str) -> str: """Redact URL so that only domain is displayed. Some services e.g. infura use path as an API key. """ parsed = urlparse(url) if parsed.port in (80, 443, None): return parsed.hostname else: return f"{parsed.hostname}:{parsed.port}"
[docs]def setup_console_logging(): """Set up coloured log output. - Helper function to have nicer logging output in tutorial scripts. - Tune down some noisy dependency library logging """ try: import coloredlogs except ImportError as e: raise RuntimeError("coloredlogs package missing - please install with pip first before running") from e level = os.environ.get("LOG_LEVEL", "info").upper() fmt = "%(asctime)s %(name)-44s %(message)s" date_fmt = "%H:%M:%S" coloredlogs.install(level=level, fmt=fmt, date_fmt=date_fmt) logging.basicConfig(level=level, handlers=[logging.StreamHandler()]) # Mute noise logging.getLogger("web3.providers.HTTPProvider").setLevel(logging.WARNING) logging.getLogger("web3.RequestManager").setLevel(logging.WARNING) logging.getLogger("urllib3.connectionpool").setLevel(logging.WARNING)