communex.misc

  1import re
  2from typing import Any, TypeVar, cast
  3
  4from communex._common import transform_stake_dmap
  5from communex.balance import to_nano
  6from communex.client import CommuneClient
  7from communex.key import check_ss58_address
  8from communex.types import (
  9    BurnConfiguration, ModuleInfoWithOptionalBalance,
 10    NetworkParams, Ss58Address, SubnetParamsMaps,
 11    SubnetParamsWithEmission, GovernanceConfiguration
 12    )
 13
 14IPFS_REGEX = re.compile(r"^Qm[1-9A-HJ-NP-Za-km-z]{44}$")
 15
 16T = TypeVar("T")
 17
 18
 19def get_map_modules(
 20    client: CommuneClient,
 21    netuid: int = 0,
 22    include_balances: bool = False,
 23) -> dict[str, ModuleInfoWithOptionalBalance]:
 24    """
 25    Gets all modules info on the network
 26    """
 27
 28    request_dict: dict[Any, Any] = {
 29        "SubspaceModule": [
 30            ("StakeFrom", []),
 31            ("Keys", [netuid]),
 32            ("Name", [netuid]),
 33            ("Address", [netuid]),
 34            ("RegistrationBlock", [netuid]),
 35            ("DelegationFee", []),
 36            ("Emission", []),
 37            ("Incentive", []),
 38            ("Dividends", []),
 39            ("LastUpdate", []),
 40            ("Metadata", [netuid]),
 41        ],
 42    }
 43    if include_balances:
 44        request_dict["System"] = [("Account", [])]
 45
 46    bulk_query = client.query_batch_map(request_dict)
 47    (
 48        ss58_to_stakefrom,
 49        uid_to_key,
 50        uid_to_name,
 51        uid_to_address,
 52        uid_to_regblock,
 53        ss58_to_delegationfee,
 54        uid_to_emission,
 55        uid_to_incentive,
 56        uid_to_dividend,
 57        uid_to_lastupdate,
 58        ss58_to_balances,
 59        ss58_to_metadata,
 60    ) = (
 61        bulk_query.get("StakeFrom", {}),
 62        bulk_query.get("Keys", {}),
 63        bulk_query["Name"],
 64        bulk_query["Address"],
 65        bulk_query["RegistrationBlock"],
 66        bulk_query["DelegationFee"],
 67        bulk_query["Emission"],
 68        bulk_query["Incentive"],
 69        bulk_query["Dividends"],
 70        bulk_query["LastUpdate"],
 71        bulk_query.get("Account", {}),
 72        bulk_query.get("Metadata", {}),
 73    )
 74    result_modules: dict[str, ModuleInfoWithOptionalBalance] = {}
 75    ss58_to_stakefrom = transform_stake_dmap(ss58_to_stakefrom)
 76    for uid, key in uid_to_key.items():
 77        key = check_ss58_address(key)
 78        name = uid_to_name[uid]
 79        address = uid_to_address[uid]
 80        emission = uid_to_emission[netuid][uid]
 81        incentive = uid_to_incentive[netuid][uid]
 82        dividend = uid_to_dividend[netuid][uid]
 83        regblock = uid_to_regblock[uid]
 84        stake_from = ss58_to_stakefrom.get(key, [])
 85        last_update = uid_to_lastupdate[netuid][uid]
 86        delegation_fee = ss58_to_delegationfee.get(
 87            key, 5
 88        )  # 5% default delegation fee
 89        metadata = ss58_to_metadata.get(key, None)
 90
 91        balance = None
 92        if include_balances and ss58_to_balances is not None:  # type: ignore
 93            balance_dict = ss58_to_balances.get(key, None)
 94            if balance_dict is not None:
 95                assert isinstance(balance_dict["data"], dict)
 96                balance = balance_dict["data"]["free"]
 97            else:
 98                balance = 0
 99        stake = sum(stake for _, stake in stake_from)
100
101        module: ModuleInfoWithOptionalBalance = {
102            "uid": uid,
103            "key": key,
104            "name": name,
105            "address": address,
106            "emission": emission,
107            "incentive": incentive,
108            "dividends": dividend,
109            "stake_from": stake_from,
110            "regblock": regblock,
111            "last_update": last_update,
112            "balance": balance,
113            "stake": stake,
114            "delegation_fee": delegation_fee,
115            "metadata": metadata,
116        }
117
118        result_modules[key] = module
119    return result_modules
120
121
122def to_snake_case(d: dict[str, T]) -> dict[str, T]:
123    """
124    Converts a dictionary with camelCase keys to snake_case keys
125    """
126    def snakerize(camel: str) -> str:
127        return re.sub(r'(?<!^)(?=[A-Z])', '_', camel).lower()
128    snaked: dict[str, T] = {snakerize(k): v for k, v in d.items()}
129    return snaked
130
131
132def get_map_subnets_params(
133    client: CommuneClient, block_hash: str | None = None
134) -> dict[int, SubnetParamsWithEmission]:
135    """
136    Gets all subnets info on the network
137    """
138    bulk_query = client.query_batch_map(
139        {
140            "SubspaceModule": [
141                ("ImmunityPeriod", []),
142                ("MinAllowedWeights", []),
143                ("MaxAllowedWeights", []),
144                ("Tempo", []),
145                ("MaxAllowedUids", []),
146                ("Founder", []),
147                ("FounderShare", []),
148                ("IncentiveRatio", []),
149                ("TrustRatio", []),
150                ("SubnetNames", []),
151                ("MaxWeightAge", []),
152                ("BondsMovingAverage", []),
153                ("MaximumSetWeightCallsPerEpoch", []),
154                ("MinValidatorStake", []),
155                ("MaxAllowedValidators", []),
156                ("ModuleBurnConfig", []),
157                ("SubnetMetadata", []),
158            ],
159            "GovernanceModule": [
160                ("SubnetGovernanceConfig", []),
161            ],
162            "SubnetEmissionModule": [
163                ("SubnetEmission", []),
164            ],
165
166        },
167        block_hash,
168    )
169    subnet_maps: SubnetParamsMaps = {
170        "netuid_to_emission": bulk_query["SubnetEmission"],
171        "netuid_to_tempo": bulk_query["Tempo"],
172        "netuid_to_min_allowed_weights": bulk_query["MinAllowedWeights"],
173        "netuid_to_max_allowed_weights": bulk_query["MaxAllowedWeights"],
174        "netuid_to_max_allowed_uids": bulk_query["MaxAllowedUids"],
175        "netuid_to_founder": bulk_query["Founder"],
176        "netuid_to_founder_share": bulk_query["FounderShare"],
177        "netuid_to_incentive_ratio": bulk_query["IncentiveRatio"],
178        "netuid_to_trust_ratio": bulk_query["TrustRatio"],
179        "netuid_to_name": bulk_query["SubnetNames"],
180        "netuid_to_max_weight_age": bulk_query["MaxWeightAge"],
181        "netuid_to_bonds_ma": bulk_query.get("BondsMovingAverage", {}),
182        "netuid_to_maximum_set_weight_calls_per_epoch": bulk_query.get("MaximumSetWeightCallsPerEpoch", {}),
183        "netuid_to_governance_configuration": bulk_query["SubnetGovernanceConfig"],
184        "netuid_to_immunity_period": bulk_query["ImmunityPeriod"],
185        "netuid_to_min_validator_stake": bulk_query.get("MinValidatorStake", {}),
186        "netuid_to_max_allowed_validators": bulk_query.get("MaxAllowedValidators", {}),
187        "netuid_to_module_burn_config": bulk_query.get("ModuleBurnConfig", {}),
188        "netuid_to_subnet_metadata": bulk_query.get("SubnetMetadata", {}),
189    }
190    result_subnets: dict[int, SubnetParamsWithEmission] = {}
191
192    for netuid, name in subnet_maps["netuid_to_name"].items():
193
194        subnet: SubnetParamsWithEmission = {
195            "name": name,
196            "founder": subnet_maps["netuid_to_founder"][netuid],
197            "founder_share": subnet_maps["netuid_to_founder_share"][netuid],
198            "incentive_ratio": subnet_maps["netuid_to_incentive_ratio"][netuid],
199            "max_allowed_uids": subnet_maps["netuid_to_max_allowed_uids"][netuid],
200            "max_allowed_weights": subnet_maps["netuid_to_max_allowed_weights"][netuid],
201            "min_allowed_weights": subnet_maps["netuid_to_min_allowed_weights"][netuid],
202            "tempo": subnet_maps["netuid_to_tempo"][netuid],
203            "trust_ratio": subnet_maps["netuid_to_trust_ratio"][netuid],
204            "emission": subnet_maps["netuid_to_emission"][netuid],
205            "max_weight_age": subnet_maps["netuid_to_max_weight_age"][netuid],
206            "bonds_ma": subnet_maps["netuid_to_bonds_ma"].get(netuid, None),
207            "maximum_set_weight_calls_per_epoch": subnet_maps["netuid_to_maximum_set_weight_calls_per_epoch"].get(netuid, 30),
208            "governance_config": subnet_maps["netuid_to_governance_configuration"][netuid],
209            "immunity_period": subnet_maps["netuid_to_immunity_period"][netuid],
210            "min_validator_stake": subnet_maps["netuid_to_min_validator_stake"].get(netuid, to_nano(50_000)),
211            "max_allowed_validators": subnet_maps["netuid_to_max_allowed_validators"].get(netuid, 50),
212            "module_burn_config": cast(BurnConfiguration, subnet_maps["netuid_to_module_burn_config"].get(netuid, None)),
213            "subnet_metadata": subnet_maps["netuid_to_subnet_metadata"].get(netuid, None),
214        }
215
216        result_subnets[netuid] = subnet
217
218    return result_subnets
219
220
221def get_global_params(c_client: CommuneClient) -> NetworkParams:
222    """
223    Returns global parameters of the whole commune ecosystem
224    """
225
226    query_all = c_client.query_batch(
227        {
228            "SubspaceModule": [
229                ("MaxNameLength", []),
230                ("MinNameLength", []),
231                ("MaxAllowedSubnets", []),
232                ("MaxAllowedModules", []),
233                ("MaxRegistrationsPerBlock", []),
234                ("MaxAllowedWeightsGlobal", []),
235                ("FloorDelegationFee", []),
236                ("FloorFounderShare", []),
237                ("MinWeightStake", []),
238                ("Kappa", []),
239                ("Rho", []),
240                ("SubnetImmunityPeriod", []),
241            ],
242            "GovernanceModule": [
243                ("GlobalGovernanceConfig", []),
244                ("GeneralSubnetApplicationCost", []),
245                ("Curator", []),
246            ]
247        }
248    )
249    global_config = cast(
250        GovernanceConfiguration,
251        query_all["GlobalGovernanceConfig"]
252    )
253    global_params: NetworkParams = {
254        "max_allowed_subnets": int(query_all["MaxAllowedSubnets"]),
255        "max_allowed_modules": int(query_all["MaxAllowedModules"]),
256        "max_registrations_per_block": int(query_all["MaxRegistrationsPerBlock"]),
257        "max_name_length": int(query_all["MaxNameLength"]),
258        "min_weight_stake": int(query_all["MinWeightStake"]),
259        "floor_delegation_fee": int(query_all["FloorDelegationFee"]),
260        "max_allowed_weights": int(query_all["MaxAllowedWeightsGlobal"]),
261        "curator": Ss58Address(query_all["Curator"]),
262        "min_name_length": int(query_all["MinNameLength"]),
263        "floor_founder_share": int(query_all["FloorFounderShare"]),
264        "general_subnet_application_cost": int(query_all["GeneralSubnetApplicationCost"]),
265        "kappa": int(query_all["Kappa"]),
266        "rho": int(query_all["Rho"]),
267        "subnet_immunity_period": int(query_all["SubnetImmunityPeriod"]),
268        "governance_config": {
269            "proposal_cost": int(global_config["proposal_cost"]),
270            "proposal_expiration": int(global_config["proposal_expiration"]),
271            "vote_mode": global_config["vote_mode"],
272            "proposal_reward_treasury_allocation": int(global_config["proposal_reward_treasury_allocation"]),
273            "max_proposal_reward_treasury_allocation": int(global_config["max_proposal_reward_treasury_allocation"]),
274            "proposal_reward_interval": int(global_config["proposal_reward_interval"])
275        }
276    }
277    return global_params
278
279
280def concat_to_local_keys(
281    balance: dict[str, int], local_key_info: dict[str, Ss58Address]
282) -> dict[str, int]:
283    key2: dict[str, int] = {
284        key_name: balance.get(key_address, 0)
285        for key_name, key_address in local_key_info.items()
286    }
287
288    return key2
289
290
291def local_keys_to_freebalance(
292    c_client: CommuneClient,
293    local_keys: dict[str, Ss58Address],
294) -> dict[str, int]:
295    query_all = c_client.query_batch_map(
296        {
297            "System": [("Account", [])],
298        }
299    )
300    balance_map = query_all["Account"]
301
302    format_balances: dict[str, int] = {
303        key: value["data"]["free"]
304        for key, value in balance_map.items()
305        if "data" in value and "free" in value["data"]
306    }
307
308    key2balance: dict[str, int] = concat_to_local_keys(format_balances, local_keys)
309
310    return key2balance
311
312
313def local_keys_to_stakedbalance(
314    c_client: CommuneClient,
315    local_keys: dict[str, Ss58Address],
316) -> dict[str, int]:
317    staketo_map = c_client.query_map_staketo()
318
319    format_stake: dict[str, int] = {
320        key: sum(stake for _, stake in value) for key, value in staketo_map.items()
321    }
322
323    key2stake: dict[str, int] = concat_to_local_keys(format_stake, local_keys)
324
325    return key2stake
326
327
328def local_keys_to_stakedfrom_balance(
329    c_client: CommuneClient,
330    local_keys: dict[str, Ss58Address],
331) -> dict[str, int]:
332    stakefrom_map = c_client.query_map_stakefrom()
333
334    format_stake: dict[str, int] = {
335        key: sum(stake for _, stake in value) for key, value in stakefrom_map.items()
336    }
337
338    key2stake: dict[str, int] = concat_to_local_keys(format_stake, local_keys)
339    key2stake = {key: stake for key, stake in key2stake.items()}
340    return key2stake
341
342
343def local_keys_allbalance(
344    c_client: CommuneClient,
345    local_keys: dict[str, Ss58Address],
346) -> tuple[dict[str, int], dict[str, int]]:
347    query_all = c_client.query_batch_map(
348        {
349            "System": [("Account", [])],
350            "SubspaceModule": [
351                ("StakeTo", []),
352            ],
353        }
354    )
355
356    balance_map, staketo_map = query_all["Account"], transform_stake_dmap(query_all["StakeTo"])
357
358    format_balances: dict[str, int] = {
359        key: value["data"]["free"]
360        for key, value in balance_map.items()
361        if "data" in value and "free" in value["data"]
362    }
363    key2balance: dict[str, int] = concat_to_local_keys(format_balances, local_keys)
364    format_stake: dict[str, int] = {
365        key: sum(stake for _, stake in value) for key, value in staketo_map.items()
366    }
367
368    key2stake: dict[str, int] = concat_to_local_keys(format_stake, local_keys)
369
370    key2balance = {
371        k: v
372        for k, v in sorted(key2balance.items(), key=lambda item: item[1], reverse=True)
373    }
374
375    key2stake = {
376        k: v
377        for k, v in sorted(key2stake.items(), key=lambda item: item[1], reverse=True)
378    }
379
380    return key2balance, key2stake
381
382
383if __name__ == "__main__":
384    from communex._common import get_node_url
385    client = CommuneClient(get_node_url(use_testnet=True))
386    get_global_params(client)