laconicd-stack/scripts/transfer-state.py
shreerang 3d7ba45796 Add playbook to send create-validator tx for subsequent nodes (#3)
Part of https://www.notion.so/Create-stacks-for-mainnet-1f2a6b22d4728034be4be2c51decf94e of cerc-io/laconicd#66
- Use calculated staking amount for running validator nodes

Co-authored-by: Shreerang Kale <shreerangkale@gmail.com>
Co-authored-by: Nabarun <nabarun@deepstacksoft.com>
Co-authored-by: Prathamesh Musale <prathamesh.musale0@gmail.com>
Reviewed-on: #3
Co-authored-by: shreerang <shreerang@noreply.git.vdb.to>
Co-committed-by: shreerang <shreerang@noreply.git.vdb.to>
2025-05-16 09:54:09 +00:00

173 lines
5.2 KiB
Python

import json
from decimal import Decimal
from collections import defaultdict
from bech32 import bech32_decode, bech32_encode, convertbits
#------
# Helper methods
def valoper_to_account_address(valoper_addr: str, new_prefix = "laconic") -> str:
hrp, data = bech32_decode(valoper_addr)
if hrp is None or data is None:
raise ValueError("Invalid bech32 address")
# Convert back from 5-bit to 8-bit
decoded = convertbits(data, 5, 8, False)
if decoded is None:
raise ValueError("Failed to convert bits")
# Re-encode with new prefix
converted_data = convertbits(decoded, 8, 5, True)
return bech32_encode(new_prefix, converted_data)
def get_min_delegation_share(state):
validators = state["app_state"]["staking"]["validators"]
delegations = state["app_state"]["staking"]["delegations"]
shares_list = []
for val in validators:
valoper_addr = val["operator_address"]
orig_delegator_addr = valoper_to_account_address(valoper_addr)
# Find delegations where the delegator is the original delegator and delegated to their validator
for delegation in delegations:
if (delegation["delegator_address"] == orig_delegator_addr and
delegation["validator_address"] == valoper_addr):
shares = int(Decimal(delegation["shares"]))
shares_list.append(shares)
break
# Find minimum
return min(shares_list)
#------
# Read testnet and mainnet states
mainnet_genesis_dir="mainnet-genesis"
testnet_state_file=f"{mainnet_genesis_dir}/testnet-state.json"
mainnet_genesis_file=f"{mainnet_genesis_dir}/config/genesis.json"
staking_amount_file=f"output/staking-amount.json"
with open(testnet_state_file) as f:
testnet_state = json.load(f)
with open(mainnet_genesis_file) as f:
mainnet_state = json.load(f)
#------
# Import required module state
mainnet_state["app_state"]["auth"] = testnet_state["app_state"]["auth"]
mainnet_state["app_state"]["bond"] = testnet_state["app_state"]["bond"]
mainnet_state["app_state"]["bank"] = testnet_state["app_state"]["bank"]
mainnet_state["app_state"]["registry"] = testnet_state["app_state"]["registry"]
mainnet_state["app_state"]["slashing"]["params"] = testnet_state["app_state"]["slashing"]["params"]
mainnet_state["consensus"]["params"] = testnet_state["consensus"]["params"]
# Update authority auction durations from 3600s to 60s
mainnet_state["app_state"]["registry"]["params"]["authority_auction_commits_duration"] = "60s"
mainnet_state["app_state"]["registry"]["params"]["authority_auction_reveals_duration"] = "60s"
#------
# Allocate back delegation stakes
# Build map of address -> extra balance to add (as integers)
deltas = {}
for delegation in testnet_state["app_state"]["staking"]["delegations"]:
addr = delegation["delegator_address"]
amount = int(Decimal(delegation["shares"]))
deltas[addr] = deltas.get(addr, 0) + amount
# Now apply the deltas to existing balances
supply_increment = 0
for balance in mainnet_state["app_state"]["bank"]["balances"]:
addr = balance["address"]
if addr in deltas:
for coin in balance["coins"]:
if coin["denom"] == "alnt":
coin["amount"] = str(int(coin["amount"]) + deltas[addr])
supply_increment += deltas[addr]
break
del deltas[addr]
# Increase the total supply
for coin in mainnet_state["app_state"]["bank"]["supply"]:
if coin["denom"] == "alnt":
coin["amount"] = str(int(coin["amount"]) + supply_increment)
#------
# Remove non-required module accounts
# Addresses to remove
addresses_to_remove = {
"bonded_tokens_pool",
"not_bonded_tokens_pool",
"distribution"
}
# Remove from auth.accounts and get their addresses
removed_addresses = set()
new_accounts = []
for account in mainnet_state["app_state"]["auth"]["accounts"]:
account_type = account.get("@type", "")
if "ModuleAccount" in account_type and account.get("name") in addresses_to_remove:
removed_addresses.add(account["base_account"]["address"])
continue
if "ModuleAccount" not in account_type:
account["sequence"] = "0"
new_accounts.append(account)
mainnet_state["app_state"]["auth"]["accounts"] = new_accounts
# Remove from bank.balances and tally removed amounts
new_balances = []
removed_amounts = defaultdict(int)
for bal in mainnet_state["app_state"]["bank"]["balances"]:
if bal["address"] in removed_addresses:
# Skip this account
for coin in bal["coins"]:
denom = coin["denom"]
amount = int(coin["amount"])
removed_amounts[denom] += amount
continue
new_balances.append(bal)
mainnet_state["app_state"]["bank"]["balances"] = new_balances
# Reduce from bank supply
new_supply = []
for coin in mainnet_state["app_state"]["bank"]["supply"]:
denom = coin["denom"]
amount = int(coin["amount"])
amount -= removed_amounts.get(denom, 0)
new_supply.append({
"denom": denom,
"amount": str(amount)
})
mainnet_state["app_state"]["bank"]["supply"] = new_supply
#------
# Find minimum delegation share for common staking amount
min_delegation_share = get_min_delegation_share(testnet_state)
with open(staking_amount_file, "w") as f:
json.dump({"common_staking_amount": min_delegation_share}, f, indent=2)
print(f"Staking amount written to {staking_amount_file}")
# Write back modified state
with open(mainnet_genesis_file, "w") as f:
json.dump(mainnet_state, f, indent=2)