Liquidity.Party/tycho-protocol-sdk
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Start using external modules

Browse Source
This commit is contained in:
Thales
2024-08-05 19:58:10 -03:00
committed by tvinagre
parent 8ea02613a2
commit d893ab264c
9 changed files with 171 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
testing/tycho-client/tycho_client/decoders.py
View File

@@ -26,10 +26,10 @@ class ThirdPartyPoolTychoDecoder:
self.trace = trace
def decode_snapshot(
self,
snapshot: dict[str, Any],
block: EVMBlock,
tokens: dict[str, EthereumToken],
self,
snapshot: dict[str, Any],
block: EVMBlock,
tokens: dict[str, EthereumToken],
) -> tuple[dict[str, ThirdPartyPool], list[str]]:
pools = {}
failed_pools = []
@@ -45,7 +45,7 @@ class ThirdPartyPoolTychoDecoder:
return pools, failed_pools
def decode_pool_state(
self, snap: dict, block: EVMBlock, tokens: dict[str, EthereumToken]
self, snap: dict, block: EVMBlock, tokens: dict[str, EthereumToken]
) -> ThirdPartyPool:
component = snap["component"]
exchange, _ = decode_tycho_exchange(component["protocol_system"])
@@ -85,20 +85,29 @@ class ThirdPartyPoolTychoDecoder:
while f"stateless_contract_addr_{index}" in static_attributes:
encoded_address = static_attributes[f"stateless_contract_addr_{index}"]
decoded = bytes.fromhex(
encoded_address[2:] if encoded_address.startswith('0x') else encoded_address).decode('utf-8')
encoded_address[2:]
if encoded_address.startswith("0x")
else encoded_address
).decode("utf-8")
if decoded.startswith("call"):
address = ThirdPartyPoolTychoDecoder.get_address_from_call(block_number, decoded)
address = ThirdPartyPoolTychoDecoder.get_address_from_call(
block_number, decoded
)
else:
address = decoded
code = static_attributes.get(f"stateless_contract_code_{index}") or get_code_for_address(address)
code = static_attributes.get(
f"stateless_contract_code_{index}"
) or get_code_for_address(address)
stateless_contracts[address] = code
index += 1
index = 0
while f"stateless_contract_addr_{index}" in attributes:
address = attributes[f"stateless_contract_addr_{index}"]
code = attributes.get(f"stateless_contract_code_{index}") or get_code_for_address(address)
code = attributes.get(
f"stateless_contract_code_{index}"
) or get_code_for_address(address)
stateless_contracts[address] = code
index += 1
return {
@@ -118,15 +127,17 @@ class ThirdPartyPoolTychoDecoder:
permanent_storage=None,
)
selector = keccak(text=decoded.split(":")[-1])[:4]
sim_result = engine.run_sim(SimulationParameters(
data=bytearray(selector),
to=decoded.split(':')[1],
block_number=block_number,
timestamp=int(time.time()),
overrides={},
caller=EXTERNAL_ACCOUNT,
value=0,
))
sim_result = engine.run_sim(
SimulationParameters(
data=bytearray(selector),
to=decoded.split(":")[1],
block_number=block_number,
timestamp=int(time.time()),
overrides={},
caller=EXTERNAL_ACCOUNT,
value=0,
)
)
address = eth_abi.decode(["address"], bytearray(sim_result.result))
return address[0]
@@ -143,10 +154,10 @@ class ThirdPartyPoolTychoDecoder:
@staticmethod
def apply_update(
pool: ThirdPartyPool,
pool_update: dict[str, Any],
balance_updates: dict[str, Any],
block: EVMBlock,
pool: ThirdPartyPool,
pool_update: dict[str, Any],
balance_updates: dict[str, Any],
block: EVMBlock,
) -> ThirdPartyPool:
# check for and apply optional state attributes
attributes = pool_update.get("updated_attributes")