Source code for eth_defi.abi

"""ABI loading from the precompiled bundle.

Provides functions to load ABI files and construct :py:class:`web3.contract.Contract` types.
The results are cached for the speedup.

We also provide some helper functions to deal with ABI encode/decode.

`See Github for available contracts ABI files <https://github.com/tradingstrategy-ai/web3-ethereum-defi/tree/master/eth_defi/abi>`_.
"""
import json
import re
from functools import lru_cache
from pathlib import Path
from typing import Optional, Sequence, Type, Union

import eth_abi
from eth_abi import decode
from eth_typing import HexAddress
from hexbytes import HexBytes
from web3 import Web3
from web3._utils.abi import get_abi_input_names, get_abi_input_types
from web3._utils.contracts import encode_abi, get_function_info
from web3.contract.contract import Contract, ContractFunction

# Cache loaded ABI files in-process memory for speedup
from web3.datastructures import AttributeDict

from eth_defi.uniswap_v2.utils import ZERO_ADDRESS
from eth_defi.utils import ZERO_ADDRESS_STR

# How big are our ABI and contract caches
_CACHE_SIZE = 512



[docs]@lru_cache(maxsize=_CACHE_SIZE)
def get_abi_by_filename(fname: str) -> dict:
    """Reads a embedded ABI file and returns it.

    Example::

        abi = get_abi_by_filename("ERC20Mock.json")

    You are most likely interested in the keys `abi` and `bytecode` of the JSON file.

    Loaded ABI files are cache in in-process memory to speed up future loading.

    Any results are cached.

    :param web3: Web3 instance
    :param fname: `JSON filename from supported contract lists <https://github.com/tradingstrategy-ai/web3-ethereum-defi/tree/master/eth_defi/abi>`_.
    :return: Full contract interface, including `bytecode`.
    """

    here = Path(__file__).resolve().parent
    abi_path = here / "abi" / Path(fname)
    with open(abi_path, "rt", encoding="utf-8") as f:
        abi = json.load(f)
    return abi




[docs]@lru_cache(maxsize=_CACHE_SIZE)
def get_contract(
    web3: Web3,
    fname: str | Path,
    bytecode: Optional[str] = None,
) -> Type[Contract]:
    """Create a Contract proxy class from our bundled contracts or filesystem..

    `See Web3.py documentation on Contract instances <https://web3py.readthedocs.io/en/stable/contracts.html#contract-deployment-example>`_.

    Any results are cached. Web3 connection is part of the cache key.

    .. note ::

        This function cannot do linking. See :py:func:`get_linked_contract`
        if the bytecode contains link markers.

    Example:

        pass

    :param web3:
        Web3 instance

    :param bytecode:
        Override bytecode payload for the contract

    :param fname:
        Solidity compiler artifact.

        Use slash prefixed path for absolute lookups.

        `JSON filename from supported contract lists <https://github.com/tradingstrategy-ai/web3-ethereum-defi/tree/master/eth_defi/abi>`_.

    :return:
        Contract proxy class
    """

    contract_interface = get_abi_by_filename(fname)
    abi = contract_interface["abi"]

    if bytecode is None:
        bytecode = contract_interface["bytecode"]

        if type(bytecode) == dict:
            # Sol 0.8 / Forge?
            # Contains keys object, sourceMap, linkReferences
            bytecode = bytecode["object"]
        else:
            # Sol 0.6 / legacy
            # Bytecode hex is directly in the key.
            pass

    Contract = web3.eth.contract(abi=abi, bytecode=bytecode)
    return Contract




[docs]def get_linked_contract(
    web3: Web3,
    fname: str | Path,
    hardhat_export_data: Optional[dict] = None,
) -> Type[Contract]:
    """Create a Contract proxy class from our bundled contracts or filesystem and links it Solidity bytecode.

    Needed when contracts contain references to libraries. The contract bytecode
    must be processed and placeholders must be replaced by the on-chain addresses
    of the deployed library contracts.

    Example:

    .. code-block:: python

        path = self.path.joinpath("artifacts/@aave/core-v3/contracts/mocks/tokens/MintableERC20.sol/MintableERC20.json")
        return get_linked_contract(web3, path, get_aave_hardhard_export())

    .. note ::

        If you do not need linking use :py:func:`get_contract` which is faster.

    :param web3:
        Web3 instance

    :param fname:
        Solidity compiler artifact.

        Use slash prefixed path for absolute lookups.

        `JSON filename from supported contract lists <https://github.com/tradingstrategy-ai/web3-ethereum-defi/tree/master/eth_defi/abi>`_.

    :param hardhat_export_data:
        Hardhat deployment export data to link bytecode.

        A JSON file generated by `hardhat deploy --export` command.

    :return:
        Contract proxy class
    """

    contract_interface = get_abi_by_filename(fname)
    abi = contract_interface["abi"]
    bytecode = contract_interface["deployedBytecode"]

    link_references = contract_interface["linkReferences"]
    bytecode = link_libraries_hardhat(bytecode, link_references, hardhat_export_data)

    Contract = web3.eth.contract(abi=abi, bytecode=bytecode)
    return Contract




[docs]def get_deployed_contract(
    web3: Web3,
    fname: str,
    address: Union[HexAddress, str],
    register_for_tracing: bool = True,
) -> Contract:
    """Get a Contract proxy objec for a contract deployed at a specific address.

    `See Web3.py documentation on Contract instances <https://web3py.readthedocs.io/en/stable/contracts.html#contract-deployment-example>`_.

    :param web3:
        Web3 instance

    :param fname:
        `JSON filename from supported contract lists <https://github.com/tradingstrategy-ai/web3-ethereum-defi/tree/master/eth_defi/abi>`_.

    :param address:
        Ethereum address of the deployed contract

    :param register_for_tracing:
        Add the contract to the deployment registry if not already there.

    :return:
        `web3.contract.Contract` proxy
    """
    assert address

    address = Web3.to_checksum_address(address)

    Contract = get_contract(web3, fname)
    contract = Contract(address)

    if register_for_tracing:
        # TODO: Currently hack around circular imports, move functoins
        from eth_defi.deploy import get_registered_contract, register_contract

        registered_contract = get_registered_contract(web3, address)
        if registered_contract is None:
            register_contract(web3, address, contract)

    return contract




[docs]def get_transaction_data_field(tx: AttributeDict) -> str:
    """Get the "Data" payload of a transaction.

    Ethereum Tester has this in tx.data while Ganache has this in tx.input.
    Yes, it is madness.

    Example:

    .. code-block::

        tx = web3.eth.get_transaction(tx_hash)
        function, input_args = router.decode_function_input(get_transaction_data_field(tx))
        print("Transaction {tx_hash} called function {function}")

    """
    if "data" in tx:
        return tx["data"]
    else:
        return tx["input"]




[docs]def encode_with_signature(function_signature: str, args: Sequence) -> bytes:
    """Mimic Solidity's abi.encodeWithSignature() in Python.

    This is a Python equivalent for `abi.encodeWithSignature()`.

    Example:

    .. code-block:: python

            payload = encode_with_signature("init(address)", [my_address])
            assert type(payload) == bytes

    :param function_signature:
        Solidity function signature that can be hashed to a selector.

        ABI fill be extractd from this signature.

    :param args:
        Argument values to be encoded.
    """

    assert type(args) in (tuple, list)

    function_selector = Web3.keccak(text=function_signature)
    selector_text = function_signature[function_signature.find("(") + 1 : function_signature.rfind(")")]
    arg_types = selector_text.split(",")
    encoded_args = eth_abi.encode(arg_types, args)
    return function_selector + encoded_args




[docs]def encode_function_args(func: ContractFunction, args: Sequence) -> bytes:
    """Mimic Solidity's abi.encodeWithSignature() in Python.

    Uses `web3.Contract.functions` prepared function as the ABI source.

    :param func:
        Function which arguments we are going to encode.

    :param args:
        Argument values to be encoded.
    """
    assert isinstance(func, ContractFunction)

    web3 = func.w3

    fn_abi, fn_selector, aligned_fn_arguments = get_function_info(
        func.fn_name,
        web3.codec,
        func.contract_abi,
        args=args,
    )
    arg_types = [t["type"] for t in fn_abi["inputs"]]
    encoded_args = eth_abi.encode(arg_types, args)
    return encoded_args




[docs]def encode_function_call(
    func: ContractFunction,
    args: Sequence,
) -> HexBytes:
    """Encode function selector + its arguments as data payload.

    Uses `web3.Contract.functions` prepared function as the ABI source.

    See also :py:func:`encode_function_args`.

    :param func:
        Function which arguments we are going to encode.

    :param args:
        Argument values to be encoded.

    :return:
        Solidity's function selector + argument payload.

    """
    w3 = func.w3
    contract_abi = func.contract_abi
    fn_abi = func.abi
    fn_identifier = func.function_identifier
    fn_abi, fn_selector, fn_arguments = get_function_info(
        # type ignored b/c fn_id here is always str b/c FallbackFn is handled above
        fn_identifier,  # type: ignore
        w3.codec,
        contract_abi,
        fn_abi,
        args,
    )
    encoded = encode_abi(w3, fn_abi, fn_arguments, fn_selector)
    return HexBytes(encoded)




[docs]def decode_function_args(
    func: ContractFunction,
    data: bytes | HexBytes,
) -> dict:
    """Decode binary CALL or CALLDATA to a Solidity function,

    Uses `web3.Contract.functions` prepared function as the ABI source.

    :param func:
        Function which arguments we are going to encode.

    :param data:
        Extracted from a transaction data field or EVM memoryo trace.

    :return:
        Ordered dict of the decoded arguments
    """
    assert isinstance(func, ContractFunction)
    fn_abi = func.abi

    arg_names = get_abi_input_names(fn_abi)
    arg_types = get_abi_input_types(fn_abi)
    arg_tuple = decode(arg_types, data)

    return dict(zip(arg_names, arg_tuple))




[docs]def humanise_decoded_arg_data(args: dict) -> dict:
    """Make decoded arguments more human readable.

    - All arguments are converted to good text types

    See :py:func:`decode_function_args`

    :return:
        Ordered dict of decoded arguments, easier to read
    """

    def _humanize(v):
        if type(v) == bytes:
            return v.hex()
        return v

    return {k: _humanize(v) for k, v in args.items()}