interface.py (5377B)
1 # standard imports 2 import logging 3 import json 4 import os 5 6 # external imports 7 from chainlib.eth.tx import ( 8 TxFactory, 9 TxFormat, 10 ) 11 from chainlib.eth.contract import ( 12 ABIContractEncoder, 13 ABIContractDecoder, 14 ABIContractType, 15 abi_decode_single, 16 ) 17 from chainlib.eth.constant import ZERO_ADDRESS 18 from chainlib.jsonrpc import JSONRPCRequest 19 from chainlib.eth.error import RequestMismatchException 20 from hexathon import ( 21 add_0x, 22 strip_0x, 23 ) 24 25 logg = logging.getLogger() 26 27 28 class AccountsIndex(TxFactory): 29 30 def __single_address_method(self, method, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC): 31 enc = ABIContractEncoder() 32 enc.method(method) 33 enc.typ(ABIContractType.ADDRESS) 34 enc.address(address) 35 data = enc.get() 36 tx = self.template(sender_address, contract_address, use_nonce=True) 37 tx = self.set_code(tx, data) 38 tx = self.finalize(tx, tx_format) 39 return tx 40 41 def add(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC): 42 return self.__single_address_method('add', contract_address, sender_address, address, tx_format) 43 44 def remove(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC): 45 return self.__single_address_method('remove', contract_address, sender_address, address, tx_format) 46 47 def activate(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC): 48 return self.__single_address_method('activate', contract_address, sender_address, address, tx_format) 49 50 def deactivate(self, contract_address, sender_address, address, tx_format=TxFormat.JSONRPC): 51 return self.__single_address_method('deactivate', contract_address, sender_address, address, tx_format) 52 53 54 def time(self, contract_address, address, sender_address=ZERO_ADDRESS, id_generator=None): 55 j = JSONRPCRequest(id_generator) 56 o = j.template() 57 o['method'] = 'eth_call' 58 enc = ABIContractEncoder() 59 enc.method('time') 60 enc.typ(ABIContractType.ADDRESS) 61 enc.address(address) 62 data = add_0x(enc.get()) 63 tx = self.template(sender_address, contract_address) 64 tx = self.set_code(tx, data) 65 o['params'].append(self.normalize(tx)) 66 return o 67 68 69 70 def have(self, contract_address, address, sender_address=ZERO_ADDRESS, id_generator=None): 71 j = JSONRPCRequest(id_generator) 72 o = j.template() 73 o['method'] = 'eth_call' 74 enc = ABIContractEncoder() 75 enc.method('have') 76 enc.typ(ABIContractType.ADDRESS) 77 enc.address(address) 78 data = add_0x(enc.get()) 79 tx = self.template(sender_address, contract_address) 80 tx = self.set_code(tx, data) 81 o['params'].append(self.normalize(tx)) 82 return o 83 84 def is_active(self, contract_address, address, sender_address=ZERO_ADDRESS, id_generator=None): 85 j = JSONRPCRequest(id_generator) 86 o = j.template() 87 o['method'] = 'eth_call' 88 enc = ABIContractEncoder() 89 enc.method('isActive') 90 enc.typ(ABIContractType.ADDRESS) 91 enc.address(address) 92 data = add_0x(enc.get()) 93 tx = self.template(sender_address, contract_address) 94 tx = self.set_code(tx, data) 95 o['params'].append(self.normalize(tx)) 96 return o 97 98 99 def entry_count(self, contract_address, sender_address=ZERO_ADDRESS, id_generator=None): 100 j = JSONRPCRequest(id_generator) 101 o = j.template() 102 o['method'] = 'eth_call' 103 enc = ABIContractEncoder() 104 enc.method('entryCount') 105 data = add_0x(enc.get()) 106 tx = self.template(sender_address, contract_address) 107 tx = self.set_code(tx, data) 108 o['params'].append(self.normalize(tx)) 109 return o 110 111 def count(self, contract_address, sender_address=ZERO_ADDRESS, id_generator=None): 112 return self.entry_count(contract_address, sender_address=sender_address, id_generator=id_generator) 113 114 def entry(self, contract_address, idx, sender_address=ZERO_ADDRESS, id_generator=None): 115 j = JSONRPCRequest(id_generator) 116 o = j.template() 117 o['method'] = 'eth_call' 118 enc = ABIContractEncoder() 119 enc.method('entry') 120 enc.typ(ABIContractType.UINT256) 121 enc.uint256(idx) 122 data = add_0x(enc.get()) 123 tx = self.template(sender_address, contract_address) 124 tx = self.set_code(tx, data) 125 o['params'].append(self.normalize(tx)) 126 return o 127 128 @classmethod 129 def parse_account(self, v): 130 return abi_decode_single(ABIContractType.ADDRESS, v) 131 132 @classmethod 133 def parse_entry_count(self, v): 134 return abi_decode_single(ABIContractType.UINT256, v) 135 136 @classmethod 137 def parse_have(self, v): 138 return abi_decode_single(ABIContractType.BOOLEAN, v) 139 140 @classmethod 141 def parse_add_request(self, v): 142 v = strip_0x(v) 143 cursor = 0 144 enc = ABIContractEncoder() 145 enc.method('add') 146 enc.typ(ABIContractType.ADDRESS) 147 r = enc.get() 148 l = len(r) 149 m = v[:l] 150 if m != r: 151 raise RequestMismatchException(v) 152 cursor += l 153 154 dec = ABIContractDecoder() 155 dec.typ(ABIContractType.ADDRESS) 156 dec.val(v[cursor:cursor + 64]) 157 r = dec.decode() 158 return r