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