Using pypechain get logs instead of logs from ethpy

src/agent0/ethpy/hyperdrive/interface/read_interface.py

@@ -290,7 +290,11 @@ def get_deploy_block(self) -> int | None:
             # We look up the chain id, and define the `from_block` based on which chain it is as the default.
             chain_id = self.web3.eth.chain_id
             # If not in lookup, we default to `earliest`
-            from_block = EARLIEST_BLOCK_LOOKUP.get(chain_id, "earliest")
+            # If not in lookup, we default to `earliest`
+            if chain_id not in EARLIEST_BLOCK_LOOKUP:
+                from_block = "earliest"
+            else:
+                from_block = EARLIEST_BLOCK_LOOKUP[chain_id]
 
             initialize_event = list(self.hyperdrive_contract.events.Initialize.get_logs_typed(from_block=from_block))
             if len(initialize_event) == 0:

src/agent0/hyperfuzz/system_fuzz/invariant_checks.py

@@ -7,6 +7,7 @@
 from typing import Any, Callable, NamedTuple
 
 from fixedpointmath import FixedPoint
+from hyperdrivetypes.types.IHyperdriveContract import CloseLongEvent, CloseShortEvent
 from web3.types import BlockData
 
 from agent0.core.hyperdrive.crash_report import (
@@ -540,32 +541,31 @@ def _check_lp_share_price(
 
     # Determine if a checkpoint was minted on the current block
     # -1 to get events from current block
-    checkpoint_events = interface.get_checkpoint_events(from_block=check_block_number - 1)
+    checkpoint_events = interface.hyperdrive_contract.events.CreateCheckpoint.get_logs_typed(
+        from_block=check_block_number - 1
+    )
     currently_minting_checkpoint = False
     for event in checkpoint_events:
-        assert "blockNumber" in event
-        if event["blockNumber"] == check_block_number:
+        if event.block_number == check_block_number:
             currently_minting_checkpoint = True
             break
 
     # Determine if matured positions were closed this timestamp
     # We look for close events on this block
     # -1 to get events from current block
-    trade_events: list[dict[str, Any]] = []
-    trade_events.extend(interface.get_close_short_events(from_block=check_block_number - 1))
-    trade_events.extend(interface.get_close_long_events(from_block=check_block_number - 1))
+    trade_events: list[CloseShortEvent | CloseLongEvent] = []
+    trade_events.extend(
+        interface.hyperdrive_contract.events.CloseShort.get_logs_typed(from_block=check_block_number - 1)
+    )
+    trade_events.extend(
+        interface.hyperdrive_contract.events.CloseLong.get_logs_typed(from_block=check_block_number - 1)
+    )
 
     closing_mature_position = False
     for event in trade_events:
-        # maturityTime should always be part of close short/long
-        assert "args" in event
-        assert "maturityTime" in event["args"]
-        assert "blockNumber" in event
         # Race condition, filter only on events from the current block
         # Check if any matured positions were closed
-        if (event["blockNumber"] == check_block_number) and (
-            mined_pool_state.block_time >= event["args"]["maturityTime"]
-        ):
+        if (event.block_number == check_block_number) and (mined_pool_state.block_time >= event.args.maturityTime):
             closing_mature_position = True
             break