Edit on GitHub

communex.module.std.whitelist_validator

 1import time
 2
 3from substrateinterface import Keypair
 4
 5from communex._common import get_node_url
 6from communex.client import CommuneClient
 7from communex.compat.key import classic_load_key
 8from communex.types import Ss58Address
 9
10# The netuid of the general subnet.
11GENERAL_NETUID = 0
12
13
14def keys_to_uids(
15    keys: dict[int, Ss58Address], target_keys: list[Ss58Address]
16) -> list[int]:
17    return [k for k, v in keys.items() if v in target_keys]
18
19
20def validaiton(client: CommuneClient, key: Keypair):
21    # Query the modules, nominated by DAO.
22    keys = client.query_map_key(netuid=GENERAL_NETUID)
23    legit_whitelist = client.query_map(
24        "LegitWhitelist", params=[], extract_value=False
25    )["LegitWhitelist"]
26    # TODO: test on production query, to verify the format, now I am going to assume it is a dict.
27    target_keys = list(legit_whitelist.keys())
28    uids_map = keys_to_uids(keys, target_keys)
29    # WIP, rigt now assuming same weights
30    weights = [1 for _ in uids_map]
31
32    client.vote(key, uids=uids_map, weights=weights, netuid=GENERAL_NETUID)
33
34
35def main(
36    client: CommuneClient,
37    key: Keypair,
38):
39    while True:
40        validaiton(client, key)
41        time.sleep(1200)
42
43
44if __name__ == "__main__":
45    client = CommuneClient(get_node_url())
46    key = classic_load_key("foo")
47    main(client, key)
GENERAL_NETUID = 0
def keys_to_uids( keys: dict[int, communex.types.Ss58Address], target_keys: list[communex.types.Ss58Address]) -> list[int]:
15def keys_to_uids(
16    keys: dict[int, Ss58Address], target_keys: list[Ss58Address]
17) -> list[int]:
18    return [k for k, v in keys.items() if v in target_keys]
def validaiton( client: communex.client.CommuneClient, key: substrateinterface.keypair.Keypair):
21def validaiton(client: CommuneClient, key: Keypair):
22    # Query the modules, nominated by DAO.
23    keys = client.query_map_key(netuid=GENERAL_NETUID)
24    legit_whitelist = client.query_map(
25        "LegitWhitelist", params=[], extract_value=False
26    )["LegitWhitelist"]
27    # TODO: test on production query, to verify the format, now I am going to assume it is a dict.
28    target_keys = list(legit_whitelist.keys())
29    uids_map = keys_to_uids(keys, target_keys)
30    # WIP, rigt now assuming same weights
31    weights = [1 for _ in uids_map]
32
33    client.vote(key, uids=uids_map, weights=weights, netuid=GENERAL_NETUID)
def main( client: communex.client.CommuneClient, key: substrateinterface.keypair.Keypair):
36def main(
37    client: CommuneClient,
38    key: Keypair,
39):
40    while True:
41        validaiton(client, key)
42        time.sleep(1200)