Switch using web3py instead brownie.

crawlers/mooncrawl/mooncrawl/state_crawler/Multicall2_abi.json

@@ -1 +1,313 @@
-[{"inputs": [{"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "aggregate", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}, {"internalType": "bytes[]", "name": "returnData", "type": "bytes[]"}], "stateMutability": "nonpayable", "type": "function"}, {"inputs": [{"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "blockAndAggregate", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}, {"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}, {"components": [{"internalType": "bool", "name": "success", "type": "bool"}, {"internalType": "bytes", "name": "returnData", "type": "bytes"}], "internalType": "struct Multicall2.Result[]", "name": "returnData", "type": "tuple[]"}], "stateMutability": "nonpayable", "type": "function"}, {"inputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}], "name": "getBlockHash", "outputs": [{"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getBlockNumber", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockCoinbase", "outputs": [{"internalType": "address", "name": "coinbase", "type": "address"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockDifficulty", "outputs": [{"internalType": "uint256", "name": "difficulty", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockGasLimit", "outputs": [{"internalType": "uint256", "name": "gaslimit", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getCurrentBlockTimestamp", "outputs": [{"internalType": "uint256", "name": "timestamp", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"internalType": "address", "name": "addr", "type": "address"}], "name": "getEthBalance", "outputs": [{"internalType": "uint256", "name": "balance", "type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "getLastBlockHash", "outputs": [{"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"internalType": "bool", "name": "requireSuccess", "type": "bool"}, {"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "tryAggregate", "outputs": [{"components": [{"internalType": "bool", "name": "success", "type": "bool"}, {"internalType": "bytes", "name": "returnData", "type": "bytes"}], "internalType": "struct Multicall2.Result[]", "name": "returnData", "type": "tuple[]"}], "stateMutability": "nonpayable", "type": "function"}, {"inputs": [{"internalType": "bool", "name": "requireSuccess", "type": "bool"}, {"components": [{"internalType": "address", "name": "target", "type": "address"}, {"internalType": "bytes", "name": "callData", "type": "bytes"}], "internalType": "struct Multicall2.Call[]", "name": "calls", "type": "tuple[]"}], "name": "tryBlockAndAggregate", "outputs": [{"internalType": "uint256", "name": "blockNumber", "type": "uint256"}, {"internalType": "bytes32", "name": "blockHash", "type": "bytes32"}, {"components": [{"internalType": "bool", "name": "success", "type": "bool"}, {"internalType": "bytes", "name": "returnData", "type": "bytes"}], "internalType": "struct Multicall2.Result[]", "name": "returnData", "type": "tuple[]"}], "stateMutability": "nonpayable", "type": "function"}]
+[
+    {
+        "inputs": [
+            {
+                "components": [
+                    {
+                        "internalType": "address",
+                        "name": "target",
+                        "type": "address"
+                    },
+                    {
+                        "internalType": "bytes",
+                        "name": "callData",
+                        "type": "bytes"
+                    }
+                ],
+                "internalType": "struct Multicall2.Call[]",
+                "name": "calls",
+                "type": "tuple[]"
+            }
+        ],
+        "name": "aggregate",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "blockNumber",
+                "type": "uint256"
+            },
+            {
+                "internalType": "bytes[]",
+                "name": "returnData",
+                "type": "bytes[]"
+            }
+        ],
+        "stateMutability": "nonpayable",
+        "type": "function"
+    },
+    {
+        "inputs": [
+            {
+                "components": [
+                    {
+                        "internalType": "address",
+                        "name": "target",
+                        "type": "address"
+                    },
+                    {
+                        "internalType": "bytes",
+                        "name": "callData",
+                        "type": "bytes"
+                    }
+                ],
+                "internalType": "struct Multicall2.Call[]",
+                "name": "calls",
+                "type": "tuple[]"
+            }
+        ],
+        "name": "blockAndAggregate",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "blockNumber",
+                "type": "uint256"
+            },
+            {
+                "internalType": "bytes32",
+                "name": "blockHash",
+                "type": "bytes32"
+            },
+            {
+                "components": [
+                    {
+                        "internalType": "bool",
+                        "name": "success",
+                        "type": "bool"
+                    },
+                    {
+                        "internalType": "bytes",
+                        "name": "returnData",
+                        "type": "bytes"
+                    }
+                ],
+                "internalType": "struct Multicall2.Result[]",
+                "name": "returnData",
+                "type": "tuple[]"
+            }
+        ],
+        "stateMutability": "nonpayable",
+        "type": "function"
+    },
+    {
+        "inputs": [
+            {
+                "internalType": "uint256",
+                "name": "blockNumber",
+                "type": "uint256"
+            }
+        ],
+        "name": "getBlockHash",
+        "outputs": [
+            {
+                "internalType": "bytes32",
+                "name": "blockHash",
+                "type": "bytes32"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [],
+        "name": "getBlockNumber",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "blockNumber",
+                "type": "uint256"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [],
+        "name": "getCurrentBlockCoinbase",
+        "outputs": [
+            {
+                "internalType": "address",
+                "name": "coinbase",
+                "type": "address"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [],
+        "name": "getCurrentBlockDifficulty",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "difficulty",
+                "type": "uint256"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [],
+        "name": "getCurrentBlockGasLimit",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "gaslimit",
+                "type": "uint256"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [],
+        "name": "getCurrentBlockTimestamp",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "timestamp",
+                "type": "uint256"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [
+            {
+                "internalType": "address",
+                "name": "addr",
+                "type": "address"
+            }
+        ],
+        "name": "getEthBalance",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "balance",
+                "type": "uint256"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [],
+        "name": "getLastBlockHash",
+        "outputs": [
+            {
+                "internalType": "bytes32",
+                "name": "blockHash",
+                "type": "bytes32"
+            }
+        ],
+        "stateMutability": "view",
+        "type": "function"
+    },
+    {
+        "inputs": [
+            {
+                "internalType": "bool",
+                "name": "requireSuccess",
+                "type": "bool"
+            },
+            {
+                "components": [
+                    {
+                        "internalType": "address",
+                        "name": "target",
+                        "type": "address"
+                    },
+                    {
+                        "internalType": "bytes",
+                        "name": "callData",
+                        "type": "bytes"
+                    }
+                ],
+                "internalType": "struct Multicall2.Call[]",
+                "name": "calls",
+                "type": "tuple[]"
+            }
+        ],
+        "name": "tryAggregate",
+        "outputs": [
+            {
+                "components": [
+                    {
+                        "internalType": "bool",
+                        "name": "success",
+                        "type": "bool"
+                    },
+                    {
+                        "internalType": "bytes",
+                        "name": "returnData",
+                        "type": "bytes"
+                    }
+                ],
+                "internalType": "struct Multicall2.Result[]",
+                "name": "returnData",
+                "type": "tuple[]"
+            }
+        ],
+        "stateMutability": "nonpayable",
+        "type": "function"
+    },
+    {
+        "inputs": [
+            {
+                "internalType": "bool",
+                "name": "requireSuccess",
+                "type": "bool"
+            },
+            {
+                "components": [
+                    {
+                        "internalType": "address",
+                        "name": "target",
+                        "type": "address"
+                    },
+                    {
+                        "internalType": "bytes",
+                        "name": "callData",
+                        "type": "bytes"
+                    }
+                ],
+                "internalType": "struct Multicall2.Call[]",
+                "name": "calls",
+                "type": "tuple[]"
+            }
+        ],
+        "name": "tryBlockAndAggregate",
+        "outputs": [
+            {
+                "internalType": "uint256",
+                "name": "blockNumber",
+                "type": "uint256"
+            },
+            {
+                "internalType": "bytes32",
+                "name": "blockHash",
+                "type": "bytes32"
+            },
+            {
+                "components": [
+                    {
+                        "internalType": "bool",
+                        "name": "success",
+                        "type": "bool"
+                    },
+                    {
+                        "internalType": "bytes",
+                        "name": "returnData",
+                        "type": "bytes"
+                    }
+                ],
+                "internalType": "struct Multicall2.Result[]",
+                "name": "returnData",
+                "type": "tuple[]"
+            }
+        ],
+        "stateMutability": "nonpayable",
+        "type": "function"
+    }
+]

crawlers/mooncrawl/mooncrawl/state_crawler/cli.py

@@ -3,26 +3,25 @@ import json
 import hashlib
 import itertools
 import logging
-from random import random
-from typing import List, Any
+from typing import List, Any, Optional
 from uuid import UUID
 
 from moonstreamdb.blockchain import AvailableBlockchainType
+from mooncrawl.moonworm_crawler.crawler import _retry_connect_web3
 from moonstreamdb.db import (
     MOONSTREAM_DB_URI_READ_ONLY,
     MOONSTREAM_POOL_SIZE,
     create_moonstream_engine,
 )
 from sqlalchemy.orm import sessionmaker
+
 from .db import view_call_to_label, commit_session
+from .Multicall2_interface import Contract as Multicall2
 from ..settings import (
     NB_CONTROLLER_ACCESS_ID,
     MOONSTREAM_STATE_CRAWLER_DB_STATEMENT_TIMEOUT_MILLIS,
 )
-
-# from .db import get_first_labeled_block_number, get_last_labeled_block_number
-from brownie import Contract, network, chain, web3
-import Multicall2
+from .web3_util import FunctionSignature
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -41,15 +40,13 @@ def make_multicall(
     multicall_calls = [
         (
             call["address"],
-            call["method"].encode_input(*call["inputs"]),
+            call["method"].encode_data(call["inputs"]).hex(),
         )
         for call in calls
     ]
 
-    multicall_result = multicall_method.call(
-        False,  # success not required
-        multicall_calls,
-        block_identifier=block_number,
+    multicall_result = multicall_method(False, calls=multicall_calls).call(
+        block_identifier=block_number
     )
 
     results = []
@@ -59,11 +56,11 @@ def make_multicall(
         if encoded_data[0]:
             results.append(
                 {
-                    "result": calls[index]["method"].decode_output(encoded_data[1]),
+                    "result": calls[index]["method"].decode_data(encoded_data[1])[0],
                     "hash": calls[index]["hash"],
                     "method": calls[index]["method"],
                     "address": calls[index]["address"],
-                    "name": calls[index]["method"].abi["name"],
+                    "name": calls[index]["method"].name,
                     "inputs": calls[index]["inputs"],
                     "call_data": multicall_calls[index][1],
                     "block_number": block_number,
@@ -74,11 +71,11 @@ def make_multicall(
         else:
             results.append(
                 {
-                    "result": None,  # calls[index]["method"].decode_output(encoded_data[1]),
+                    "result": calls[index]["method"].decode_data(encoded_data[1]),
                     "hash": calls[index]["hash"],
                     "method": calls[index]["method"],
                     "address": calls[index]["address"],
-                    "name": calls[index]["method"].abi["name"],
+                    "name": calls[index]["method"].name,
                     "inputs": calls[index]["inputs"],
                     "call_data": multicall_calls[index][1],
                     "block_number": block_number,
@@ -131,8 +128,8 @@ def crawl_calls_level(
             calls_of_level.append(
                 {
                     "address": call["address"],
-                    "method": interfaces[call["address"]].get_method_object(
-                        interfaces[call["address"]].signatures[call["name"]]
+                    "method": FunctionSignature(
+                        interfaces[call["address"]].get_function_by_name(call["name"])
                     ),
                     "hash": call["generated_hash"],
                     "inputs": call_parameters,
@@ -168,27 +165,30 @@ def crawl_calls_level(
         commit_session(db_session)
 
 
-def parse_jobs(jobs, blockchain_type, block_number):
+def parse_jobs(
+    jobs: List[Any],
+    blockchain_type: AvailableBlockchainType,
+    block_number: Optional[int],
+    batch_size: int,
+    access_id: UUID,
+):
 
     contracts_ABIs = {}
     contracts_methods = {}
     calls = {0: []}
 
-    blockchain_type_to_brownie_network = {
-        AvailableBlockchainType.POLYGON: "polygon-main",
-        AvailableBlockchainType.MUMBAI: "polygon-test",
-    }
-
-    network.connect(blockchain_type_to_brownie_network[blockchain_type])
+    web3_client = _retry_connect_web3(
+        blockchain_type=blockchain_type, access_id=access_id
+    )
 
     if block_number is None:
-        block_number = len(chain) - 1
+        block_number = web3_client.eth.get_block("latest").number
 
-    block_timestamp = web3.eth.get_block(block_number).timestamp
+    block_timestamp = web3_client.eth.get_block(block_number).timestamp
 
-    multicaller = Multicall2.Multicall2(Multicall2_address)
+    multicaller = Multicall2(web3=web3_client, contract_address=Multicall2_address)
 
-    multicall_method = multicaller.contract.tryAggregate
+    multicall_method = multicaller.tryAggregate
 
     def recursive_unpack(method_abi: Any, level: int = 0) -> Any:
         have_subcalls = False
@@ -259,14 +259,13 @@ def parse_jobs(jobs, blockchain_type, block_number):
             abis.append(contracts_ABIs[contract_address][method_hash])
 
         # generate interface
-        interfaces[contract_address] = Contract.from_abi(
-            random(), contract_address, abis
+        interfaces[contract_address] = web3_client.eth.contract(
+            address=contract_address, abi=abis
         )
 
     responces = {}
 
     # # create chunks of calls
-    batch_size = 500
 
     # reverse call_tree
     call_tree_levels = sorted(calls.keys(), reverse=True)[:-1]
@@ -353,7 +352,9 @@ def handle_crawl(args: argparse.Namespace) -> None:
 
     blockchain_type = AvailableBlockchainType(args.blockchain_type)
 
-    parse_jobs([my_job], blockchain_type, args.block_number)
+    parse_jobs(
+        [my_job], blockchain_type, args.block_number, args.batch_size, args.access_id
+    )
 
 
 def parse_abi(args: argparse.Namespace) -> None:
@@ -402,6 +403,13 @@ def main() -> None:
     view_state_crawler_parser.add_argument(
         "--block-number", "-N", type=str, help="Block number."
     )
+    view_state_crawler_parser.add_argument(
+        "--batch-size",
+        "-s",
+        type=int,
+        default=500,
+        help="Size of chunks wich send to Multicall2 contract.",
+    )
     view_state_crawler_parser.set_defaults(func=handle_crawl)
 
     generate_view_parser = subparsers.add_parser(

crawlers/mooncrawl/mooncrawl/state_crawler/web3_util.py

@@ -0,0 +1,247 @@
+import getpass
+import os
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+
+import web3
+from eth_account.account import Account  # type: ignore
+from eth_abi import encode_single, decode_single
+from eth_typing.evm import ChecksumAddress
+from eth_utils import function_signature_to_4byte_selector
+from hexbytes.main import HexBytes
+from web3 import Web3
+from web3.contract import Contract, ContractFunction
+from web3.providers.ipc import IPCProvider
+from web3.providers.rpc import HTTPProvider
+from web3.types import ABI, Nonce, TxParams, TxReceipt, Wei
+from web3._utils.abi import normalize_event_input_types
+
+
+class ContractConstructor:
+    def __init__(self, *args: Any):
+        self.args = args
+
+
+def build_transaction(
+    web3: Web3,
+    builder: Union[ContractFunction, Any],
+    sender: ChecksumAddress,
+) -> Union[TxParams, Any]:
+    """
+    Builds transaction json with the given arguments. It is not submitting transaction
+    Arguments:
+    - web3: Web3 client
+    - builder: ContractFunction or other class that has method buildTransaction(TxParams)
+    - sender: `from` value of transaction, address which is sending this transaction
+    - maxFeePerGas: Optional, max priority fee for dynamic fee transactions in Wei
+    - maxPriorityFeePerGas: Optional the part of the fee that goes to the miner
+    """
+
+    transaction = builder.buildTransaction(
+        {
+            "from": sender,
+            "nonce": get_nonce(web3, sender),
+        }
+    )
+    return transaction
+
+
+def get_nonce(web3: Web3, address: ChecksumAddress) -> Nonce:
+    """
+    Returns Nonce: number of transactions for given address
+    """
+    nonce = web3.eth.get_transaction_count(address)
+    return nonce
+
+
+def submit_transaction(
+    web3: Web3, transaction: Union[TxParams, Any], signer_private_key: str
+) -> HexBytes:
+
+    """
+    Signs and submits json transaction to blockchain from the name of signer
+    """
+    signed_transaction = web3.eth.account.sign_transaction(
+        transaction, private_key=signer_private_key
+    )
+    return submit_signed_raw_transaction(web3, signed_transaction.rawTransaction)
+
+
+def submit_signed_raw_transaction(
+    web3: Web3, signed_raw_transaction: HexBytes
+) -> HexBytes:
+    """
+    Submits already signed raw transaction.
+    """
+    transaction_hash = web3.eth.send_raw_transaction(signed_raw_transaction)
+    return transaction_hash
+
+
+def wait_for_transaction_receipt(web3: Web3, transaction_hash: HexBytes):
+    return web3.eth.wait_for_transaction_receipt(transaction_hash)
+
+
+def deploy_contract(
+    web3: Web3,
+    contract_bytecode: str,
+    contract_abi: List[Dict[str, Any]],
+    deployer: ChecksumAddress,
+    deployer_private_key: str,
+    constructor_arguments: Optional[List[Any]] = None,
+) -> Tuple[HexBytes, ChecksumAddress]:
+    """
+    Deploys smart contract to blockchain
+    Arguments:
+    - web3: web3 client
+    - contract_bytecode: Compiled smart contract bytecode
+    - contract_abi: Json abi of contract. Must include `constructor` function
+    - deployer: Address which is deploying contract. Deployer will pay transaction fee
+    - deployer_private_key: Private key of deployer. Needed for signing and submitting transaction
+    - constructor_arguments: arguments that are passed to `constructor` function  of the smart contract
+    """
+    contract = web3.eth.contract(abi=contract_abi, bytecode=contract_bytecode)
+    if constructor_arguments is None:
+        transaction = build_transaction(web3, contract.constructor(), deployer)
+    else:
+        transaction = build_transaction(
+            web3, contract.constructor(*constructor_arguments), deployer
+        )
+
+    transaction_hash = submit_transaction(web3, transaction, deployer_private_key)
+    transaction_receipt = wait_for_transaction_receipt(web3, transaction_hash)
+    contract_address = transaction_receipt.contractAddress
+    return transaction_hash, web3.toChecksumAddress(contract_address)
+
+
+def deploy_contract_from_constructor_function(
+    web3: Web3,
+    contract_bytecode: str,
+    contract_abi: List[Dict[str, Any]],
+    deployer: ChecksumAddress,
+    deployer_private_key: str,
+    constructor: ContractConstructor,
+) -> Tuple[HexBytes, ChecksumAddress]:
+    """
+    Deploys smart contract to blockchain from constructor ContractFunction
+    Arguments:
+    - web3: web3 client
+    - contract_bytecode: Compiled smart contract bytecode
+    - contract_abi: Json abi of contract. Must include `constructor` function
+    - deployer: Address which is deploying contract. Deployer will pay transaction fee
+    - deployer_private_key: Private key of deployer. Needed for signing and submitting transaction
+    - constructor:`constructor` function  of the smart contract
+    """
+    contract = web3.eth.contract(abi=contract_abi, bytecode=contract_bytecode)
+    transaction = build_transaction(
+        web3, contract.constructor(*constructor.args), deployer
+    )
+
+    transaction_hash = submit_transaction(web3, transaction, deployer_private_key)
+    transaction_receipt = wait_for_transaction_receipt(web3, transaction_hash)
+    contract_address = transaction_receipt.contractAddress
+    return transaction_hash, web3.toChecksumAddress(contract_address)
+
+
+def decode_transaction_input(web3: Web3, transaction_input: str, abi: Dict[str, Any]):
+    contract = web3.eth.contract(abi=abi)
+    return contract.decode_function_input(transaction_input)
+
+
+def read_keys_from_cli() -> Tuple[ChecksumAddress, str]:
+    private_key = getpass.getpass(prompt="Enter private key of your address:")
+    account = Account.from_key(private_key)
+    return (Web3.toChecksumAddress(account.address), private_key)
+
+
+def read_keys_from_env() -> Tuple[ChecksumAddress, str]:
+    private_key = os.environ.get("MOONWORM_ETHEREUM_ADDRESS_PRIVATE_KEY")
+    if private_key is None:
+        raise ValueError(
+            "MOONWORM_ETHEREUM_ADDRESS_PRIVATE_KEY env variable is not set"
+        )
+    try:
+        account = Account.from_key(private_key)
+        return (Web3.toChecksumAddress(account.address), private_key)
+    except:
+        raise ValueError(
+            "Failed to initiate account from MOONWORM_ETHEREUM_ADDRESS_PRIVATE_KEY"
+        )
+
+
+def connect(web3_uri: str) -> Web3:
+    web3_provider: Union[IPCProvider, HTTPProvider] = Web3.IPCProvider()
+    if web3_uri.startswith("http://") or web3_uri.startswith("https://"):
+        web3_provider = Web3.HTTPProvider(web3_uri)
+    else:
+        web3_provider = Web3.IPCProvider(web3_uri)
+    web3_client = Web3(web3_provider)
+    return web3_client
+
+
+def read_web3_provider_from_env() -> Web3:
+    provider_path = os.environ.get("MOONWORM_WEB3_PROVIDER_URI")
+    if provider_path is None:
+        raise ValueError("MOONWORM_WEB3_PROVIDER_URI env variable is not set")
+    return connect(provider_path)
+
+
+def read_web3_provider_from_cli() -> Web3:
+    provider_path = input("Enter web3 uri path: ")
+    return connect(provider_path)
+
+
+def cast_to_python_type(evm_type: str) -> Callable:
+    if evm_type.startswith(("uint", "int")):
+        return int
+    elif evm_type.startswith("bytes"):
+        return bytes
+    elif evm_type == "string":
+        return str
+    elif evm_type == "address":
+        return Web3.toChecksumAddress
+    elif evm_type == "bool":
+        return bool
+    else:
+        raise ValueError(f"Cannot convert to python type {evm_type}")
+
+
+class FunctionInput:
+    def __init__(self, name: str, value: any, solidity_type: str):
+        self.name = name
+        self.value = value
+        self.solidity_type = solidity_type
+
+
+class FunctionSignature:
+    def __init__(self, function: ContractFunction):
+        self.name = function.abi["name"]
+        self.inputs = [
+            {"name": arg["name"], "type": arg["type"]}
+            for arg in normalize_event_input_types(function.abi.get("inputs", []))
+        ]
+        self.input_types_signature = "({})".format(
+            ",".join([inp["type"] for inp in self.inputs])
+        )
+        self.output_types_signature = "({})".format(
+            ",".join(
+                [
+                    arg["type"]
+                    for arg in normalize_event_input_types(
+                        function.abi.get("outputs", [])
+                    )
+                ]
+            )
+        )
+
+        self.signature = "{}{}".format(self.name, self.input_types_signature)
+
+        self.fourbyte = function_signature_to_4byte_selector(self.signature)
+
+    def encode_data(self, args=None) -> str:
+        return (
+            self.fourbyte + encode_single(self.input_types_signature, args)
+            if args
+            else self.fourbyte
+        )
+
+    def decode_data(self, output):
+        return decode_single(self.output_types_signature, output)