cicnet/cic-cli
5
1
Fork 0
Code Issues Pull Requests Projects Releases 18 Wiki Activity

Factor out code common for extensions

Browse Source
This commit is contained in:
nolash 2021-10-12 08:39:20 +02:00
parent 39becba3b2
commit 68867d09d2
Signed by: lash
GPG Key ID: 21D2E7BB88C2A746
4 changed files with 141 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cic/cmd/export.py
View File

@ -18,6 +18,7 @@ logg = logging.getLogger(__name__)
def process_args(argparser):
argparser.add_argument('-d', '--directory', type=str, dest='directory', default='.', help='directory')
argparser.add_argument('-o', '--output-directory', type=str, dest='output_directory', help='output directory')
argparser.add_argument('-y', '--signer', type=str, dest='y', help='target-specific signer to use for export')
argparser.add_argument('-p', type=str, help='RPC endpoint')
argparser.add_argument('target', type=str, help='target network type')
@ -80,4 +81,5 @@ def execute(config, eargs):
c.apply_token(ct)
p = Processor(proof=cp, attachment=ca, metadata=cm, extensions=[eargs.target])
c.process(writer=writers['ext'])
c.process(writer=writers['ext'](path=eargs.output_directory))

74
cic/ext/eth/__init__.py
View File

@ -28,25 +28,18 @@ from eth_address_declarator.declarator import AddressDeclarator
# local imports
from cic.ext.eth.rpc import parse_adapter
from cic.extension import Extension
logg = logging.getLogger(__name__)
class CICEth:
class CICEth(Extension):
def __init__(self, chain_spec, resources, proof, signer=None, rpc=None, fee_oracle=None, outputs_writer=None):
def __init__(self, chain_spec, resources, proof, signer=None, rpc=None, outputs_writer=None, fee_oracle=None):
"""resources will be modified
"""
self.resources = resources
self.proof = proof
self.chain_spec = chain_spec
self.signer = signer
self.rpc = rpc
super(CICEth, self).__init__(chain_spec, resources, proof, signer=signer, rpc=rpc, outputs_writer=outputs_writer)
self.fee_oracle = fee_oracle
self.token_details = None
self.token_address = None
self.token_code = None
self.outputs = []
self.tx_format = TxFormat.RAW_ARGS
self.outputs_writer = outputs_writer
if self.rpc != None:
@ -55,22 +48,6 @@ class CICEth:
self.tx_format = TxFormat.RLP_SIGNED
def apply_token(self, token):
return self.prepare_token(token.name, token.symbol, token.precision, token.code)
def prepare_token(self, name, symbol, precision, code, extra=[], extra_types=[], positions=None):
self.token_details = {
'name': name,
'symbol': symbol,
'precision': precision,
'code': code or '',
'extra': extra,
'extra_types': extra_types,
'positions': positions,
}
def __detect_arg_type(self, v):
typ = None
try:
@ -152,7 +129,9 @@ class CICEth:
for i in positions:
getattr(enc, args_types[i])(args[i])
code = self.token_details['code'] + enc.get()
code = enc.get()
if self.token_code != None:
code = self.token_code + code
logg.debug('resource {}'.format(self.resources))
signer_address = add_0x(to_checksum_address(self.resources['token']['key_account']))
@ -185,19 +164,20 @@ class CICEth:
if writer == None:
writer = self.outputs_writer
signer_address = self.resources['token_index']['key_address']
contract_address = self.resources['token_index']['reference']
signer_address = add_0x(to_checksum_address(self.resources['token_index']['key_account']))
contract_address = add_0x(to_checksum_address(self.resources['token_index']['reference']))
gas_oracle = OverrideGasOracle(limit=TokenUniqueSymbolIndex.gas(), conn=self.rpc)
nonce_oracle = None
if self.rpc != None:
nonce_oracle = RPCNonceOracle(signer_address, conn=self.rpc)
nonce_oracle = RPCNonceOracle(add_0x(signer_address), conn=self.rpc)
c = TokenUniqueSymbolIndex(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
o = c.register(contract_address, signer_address, self.token_address, tx_format=self.tx_format)
r = None
if self.rpc != None:
r = self.rpc.do(o[1])
self.rpc.wait(r)
elif self.signer != None:
r = o[1]
else:
@ -211,8 +191,8 @@ class CICEth:
if writer == None:
writer = self.outputs_writer
signer_address = self.resources['address_declarator']['key_address']
contract_address = self.resources['address_declarator']['reference']
signer_address = add_0x(to_checksum_address(self.resources['address_declarator']['key_account']))
contract_address = add_0x(to_checksum_address(self.resources['address_declarator']['reference']))
gas_oracle = OverrideGasOracle(limit=AddressDeclarator.gas(), conn=self.rpc)
nonce_oracle = None
@ -229,39 +209,25 @@ class CICEth:
r = None
if self.rpc != None:
r = self.rpc.do(o[1])
self.rpc.wait(r)
elif self.signer != None:
r = o[1]
else:
r = o
self.add_outputs(k, r)
results.append(r)
v = r.encode('utf-8')
if writer != None:
writer.write(k, r.encode('utf-8'))
writer.write(k, v)
return results
def process(self, writer=None):
if writer == None:
writer = self.outputs_writer
def prepare_extension(self):
super(CICEth, self).prepare_extension()
tasks = []
self.token_address = self.resources['token']['reference']
if self.token_address == None:
tasks.append('token')
for k in self.resources.keys():
if k == 'token':
continue
if self.resources[k]['reference'] != None:
tasks.append(k)
for task in tasks:
logg.debug('ciceth adapter process {}'.format(task))
r = getattr(self, 'process_' + task)(writer=writer)
return self.token_address
if self.token_address != None:
self.token_address = add_0x(to_checksum_address(self.token_address))
def new(chain_spec, resources, proof, signer_hint=None, rpc=None):

112
cic/extension.py Normal file
View File

@ -0,0 +1,112 @@
# standard imports
import logging
# external imports
from hexathon import valid as valid_hex
logg = logging.getLogger(__name__)
class Extension:
def __init__(self, chain_spec, resources, proof, signer=None, rpc=None, outputs_writer=None):
self.resources = resources
self.proof = proof
self.chain_spec = chain_spec
self.signer = signer
self.rpc = rpc
self.token_details = None
self.token_address = None
self.token_code = None
self.outputs = []
# TODO: apply / prepare token can be factored out
def apply_token(self, token):
return self.prepare_token(token.name, token.symbol, token.precision, token.code)
def prepare_token(self, name, symbol, precision, code, extra=[], extra_types=[], positions=None):
self.token_details = {
'name': name,
'symbol': symbol,
'precision': precision,
'code': code,
'extra': extra,
'extra_types': extra_types,
'positions': positions,
}
def prepare_extension(self):
pass
def parse_code_as_file(self, v):
try:
f = open(v, 'r')
r = f.read()
f.close()
self.parse_code_as_hex(r)
except FileNotFoundError:
logg.debug('could not parse code as file: {}'.format(e))
pass
except IsADirectoryError:
logg.debug('could not parse code as file: {}'.format(e))
pass
def parse_code_as_hex(self, v):
try:
self.token_code = valid_hex(v)
except ValueError as e:
logg.debug('could not parse code as hex: {}'.format(e))
pass
def load_code(self, hint=None):
code = self.token_details['code']
if hint == 'hex':
self.token_code = valid_hex(code)
for m in [
self.parse_code_as_hex,
self.parse_code_as_file,
]:
m(code)
if self.token_code != None:
break
if self.token_code == None:
raise RuntimeError('could not successfully parse token code')
return self.token_code
def process(self, writer=None):
if writer == None:
writer = self.outputs_writer
tasks = []
self.token_address = self.resources['token']['reference']
if self.token_address == None:
if self.token_details['code'] == None:
raise RuntimeError('neither token address nor token code has been set')
self.load_code()
tasks.append('token')
for k in self.resources.keys():
if k == 'token':
continue
if self.resources[k]['reference'] != None:
tasks.append(k)
self.prepare_extension()
for task in tasks:
logg.debug('ciceth adapter process {}'.format(task))
r = getattr(self, 'process_' + task)(writer=writer)
return self.token_address

7
cic/output.py
View File

@ -5,11 +5,14 @@ import sys
class OutputWriter:
def __init__(self, *args, **kwargs):
pass
def write(self, k, v):
raise NotImplementedError()
class StdoutWriter:
class StdoutWriter(OutputWriter):
def write(self, k, v):
sys.stdout.write('{}\t{}\n'.format(k, v))
@ -17,7 +20,7 @@ class StdoutWriter:
class KVWriter(OutputWriter):
def __init__(self, path):
def __init__(self, path=None, *args, **kwargs):
os.stat(path)
self.path = path