119 lines
3.3 KiB
Python
119 lines
3.3 KiB
Python
# standard imports
|
|
import logging
|
|
|
|
# external imports
|
|
from hexathon import valid as valid_hex
|
|
|
|
# local imports
|
|
from cic.output import StdoutWriter
|
|
|
|
logg = logging.getLogger(__name__)
|
|
|
|
|
|
class Extension:
|
|
|
|
def __init__(self, chain_spec, resources, proof, signer=None, rpc=None, outputs_writer=StdoutWriter()):
|
|
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 = []
|
|
self.outputs_writer = outputs_writer
|
|
|
|
|
|
# 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, token.supply)
|
|
|
|
|
|
def prepare_token(self, name, symbol, precision, code, supply, extra=[], extra_types=[], positions=None):
|
|
self.token_details = {
|
|
'name': name,
|
|
'symbol': symbol,
|
|
'precision': precision,
|
|
'code': code,
|
|
'supply': supply,
|
|
'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']
|
|
|
|
# TODO: get token details when token address is not none
|
|
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, self.token_details['symbol'])
|