Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def process(self):
return self.get_next_bytes(self.element_count).decode()
class ChangesTrieRoot(Bytes):
pass
class SealV0(Struct):
type_string = '(u64, Signature)'
type_mapping = (('slot', 'u64'), ('signature', 'Signature'))
class Consensus(Struct):
type_string = '(ConsensusEngineId, Vec)'
type_mapping = (('engine', 'ConsensusEngineId'), ('data', 'HexBytes'))
class Seal(Struct):
type_string = '(ConsensusEngineId, Bytes)'
type_mapping = (('engine', 'ConsensusEngineId'), ('data', 'HexBytes'))
class PreRuntime(Struct):
type_string = '(ConsensusEngineId, Bytes)'
type_mapping = (('engine', 'ConsensusEngineId'), ('data', 'HexBytes'))
pass
class PrefabWasmModule(Struct):
type_string = 'wasm::PrefabWasmModule'
type_mapping = (
('scheduleVersion', 'Compact'),
('initial', 'Compact'),
('maximum', 'Compact'),
('_reserved', 'Option'),
('code', 'Bytes'),
)
class SessionKeysSubstrate(Struct):
type_mapping = (
('grandpa', 'AccountId'),
('babe', 'AccountId'),
('im_online', 'AccountId'),
)
class LegacyKeys(Struct):
type_mapping = (
('grandpa', 'AccountId'),
('babe', 'AccountId'),
)
# Encode call params
if len(self.call_function.args) > 0:
for arg in self.call_function.args:
if arg.name not in value['call_args']:
raise ValueError('Parameter \'{}\' not specified'.format(arg.name))
else:
param_value = value['call_args'][arg.name]
arg_obj = self.get_decoder_class(arg.type, metadata=self.metadata)
data += arg_obj.encode(param_value)
return data
class ProposalPreimage(Struct):
type_string = '(Vec, AccountId, BalanceOf, BlockNumber)'
type_mapping = (
("proposal", "HexBytes"),
("registredBy", "AccountId"),
("deposit", "BalanceOf"),
("blockNumber", "BlockNumber")
)
def process(self):
result = {}
for key, data_type in self.type_mapping:
result[key] = self.process_type(data_type, metadata=self.metadata).value
# Replace HexBytes with actual proposal
result['proposal'] = Proposal(ScaleBytes(result['proposal']), metadata=self.metadata).decode()
result = {}
for key, data_type in self.type_mapping:
result[key] = self.process_type(data_type, metadata=self.metadata).value
# Replace HexBytes with actual proposal
result['proposal'] = Proposal(ScaleBytes(result['proposal']), metadata=self.metadata).decode()
return result
class Proposal(BoxProposal):
type_string = '>::Proposal'
class ValidatorPrefs(Struct):
type_string = '(Compact)'
type_mapping = (('commission', 'Compact'),)
class ValidatorPrefsLegacy(Struct):
type_string = '(Compact,Compact)'
type_mapping = (('unstakeThreshold', 'Compact'), ('validatorPayment', 'Compact'))
class Linkage(Struct):
type_string = 'Linkage'
type_mapping = (
('previous', 'Option'),
type_mapping = (
('grandpa', 'AccountId'),
)
class QueuedKeys(Struct):
type_string = '(ValidatorId, Keys)'
type_mapping = (
('validator', 'ValidatorId'),
('keys', 'Keys'),
)
class LegacyQueuedKeys(Struct):
type_string = '(ValidatorId, LegacyKeys)'
type_mapping = (
('validator', 'ValidatorId'),
('keys', 'LegacyKeys'),
)
class EdgewareQueuedKeys(Struct):
type_string = '(ValidatorId, EdgewareKeys)'
type_mapping = (
('validator', 'ValidatorId'),
('keys', 'EdgewareKeys'),
('maximum', 'Compact'),
('_reserved', 'Option'),
('code', 'Bytes'),
)
class SessionKeysSubstrate(Struct):
type_mapping = (
('grandpa', 'AccountId'),
('babe', 'AccountId'),
('im_online', 'AccountId'),
)
class LegacyKeys(Struct):
type_mapping = (
('grandpa', 'AccountId'),
('babe', 'AccountId'),
)
class EdgewareKeys(Struct):
type_mapping = (
('grandpa', 'AccountId'),
)
class QueuedKeys(Struct):
type_string = '(ValidatorId, Keys)'
result['proposal'] = Proposal(ScaleBytes(result['proposal']), metadata=self.metadata).decode()
return result
class Proposal(BoxProposal):
type_string = '>::Proposal'
class ValidatorPrefs(Struct):
type_string = '(Compact)'
type_mapping = (('commission', 'Compact'),)
class ValidatorPrefsLegacy(Struct):
type_string = '(Compact,Compact)'
type_mapping = (('unstakeThreshold', 'Compact'), ('validatorPayment', 'Compact'))
class Linkage(Struct):
type_string = 'Linkage'
type_mapping = (
('previous', 'Option'),
('next', 'Option')
)
class GenericAccountId(H256):
class LegacyKeys(Struct):
type_mapping = (
('grandpa', 'AccountId'),
('babe', 'AccountId'),
)
class EdgewareKeys(Struct):
type_mapping = (
('grandpa', 'AccountId'),
)
class QueuedKeys(Struct):
type_string = '(ValidatorId, Keys)'
type_mapping = (
('validator', 'ValidatorId'),
('keys', 'Keys'),
)
class LegacyQueuedKeys(Struct):
type_string = '(ValidatorId, LegacyKeys)'
type_mapping = (
('validator', 'ValidatorId'),
('keys', 'LegacyKeys'),
self.ss58_address = None
super().__init__(data, sub_type, metadata)
def process_encode(self, value):
if value[0:2] != '0x' and len(value) == 47:
from scalecodec.utils.ss58 import ss58_decode
self.ss58_address = value
value = '0x{}'.format(ss58_decode(value))
return super().process_encode(value)
class GenericAccountIndex(U32):
pass
class KeyValue(Struct):
type_string = '(Vec, Vec)'
type_mapping = (('key', 'Vec'), ('value', 'Vec'))
class NewAccountOutcome(CompactU32):
type_string = 'NewAccountOutcome'
class Vec(ScaleType):
def __init__(self, data=None, **kwargs):
self.elements = []
super().__init__(data, **kwargs)
def process(self):
element_count = self.process_type('Compact').value
pass
class SealV0(Struct):
type_string = '(u64, Signature)'
type_mapping = (('slot', 'u64'), ('signature', 'Signature'))
class Consensus(Struct):
type_string = '(ConsensusEngineId, Vec)'
type_mapping = (('engine', 'ConsensusEngineId'), ('data', 'HexBytes'))
class Seal(Struct):
type_string = '(ConsensusEngineId, Bytes)'
type_mapping = (('engine', 'ConsensusEngineId'), ('data', 'HexBytes'))
class PreRuntime(Struct):
type_string = '(ConsensusEngineId, Bytes)'
type_mapping = (('engine', 'ConsensusEngineId'), ('data', 'HexBytes'))
class LogDigest(Enum):
value_list = ['Other', 'AuthoritiesChange', 'ChangesTrieRoot', 'SealV0', 'Consensus', 'Seal', 'PreRuntime']
def __init__(self, data, **kwargs):