interface.py (5325B)
1 # standard imports 2 import logging 3 import json 4 import os 5 6 # external imports 7 from hexathon import ( 8 strip_0x, 9 add_0x, 10 ) 11 from chainlib.eth.tx import ( 12 TxFormat, 13 TxFactory, 14 ) 15 from chainlib.eth.contract import ( 16 ABIContractEncoder, 17 ABIContractType, 18 abi_decode_single, 19 ) 20 from chainlib.jsonrpc import JSONRPCRequest 21 from chainlib.eth.constant import ZERO_ADDRESS 22 from chainlib.block import BlockSpec 23 from chainlib.eth.jsonrpc import to_blockheight_param 24 25 logg = logging.getLogger(__name__) 26 27 28 class Declarator(TxFactory): 29 30 def add_declaration(self, contract_address, sender_address, subject_address, proof, topic=None, tx_format=TxFormat.JSONRPC): 31 enc = ABIContractEncoder() 32 enc.method('addDeclaration') 33 enc.typ(ABIContractType.ADDRESS) 34 enc.typ(ABIContractType.BYTES32) 35 if topic != None: 36 enc.typ(ABIContractType.BYTES32) 37 enc.address(subject_address) 38 enc.bytes32(proof) 39 if topic != None: 40 enc.bytes32(topic) 41 data = enc.get() 42 tx = self.template(sender_address, contract_address, use_nonce=True) 43 tx = self.set_code(tx, data) 44 tx = self.finalize(tx, tx_format) 45 return tx 46 47 48 def declarator_count(self, contract_address, subject_address, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST): 49 j = JSONRPCRequest(id_generator) 50 o = j.template() 51 o['method'] = 'eth_call' 52 enc = ABIContractEncoder() 53 enc.method('declaratorCount') 54 enc.typ(ABIContractType.ADDRESS) 55 enc.address(subject_address) 56 data = add_0x(enc.get()) 57 tx = self.template(sender_address, contract_address) 58 tx = self.set_code(tx, data) 59 o['params'].append(self.normalize(tx)) 60 height = to_blockheight_param(height) 61 o['params'].append(height) 62 o = j.finalize(o) 63 return o 64 65 66 def declaration(self, contract_address, declarator_address, subject_address, topic=None, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST): 67 j = JSONRPCRequest(id_generator) 68 o = j.template() 69 o['method'] = 'eth_call' 70 enc = ABIContractEncoder() 71 enc.method('declaration') 72 enc.typ(ABIContractType.ADDRESS) 73 enc.typ(ABIContractType.ADDRESS) 74 if topic != None: 75 enc.typ(ABIContractType.BYTES32) 76 enc.address(declarator_address) 77 enc.address(subject_address) 78 if topic != None: 79 enc.bytes32(topic) 80 data = add_0x(enc.get()) 81 tx = self.template(sender_address, contract_address) 82 tx = self.set_code(tx, data) 83 o['params'].append(self.normalize(tx)) 84 height = to_blockheight_param(height) 85 o['params'].append(height) 86 o = j.finalize(o) 87 return o 88 89 90 def declaration_address_at(self, contract_address, declarator_address, idx, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST): 91 j = JSONRPCRequest(id_generator) 92 o = j.template() 93 o['method'] = 'eth_call' 94 enc = ABIContractEncoder() 95 enc.method('declarationAddressAt') 96 enc.typ(ABIContractType.ADDRESS) 97 enc.typ(ABIContractType.UINT256) 98 enc.address(declarator_address) 99 enc.uint256(idx) 100 data = add_0x(enc.get()) 101 tx = self.template(sender_address, contract_address) 102 tx = self.set_code(tx, data) 103 o['params'].append(self.normalize(tx)) 104 height = to_blockheight_param(height) 105 o['params'].append(height) 106 o = j.finalize(o) 107 return o 108 109 110 def declarator_address_at(self, contract_address, subject_address, idx, sender_address=ZERO_ADDRESS, id_generator=None, height=BlockSpec.LATEST): 111 j = JSONRPCRequest(id_generator) 112 o = j.template() 113 o['method'] = 'eth_call' 114 enc = ABIContractEncoder() 115 enc.method('declaratorAddressAt') 116 enc.typ(ABIContractType.ADDRESS) 117 enc.typ(ABIContractType.UINT256) 118 enc.address(subject_address) 119 enc.uint256(idx) 120 data = add_0x(enc.get()) 121 tx = self.template(sender_address, contract_address) 122 tx = self.set_code(tx, data) 123 o['params'].append(self.normalize(tx)) 124 height = to_blockheight_param(height) 125 o['params'].append(height) 126 o = j.finalize(o) 127 return o 128 129 130 @classmethod 131 def parse_declarator_count(self, v): 132 return abi_decode_single(ABIContractType.UINT256, v) 133 134 135 @classmethod 136 def parse_declaration(self, v): 137 cursor = 0 138 r = [] 139 try: 140 v = strip_0x(v) 141 except ValueError: 142 return r 143 position = int.from_bytes(bytes.fromhex(v[cursor:cursor+64]), 'big') 144 cursor += (position * 2) 145 length = int.from_bytes(bytes.fromhex(v[cursor:cursor+64]), 'big') 146 cursor += 64 147 for i in range(length): 148 r.append(v[cursor:cursor+64]) 149 cursor += 64 150 return r 151 152 153 @classmethod 154 def parse_declaration_address_at(self, v): 155 return abi_decode_single(ABIContractType.ADDRESS, v) 156 157 158 @classmethod 159 def parse_declarator_address_at(self, v): 160 return abi_decode_single(ABIContractType.ADDRESS, v)