utils

Documentation for eth_defi.utils Python module.

Bunch of random utilities.

Functions

addr(address)

Convert various address formats to HexAddress.

chunked(iterable, chunk_size)

find_free_port([min_port, max_port, max_attempt])

Find a free localhost port to bind.

from_unix_timestamp(timestamp)

Convert UNIX seconds since epoch to naive Python datetime.

get_url_domain(url)

Redact URL so that only domain is displayed.

is_good_multichain_address(address)

Check if a vault address has a recognised format.

is_localhost_port_listening(port[, host])

Check if a localhost is running a server already.

sanitise_string(s[, max_length])

Remove null characters.

setup_console_logging([default_log_level, ...])

Set up coloured log output.

shutdown_hard(process[, log_level, block, ...])

Kill Psutil process.

to_unix_timestamp(dt)

Convert Python UTC datetime to UNIX seconds since epoch.

wait_other_writers(path[, timeout])

Wait other potential writers writing the same file.

Classes

ThreadColourFormatter

Log formatter that assigns a unique ANSI colour to each thread name.
is_good_multichain_address(address)

Check if a vault address has a recognised format.

  • EVM vaults use 0x-prefixed hex addresses

  • Non-EVM protocols like GRVT use platform-specific IDs (e.g. VLT:xxx)

  • Lighter pools use synthetic IDs (e.g. lighter-pool-281474976710654)

Parameters

address (str) – The vault address string to validate.

Returns

True if the address starts with a known prefix.

Return type

bool

sanitise_string(s, max_length=None)

Remove null characters.

Parameters

  • s (str) –

  • max_length (int | None) –

Return type

str

is_localhost_port_listening(port, host='localhost')

Check if a localhost is running a server already.

See https://www.adamsmith.haus/python/answers/how-to-check-if-a-network-port-is-open-in-python

Returns

True if there is a process occupying the port

Parameters

port (int) –

Return type

bool

find_free_port(min_port=20000, max_port=40000, max_attempt=20)

Find a free localhost port to bind.

Does by random.

Note

Subject to race condition, but should be rareish.

Parameters

  • min_port (int) – Minimum port range

  • max_port (int) – Maximum port range

  • max_attempt (int) – Give up and die with an exception if no port found after this many attempts.

Returns

Free port number

Return type

int

shutdown_hard(process, log_level=None, block=True, block_timeout=30, check_port=None)

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

Parameters

  • process (psutil.Popen) – Process to kill

  • block –

    Block the execution until the process has terminated.

    You must give check_port option to ensure we enforce the shutdown.

  • block_timeout – How long we give for process to clean up after itself

  • log_level (Optional[int]) – If set, dump anything in Anvil stdout to the Python logging using level INFO.

  • check_port (Optional[int]) – Check that TCP/IP localhost port is freed after shutdown

Returns

stdout, stderr as string

Return type

tuple[bytes, bytes]

to_unix_timestamp(dt)

Convert Python UTC datetime to UNIX seconds since epoch.

Example:

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
Parameters

dt (datetime.datetime) – Python datetime to convert

Returns

Datetime as seconds since 1970-1-1

Return type

float

from_unix_timestamp(timestamp)

Convert UNIX seconds since epoch to naive Python datetime.

Parameters

timestamp (float) – Timestamp in since 1970-1-1 as float or int as seconds

Returns

Naive Python datetime in UTC timezone (tzinfo is None, but the time is in UTC)

Return type

datetime.datetime

get_url_domain(url)

Redact URL so that only domain is displayed.

Some services e.g. infura use path as an API key.

Parameters

url (str) –

Return type

str

class ThreadColourFormatter

Bases: logging.Formatter

Log formatter that assigns a unique ANSI colour to each thread name.

Wraps an existing formatter (e.g. the one installed by coloredlogs) and replaces the plain thread name in the formatted output with a colour-coded version. This preserves all other colours (timestamp, logger name, level) that the underlying formatter provides.

When no inner formatter is given, falls back to standard logging.Formatter behaviour.

Colours are drawn from a palette of bold ANSI codes and assigned on first encounter, cycling if more threads appear than palette entries.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

__init__(inner=None, fmt=None, datefmt=None)

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

Parameters

inner (logging.Formatter | None) –

format(record)

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

Parameters

record (logging.LogRecord) –

Return type

str

converter()
localtime([seconds]) -> (tm_year,tm_mon,tm_mday,tm_hour,tm_min,

tm_sec,tm_wday,tm_yday,tm_isdst)

Convert seconds since the Epoch to a time tuple expressing local time. When ‘seconds’ is not passed in, convert the current time instead.

formatException(ei)

Format and return the specified exception information as a string.

This default implementation just uses traceback.print_exception()

formatStack(stack_info)

This method is provided as an extension point for specialized formatting of stack information.

The input data is a string as returned from a call to traceback.print_stack(), but with the last trailing newline removed.

The base implementation just returns the value passed in.

formatTime(record, datefmt=None)

Return the creation time of the specified LogRecord as formatted text.

This method should be called from format() by a formatter which wants to make use of a formatted time. This method can be overridden in formatters to provide for any specific requirement, but the basic behaviour is as follows: if datefmt (a string) is specified, it is used with time.strftime() to format the creation time of the record. Otherwise, an ISO8601-like (or RFC 3339-like) format is used. The resulting string is returned. This function uses a user-configurable function to convert the creation time to a tuple. By default, time.localtime() is used; to change this for a particular formatter instance, set the ‘converter’ attribute to a function with the same signature as time.localtime() or time.gmtime(). To change it for all formatters, for example if you want all logging times to be shown in GMT, set the ‘converter’ attribute in the Formatter class.

usesTime()

Check if the format uses the creation time of the record.

setup_console_logging(default_log_level='warning', simplified_logging=False, log_file=None, std_out_log_level=None, only_log_file=False, clear_log_file=True, coloured_threads=False)

Set up coloured log output.

  • Helper function to have nicer logging output in tutorial scripts.

  • Tune down some noisy dependency library logging

Parameters

  • log_file (str | pathlib.Path) – Output both console and this log file.

  • coloured_threads – When True, each thread name in the log output gets a unique ANSI colour so interleaved parallel logs are easy to follow visually.

  • std_out_log_level (Optional[int]) –

Returns

Root logger

Return type

logging.Logger

addr(address)

Convert various address formats to HexAddress.

Args:

address: Can be a string, HexAddress, or HexStr

Returns:

HexAddress object

Parameters

address (Union[str, eth_typing.evm.HexAddress, eth_typing.encoding.HexStr]) –

Return type

eth_typing.evm.HexAddress

wait_other_writers(path, timeout=120)

Wait other potential writers writing the same file.

  • Work around issues when parallel unit tests and such try to write the same file

Example:

import urllib
import tempfile

import pytest
import pandas as pd


@pytest.fixture()
def my_cached_test_data_frame() -> pd.DataFrame:
    # Al tests use a cached dataset stored in the /tmp directory
    path = os.path.join(tempfile.gettempdir(), "my_shared_data.parquet")

    with wait_other_writers(path):
        # Read result from the previous writer
        if not path.exists(path):
            # Download and write to cache
            urllib.request.urlretrieve("https://example.com", path)

        return pd.read_parquet(path)
Parameters

  • path (pathlib.Path | str) – File that is being written

  • timeout (int) –

    How many seconds wait to acquire the lock file.

    Default 2 minutes.

Raises

filelock.Timeout – If the file writer is stuck with the lock.