Select Git revision
functions.py
Hugo Trentesaux authored
- export timestamp vs start timestamp - initial monetary mass - TODO treasury to collect ignored money - membership expiration date - warning if expired certs
functions.py 6.60 KiB
from lib.utility import *
from time import time
def load_json_url(path):
with open(path, "r") as fd:
return json.load(fd)
def get_wallets_data():
# Get wallets balances data
wallets_data = load_json_url("inputs/wallets.json")
wallets = {}
ignored_money = 0
for wallet in wallets_data:
balance = wallet["value"]["balance"]
if "&&" in wallet["key"]:
ignored_money += balance
continue
pubkey = wallet["key"].split("(")[1].split(")")[0]
# Remove pubkeys > 32 bytes
# d2meevcahfts2gqmvmrw5hzi25jddikk4nc4u1fkwrau
# afv1d5xa7fcdhcta1bqfq3pwvwem16gw67qj37obgnsv
# dyvfr3fqbs6g1yrktbstzxdkj3n7c5hrqf9buddzne4o
# 11111111111111111111111111111111111111111111
# jUPLL2BgY2QpheWEY3R13edV2Y4tvQMCXjJVM8PGDvyd
# gatrpfmgsuez193bja5snivz3dsvsqn5kcm4ydtpknsy
pubkey_bytes = base58.b58decode(pubkey)
pubkey_lenght = len(pubkey_bytes)
if pubkey_lenght > 32 or balance == 0:
ignored_money += balance
continue
wallets.update({v1_pubkey_to_v2_address(pubkey): int(balance)})
# add ignored money to treasury
# FIXME get treasury address
wallets["5EYCAe5ijiYfyeZ2JJCGq56LmPyNRAKzpG4QkoQkkQNB5e6Z"] = ignored_money
return wallets
def get_membership_expiry():
"""get membership expiry from input file"""
# Get Dex membership data
membership_data = load_json_url("inputs/membership.json")
membership_expiry = {}
for membership in membership_data:
membership_expiry[membership["key"]] = membership["value"][0]["expires_on"]
return membership_expiry
def get_identities_and_wallets(last_block_time, start_timestamp):
"""get identities with certifications and wallets with their balance
last_block_time is the time of the last exported block
start_timestamp is the timestamp of the v2 genesis
used to estimate cert expiration in number of blocks
"""
# initialize
blocs = {}
identity_names = {}
identities = {}
# Get wallets data
wallets = get_wallets_data()
# Get membership expiry
membership_expiry = get_membership_expiry()
# Get Dex idty data
idty_data = load_json_url("inputs/idty.json")
# Get Dex certs data
certs_data = load_json_url("inputs/certs.json")
# Get blocs number with dates
blocs_data = load_json_url("inputs/blocs.json")
# Get identities switches
addresses_switches = load_json("custom/addresses_switches.json")
# Get custom identities
custom_identities = load_json("custom/identities.json")
# TODO make sure that index respects order of arrival
# Get identities names by pubkey
for idty in idty_data:
pubkey = idty["key"]
address = v1_pubkey_to_v2_address(pubkey)
value = idty["value"][0]
index = value["wotb_id"] + 1
uid = value["uid"]
is_member = value["member"]
identity_names[pubkey] = uid
membership_expire_on = date_to_bloc_number(membership_expiry[pubkey], start_timestamp)
if membership_expire_on < 0 : membership_expire_on = 0 # forget old expiry date
# add address and balance to identity
if address not in wallets:
balance = 0
else:
balance = wallets[address]
# remove identity from wallet
# (only let simple wallets)
del wallets[address]
# fill in identity entry
identities[uid] = {
"index": index,
"owner_key": address,
"balance": balance,
"membership_expire_on": membership_expire_on if is_member else 0,
"next_cert_issuable_on": 0, # initialized to zero, modified later
"certs_received": {},
}
# Switch address to custom if exist
if uid in addresses_switches.keys():
identities[uid]["owner_key"] = addresses_switches[uid]["v2"]
identities[uid]["old_owner_key"] = v1_pubkey_to_v2_address(
addresses_switches[uid]["v1"]
)
# Add custom identities
identities.update(custom_identities)
# get info from block
for bloc in blocs_data:
blocs[int(bloc["key"])] = bloc["value"]["medianTime"]
# Generate identities Ğ1v2 genesis json bloc
# certs are stored per issuer in input file
# certs are stored per receiver in output file
print(" parse certification...")
for issuer in certs_data:
i_pubkey = issuer["key"]
i_uid = identity_names[i_pubkey]
i_address = v1_pubkey_to_v2_address(i_pubkey)
for cert in issuer["value"]["issued"]:
# if certification expired, skip silently
if cert["expired_on"] != 0 :
continue
r_pubkey = cert["receiver"]
r_uid = identity_names[r_pubkey]
r_address = v1_pubkey_to_v2_address(r_pubkey)
# get expiration of certification
# timestamp of cert creation
created_at = blocs[cert["created_on"]]
# block of cert creation
created_on = date_to_bloc_number(created_at, start_timestamp)
# block of next issuable cert
next_issuable_on = created_on + CERT_PERIOD
# timestamp of cert expiration
cert_expire_at = cert["expires_on"]
cert_expire_on = date_to_bloc_number(cert_expire_at, start_timestamp)
# if certification expiration date is before export,
# it is a renewed certification and can be ignored
if cert_expire_at < last_block_time:
continue
# certifications can also have expired between export and start_timestamp
# in this case we display a warning because these missing certification
# could lead to a genesis with not enough certifications received for some members
if cert_expire_at <= start_timestamp:
print(f"⚠️ {i_uid} → {r_uid} cert expired between export and start")
continue
# bump the next issuable date if necessary
elif next_issuable_on > identities[i_uid]["next_cert_issuable_on"]:
identities[i_uid]["next_cert_issuable_on"] = next_issuable_on
# add received certification to identity
identities[r_uid]["certs_received"][i_uid] = cert_expire_on
return [identities, wallets]
def get_smiths():
final_smiths = {}
smiths_brut = load_json("custom/smiths.json")
for smith in smiths_brut:
final_smiths.update(smith)
return final_smiths
def get_technical_committee():
return load_json("custom/technical_committee.json")