feat: namedtuple decoding for vvmcontract structs

boa/contracts/abi/abi_contract.py

@@ -1,11 +1,11 @@
-from collections import defaultdict
+from collections import defaultdict, namedtuple
 from functools import cached_property
 from typing import Any, Optional, Union
 from warnings import warn
 
 from eth.abc import ComputationAPI
 from vyper.semantics.analysis.base import FunctionVisibility, StateMutability
-from vyper.utils import method_id
+from vyper.utils import keccak256, method_id
 
 from boa.contracts.base_evm_contract import (
     BoaError,
@@ -47,7 +47,7 @@ def argument_count(self) -> int:
 
     @property
     def signature(self) -> str:
-        return f"({_format_abi_type(self.argument_types)})"
+        return _format_abi_type(self.argument_types)
 
     @cached_property
     def return_type(self) -> list:
@@ -138,9 +138,9 @@ def __call__(self, *args, value=0, gas=None, sender=None, **kwargs):
             case ():
                 return None
             case (single,):
-                return single
+                return _parse_complex(self._abi["outputs"][0], single, name=self.name)
             case multiple:
-                return tuple(multiple)
+                return _parse_complex(self._abi["outputs"], multiple, name=self.name)
 
 
 class ABIOverload:
@@ -234,13 +234,15 @@ def __init__(
         name: str,
         abi: list[dict],
         functions: list[ABIFunction],
+        events: list[dict],
         address: Address,
         filename: Optional[str] = None,
         env=None,
     ):
         super().__init__(name, env, filename=filename, address=address)
         self._abi = abi
         self._functions = functions
+        self._events = events
 
         self._bytecode = self.env.get_code(address)
         if not self._bytecode:
@@ -276,6 +278,76 @@ def method_id_map(self):
             if not function.is_constructor
         }
 
+    @cached_property
+    def event_for(self):
+        # [{"name": "Bar", "inputs":
+        #   [{"name": "x", "type": "uint256", "indexed": false},
+        #   {"name": "y", "type": "tuple", "components": [{"name": "x", "type": "uint256"}], "indexed": false}],
+        # "anonymous": false, "type": "event"},
+        # }]
+        ret = {}
+        for event_abi in self._events:
+            event_signature = ",".join(
+                _abi_from_json(item) for item in event_abi["inputs"]
+            )
+            event_name = event_abi["name"]
+            event_signature = f"{event_name}({event_signature})"
+            event_id = int(keccak256(event_signature.encode()).hex(), 16)
+            ret[event_id] = event_abi
+        return ret
+
+    def decode_log(self, log_entry):
+        # low level log id
+        _log_id, address, topics, data = log_entry
+        assert self._address.canonical_address == address
+        event_hash = topics[0]
+        event_abi = self.event_for[event_hash]
+
+        topic_abis = []
+        arg_abis = []
+        tuple_names = []
+        for item_abi in event_abi["inputs"]:
+            is_topic = item_abi["indexed"]
+            assert isinstance(is_topic, bool)
+            if not is_topic:
+                arg_abis.append(item_abi)
+            else:
+                topic_abis.append(item_abi)
+
+            tuple_names.append(item_abi["name"])
+
+        tuple_typ = namedtuple(event_abi["name"], tuple_names)
+
+        decoded_topics = []
+        for topic_abi, t in zip(topic_abis, topics[1:]):
+            # convert to bytes for abi decoder
+            encoded_topic = t.to_bytes(32, "big")
+            decoded_topics.append(abi_decode(_abi_from_json(topic_abi), encoded_topic))
+
+        args_selector = _format_abi_type(
+            [_abi_from_json(arg_abi) for arg_abi in arg_abis]
+        )
+
+        decoded_args = abi_decode(args_selector, data)
+
+        t_i = 0
+        a_i = 0
+        xs = []
+        # re-align the evm topic + args lists with the way they appear in the abi
+        # ex. Transfer(indexed address, address, indexed address)
+        for item_abi in event_abi["inputs"]:
+            is_topic = item_abi["indexed"]
+            if is_topic:
+                # topic abi is currently never complex, but use _parse_complex as
+                # future-proofing mechanism
+                xs.append(_parse_complex(topic_abis[t_i], decoded_topics[t_i]))
+                t_i += 1
+            else:
+                xs.append(_parse_complex(arg_abis[a_i], decoded_args[a_i]))
+                a_i += 1
+
+        return tuple_typ(*xs)
+
     def marshal_to_python(self, computation, abi_type: list[str]) -> tuple[Any, ...]:
         """
         Convert the output of a contract call to a Python object.
@@ -286,7 +358,7 @@ def marshal_to_python(self, computation, abi_type: list[str]) -> tuple[Any, ...]
         if computation.is_error:
             return self.handle_error(computation)
 
-        schema = f"({_format_abi_type(abi_type)})"
+        schema = _format_abi_type(abi_type)
         try:
             return abi_decode(schema, computation.output)
         except ABIError as e:
@@ -360,6 +432,10 @@ def functions(self):
             if item.get("type") == "function"
         ]
 
+    @property
+    def events(self):
+        return [item for item in self.abi if item.get("type") == "event"]
+
     @classmethod
     def from_abi_dict(cls, abi, name="<anonymous contract>", filename=None):
         return cls(name, abi, filename)
@@ -370,7 +446,7 @@ def at(self, address: Address | str) -> ABIContract:
         """
         address = Address(address)
         contract = ABIContract(
-            self._name, self._abi, self.functions, address, self.filename
+            self._name, self._abi, self.functions, self.events, address, self.filename
         )
 
         contract.env.register_contract(address, contract)
@@ -390,15 +466,15 @@ def __repr__(self):
 
     @cached_property
     def args_abi_type(self):
-        return f"({_format_abi_type(self.function.argument_types)})"
+        return _format_abi_type(self.function.argument_types)
 
     @cached_property
     def _argument_names(self) -> list[str]:
         return [arg["name"] for arg in self.function._abi["inputs"]]
 
     @cached_property
     def return_abi_type(self):
-        return f"({_format_abi_type(self.function.return_type)})"
+        return _format_abi_type(self.function.return_type)
 
 
 def _abi_from_json(abi: dict) -> str:
@@ -407,6 +483,12 @@ def _abi_from_json(abi: dict) -> str:
     :param abi: The ABI type to parse.
     :return: The schema string for the given abi type.
     """
+    # {"stateMutability": "view", "type": "function", "name": "foo",
+    # "inputs": [],
+    # "outputs": [{"name": "", "type": "tuple",
+    #    "components": [{"name": "x", "type": "uint256"}]}]
+    # }
+
     if "components" in abi:
         components = ",".join([_abi_from_json(item) for item in abi["components"]])
         if abi["type"].startswith("tuple"):
@@ -416,11 +498,33 @@ def _abi_from_json(abi: dict) -> str:
     return abi["type"]
 
 
+def _parse_complex(abi: dict, value: Any, name=None) -> str:
+    """
+    Parses an ABI type into its schema string.
+    :param abi: The ABI type to parse.
+    :return: The schema string for the given abi type.
+    """
+    # simple case
+    if "components" not in abi:
+        return value
+
+    # complex case
+    # construct a namedtuple type on the fly
+    components = abi["components"]
+    typname = name or abi["name"] or "user_struct"
+    component_names = [item["name"] for item in components]
+    typ = namedtuple(typname, component_names)
+    components_parsed = [
+        _parse_complex(item_abi, item) for (item_abi, item) in zip(components, value)
+    ]
+    return typ(*components_parsed)
+
+
 def _format_abi_type(types: list) -> str:
     """
     Converts a list of ABI types into a comma-separated string.
     """
-    return ",".join(
-        item if isinstance(item, str) else f"({_format_abi_type(item)})"
-        for item in types
+    ret = ",".join(
+        item if isinstance(item, str) else _format_abi_type(item) for item in types
     )
+    return f"({ret})"

boa/contracts/base_evm_contract.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 from eth.abc import ComputationAPI
 
@@ -13,6 +13,11 @@
     from boa.vm.py_evm import titanoboa_computation
 
 
+@dataclass
+class RawEvent:
+    event_data: Any
+
+
 class _BaseEVMContract:
     """
     Base class for EVM (Ethereum Virtual Machine) contract:
@@ -57,6 +62,37 @@ def address(self) -> Address:
             raise RuntimeError("Contract address is not set")
         return self._address
 
+    # ## handling events
+    def _get_logs(self, computation, include_child_logs):
+        if computation is None:
+            return []
+
+        if include_child_logs:
+            return list(computation.get_raw_log_entries())
+
+        return computation._log_entries
+
+    def get_logs(self, computation=None, include_child_logs=True):
+        if computation is None:
+            computation = self._computation
+
+        entries = self._get_logs(computation, include_child_logs)
+
+        # py-evm log format is (log_id, topics, data)
+        # sort on log_id
+        entries = sorted(entries)
+
+        ret = []
+        for e in entries:
+            logger_address = e[1]
+            c = self.env.lookup_contract(logger_address)
+            if c is not None:
+                ret.append(c.decode_log(e))
+            else:
+                ret.append(RawEvent(e))
+
+        return ret
+
 
 class StackTrace(list):  # list[str|ErrorDetail]
     def __str__(self):

boa/contracts/vvm/vvm_contract.py

@@ -51,10 +51,19 @@ def deploy(self, *args, contract_name=None, env=None, **kwargs):
         if env is None:
             env = Env.get_singleton()
 
-        address, _ = env.deploy_code(bytecode=self.bytecode + encoded_args, **kwargs)
+        address, computation = env.deploy(
+            bytecode=self.bytecode + encoded_args, **kwargs
+        )
 
         # TODO: pass thru contract_name
-        return self.at(address)
+        # NOTE: if computation.is_error, `self.at()` will raise a warning
+        # in the future we should refactor so that the warning is silenced
+        ret = self.at(address)
+
+        if computation.is_error:
+            ret.handle_error(computation)
+
+        return ret
 
     @cached_property
     def _blueprint_deployer(self):
@@ -75,10 +84,13 @@ def deploy_as_blueprint(self, env=None, blueprint_preamble=None, **kwargs):
         blueprint_bytecode = generate_blueprint_bytecode(
             self.bytecode, blueprint_preamble
         )
-        address, _ = env.deploy_code(bytecode=blueprint_bytecode, **kwargs)
+        address, computation = env.deploy(bytecode=blueprint_bytecode, **kwargs)
 
         ret = self._blueprint_deployer.at(address)
 
+        if computation.is_error:
+            ret.handle_error(computation)
+
         env.register_blueprint(self.bytecode, ret)
         return ret
 

boa/contracts/vyper/event.py

@@ -28,8 +28,3 @@ def __repr__(self):
 
         args = ", ".join(f"{k}={v}" for k, v in b)
         return f"{self.event_type.name}({args})"
-
-
-@dataclass
-class RawEvent:
-    event_data: Any

boa/contracts/vyper/vyper_contract.py

@@ -51,7 +51,7 @@
     ByteAddressableStorage,
     decode_vyper_object,
 )
-from boa.contracts.vyper.event import Event, RawEvent
+from boa.contracts.vyper.event import Event
 from boa.contracts.vyper.ir_executor import executor_from_ir
 from boa.environment import Env
 from boa.profiling import cache_gas_used_for_computation
@@ -711,37 +711,6 @@ def trace_source(self, computation) -> Optional["VyperTraceSource"]:
             return None
         return VyperTraceSource(self, node, method_id=computation.msg.data[:4])
 
-    # ## handling events
-    def _get_logs(self, computation, include_child_logs):
-        if computation is None:
-            return []
-
-        if include_child_logs:
-            return list(computation.get_raw_log_entries())
-
-        return computation._log_entries
-
-    def get_logs(self, computation=None, include_child_logs=True):
-        if computation is None:
-            computation = self._computation
-
-        entries = self._get_logs(computation, include_child_logs)
-
-        # py-evm log format is (log_id, topics, data)
-        # sort on log_id
-        entries = sorted(entries)
-
-        ret = []
-        for e in entries:
-            logger_address = e[1]
-            c = self.env.lookup_contract(logger_address)
-            if c is not None:
-                ret.append(c.decode_log(e))
-            else:
-                ret.append(RawEvent(e))
-
-        return ret
-
     @cached_property
     def event_for(self):
         module_t = self.compiler_data.global_ctx

boa/interpret.py

@@ -273,14 +273,21 @@ def load_partial(filename: str, compiler_args=None):
         )
 
 
-def _loads_partial_vvm(source_code: str, version: Version, filename: str):
+def _loads_partial_vvm(
+    source_code: str, version: Version, filename: str, base_path=None
+):
     global _disk_cache
 
+    if base_path is None:
+        base_path = Path(".")
+
     # install the requested version if not already installed
     vvm.install_vyper(version=version)
 
     def _compile():
-        compiled_src = vvm.compile_source(source_code, vyper_version=version)
+        compiled_src = vvm.compile_source(
+            source_code, vyper_version=version, base_path=base_path
+        )
         compiler_output = compiled_src["<stdin>"]
         return VVMDeployer.from_compiler_output(compiler_output, filename=filename)