communex.client
1import json 2import queue 3from concurrent.futures import Future, ThreadPoolExecutor 4from contextlib import contextmanager 5from copy import deepcopy 6from dataclasses import dataclass 7from typing import Any, Mapping, TypeVar, cast 8 9from substrateinterface import ExtrinsicReceipt # type: ignore 10from substrateinterface import Keypair # type: ignore 11from substrateinterface import SubstrateInterface # type: ignore 12from substrateinterface.storage import StorageKey # type: ignore 13 14from communex._common import transform_stake_dmap 15from communex.errors import ChainTransactionError, NetworkQueryError 16from communex.types import NetworkParams, Ss58Address, SubnetParams 17 18# TODO: InsufficientBalanceError, MismatchedLengthError etc 19 20MAX_REQUEST_SIZE = 9_000_000 21 22 23@dataclass 24class Chunk: 25 batch_requests: list[tuple[Any, Any]] 26 prefix_list: list[list[str]] 27 fun_params: list[tuple[Any, Any, Any, Any, str]] 28 29 30T1 = TypeVar("T1") 31T2 = TypeVar("T2") 32 33 34class CommuneClient: 35 """ 36 A client for interacting with Commune network nodes, querying storage, 37 submitting transactions, etc. 38 39 Attributes: 40 wait_for_finalization: Whether to wait for transaction finalization. 41 42 Example: 43 ```py 44 client = CommuneClient() 45 client.query(name='function_name', params=['param1', 'param2']) 46 ``` 47 48 Raises: 49 AssertionError: If the maximum connections value is less than or equal 50 to zero. 51 """ 52 53 wait_for_finalization: bool 54 _num_connections: int 55 _connection_queue: queue.Queue[SubstrateInterface] 56 url: str 57 58 def __init__( 59 self, 60 url: str, 61 num_connections: int = 1, 62 wait_for_finalization: bool = False, 63 timeout: int | None = None, 64 ): 65 """ 66 Args: 67 url: The URL of the network node to connect to. 68 num_connections: The number of websocket connections to be opened. 69 """ 70 assert num_connections > 0 71 self._num_connections = num_connections 72 self.wait_for_finalization = wait_for_finalization 73 self._connection_queue = queue.Queue(num_connections) 74 self.url = url 75 ws_options: dict[str, int] = {} 76 if timeout is not None: 77 ws_options["timeout"] = timeout 78 self.ws_options = ws_options 79 for _ in range(num_connections): 80 self._connection_queue.put( 81 SubstrateInterface(url, ws_options=ws_options) 82 ) 83 84 @property 85 def connections(self) -> int: 86 """ 87 Gets the maximum allowed number of simultaneous connections to the 88 network node. 89 """ 90 return self._num_connections 91 92 @contextmanager 93 def get_conn(self, timeout: float | None = None, init: bool = False): 94 """ 95 Context manager to get a connection from the pool. 96 97 Tries to get a connection from the pool queue. If the queue is empty, 98 it blocks for `timeout` seconds until a connection is available. If 99 `timeout` is None, it blocks indefinitely. 100 101 Args: 102 timeout: The maximum time in seconds to wait for a connection. 103 104 Yields: 105 The connection object from the pool. 106 107 Raises: 108 QueueEmptyError: If no connection is available within the timeout 109 period. 110 """ 111 conn = self._connection_queue.get(timeout=timeout) 112 if init: 113 conn.init_runtime() # type: ignore 114 try: 115 if conn.websocket and conn.websocket.connected: # type: ignore 116 yield conn 117 else: 118 conn = SubstrateInterface(self.url, ws_options=self.ws_options) 119 yield conn 120 finally: 121 self._connection_queue.put(conn) 122 123 def _get_storage_keys( 124 self, 125 storage: str, 126 queries: list[tuple[str, list[Any]]], 127 block_hash: str | None, 128 ): 129 130 send: list[tuple[str, list[Any]]] = [] 131 prefix_list: list[Any] = [] 132 133 key_idx = 0 134 with self.get_conn(init=True) as substrate: 135 for function, params in queries: 136 storage_key = StorageKey.create_from_storage_function( # type: ignore 137 storage, function, params, runtime_config=substrate.runtime_config, metadata=substrate.metadata # type: ignore 138 ) 139 140 prefix = storage_key.to_hex() 141 prefix_list.append(prefix) 142 send.append(("state_getKeys", [prefix, block_hash])) 143 key_idx += 1 144 return send, prefix_list 145 146 def _get_lists( 147 self, 148 storage_module: str, 149 queries: list[tuple[str, list[Any]]], 150 substrate: SubstrateInterface, 151 ) -> list[tuple[Any, Any, Any, Any, str]]: 152 """ 153 Generates a list of tuples containing parameters for each storage function based on the given functions and substrate interface. 154 155 Args: 156 functions (dict[str, list[query_call]]): A dictionary where keys are storage module names and values are lists of tuples. 157 Each tuple consists of a storage function name and its parameters. 158 substrate: An instance of the SubstrateInterface class used to interact with the substrate. 159 160 Returns: 161 A list of tuples in the format `(value_type, param_types, key_hashers, params, storage_function)` for each storage function in the given functions. 162 163 Example: 164 >>> _get_lists( 165 functions={'storage_module': [('storage_function', ['param1', 'param2'])]}, 166 substrate=substrate_instance 167 ) 168 [('value_type', 'param_types', 'key_hashers', ['param1', 'param2'], 'storage_function'), ...] 169 """ 170 171 function_parameters: list[tuple[Any, Any, Any, Any, str]] = [] 172 173 metadata_pallet = substrate.metadata.get_metadata_pallet( # type: ignore 174 storage_module 175 ) 176 for storage_function, params in queries: 177 storage_item = metadata_pallet.get_storage_function( # type: ignore 178 storage_function 179 ) 180 181 value_type = storage_item.get_value_type_string() # type: ignore 182 param_types = storage_item.get_params_type_string() # type: ignore 183 key_hashers = storage_item.get_param_hashers() # type: ignore 184 function_parameters.append( 185 ( 186 value_type, 187 param_types, 188 key_hashers, 189 params, 190 storage_function, 191 ) # type: ignore 192 ) 193 return function_parameters 194 195 def _send_batch( 196 self, 197 batch_payload: list[Any], 198 request_ids: list[int], 199 extract_result: bool = True, 200 ): 201 """ 202 Sends a batch of requests to the substrate and collects the results. 203 204 Args: 205 substrate: An instance of the substrate interface. 206 batch_payload: The payload of the batch request. 207 request_ids: A list of request IDs for tracking responses. 208 results: A list to store the results of the requests. 209 extract_result: Whether to extract the result from the response. 210 211 Raises: 212 NetworkQueryError: If there is an `error` in the response message. 213 214 Note: 215 No explicit return value as results are appended to the provided 'results' list. 216 """ 217 results: list[str | dict[Any, Any]] = [] 218 with self.get_conn(init=True) as substrate: 219 try: 220 221 substrate.websocket.send( # type: ignore 222 json.dumps(batch_payload) 223 ) # type: ignore 224 except NetworkQueryError: 225 pass 226 while len(results) < len(request_ids): 227 received_messages = json.loads( 228 substrate.websocket.recv() # type: ignore 229 ) # type: ignore 230 if isinstance(received_messages, dict): 231 received_messages: list[dict[Any, Any]] = [received_messages] 232 233 for message in received_messages: 234 if message.get("id") in request_ids: 235 if extract_result: 236 try: 237 results.append(message["result"]) 238 except Exception: 239 raise ( 240 RuntimeError( 241 f"Error extracting result from message: {message}" 242 ) 243 ) 244 else: 245 results.append(message) 246 if "error" in message: 247 raise NetworkQueryError(message["error"]) 248 249 return results 250 251 def _make_request_smaller( 252 self, 253 batch_request: list[tuple[T1, T2]], 254 prefix_list: list[list[str]], 255 fun_params: list[tuple[Any, Any, Any, Any, str]], 256 ) -> tuple[list[list[tuple[T1, T2]]], list[Chunk]]: 257 """ 258 Splits a batch of requests into smaller batches, each not exceeding the specified maximum size. 259 260 Args: 261 batch_request: A list of requests to be sent in a batch. 262 max_size: Maximum size of each batch in bytes. 263 264 Returns: 265 A list of smaller request batches. 266 267 Example: 268 >>> _make_request_smaller(batch_request=[('method1', 'params1'), ('method2', 'params2')], max_size=1000) 269 [[('method1', 'params1')], [('method2', 'params2')]] 270 """ 271 assert len(prefix_list) == len(fun_params) == len(batch_request) 272 273 def estimate_size(request: tuple[T1, T2]): 274 """Convert the batch request to a string and measure its length""" 275 return len(json.dumps(request)) 276 277 # Initialize variables 278 result: list[list[tuple[T1, T2]]] = [] 279 current_batch = [] 280 current_prefix_batch = [] 281 current_params_batch = [] 282 current_size = 0 283 284 chunk_list: list[Chunk] = [] 285 286 # Iterate through each request in the batch 287 for request, prefix, params in zip(batch_request, prefix_list, fun_params): 288 request_size = estimate_size(request) 289 290 # Check if adding this request exceeds the max size 291 if current_size + request_size > MAX_REQUEST_SIZE: 292 # If so, start a new batch 293 294 # Essentiatly checks that it's not the first iteration 295 if current_batch: 296 chunk = Chunk( 297 current_batch, current_prefix_batch, current_params_batch 298 ) 299 chunk_list.append(chunk) 300 result.append(current_batch) 301 302 current_batch = [request] 303 current_prefix_batch = [prefix] 304 current_params_batch = [params] 305 current_size = request_size 306 else: 307 # Otherwise, add to the current batch 308 current_batch.append(request) 309 current_size += request_size 310 current_prefix_batch.append(prefix) 311 current_params_batch.append(params) 312 313 # Add the last batch if it's not empty 314 if current_batch: 315 result.append(current_batch) 316 chunk = Chunk(current_batch, current_prefix_batch, current_params_batch) 317 chunk_list.append(chunk) 318 319 return result, chunk_list 320 321 def _are_changes_equal(self, change_a: Any, change_b: Any): 322 for (a, b), (c, d) in zip(change_a, change_b): 323 if a != c or b != d: 324 return False 325 326 def _rpc_request_batch( 327 self, batch_requests: list[tuple[str, list[Any]]], extract_result: bool = True 328 ) -> list[str]: 329 """ 330 Sends batch requests to the substrate node using multiple threads and collects the results. 331 332 Args: 333 substrate: An instance of the substrate interface. 334 batch_requests : A list of requests to be sent in batches. 335 max_size: Maximum size of each batch in bytes. 336 extract_result: Whether to extract the result from the response message. 337 338 Returns: 339 A list of results from the batch requests. 340 341 Example: 342 >>> _rpc_request_batch(substrate_instance, [('method1', ['param1']), ('method2', ['param2'])]) 343 ['result1', 'result2', ...] 344 """ 345 346 chunk_results: list[Any] = [] 347 # smaller_requests = self._make_request_smaller(batch_requests) 348 request_id = 0 349 with ThreadPoolExecutor() as executor: 350 futures: list[Future[list[str | dict[Any, Any]]]] = [] 351 for chunk in [batch_requests]: 352 request_ids: list[int] = [] 353 batch_payload: list[Any] = [] 354 for method, params in chunk: 355 request_id += 1 356 request_ids.append(request_id) 357 batch_payload.append( 358 { 359 "jsonrpc": "2.0", 360 "method": method, 361 "params": params, 362 "id": request_id, 363 } 364 ) 365 366 futures.append( 367 executor.submit( 368 self._send_batch, 369 batch_payload=batch_payload, 370 request_ids=request_ids, 371 extract_result=extract_result, 372 ) 373 ) 374 for future in futures: 375 resul = future.result() 376 chunk_results.append(resul) 377 return chunk_results 378 379 def _rpc_request_batch_chunked( 380 self, chunk_requests: list[Chunk], extract_result: bool = True 381 ): 382 """ 383 Sends batch requests to the substrate node using multiple threads and collects the results. 384 385 Args: 386 substrate: An instance of the substrate interface. 387 batch_requests : A list of requests to be sent in batches. 388 max_size: Maximum size of each batch in bytes. 389 extract_result: Whether to extract the result from the response message. 390 391 Returns: 392 A list of results from the batch requests. 393 394 Example: 395 >>> _rpc_request_batch(substrate_instance, [('method1', ['param1']), ('method2', ['param2'])]) 396 ['result1', 'result2', ...] 397 """ 398 399 def split_chunks(chunk: Chunk, chunk_info: list[Chunk], chunk_info_idx: int): 400 manhattam_chunks: list[tuple[Any, Any]] = [] 401 mutaded_chunk_info = deepcopy(chunk_info) 402 max_n_keys = 35000 403 for query in chunk.batch_requests: 404 result_keys = query[1][0] 405 keys_amount = len(result_keys) 406 if keys_amount > max_n_keys: 407 mutaded_chunk_info.pop(chunk_info_idx) 408 for i in range(0, keys_amount, max_n_keys): 409 new_chunk = deepcopy(chunk) 410 splitted_keys = result_keys[i: i + max_n_keys] 411 splitted_query = deepcopy(query) 412 splitted_query[1][0] = splitted_keys 413 new_chunk.batch_requests = [splitted_query] 414 manhattam_chunks.append(splitted_query) 415 mutaded_chunk_info.insert(chunk_info_idx, new_chunk) 416 else: 417 manhattam_chunks.append(query) 418 return manhattam_chunks, mutaded_chunk_info 419 420 assert len(chunk_requests) > 0 421 mutated_chunk_info: list[Chunk] = [] 422 chunk_results: list[Any] = [] 423 # smaller_requests = self._make_request_smaller(batch_requests) 424 request_id = 0 425 426 with ThreadPoolExecutor() as executor: 427 futures: list[Future[list[str | dict[Any, Any]]]] = [] 428 for idx, macro_chunk in enumerate(chunk_requests): 429 _, mutated_chunk_info = split_chunks(macro_chunk, chunk_requests, idx) 430 for chunk in mutated_chunk_info: 431 request_ids: list[int] = [] 432 batch_payload: list[Any] = [] 433 for method, params in chunk.batch_requests: 434 # for method, params in micro_chunk: 435 request_id += 1 436 request_ids.append(request_id) 437 batch_payload.append( 438 { 439 "jsonrpc": "2.0", 440 "method": method, 441 "params": params, 442 "id": request_id, 443 } 444 ) 445 futures.append( 446 executor.submit( 447 self._send_batch, 448 batch_payload=batch_payload, 449 request_ids=request_ids, 450 extract_result=extract_result, 451 ) 452 ) 453 for future in futures: 454 resul = future.result() 455 chunk_results.append(resul) 456 return chunk_results, mutated_chunk_info 457 458 def _decode_response( 459 self, 460 response: list[str], 461 function_parameters: list[tuple[Any, Any, Any, Any, str]], 462 prefix_list: list[Any], 463 block_hash: str, 464 ) -> dict[str, dict[Any, Any]]: 465 """ 466 Decodes a response from the substrate interface and organizes the data into a dictionary. 467 468 Args: 469 response: A list of encoded responses from a substrate query. 470 function_parameters: A list of tuples containing the parameters for each storage function. 471 last_keys: A list of the last keys used in the substrate query. 472 prefix_list: A list of prefixes used in the substrate query. 473 substrate: An instance of the SubstrateInterface class. 474 block_hash: The hash of the block to be queried. 475 476 Returns: 477 A dictionary where each key is a storage function name and the value is another dictionary. 478 This inner dictionary's key is the decoded key from the response and the value is the corresponding decoded value. 479 480 Raises: 481 ValueError: If an unsupported hash type is encountered in the `concat_hash_len` function. 482 483 Example: 484 >>> _decode_response( 485 response=[...], 486 function_parameters=[...], 487 last_keys=[...], 488 prefix_list=[...], 489 substrate=substrate_instance, 490 block_hash="0x123..." 491 ) 492 {'storage_function_name': {decoded_key: decoded_value, ...}, ...} 493 """ 494 495 def get_item_key_value(item_key: tuple[Any, ...] | Any) -> tuple[Any, ...] | Any: 496 if isinstance(item_key, tuple): 497 return tuple(k.value for k in item_key) 498 return item_key.value 499 500 def concat_hash_len(key_hasher: str) -> int: 501 """ 502 Determines the length of the hash based on the given key hasher type. 503 504 Args: 505 key_hasher: The type of key hasher. 506 507 Returns: 508 The length of the hash corresponding to the given key hasher type. 509 510 Raises: 511 ValueError: If the key hasher type is not supported. 512 513 Example: 514 >>> concat_hash_len("Blake2_128Concat") 515 16 516 """ 517 518 if key_hasher == "Blake2_128Concat": 519 return 16 520 elif key_hasher == "Twox64Concat": 521 return 8 522 elif key_hasher == "Identity": 523 return 0 524 else: 525 raise ValueError("Unsupported hash type") 526 527 assert len(response) == len(function_parameters) == len(prefix_list) 528 result_dict: dict[str, dict[Any, Any]] = {} 529 for res, fun_params_tuple, prefix in zip( 530 response, function_parameters, prefix_list 531 ): 532 if not res: 533 continue 534 res = res[0] 535 changes = res["changes"] # type: ignore 536 value_type, param_types, key_hashers, params, storage_function = ( 537 fun_params_tuple 538 ) 539 with self.get_conn(init=True) as substrate: 540 for item in changes: 541 # Determine type string 542 key_type_string: list[Any] = [] 543 for n in range(len(params), len(param_types)): 544 key_type_string.append( 545 f"[u8; {concat_hash_len(key_hashers[n])}]" 546 ) 547 key_type_string.append(param_types[n]) 548 549 item_key_obj = substrate.decode_scale( # type: ignore 550 type_string=f"({', '.join(key_type_string)})", 551 scale_bytes="0x" + item[0][len(prefix):], 552 return_scale_obj=True, 553 block_hash=block_hash, 554 ) 555 # strip key_hashers to use as item key 556 if len(param_types) - len(params) == 1: 557 item_key = item_key_obj.value_object[1] # type: ignore 558 else: 559 item_key = tuple( # type: ignore 560 item_key_obj.value_object[key + 1] # type: ignore 561 for key in range( # type: ignore 562 len(params), len(param_types) + 1, 2 563 ) 564 ) 565 566 item_value = substrate.decode_scale( # type: ignore 567 type_string=value_type, 568 scale_bytes=item[1], 569 return_scale_obj=True, 570 block_hash=block_hash, 571 ) 572 result_dict.setdefault(storage_function, {}) 573 key = get_item_key_value(item_key) # type: ignore 574 result_dict[storage_function][key] = item_value.value # type: ignore 575 576 return result_dict 577 578 def query_batch( 579 self, functions: dict[str, list[tuple[str, list[Any]]]] 580 ) -> dict[str, str]: 581 """ 582 Executes batch queries on a substrate and returns results in a dictionary format. 583 584 Args: 585 substrate: An instance of SubstrateInterface to interact with the substrate. 586 functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls (function name and parameters). 587 588 Returns: 589 A dictionary where keys are storage function names and values are the query results. 590 591 Raises: 592 Exception: If no result is found from the batch queries. 593 594 Example: 595 >>> query_batch(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) 596 {'function_name': 'query_result', ...} 597 """ 598 599 result: dict[str, str] = {} 600 if not functions: 601 raise Exception("No result") 602 with self.get_conn(init=True) as substrate: 603 for module, queries in functions.items(): 604 storage_keys: list[Any] = [] 605 for fn, params in queries: 606 storage_function = substrate.create_storage_key( # type: ignore 607 pallet=module, storage_function=fn, params=params 608 ) 609 storage_keys.append(storage_function) 610 611 block_hash = substrate.get_block_hash() 612 responses: list[Any] = substrate.query_multi( # type: ignore 613 storage_keys=storage_keys, block_hash=block_hash 614 ) 615 616 for item in responses: 617 fun = item[0] 618 query = item[1] 619 storage_fun = fun.storage_function 620 result[storage_fun] = query.value 621 622 return result 623 624 def query_batch_map( 625 self, 626 functions: dict[str, list[tuple[str, list[Any]]]], 627 block_hash: str | None = None, 628 ) -> dict[str, dict[Any, Any]]: 629 """ 630 Queries multiple storage functions using a map batch approach and returns the combined result. 631 632 Args: 633 substrate: An instance of SubstrateInterface for substrate interaction. 634 functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls. 635 636 Returns: 637 The combined result of the map batch query. 638 639 Example: 640 >>> query_batch_map(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) 641 # Returns the combined result of the map batch query 642 """ 643 multi_result: dict[str, dict[Any, Any]] = {} 644 645 def recursive_update( 646 d: dict[str, dict[T1, T2] | dict[str, Any]], 647 u: Mapping[str, dict[Any, Any] | str], 648 ) -> dict[str, dict[T1, T2]]: 649 for k, v in u.items(): 650 if isinstance(v, dict): 651 d[k] = recursive_update(d.get(k, {}), v) # type: ignore 652 else: 653 d[k] = v # type: ignore 654 return d # type: ignore 655 656 def get_page(): 657 send, prefix_list = self._get_storage_keys(storage, queries, block_hash) 658 with self.get_conn(init=True) as substrate: 659 function_parameters = self._get_lists(storage, queries, substrate) 660 responses = self._rpc_request_batch(send) 661 # assumption because send is just the storage_function keys 662 # so it should always be really small regardless of the amount of queries 663 assert len(responses) == 1 664 res = responses[0] 665 built_payload: list[tuple[str, list[Any]]] = [] 666 for result_keys in res: 667 built_payload.append( 668 ("state_queryStorageAt", [result_keys, block_hash]) 669 ) 670 _, chunks_info = self._make_request_smaller( 671 built_payload, prefix_list, function_parameters 672 ) 673 chunks_response, chunks_info = self._rpc_request_batch_chunked(chunks_info) 674 return chunks_response, chunks_info 675 676 if not block_hash: 677 with self.get_conn(init=True) as substrate: 678 block_hash = substrate.get_block_hash() 679 for storage, queries in functions.items(): 680 chunks, chunks_info = get_page() 681 # if this doesn't happen something is wrong on the code 682 # and we won't be able to decode the data properly 683 assert len(chunks) == len(chunks_info) 684 for chunk_info, response in zip(chunks_info, chunks): 685 storage_result = self._decode_response( 686 response, chunk_info.fun_params, chunk_info.prefix_list, block_hash 687 ) 688 multi_result = recursive_update(multi_result, storage_result) 689 690 return multi_result 691 692 def query( 693 self, 694 name: str, 695 params: list[Any] = [], 696 module: str = "SubspaceModule", 697 block_hash: str | None = None, 698 ) -> Any: 699 """ 700 Queries a storage function on the network. 701 702 Sends a query to the network and retrieves data from a 703 specified storage function. 704 705 Args: 706 name: The name of the storage function to query. 707 params: The parameters to pass to the storage function. 708 module: The module where the storage function is located. 709 710 Returns: 711 The result of the query from the network. 712 713 Raises: 714 NetworkQueryError: If the query fails or is invalid. 715 """ 716 717 result = self.query_batch({module: [(name, params)]}) 718 719 return result[name] 720 721 def query_map( 722 self, 723 name: str, 724 params: list[Any] = [], 725 module: str = "SubspaceModule", 726 extract_value: bool = True, 727 block_hash: str | None = None, 728 ) -> dict[Any, Any]: 729 """ 730 Queries a storage map from a network node. 731 732 Args: 733 name: The name of the storage map to query. 734 params: A list of parameters for the query. 735 module: The module in which the storage map is located. 736 737 Returns: 738 A dictionary representing the key-value pairs 739 retrieved from the storage map. 740 741 Raises: 742 QueryError: If the query to the network fails or is invalid. 743 """ 744 745 result = self.query_batch_map({module: [(name, params)]}, block_hash) 746 747 if extract_value: 748 return {k.value: v.value for k, v in result} # type: ignore 749 750 return result 751 752 def compose_call( 753 self, 754 fn: str, 755 params: dict[str, Any], 756 key: Keypair | None, 757 module: str = "SubspaceModule", 758 wait_for_inclusion: bool = True, 759 wait_for_finalization: bool | None = None, 760 sudo: bool = False, 761 unsigned: bool = False, 762 ) -> ExtrinsicReceipt: 763 """ 764 Composes and submits a call to the network node. 765 766 Composes and signs a call with the provided keypair, and submits it to 767 the network. The call can be a standard extrinsic or a sudo extrinsic if 768 elevated permissions are required. The method can optionally wait for 769 the call's inclusion in a block and/or its finalization. 770 771 Args: 772 fn: The function name to call on the network. 773 params: A dictionary of parameters for the call. 774 key: The keypair for signing the extrinsic. 775 module: The module containing the function. 776 wait_for_inclusion: Wait for the call's inclusion in a block. 777 wait_for_finalization: Wait for the transaction's finalization. 778 sudo: Execute the call as a sudo (superuser) operation. 779 780 Returns: 781 The receipt of the submitted extrinsic, if 782 `wait_for_inclusion` is True. Otherwise, returns a string 783 identifier of the extrinsic. 784 785 Raises: 786 ChainTransactionError: If the transaction fails. 787 """ 788 789 if key is None and not unsigned: 790 raise ValueError("Key must be provided for signed extrinsics.") 791 792 with self.get_conn() as substrate: 793 if wait_for_finalization is None: 794 wait_for_finalization = self.wait_for_finalization 795 796 call = substrate.compose_call( # type: ignore 797 call_module=module, call_function=fn, call_params=params 798 ) 799 if sudo: 800 call = substrate.compose_call( # type: ignore 801 call_module="Sudo", 802 call_function="sudo", 803 call_params={ 804 "call": call.value, # type: ignore 805 }, 806 ) 807 808 if not unsigned: 809 extrinsic = substrate.create_signed_extrinsic( # type: ignore 810 call=call, keypair=key # type: ignore 811 ) # type: ignore 812 else: 813 extrinsic = substrate.create_unsigned_extrinsic(call=call) # type: ignore 814 815 response = substrate.submit_extrinsic( 816 extrinsic=extrinsic, 817 wait_for_inclusion=wait_for_inclusion, 818 wait_for_finalization=wait_for_finalization, 819 ) 820 if wait_for_inclusion: 821 if not response.is_success: 822 raise ChainTransactionError( 823 response.error_message, response # type: ignore 824 ) 825 826 return response 827 828 def compose_call_multisig( 829 self, 830 fn: str, 831 params: dict[str, Any], 832 key: Keypair, 833 signatories: list[Ss58Address], 834 threshold: int, 835 module: str = "SubspaceModule", 836 wait_for_inclusion: bool = True, 837 wait_for_finalization: bool | None = None, 838 sudo: bool = False, 839 era: dict[str, int] | None = None, 840 ) -> ExtrinsicReceipt: 841 """ 842 Composes and submits a multisignature call to the network node. 843 844 This method allows the composition and submission of a call that 845 requires multiple signatures for execution, known as a multisignature 846 call. It supports specifying signatories, a threshold of signatures for 847 the call's execution, and an optional era for the call's mortality. The 848 call can be a standard extrinsic, a sudo extrinsic for elevated 849 permissions, or a multisig extrinsic if multiple signatures are 850 required. Optionally, the method can wait for the call's inclusion in a 851 block and/or its finalization. Make sure to pass all keys, 852 that are part of the multisignature. 853 854 Args: 855 fn: The function name to call on the network. params: A dictionary 856 of parameters for the call. key: The keypair for signing the 857 extrinsic. signatories: List of SS58 addresses of the signatories. 858 Include ALL KEYS that are part of the multisig. threshold: The 859 minimum number of signatories required to execute the extrinsic. 860 module: The module containing the function to call. 861 wait_for_inclusion: Whether to wait for the call's inclusion in a 862 block. wait_for_finalization: Whether to wait for the transaction's 863 finalization. sudo: Execute the call as a sudo (superuser) 864 operation. era: Specifies the call's mortality in terms of blocks in 865 the format 866 {'period': amount_blocks}. If omitted, the extrinsic is 867 immortal. 868 869 Returns: 870 The receipt of the submitted extrinsic if `wait_for_inclusion` is 871 True. Otherwise, returns a string identifier of the extrinsic. 872 873 Raises: 874 ChainTransactionError: If the transaction fails. 875 """ 876 877 # getting the call ready 878 with self.get_conn() as substrate: 879 if wait_for_finalization is None: 880 wait_for_finalization = self.wait_for_finalization 881 882 # prepares the `GenericCall` object 883 call = substrate.compose_call( # type: ignore 884 call_module=module, call_function=fn, call_params=params 885 ) 886 if sudo: 887 call = substrate.compose_call( # type: ignore 888 call_module="Sudo", 889 call_function="sudo", 890 call_params={ 891 "call": call.value, # type: ignore 892 }, 893 ) 894 895 # modify the rpc methods at runtime, to allow for correct payment 896 # fee calculation parity has a bug in this version, 897 # where the method has to be removed 898 rpc_methods = substrate.config.get("rpc_methods") # type: ignore 899 900 if "state_call" in rpc_methods: # type: ignore 901 rpc_methods.remove("state_call") # type: ignore 902 903 # create the multisig account 904 multisig_acc = substrate.generate_multisig_account( # type: ignore 905 signatories, threshold 906 ) 907 908 # send the multisig extrinsic 909 extrinsic = substrate.create_multisig_extrinsic( # type: ignore 910 call=call, # type: ignore 911 keypair=key, 912 multisig_account=multisig_acc, # type: ignore 913 era=era, # type: ignore 914 ) # type: ignore 915 916 response = substrate.submit_extrinsic( 917 extrinsic=extrinsic, 918 wait_for_inclusion=wait_for_inclusion, 919 wait_for_finalization=wait_for_finalization, 920 ) 921 922 if wait_for_inclusion: 923 if not response.is_success: 924 raise ChainTransactionError( 925 response.error_message, response # type: ignore 926 ) 927 928 return response 929 930 def transfer( 931 self, 932 key: Keypair, 933 amount: int, 934 dest: Ss58Address, 935 ) -> ExtrinsicReceipt: 936 """ 937 Transfers a specified amount of tokens from the signer's account to the 938 specified account. 939 940 Args: 941 key: The keypair associated with the sender's account. 942 amount: The amount to transfer, in nanotokens. 943 dest: The SS58 address of the recipient. 944 945 Returns: 946 A receipt of the transaction. 947 948 Raises: 949 InsufficientBalanceError: If the sender's account does not have 950 enough balance. 951 ChainTransactionError: If the transaction fails. 952 """ 953 954 params = {"dest": dest, "value": amount} 955 956 return self.compose_call( 957 module="Balances", fn="transfer_keep_alive", params=params, key=key 958 ) 959 960 def transfer_multiple( 961 self, 962 key: Keypair, 963 destinations: list[Ss58Address], 964 amounts: list[int], 965 netuid: str | int = 0, 966 ) -> ExtrinsicReceipt: 967 """ 968 Transfers specified amounts of tokens from the signer's account to 969 multiple target accounts. 970 971 The `destinations` and `amounts` lists must be of the same length. 972 973 Args: 974 key: The keypair associated with the sender's account. 975 destinations: A list of SS58 addresses of the recipients. 976 amounts: Amount to transfer to each recipient, in nanotokens. 977 netuid: The network identifier. 978 979 Returns: 980 A receipt of the transaction. 981 982 Raises: 983 InsufficientBalanceError: If the sender's account does not have 984 enough balance for all transfers. 985 ChainTransactionError: If the transaction fails. 986 """ 987 988 assert len(destinations) == len(amounts) 989 990 # extract existential deposit from amounts 991 existential_deposit = self.get_existential_deposit() 992 amounts = [a - existential_deposit for a in amounts] 993 994 params = { 995 "netuid": netuid, 996 "destinations": destinations, 997 "amounts": amounts, 998 } 999 1000 return self.compose_call( 1001 module="SubspaceModule", fn="transfer_multiple", params=params, key=key 1002 ) 1003 1004 def stake( 1005 self, 1006 key: Keypair, 1007 amount: int, 1008 dest: Ss58Address, 1009 ) -> ExtrinsicReceipt: 1010 """ 1011 Stakes the specified amount of tokens to a module key address. 1012 1013 Args: 1014 key: The keypair associated with the staker's account. 1015 amount: The amount of tokens to stake, in nanotokens. 1016 dest: The SS58 address of the module key to stake to. 1017 netuid: The network identifier. 1018 1019 Returns: 1020 A receipt of the staking transaction. 1021 1022 Raises: 1023 InsufficientBalanceError: If the staker's account does not have 1024 enough balance. 1025 ChainTransactionError: If the transaction fails. 1026 """ 1027 1028 params = {"amount": amount, "module_key": dest} 1029 1030 return self.compose_call(fn="add_stake", params=params, key=key) 1031 1032 def unstake( 1033 self, 1034 key: Keypair, 1035 amount: int, 1036 dest: Ss58Address, 1037 ) -> ExtrinsicReceipt: 1038 """ 1039 Unstakes the specified amount of tokens from a module key address. 1040 1041 Args: 1042 key: The keypair associated with the unstaker's account. 1043 amount: The amount of tokens to unstake, in nanotokens. 1044 dest: The SS58 address of the module key to unstake from. 1045 netuid: The network identifier. 1046 1047 Returns: 1048 A receipt of the unstaking transaction. 1049 1050 Raises: 1051 InsufficientStakeError: If the staked key does not have enough 1052 staked tokens by the signer key. 1053 ChainTransactionError: If the transaction fails. 1054 """ 1055 1056 params = {"amount": amount, "module_key": dest} 1057 return self.compose_call(fn="remove_stake", params=params, key=key) 1058 1059 def update_module( 1060 self, 1061 key: Keypair, 1062 name: str, 1063 address: str, 1064 metadata: str | None = None, 1065 delegation_fee: int = 20, 1066 netuid: int = 0, 1067 ) -> ExtrinsicReceipt: 1068 """ 1069 Updates the parameters of a registered module. 1070 1071 The delegation fee must be an integer between 0 and 100. 1072 1073 Args: 1074 key: The keypair associated with the module's account. 1075 name: The new name for the module. If None, the name is not updated. 1076 address: The new address for the module. 1077 If None, the address is not updated. 1078 delegation_fee: The new delegation fee for the module, 1079 between 0 and 100. 1080 netuid: The network identifier. 1081 1082 Returns: 1083 A receipt of the module update transaction. 1084 1085 Raises: 1086 InvalidParameterError: If the provided parameters are invalid. 1087 ChainTransactionError: If the transaction fails. 1088 """ 1089 1090 assert isinstance(delegation_fee, int) 1091 1092 params = { 1093 "netuid": netuid, 1094 "name": name, 1095 "address": address, 1096 "delegation_fee": delegation_fee, 1097 "metadata": metadata, 1098 } 1099 1100 response = self.compose_call("update_module", params=params, key=key) 1101 1102 return response 1103 1104 def register_module( 1105 self, 1106 key: Keypair, 1107 name: str, 1108 address: str | None = None, 1109 subnet: str = "Rootnet", 1110 metadata: str | None = None, 1111 ) -> ExtrinsicReceipt: 1112 """ 1113 Registers a new module in the network. 1114 1115 Args: 1116 key: The keypair used for registering the module. 1117 name: The name of the module. If None, a default or previously 1118 set name is used. # How does this work? 1119 address: The address of the module. If None, a default or 1120 previously set address is used. # How does this work? 1121 subnet: The network subnet to register the module in. 1122 min_stake: The minimum stake required for the module, in nanotokens. 1123 If None, a default value is used. 1124 1125 Returns: 1126 A receipt of the registration transaction. 1127 1128 Raises: 1129 InvalidParameterError: If the provided parameters are invalid. 1130 ChainTransactionError: If the transaction fails. 1131 """ 1132 1133 key_addr = key.ss58_address 1134 1135 params = { 1136 "network_name": subnet, 1137 "address": address, 1138 "name": name, 1139 "module_key": key_addr, 1140 "metadata": metadata, 1141 } 1142 1143 response = self.compose_call("register", params=params, key=key) 1144 return response 1145 1146 def deregister_module(self, key: Keypair, netuid: int) -> ExtrinsicReceipt: 1147 """ 1148 Deregisters a module from the network. 1149 1150 Args: 1151 key: The keypair associated with the module's account. 1152 netuid: The network identifier. 1153 1154 Returns: 1155 A receipt of the module deregistration transaction. 1156 1157 Raises: 1158 ChainTransactionError: If the transaction fails. 1159 """ 1160 1161 params = {"netuid": netuid} 1162 1163 response = self.compose_call("deregister", params=params, key=key) 1164 1165 return response 1166 1167 def register_subnet(self, key: Keypair, name: str, metadata: str | None = None) -> ExtrinsicReceipt: 1168 """ 1169 Registers a new subnet in the network. 1170 1171 Args: 1172 key (Keypair): The keypair used for registering the subnet. 1173 name (str): The name of the subnet to be registered. 1174 metadata (str | None, optional): Additional metadata for the subnet. Defaults to None. 1175 1176 Returns: 1177 ExtrinsicReceipt: A receipt of the subnet registration transaction. 1178 1179 Raises: 1180 ChainTransactionError: If the transaction fails. 1181 """ 1182 1183 params = { 1184 "name": name, 1185 "metadata": metadata, 1186 } 1187 1188 response = self.compose_call("register_subnet", params=params, key=key) 1189 1190 return response 1191 1192 def vote( 1193 self, 1194 key: Keypair, 1195 uids: list[int], 1196 weights: list[int], 1197 netuid: int = 0, 1198 ) -> ExtrinsicReceipt: 1199 """ 1200 Casts votes on a list of module UIDs with corresponding weights. 1201 1202 The length of the UIDs list and the weights list should be the same. 1203 Each weight corresponds to the UID at the same index. 1204 1205 Args: 1206 key: The keypair used for signing the vote transaction. 1207 uids: A list of module UIDs to vote on. 1208 weights: A list of weights corresponding to each UID. 1209 netuid: The network identifier. 1210 1211 Returns: 1212 A receipt of the voting transaction. 1213 1214 Raises: 1215 InvalidParameterError: If the lengths of UIDs and weights lists 1216 do not match. 1217 ChainTransactionError: If the transaction fails. 1218 """ 1219 1220 assert len(uids) == len(weights) 1221 1222 params = { 1223 "uids": uids, 1224 "weights": weights, 1225 "netuid": netuid, 1226 } 1227 1228 response = self.compose_call("set_weights", params=params, key=key) 1229 1230 return response 1231 1232 def update_subnet( 1233 self, 1234 key: Keypair, 1235 params: SubnetParams, 1236 netuid: int = 0, 1237 ) -> ExtrinsicReceipt: 1238 """ 1239 Update a subnet's configuration. 1240 1241 It requires the founder key for authorization. 1242 1243 Args: 1244 key: The founder keypair of the subnet. 1245 params: The new parameters for the subnet. 1246 netuid: The network identifier. 1247 1248 Returns: 1249 A receipt of the subnet update transaction. 1250 1251 Raises: 1252 AuthorizationError: If the key is not authorized. 1253 ChainTransactionError: If the transaction fails. 1254 """ 1255 1256 general_params = dict(params) 1257 general_params["netuid"] = netuid 1258 if "metadata" not in general_params: 1259 general_params["metadata"] = None 1260 1261 response = self.compose_call( 1262 fn="update_subnet", 1263 params=general_params, 1264 key=key, 1265 ) 1266 1267 return response 1268 1269 def transfer_stake( 1270 self, 1271 key: Keypair, 1272 amount: int, 1273 from_module_key: Ss58Address, 1274 dest_module_address: Ss58Address, 1275 ) -> ExtrinsicReceipt: 1276 """ 1277 Realocate staked tokens from one staked module to another module. 1278 1279 Args: 1280 key: The keypair associated with the account that is delegating the tokens. 1281 amount: The amount of staked tokens to transfer, in nanotokens. 1282 from_module_key: The SS58 address of the module you want to transfer from (currently delegated by the key). 1283 dest_module_address: The SS58 address of the destination (newly delegated key). 1284 netuid: The network identifier. 1285 1286 Returns: 1287 A receipt of the stake transfer transaction. 1288 1289 Raises: 1290 InsufficientStakeError: If the source module key does not have 1291 enough staked tokens. ChainTransactionError: If the transaction 1292 fails. 1293 """ 1294 1295 amount = amount - self.get_existential_deposit() 1296 1297 params = { 1298 "amount": amount, 1299 "module_key": from_module_key, 1300 "new_module_key": dest_module_address, 1301 } 1302 1303 response = self.compose_call("transfer_stake", key=key, params=params) 1304 1305 return response 1306 1307 def multiunstake( 1308 self, 1309 key: Keypair, 1310 keys: list[Ss58Address], 1311 amounts: list[int], 1312 ) -> ExtrinsicReceipt: 1313 """ 1314 Unstakes tokens from multiple module keys. 1315 1316 And the lists `keys` and `amounts` must be of the same length. Each 1317 amount corresponds to the module key at the same index. 1318 1319 Args: 1320 key: The keypair associated with the unstaker's account. 1321 keys: A list of SS58 addresses of the module keys to unstake from. 1322 amounts: A list of amounts to unstake from each module key, 1323 in nanotokens. 1324 netuid: The network identifier. 1325 1326 Returns: 1327 A receipt of the multi-unstaking transaction. 1328 1329 Raises: 1330 MismatchedLengthError: If the lengths of keys and amounts lists do 1331 not match. InsufficientStakeError: If any of the module keys do not 1332 have enough staked tokens. ChainTransactionError: If the transaction 1333 fails. 1334 """ 1335 1336 assert len(keys) == len(amounts) 1337 1338 params = {"module_keys": keys, "amounts": amounts} 1339 1340 response = self.compose_call("remove_stake_multiple", params=params, key=key) 1341 1342 return response 1343 1344 def multistake( 1345 self, 1346 key: Keypair, 1347 keys: list[Ss58Address], 1348 amounts: list[int], 1349 ) -> ExtrinsicReceipt: 1350 """ 1351 Stakes tokens to multiple module keys. 1352 1353 The lengths of the `keys` and `amounts` lists must be the same. Each 1354 amount corresponds to the module key at the same index. 1355 1356 Args: 1357 key: The keypair associated with the staker's account. 1358 keys: A list of SS58 addresses of the module keys to stake to. 1359 amounts: A list of amounts to stake to each module key, 1360 in nanotokens. 1361 netuid: The network identifier. 1362 1363 Returns: 1364 A receipt of the multi-staking transaction. 1365 1366 Raises: 1367 MismatchedLengthError: If the lengths of keys and amounts lists 1368 do not match. 1369 ChainTransactionError: If the transaction fails. 1370 """ 1371 1372 assert len(keys) == len(amounts) 1373 1374 params = { 1375 "module_keys": keys, 1376 "amounts": amounts, 1377 } 1378 1379 response = self.compose_call("add_stake_multiple", params=params, key=key) 1380 1381 return response 1382 1383 def add_profit_shares( 1384 self, 1385 key: Keypair, 1386 keys: list[Ss58Address], 1387 shares: list[int], 1388 ) -> ExtrinsicReceipt: 1389 """ 1390 Allocates profit shares to multiple keys. 1391 1392 The lists `keys` and `shares` must be of the same length, 1393 with each share amount corresponding to the key at the same index. 1394 1395 Args: 1396 key: The keypair associated with the account 1397 distributing the shares. 1398 keys: A list of SS58 addresses to allocate shares to. 1399 shares: A list of share amounts to allocate to each key, 1400 in nanotokens. 1401 1402 Returns: 1403 A receipt of the profit sharing transaction. 1404 1405 Raises: 1406 MismatchedLengthError: If the lengths of keys and shares 1407 lists do not match. 1408 ChainTransactionError: If the transaction fails. 1409 """ 1410 1411 assert len(keys) == len(shares) 1412 1413 params = {"keys": keys, "shares": shares} 1414 1415 response = self.compose_call("add_profit_shares", params=params, key=key) 1416 1417 return response 1418 1419 def add_subnet_proposal( 1420 self, key: Keypair, 1421 params: dict[str, Any], 1422 ipfs: str, 1423 netuid: int = 0 1424 ) -> ExtrinsicReceipt: 1425 """ 1426 Submits a proposal for creating or modifying a subnet within the 1427 network. 1428 1429 The proposal includes various parameters like the name, founder, share 1430 allocations, and other subnet-specific settings. 1431 1432 Args: 1433 key: The keypair used for signing the proposal transaction. 1434 params: The parameters for the subnet proposal. 1435 netuid: The network identifier. 1436 1437 Returns: 1438 A receipt of the subnet proposal transaction. 1439 1440 Raises: 1441 InvalidParameterError: If the provided subnet 1442 parameters are invalid. 1443 ChainTransactionError: If the transaction fails. 1444 """ 1445 1446 general_params = dict(params) 1447 general_params["netuid"] = netuid 1448 general_params["data"] = ipfs 1449 if "metadata" not in general_params: 1450 general_params["metadata"] = None 1451 1452 # general_params["burn_config"] = json.dumps(general_params["burn_config"]) 1453 response = self.compose_call( 1454 fn="add_subnet_params_proposal", 1455 params=general_params, 1456 key=key, 1457 module="GovernanceModule", 1458 ) 1459 1460 return response 1461 1462 def add_custom_proposal( 1463 self, 1464 key: Keypair, 1465 cid: str, 1466 ) -> ExtrinsicReceipt: 1467 1468 params = {"data": cid} 1469 1470 response = self.compose_call( 1471 fn="add_global_custom_proposal", 1472 params=params, 1473 key=key, 1474 module="GovernanceModule", 1475 ) 1476 return response 1477 1478 def add_custom_subnet_proposal( 1479 self, 1480 key: Keypair, 1481 cid: str, 1482 netuid: int = 0, 1483 ) -> ExtrinsicReceipt: 1484 """ 1485 Submits a proposal for creating or modifying a custom subnet within the 1486 network. 1487 1488 The proposal includes various parameters like the name, founder, share 1489 allocations, and other subnet-specific settings. 1490 1491 Args: 1492 key: The keypair used for signing the proposal transaction. 1493 params: The parameters for the subnet proposal. 1494 netuid: The network identifier. 1495 1496 Returns: 1497 A receipt of the subnet proposal transaction. 1498 """ 1499 1500 params = { 1501 "data": cid, 1502 "netuid": netuid, 1503 } 1504 1505 response = self.compose_call( 1506 fn="add_subnet_custom_proposal", 1507 params=params, 1508 key=key, 1509 module="GovernanceModule", 1510 ) 1511 1512 return response 1513 1514 def add_global_proposal( 1515 self, 1516 key: Keypair, 1517 params: NetworkParams, 1518 cid: str | None, 1519 ) -> ExtrinsicReceipt: 1520 """ 1521 Submits a proposal for altering the global network parameters. 1522 1523 Allows for the submission of a proposal to 1524 change various global parameters 1525 of the network, such as emission rates, rate limits, and voting 1526 thresholds. It is used to 1527 suggest changes that affect the entire network's operation. 1528 1529 Args: 1530 key: The keypair used for signing the proposal transaction. 1531 params: A dictionary containing global network parameters 1532 like maximum allowed subnets, modules, 1533 transaction rate limits, and others. 1534 1535 Returns: 1536 A receipt of the global proposal transaction. 1537 1538 Raises: 1539 InvalidParameterError: If the provided network 1540 parameters are invalid. 1541 ChainTransactionError: If the transaction fails. 1542 """ 1543 general_params = cast(dict[str, Any], params) 1544 cid = cid or "" 1545 general_params["data"] = cid 1546 1547 response = self.compose_call( 1548 fn="add_global_params_proposal", 1549 params=general_params, 1550 key=key, 1551 module="GovernanceModule", 1552 ) 1553 1554 return response 1555 1556 def vote_on_proposal( 1557 self, 1558 key: Keypair, 1559 proposal_id: int, 1560 agree: bool, 1561 ) -> ExtrinsicReceipt: 1562 """ 1563 Casts a vote on a specified proposal within the network. 1564 1565 Args: 1566 key: The keypair used for signing the vote transaction. 1567 proposal_id: The unique identifier of the proposal to vote on. 1568 1569 Returns: 1570 A receipt of the voting transaction in nanotokens. 1571 1572 Raises: 1573 InvalidProposalIDError: If the provided proposal ID does not 1574 exist or is invalid. 1575 ChainTransactionError: If the transaction fails. 1576 """ 1577 1578 params = {"proposal_id": proposal_id, "agree": agree} 1579 1580 response = self.compose_call( 1581 "vote_proposal", 1582 key=key, 1583 params=params, 1584 module="GovernanceModule", 1585 ) 1586 1587 return response 1588 1589 def unvote_on_proposal( 1590 self, 1591 key: Keypair, 1592 proposal_id: int, 1593 ) -> ExtrinsicReceipt: 1594 """ 1595 Retracts a previously cast vote on a specified proposal. 1596 1597 Args: 1598 key: The keypair used for signing the unvote transaction. 1599 proposal_id: The unique identifier of the proposal to withdraw the 1600 vote from. 1601 1602 Returns: 1603 A receipt of the unvoting transaction in nanotokens. 1604 1605 Raises: 1606 InvalidProposalIDError: If the provided proposal ID does not 1607 exist or is invalid. 1608 ChainTransactionError: If the transaction fails to be processed, or 1609 if there was no prior vote to retract. 1610 """ 1611 1612 params = {"proposal_id": proposal_id} 1613 1614 response = self.compose_call( 1615 "remove_vote_proposal", 1616 key=key, 1617 params=params, 1618 module="GovernanceModule", 1619 ) 1620 1621 return response 1622 1623 def enable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1624 """ 1625 Enables vote power delegation for the signer's account. 1626 1627 Args: 1628 key: The keypair used for signing the delegation transaction. 1629 1630 Returns: 1631 A receipt of the vote power delegation transaction. 1632 1633 Raises: 1634 ChainTransactionError: If the transaction fails. 1635 """ 1636 1637 response = self.compose_call( 1638 "enable_vote_power_delegation", 1639 params={}, 1640 key=key, 1641 module="GovernanceModule", 1642 ) 1643 1644 return response 1645 1646 def disable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1647 """ 1648 Disables vote power delegation for the signer's account. 1649 1650 Args: 1651 key: The keypair used for signing the delegation transaction. 1652 1653 Returns: 1654 A receipt of the vote power delegation transaction. 1655 1656 Raises: 1657 ChainTransactionError: If the transaction fails. 1658 """ 1659 1660 response = self.compose_call( 1661 "disable_vote_power_delegation", 1662 params={}, 1663 key=key, 1664 module="GovernanceModule", 1665 ) 1666 1667 return response 1668 1669 def add_dao_application( 1670 self, key: Keypair, application_key: Ss58Address, data: str 1671 ) -> ExtrinsicReceipt: 1672 """ 1673 Submits a new application to the general subnet DAO. 1674 1675 Args: 1676 key: The keypair used for signing the application transaction. 1677 application_key: The SS58 address of the application key. 1678 data: The data associated with the application. 1679 1680 Returns: 1681 A receipt of the application transaction. 1682 1683 Raises: 1684 ChainTransactionError: If the transaction fails. 1685 """ 1686 1687 params = {"application_key": application_key, "data": data} 1688 1689 response = self.compose_call( 1690 "add_dao_application", module="GovernanceModule", key=key, 1691 params=params 1692 ) 1693 1694 return response 1695 1696 def query_map_curator_applications(self) -> dict[str, dict[str, str]]: 1697 query_result = self.query_map( 1698 "CuratorApplications", module="GovernanceModule", params=[], 1699 extract_value=False 1700 ) 1701 applications = query_result.get("CuratorApplications", {}) 1702 return applications 1703 1704 def query_map_proposals( 1705 self, extract_value: bool = False 1706 ) -> dict[int, dict[str, Any]]: 1707 """ 1708 Retrieves a mappping of proposals from the network. 1709 1710 Queries the network and returns a mapping of proposal IDs to 1711 their respective parameters. 1712 1713 Returns: 1714 A dictionary mapping proposal IDs 1715 to dictionaries of their parameters. 1716 1717 Raises: 1718 QueryError: If the query to the network fails or is invalid. 1719 """ 1720 1721 return self.query_map( 1722 "Proposals", extract_value=extract_value, module="GovernanceModule" 1723 )["Proposals"] 1724 1725 def query_map_weights( 1726 self, netuid: int = 0, extract_value: bool = False 1727 ) -> dict[int, list[tuple[int, int]]] | None: 1728 """ 1729 Retrieves a mapping of weights for keys on the network. 1730 1731 Queries the network and returns a mapping of key UIDs to 1732 their respective weights. 1733 1734 Args: 1735 netuid: The network UID from which to get the weights. 1736 1737 Returns: 1738 A dictionary mapping key UIDs to lists of their weights. 1739 1740 Raises: 1741 QueryError: If the query to the network fails or is invalid. 1742 """ 1743 1744 weights_dict = self.query_map( 1745 "Weights", 1746 [netuid], 1747 extract_value=extract_value 1748 ).get("Weights") 1749 return weights_dict 1750 1751 def query_map_key( 1752 self, 1753 netuid: int = 0, 1754 extract_value: bool = False, 1755 ) -> dict[int, Ss58Address]: 1756 """ 1757 Retrieves a map of keys from the network. 1758 1759 Fetches a mapping of key UIDs to their associated 1760 addresses on the network. 1761 The query can be targeted at a specific network UID if required. 1762 1763 Args: 1764 netuid: The network UID from which to get the keys. 1765 1766 Returns: 1767 A dictionary mapping key UIDs to their addresses. 1768 1769 Raises: 1770 QueryError: If the query to the network fails or is invalid. 1771 """ 1772 return self.query_map("Keys", [netuid], extract_value=extract_value)["Keys"] 1773 1774 def query_map_address( 1775 self, netuid: int = 0, extract_value: bool = False 1776 ) -> dict[int, str]: 1777 """ 1778 Retrieves a map of key addresses from the network. 1779 1780 Queries the network for a mapping of key UIDs to their addresses. 1781 1782 Args: 1783 netuid: The network UID from which to get the addresses. 1784 1785 Returns: 1786 A dictionary mapping key UIDs to their addresses. 1787 1788 Raises: 1789 QueryError: If the query to the network fails or is invalid. 1790 """ 1791 1792 return self.query_map("Address", [netuid], extract_value=extract_value)[ 1793 "Address" 1794 ] 1795 1796 def query_map_emission(self, extract_value: bool = False) -> dict[int, list[int]]: 1797 """ 1798 Retrieves a map of emissions for keys on the network. 1799 1800 Queries the network to get a mapping of 1801 key UIDs to their emission values. 1802 1803 Returns: 1804 A dictionary mapping key UIDs to lists of their emission values. 1805 1806 Raises: 1807 QueryError: If the query to the network fails or is invalid. 1808 """ 1809 1810 return self.query_map("Emission", extract_value=extract_value)["Emission"] 1811 1812 def query_map_pending_emission(self, extract_value: bool = False) -> int: 1813 """ 1814 Retrieves a map of pending emissions for the subnets. 1815 1816 Queries the network for a mapping of subnet UIDs to their pending emission values. 1817 1818 Returns: 1819 A dictionary mapping subnet UIDs to their pending emission values. 1820 1821 Raises: 1822 QueryError: If the query to the network fails or is invalid. 1823 """ 1824 return self.query_map("PendingEmission", extract_value=extract_value, module="SubnetEmissionModule")["PendingEmission"] 1825 1826 def query_map_subnet_emission(self, extract_value: bool = False) -> dict[int, int]: 1827 """ 1828 Retrieves a map of subnet emissions for the network. 1829 1830 Queries the network for a mapping of subnet UIDs to their emission values. 1831 1832 Returns: 1833 A dictionary mapping subnet UIDs to their emission values. 1834 1835 Raises: 1836 QueryError: If the query to the network fails or is invalid. 1837 """ 1838 1839 return self.query_map("SubnetEmission", extract_value=extract_value, module="SubnetEmissionModule")["SubnetEmission"] 1840 1841 def query_map_subnet_consensus(self, extract_value: bool = False) -> dict[int, str]: 1842 """ 1843 Retrieves a map of subnet consensus types for the network. 1844 1845 Queries the network for a mapping of subnet UIDs to their consensus types. 1846 1847 Returns: 1848 A dictionary mapping subnet UIDs to their consensus types. 1849 1850 Raises: 1851 QueryError: If the query to the network fails or is invalid. 1852 """ 1853 1854 return self.query_map("SubnetConsensusType", extract_value=extract_value, module="SubnetEmissionModule")["SubnetConsensusType"] 1855 1856 def query_map_incentive(self, extract_value: bool = False) -> dict[int, list[int]]: 1857 """ 1858 Retrieves a mapping of incentives for keys on the network. 1859 1860 Queries the network and returns a mapping of key UIDs to 1861 their respective incentive values. 1862 1863 Returns: 1864 A dictionary mapping key UIDs to lists of their incentive values. 1865 1866 Raises: 1867 QueryError: If the query to the network fails or is invalid. 1868 """ 1869 1870 return self.query_map("Incentive", extract_value=extract_value)["Incentive"] 1871 1872 def query_map_dividend(self, extract_value: bool = False) -> dict[int, list[int]]: 1873 """ 1874 Retrieves a mapping of dividends for keys on the network. 1875 1876 Queries the network for a mapping of key UIDs to 1877 their dividend values. 1878 1879 Returns: 1880 A dictionary mapping key UIDs to lists of their dividend values. 1881 1882 Raises: 1883 QueryError: If the query to the network fails or is invalid. 1884 """ 1885 1886 return self.query_map("Dividends", extract_value=extract_value)["Dividends"] 1887 1888 def query_map_regblock( 1889 self, netuid: int = 0, extract_value: bool = False 1890 ) -> dict[int, int]: 1891 """ 1892 Retrieves a mapping of registration blocks for keys on the network. 1893 1894 Queries the network for a mapping of key UIDs to 1895 the blocks where they were registered. 1896 1897 Args: 1898 netuid: The network UID from which to get the registration blocks. 1899 1900 Returns: 1901 A dictionary mapping key UIDs to their registration blocks. 1902 1903 Raises: 1904 QueryError: If the query to the network fails or is invalid. 1905 """ 1906 1907 return self.query_map( 1908 "RegistrationBlock", [netuid], extract_value=extract_value 1909 )["RegistrationBlock"] 1910 1911 def query_map_lastupdate(self, extract_value: bool = False) -> dict[int, list[int]]: 1912 """ 1913 Retrieves a mapping of the last update times for keys on the network. 1914 1915 Queries the network for a mapping of key UIDs to their last update times. 1916 1917 Returns: 1918 A dictionary mapping key UIDs to lists of their last update times. 1919 1920 Raises: 1921 QueryError: If the query to the network fails or is invalid. 1922 """ 1923 1924 return self.query_map("LastUpdate", extract_value=extract_value)["LastUpdate"] 1925 1926 def query_map_stakefrom( 1927 self, extract_value: bool = False 1928 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1929 """ 1930 Retrieves a mapping of stakes from various sources for keys on the network. 1931 1932 Queries the network to obtain a mapping of key addresses to the sources 1933 and amounts of stakes they have received. 1934 1935 Args: 1936 netuid: The network UID from which to get the stakes. 1937 1938 Returns: 1939 A dictionary mapping key addresses to lists of tuples 1940 (module_key_address, amount). 1941 1942 Raises: 1943 QueryError: If the query to the network fails or is invalid. 1944 """ 1945 1946 result = self.query_map("StakeFrom", [], extract_value=extract_value)[ 1947 "StakeFrom" 1948 ] 1949 1950 return transform_stake_dmap(result) 1951 1952 def query_map_staketo( 1953 self, extract_value: bool = False 1954 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1955 """ 1956 Retrieves a mapping of stakes to destinations for keys on the network. 1957 1958 Queries the network for a mapping of key addresses to the destinations 1959 and amounts of stakes they have made. 1960 1961 Args: 1962 netuid: The network UID from which to get the stakes. 1963 1964 Returns: 1965 A dictionary mapping key addresses to lists of tuples 1966 (module_key_address, amount). 1967 1968 Raises: 1969 QueryError: If the query to the network fails or is invalid. 1970 """ 1971 1972 result = self.query_map("StakeTo", [], extract_value=extract_value)[ 1973 "StakeTo" 1974 ] 1975 return transform_stake_dmap(result) 1976 1977 def query_map_delegationfee( 1978 self, netuid: int = 0, extract_value: bool = False 1979 ) -> dict[str, int]: 1980 """ 1981 Retrieves a mapping of delegation fees for keys on the network. 1982 1983 Queries the network to obtain a mapping of key addresses to their 1984 respective delegation fees. 1985 1986 Args: 1987 netuid: The network UID to filter the delegation fees. 1988 1989 Returns: 1990 A dictionary mapping key addresses to their delegation fees. 1991 1992 Raises: 1993 QueryError: If the query to the network fails or is invalid. 1994 """ 1995 1996 return self.query_map("DelegationFee", [netuid], extract_value=extract_value)[ 1997 "DelegationFee" 1998 ] 1999 2000 def query_map_tempo(self, extract_value: bool = False) -> dict[int, int]: 2001 """ 2002 Retrieves a mapping of tempo settings for the network. 2003 2004 Queries the network to obtain the tempo (rate of reward distributions) 2005 settings for various network subnets. 2006 2007 Returns: 2008 A dictionary mapping network UIDs to their tempo settings. 2009 2010 Raises: 2011 QueryError: If the query to the network fails or is invalid. 2012 """ 2013 2014 return self.query_map("Tempo", extract_value=extract_value)["Tempo"] 2015 2016 def query_map_immunity_period(self, extract_value: bool) -> dict[int, int]: 2017 """ 2018 Retrieves a mapping of immunity periods for the network. 2019 2020 Queries the network for the immunity period settings, 2021 which represent the time duration during which modules 2022 can not get deregistered. 2023 2024 Returns: 2025 A dictionary mapping network UIDs to their immunity period settings. 2026 2027 Raises: 2028 QueryError: If the query to the network fails or is invalid. 2029 """ 2030 2031 return self.query_map("ImmunityPeriod", extract_value=extract_value)[ 2032 "ImmunityPeriod" 2033 ] 2034 2035 def query_map_min_allowed_weights( 2036 self, extract_value: bool = False 2037 ) -> dict[int, int]: 2038 """ 2039 Retrieves a mapping of minimum allowed weights for the network. 2040 2041 Queries the network to obtain the minimum allowed weights, 2042 which are the lowest permissible weight values that can be set by 2043 validators. 2044 2045 Returns: 2046 A dictionary mapping network UIDs to 2047 their minimum allowed weight values. 2048 2049 Raises: 2050 QueryError: If the query to the network fails or is invalid. 2051 """ 2052 2053 return self.query_map("MinAllowedWeights", extract_value=extract_value)[ 2054 "MinAllowedWeights" 2055 ] 2056 2057 def query_map_max_allowed_weights( 2058 self, extract_value: bool = False 2059 ) -> dict[int, int]: 2060 """ 2061 Retrieves a mapping of maximum allowed weights for the network. 2062 2063 Queries the network for the maximum allowed weights, 2064 which are the highest permissible 2065 weight values that can be set by validators. 2066 2067 Returns: 2068 A dictionary mapping network UIDs to 2069 their maximum allowed weight values. 2070 2071 Raises: 2072 QueryError: If the query to the network fails or is invalid. 2073 """ 2074 2075 return self.query_map("MaxAllowedWeights", extract_value=extract_value)[ 2076 "MaxAllowedWeights" 2077 ] 2078 2079 def query_map_max_allowed_uids(self, extract_value: bool = False) -> dict[int, int]: 2080 """ 2081 Queries the network for the maximum number of allowed user IDs (UIDs) 2082 for each network subnet. 2083 2084 Fetches a mapping of network subnets to their respective 2085 limits on the number of user IDs that can be created or used. 2086 2087 Returns: 2088 A dictionary mapping network UIDs (unique identifiers) to their 2089 maximum allowed number of UIDs. 2090 Each entry represents a network subnet 2091 with its corresponding UID limit. 2092 2093 Raises: 2094 QueryError: If the query to the network fails or is invalid. 2095 """ 2096 2097 return self.query_map("MaxAllowedUids", extract_value=extract_value)[ 2098 "MaxAllowedUids" 2099 ] 2100 2101 def query_map_min_stake(self, extract_value: bool = False) -> dict[int, int]: 2102 """ 2103 Retrieves a mapping of minimum allowed stake on the network. 2104 2105 Queries the network to obtain the minimum number of stake, 2106 which is represented in nanotokens. 2107 2108 Returns: 2109 A dictionary mapping network UIDs to 2110 their minimum allowed stake values. 2111 2112 Raises: 2113 QueryError: If the query to the network fails or is invalid. 2114 """ 2115 2116 return self.query_map("MinStake", extract_value=extract_value)["MinStake"] 2117 2118 def query_map_max_stake(self, extract_value: bool = False) -> dict[int, int]: 2119 """ 2120 Retrieves a mapping of the maximum stake values for the network. 2121 2122 Queries the network for the maximum stake values across various s 2123 ubnets of the network. 2124 2125 Returns: 2126 A dictionary mapping network UIDs to their maximum stake values. 2127 2128 Raises: 2129 QueryError: If the query to the network fails or is invalid. 2130 """ 2131 2132 return self.query_map("MaxStake", extract_value=extract_value)["MaxStake"] 2133 2134 def query_map_founder(self, extract_value: bool = False) -> dict[int, str]: 2135 """ 2136 Retrieves a mapping of founders for the network. 2137 2138 Queries the network to obtain the founders associated with 2139 various subnets. 2140 2141 Returns: 2142 A dictionary mapping network UIDs to their respective founders. 2143 2144 Raises: 2145 QueryError: If the query to the network fails or is invalid. 2146 """ 2147 2148 return self.query_map("Founder", extract_value=extract_value)["Founder"] 2149 2150 def query_map_founder_share(self, extract_value: bool = False) -> dict[int, int]: 2151 """ 2152 Retrieves a mapping of founder shares for the network. 2153 2154 Queries the network for the share percentages 2155 allocated to founders across different subnets. 2156 2157 Returns: 2158 A dictionary mapping network UIDs to their founder share percentages. 2159 2160 Raises: 2161 QueryError: If the query to the network fails or is invalid. 2162 """ 2163 2164 return self.query_map("FounderShare", extract_value=extract_value)[ 2165 "FounderShare" 2166 ] 2167 2168 def query_map_incentive_ratio(self, extract_value: bool = False) -> dict[int, int]: 2169 """ 2170 Retrieves a mapping of incentive ratios for the network. 2171 2172 Queries the network for the incentive ratios, 2173 which are the proportions of rewards or incentives 2174 allocated in different subnets of the network. 2175 2176 Returns: 2177 A dictionary mapping network UIDs to their incentive ratios. 2178 2179 Raises: 2180 QueryError: If the query to the network fails or is invalid. 2181 """ 2182 2183 return self.query_map("IncentiveRatio", extract_value=extract_value)[ 2184 "IncentiveRatio" 2185 ] 2186 2187 def query_map_trust_ratio(self, extract_value: bool = False) -> dict[int, int]: 2188 """ 2189 Retrieves a mapping of trust ratios for the network. 2190 2191 Queries the network for trust ratios, 2192 indicative of the level of trust or credibility assigned 2193 to different subnets of the network. 2194 2195 Returns: 2196 A dictionary mapping network UIDs to their trust ratios. 2197 2198 Raises: 2199 QueryError: If the query to the network fails or is invalid. 2200 """ 2201 2202 return self.query_map("TrustRatio", extract_value=extract_value)["TrustRatio"] 2203 2204 def query_map_vote_mode_subnet(self, extract_value: bool = False) -> dict[int, str]: 2205 """ 2206 Retrieves a mapping of vote modes for subnets within the network. 2207 2208 Queries the network for the voting modes used in different 2209 subnets, which define the methodology or approach of voting within those 2210 subnets. 2211 2212 Returns: 2213 A dictionary mapping network UIDs to their vote 2214 modes for subnets. 2215 2216 Raises: 2217 QueryError: If the query to the network fails or is invalid. 2218 """ 2219 2220 return self.query_map("VoteModeSubnet", extract_value=extract_value)[ 2221 "VoteModeSubnet" 2222 ] 2223 2224 def query_map_legit_whitelist( 2225 self, extract_value: bool = False 2226 ) -> dict[Ss58Address, int]: 2227 """ 2228 Retrieves a mapping of whitelisted addresses for the network. 2229 2230 Queries the network for a mapping of whitelisted addresses 2231 and their respective legitimacy status. 2232 2233 Returns: 2234 A dictionary mapping addresses to their legitimacy status. 2235 2236 Raises: 2237 QueryError: If the query to the network fails or is invalid. 2238 """ 2239 2240 return self.query_map( 2241 "LegitWhitelist", module="GovernanceModule", extract_value=extract_value)[ 2242 "LegitWhitelist" 2243 ] 2244 2245 def query_map_subnet_names(self, extract_value: bool = False) -> dict[int, str]: 2246 """ 2247 Retrieves a mapping of subnet names within the network. 2248 2249 Queries the network for the names of various subnets, 2250 providing an overview of the different 2251 subnets within the network. 2252 2253 Returns: 2254 A dictionary mapping network UIDs to their subnet names. 2255 2256 Raises: 2257 QueryError: If the query to the network fails or is invalid. 2258 """ 2259 2260 return self.query_map("SubnetNames", extract_value=extract_value)["SubnetNames"] 2261 2262 def query_map_balances( 2263 self, extract_value: bool = False, block_hash: str | None = None 2264 ) -> dict[str, dict[str, int | dict[str, int | float]]]: 2265 """ 2266 Retrieves a mapping of account balances within the network. 2267 2268 Queries the network for the balances associated with different accounts. 2269 It provides detailed information including various types of 2270 balances for each account. 2271 2272 Returns: 2273 A dictionary mapping account addresses to their balance details. 2274 2275 Raises: 2276 QueryError: If the query to the network fails or is invalid. 2277 """ 2278 2279 return self.query_map("Account", module="System", extract_value=extract_value, block_hash=block_hash)[ 2280 "Account" 2281 ] 2282 2283 def query_map_registration_blocks( 2284 self, netuid: int = 0, extract_value: bool = False 2285 ) -> dict[int, int]: 2286 """ 2287 Retrieves a mapping of registration blocks for UIDs on the network. 2288 2289 Queries the network to find the block numbers at which various 2290 UIDs were registered. 2291 2292 Args: 2293 netuid: The network UID from which to get the registrations. 2294 2295 Returns: 2296 A dictionary mapping UIDs to their registration block numbers. 2297 2298 Raises: 2299 QueryError: If the query to the network fails or is invalid. 2300 """ 2301 2302 return self.query_map( 2303 "RegistrationBlock", [netuid], extract_value=extract_value 2304 )["RegistrationBlock"] 2305 2306 def query_map_name( 2307 self, netuid: int = 0, extract_value: bool = False 2308 ) -> dict[int, str]: 2309 """ 2310 Retrieves a mapping of names for keys on the network. 2311 2312 Queries the network for the names associated with different keys. 2313 It provides a mapping of key UIDs to their registered names. 2314 2315 Args: 2316 netuid: The network UID from which to get the names. 2317 2318 Returns: 2319 A dictionary mapping key UIDs to their names. 2320 2321 Raises: 2322 QueryError: If the query to the network fails or is invalid. 2323 """ 2324 2325 return self.query_map("Name", [netuid], extract_value=extract_value)["Name"] 2326 2327 # == QUERY FUNCTIONS == # 2328 2329 def get_immunity_period(self, netuid: int = 0) -> int: 2330 """ 2331 Queries the network for the immunity period setting. 2332 2333 The immunity period is a time duration during which a module 2334 can not be deregistered from the network. 2335 Fetches the immunity period for a specified network subnet. 2336 2337 Args: 2338 netuid: The network UID for which to query the immunity period. 2339 2340 Returns: 2341 The immunity period setting for the specified network subnet. 2342 2343 Raises: 2344 QueryError: If the query to the network fails or is invalid. 2345 """ 2346 2347 return self.query( 2348 "ImmunityPeriod", 2349 params=[netuid], 2350 ) 2351 2352 def get_max_set_weights_per_epoch(self): 2353 return self.query("MaximumSetWeightCallsPerEpoch") 2354 2355 def get_min_allowed_weights(self, netuid: int = 0) -> int: 2356 """ 2357 Queries the network for the minimum allowed weights setting. 2358 2359 Retrieves the minimum weight values that are possible to set 2360 by a validator within a specific network subnet. 2361 2362 Args: 2363 netuid: The network UID for which to query the minimum allowed 2364 weights. 2365 2366 Returns: 2367 The minimum allowed weight values for the specified network 2368 subnet. 2369 2370 Raises: 2371 QueryError: If the query to the network fails or is invalid. 2372 """ 2373 2374 return self.query( 2375 "MinAllowedWeights", 2376 params=[netuid], 2377 ) 2378 2379 def get_dao_treasury_address(self) -> Ss58Address: 2380 return self.query("DaoTreasuryAddress", module="GovernanceModule") 2381 2382 def get_max_allowed_weights(self, netuid: int = 0) -> int: 2383 """ 2384 Queries the network for the maximum allowed weights setting. 2385 2386 Retrieves the maximum weight values that are possible to set 2387 by a validator within a specific network subnet. 2388 2389 Args: 2390 netuid: The network UID for which to query the maximum allowed 2391 weights. 2392 2393 Returns: 2394 The maximum allowed weight values for the specified network 2395 subnet. 2396 2397 Raises: 2398 QueryError: If the query to the network fails or is invalid. 2399 """ 2400 2401 return self.query("MaxAllowedWeights", params=[netuid]) 2402 2403 def get_max_allowed_uids(self, netuid: int = 0) -> int: 2404 """ 2405 Queries the network for the maximum allowed UIDs setting. 2406 2407 Fetches the upper limit on the number of user IDs that can 2408 be allocated or used within a specific network subnet. 2409 2410 Args: 2411 netuid: The network UID for which to query the maximum allowed UIDs. 2412 2413 Returns: 2414 The maximum number of allowed UIDs for the specified network subnet. 2415 2416 Raises: 2417 QueryError: If the query to the network fails or is invalid. 2418 """ 2419 2420 return self.query("MaxAllowedUids", params=[netuid]) 2421 2422 def get_name(self, netuid: int = 0) -> str: 2423 """ 2424 Queries the network for the name of a specific subnet. 2425 2426 Args: 2427 netuid: The network UID for which to query the name. 2428 2429 Returns: 2430 The name of the specified network subnet. 2431 2432 Raises: 2433 QueryError: If the query to the network fails or is invalid. 2434 """ 2435 2436 return self.query("Name", params=[netuid]) 2437 2438 def get_subnet_name(self, netuid: int = 0) -> str: 2439 """ 2440 Queries the network for the name of a specific subnet. 2441 2442 Args: 2443 netuid: The network UID for which to query the name. 2444 2445 Returns: 2446 The name of the specified network subnet. 2447 2448 Raises: 2449 QueryError: If the query to the network fails or is invalid. 2450 """ 2451 2452 return self.query("SubnetNames", params=[netuid]) 2453 2454 def get_global_dao_treasury(self): 2455 return self.query("GlobalDaoTreasury", module="GovernanceModule") 2456 2457 def get_n(self, netuid: int = 0) -> int: 2458 """ 2459 Queries the network for the 'N' hyperparameter, which represents how 2460 many modules are on the network. 2461 2462 Args: 2463 netuid: The network UID for which to query the 'N' hyperparameter. 2464 2465 Returns: 2466 The value of the 'N' hyperparameter for the specified network 2467 subnet. 2468 2469 Raises: 2470 QueryError: If the query to the network fails or is invalid. 2471 """ 2472 2473 return self.query("N", params=[netuid]) 2474 2475 def get_tempo(self, netuid: int = 0) -> int: 2476 """ 2477 Queries the network for the tempo setting, measured in blocks, for the 2478 specified subnet. 2479 2480 Args: 2481 netuid: The network UID for which to query the tempo. 2482 2483 Returns: 2484 The tempo setting for the specified subnet. 2485 2486 Raises: 2487 QueryError: If the query to the network fails or is invalid. 2488 """ 2489 2490 return self.query("Tempo", params=[netuid]) 2491 2492 def get_total_free_issuance(self, block_hash: str | None = None) -> int: 2493 """ 2494 Queries the network for the total free issuance. 2495 2496 Fetches the total amount of free issuance tokens available 2497 2498 Returns: 2499 The total free issuance amount. 2500 2501 Raises: 2502 QueryError: If the query to the network fails or is invalid. 2503 """ 2504 2505 return self.query("TotalIssuance", module="Balances", block_hash=block_hash) 2506 2507 def get_total_stake(self, block_hash: str | None = None) -> int: 2508 """ 2509 Retrieves a mapping of total stakes for keys on the network. 2510 2511 Queries the network for a mapping of key UIDs to their total stake amounts. 2512 2513 Returns: 2514 A dictionary mapping key UIDs to their total stake amounts. 2515 2516 Raises: 2517 QueryError: If the query to the network fails or is invalid. 2518 """ 2519 2520 return self.query("TotalStake", block_hash=block_hash) 2521 2522 def get_registrations_per_block(self): 2523 """ 2524 Queries the network for the number of registrations per block. 2525 2526 Fetches the number of registrations that are processed per 2527 block within the network. 2528 2529 Returns: 2530 The number of registrations processed per block. 2531 2532 Raises: 2533 QueryError: If the query to the network fails or is invalid. 2534 """ 2535 2536 return self.query( 2537 "RegistrationsPerBlock", 2538 ) 2539 2540 def max_registrations_per_block(self, netuid: int = 0): 2541 """ 2542 Queries the network for the maximum number of registrations per block. 2543 2544 Retrieves the upper limit of registrations that can be processed in 2545 each block within a specific network subnet. 2546 2547 Args: 2548 netuid: The network UID for which to query. 2549 2550 Returns: 2551 The maximum number of registrations per block for 2552 the specified network subnet. 2553 2554 Raises: 2555 QueryError: If the query to the network fails or is invalid. 2556 """ 2557 2558 return self.query( 2559 "MaxRegistrationsPerBlock", 2560 params=[netuid], 2561 ) 2562 2563 def get_proposal(self, proposal_id: int = 0): 2564 """ 2565 Queries the network for a specific proposal. 2566 2567 Args: 2568 proposal_id: The ID of the proposal to query. 2569 2570 Returns: 2571 The details of the specified proposal. 2572 2573 Raises: 2574 QueryError: If the query to the network fails, is invalid, 2575 or if the proposal ID does not exist. 2576 """ 2577 2578 return self.query( 2579 "Proposals", 2580 params=[proposal_id], 2581 ) 2582 2583 def get_trust(self, netuid: int = 0): 2584 """ 2585 Queries the network for the trust setting of a specific network subnet. 2586 2587 Retrieves the trust level or score, which may represent the 2588 level of trustworthiness or reliability within a 2589 particular network subnet. 2590 2591 Args: 2592 netuid: The network UID for which to query the trust setting. 2593 2594 Returns: 2595 The trust level or score for the specified network subnet. 2596 2597 Raises: 2598 QueryError: If the query to the network fails or is invalid. 2599 """ 2600 2601 return self.query( 2602 "Trust", 2603 params=[netuid], 2604 ) 2605 2606 def get_uids(self, key: Ss58Address, netuid: int = 0) -> bool | None: 2607 """ 2608 Queries the network for module UIDs associated with a specific key. 2609 2610 Args: 2611 key: The key address for which to query UIDs. 2612 netuid: The network UID within which to search for the key. 2613 2614 Returns: 2615 A list of UIDs associated with the specified key. 2616 2617 Raises: 2618 QueryError: If the query to the network fails or is invalid. 2619 """ 2620 2621 return self.query( 2622 "Uids", 2623 params=[netuid, key], 2624 ) 2625 2626 def get_unit_emission(self) -> int: 2627 """ 2628 Queries the network for the unit emission setting. 2629 2630 Retrieves the unit emission value, which represents the 2631 emission rate or quantity for the $COMM token. 2632 2633 Returns: 2634 The unit emission value in nanos for the network. 2635 2636 Raises: 2637 QueryError: If the query to the network fails or is invalid. 2638 """ 2639 2640 return self.query("UnitEmission", module="SubnetEmissionModule") 2641 2642 def get_tx_rate_limit(self) -> int: 2643 """ 2644 Queries the network for the transaction rate limit. 2645 2646 Retrieves the rate limit for transactions within the network, 2647 which defines the maximum number of transactions that can be 2648 processed within a certain timeframe. 2649 2650 Returns: 2651 The transaction rate limit for the network. 2652 2653 Raises: 2654 QueryError: If the query to the network fails or is invalid. 2655 """ 2656 2657 return self.query( 2658 "TxRateLimit", 2659 ) 2660 2661 def get_subnet_burn(self) -> int: 2662 """Queries the network for the subnet burn value. 2663 2664 Retrieves the subnet burn value from the network, which represents 2665 the amount of tokens that are burned (permanently removed from 2666 circulation) for subnet-related operations. 2667 2668 Returns: 2669 int: The subnet burn value. 2670 2671 Raises: 2672 QueryError: If the query to the network fails or returns invalid data. 2673 """ 2674 2675 return self.query( 2676 "SubnetBurn", 2677 ) 2678 2679 def get_burn_rate(self) -> int: 2680 """ 2681 Queries the network for the burn rate setting. 2682 2683 Retrieves the burn rate, which represents the rate at 2684 which the $COMM token is permanently 2685 removed or 'burned' from circulation. 2686 2687 Returns: 2688 The burn rate for the network. 2689 2690 Raises: 2691 QueryError: If the query to the network fails or is invalid. 2692 """ 2693 2694 return self.query( 2695 "BurnRate", 2696 params=[], 2697 ) 2698 2699 def get_burn(self, netuid: int = 0) -> int: 2700 """ 2701 Queries the network for the burn setting. 2702 2703 Retrieves the burn value, which represents the amount of the 2704 $COMM token that is 'burned' or permanently removed from 2705 circulation. 2706 2707 Args: 2708 netuid: The network UID for which to query the burn value. 2709 2710 Returns: 2711 The burn value for the specified network subnet. 2712 2713 Raises: 2714 QueryError: If the query to the network fails or is invalid. 2715 """ 2716 2717 return self.query("Burn", params=[netuid]) 2718 2719 def get_min_burn(self) -> int: 2720 """ 2721 Queries the network for the minimum burn setting. 2722 2723 Retrieves the minimum burn value, indicating the lowest 2724 amount of the $COMM tokens that can be 'burned' or 2725 permanently removed from circulation. 2726 2727 Returns: 2728 The minimum burn value for the network. 2729 2730 Raises: 2731 QueryError: If the query to the network fails or is invalid. 2732 """ 2733 2734 return self.query( 2735 "BurnConfig", 2736 params=[], 2737 )["min_burn"] 2738 2739 def get_min_weight_stake(self) -> int: 2740 """ 2741 Queries the network for the minimum weight stake setting. 2742 2743 Retrieves the minimum weight stake, which represents the lowest 2744 stake weight that is allowed for certain operations or 2745 transactions within the network. 2746 2747 Returns: 2748 The minimum weight stake for the network. 2749 2750 Raises: 2751 QueryError: If the query to the network fails or is invalid. 2752 """ 2753 2754 return self.query("MinWeightStake", params=[]) 2755 2756 def get_vote_mode_global(self) -> str: 2757 """ 2758 Queries the network for the global vote mode setting. 2759 2760 Retrieves the global vote mode, which defines the overall voting 2761 methodology or approach used across the network in default. 2762 2763 Returns: 2764 The global vote mode setting for the network. 2765 2766 Raises: 2767 QueryError: If the query to the network fails or is invalid. 2768 """ 2769 2770 return self.query( 2771 "VoteModeGlobal", 2772 ) 2773 2774 def get_max_proposals(self) -> int: 2775 """ 2776 Queries the network for the maximum number of proposals allowed. 2777 2778 Retrieves the upper limit on the number of proposals that can be 2779 active or considered at any given time within the network. 2780 2781 Returns: 2782 The maximum number of proposals allowed on the network. 2783 2784 Raises: 2785 QueryError: If the query to the network fails or is invalid. 2786 """ 2787 2788 return self.query( 2789 "MaxProposals", 2790 ) 2791 2792 def get_max_registrations_per_block(self) -> int: 2793 """ 2794 Queries the network for the maximum number of registrations per block. 2795 2796 Retrieves the maximum number of registrations that can 2797 be processed in each block within the network. 2798 2799 Returns: 2800 The maximum number of registrations per block on the network. 2801 2802 Raises: 2803 QueryError: If the query to the network fails or is invalid. 2804 """ 2805 2806 return self.query( 2807 "MaxRegistrationsPerBlock", 2808 params=[], 2809 ) 2810 2811 def get_max_name_length(self) -> int: 2812 """ 2813 Queries the network for the maximum length allowed for names. 2814 2815 Retrieves the maximum character length permitted for names 2816 within the network. Such as the module names 2817 2818 Returns: 2819 The maximum length allowed for names on the network. 2820 2821 Raises: 2822 QueryError: If the query to the network fails or is invalid. 2823 """ 2824 2825 return self.query( 2826 "MaxNameLength", 2827 params=[], 2828 ) 2829 2830 def get_global_vote_threshold(self) -> int: 2831 """ 2832 Queries the network for the global vote threshold. 2833 2834 Retrieves the global vote threshold, which is the critical value or 2835 percentage required for decisions in the network's governance process. 2836 2837 Returns: 2838 The global vote threshold for the network. 2839 2840 Raises: 2841 QueryError: If the query to the network fails or is invalid. 2842 """ 2843 2844 return self.query( 2845 "GlobalVoteThreshold", 2846 ) 2847 2848 def get_max_allowed_subnets(self) -> int: 2849 """ 2850 Queries the network for the maximum number of allowed subnets. 2851 2852 Retrieves the upper limit on the number of subnets that can 2853 be created or operated within the network. 2854 2855 Returns: 2856 The maximum number of allowed subnets on the network. 2857 2858 Raises: 2859 QueryError: If the query to the network fails or is invalid. 2860 """ 2861 2862 return self.query( 2863 "MaxAllowedSubnets", 2864 params=[], 2865 ) 2866 2867 def get_max_allowed_modules(self) -> int: 2868 """ 2869 Queries the network for the maximum number of allowed modules. 2870 2871 Retrieves the upper limit on the number of modules that 2872 can be registered within the network. 2873 2874 Returns: 2875 The maximum number of allowed modules on the network. 2876 2877 Raises: 2878 QueryError: If the query to the network fails or is invalid. 2879 """ 2880 2881 return self.query( 2882 "MaxAllowedModules", 2883 params=[], 2884 ) 2885 2886 def get_min_stake(self, netuid: int = 0) -> int: 2887 """ 2888 Queries the network for the minimum stake required to register a key. 2889 2890 Retrieves the minimum amount of stake necessary for 2891 registering a key within a specific network subnet. 2892 2893 Args: 2894 netuid: The network UID for which to query the minimum stake. 2895 2896 Returns: 2897 The minimum stake required for key registration in nanos. 2898 2899 Raises: 2900 QueryError: If the query to the network fails or is invalid. 2901 """ 2902 2903 return self.query("MinStake", params=[netuid]) 2904 2905 def get_stakefrom( 2906 self, 2907 key: Ss58Address, 2908 ) -> dict[str, int]: 2909 """ 2910 Retrieves the stake amounts from all stakers to a specific staked address. 2911 2912 Queries the network for the stakes received by a particular staked address 2913 from all stakers. 2914 2915 Args: 2916 key: The address of the key receiving the stakes. 2917 2918 Returns: 2919 A dictionary mapping staker addresses to their respective stake amounts. 2920 2921 Raises: 2922 QueryError: If the query to the network fails or is invalid. 2923 """ 2924 2925 # Has to use query map in order to iterate through the storage prefix. 2926 return self.query_map("StakeFrom", [key], extract_value=False).get("StakeFrom", {}) 2927 2928 def get_staketo( 2929 self, 2930 key: Ss58Address, 2931 ) -> dict[str, int]: 2932 """ 2933 Retrieves the stake amounts provided by a specific staker to all staked addresses. 2934 2935 Queries the network for the stakes provided by a particular staker to 2936 all staked addresses. 2937 2938 Args: 2939 key: The address of the key providing the stakes. 2940 2941 Returns: 2942 A dictionary mapping staked addresses to their respective received stake amounts. 2943 2944 Raises: 2945 QueryError: If the query to the network fails or is invalid. 2946 """ 2947 2948 # Has to use query map in order to iterate through the storage prefix. 2949 return self.query_map("StakeTo", [key], extract_value=False).get("StakeTo", {}) 2950 2951 def get_balance( 2952 self, 2953 addr: Ss58Address, 2954 ) -> int: 2955 """ 2956 Retrieves the balance of a specific key. 2957 2958 Args: 2959 addr: The address of the key to query the balance for. 2960 2961 Returns: 2962 The balance of the specified key. 2963 2964 Raises: 2965 QueryError: If the query to the network fails or is invalid. 2966 """ 2967 2968 result = self.query("Account", module="System", params=[addr]) 2969 2970 return result["data"]["free"] 2971 2972 def get_block(self, block_hash: str | None = None) -> dict[Any, Any] | None: 2973 """ 2974 Retrieves information about a specific block in the network. 2975 2976 Queries the network for details about a block, such as its number, 2977 hash, and other relevant information. 2978 2979 Returns: 2980 The requested information about the block, 2981 or None if the block does not exist 2982 or the information is not available. 2983 2984 Raises: 2985 QueryError: If the query to the network fails or is invalid. 2986 """ 2987 2988 with self.get_conn() as substrate: 2989 block: dict[Any, Any] | None = substrate.get_block( # type: ignore 2990 block_hash # type: ignore 2991 ) 2992 2993 return block 2994 2995 def get_existential_deposit(self, block_hash: str | None = None) -> int: 2996 """ 2997 Retrieves the existential deposit value for the network. 2998 2999 The existential deposit is the minimum balance that must be maintained 3000 in an account to prevent it from being purged. Denotated in nano units. 3001 3002 Returns: 3003 The existential deposit value in nano units. 3004 Note: 3005 The value returned is a fixed value defined in the 3006 client and may not reflect changes in the network's configuration. 3007 """ 3008 3009 with self.get_conn() as substrate: 3010 result: int = substrate.get_constant( # type: ignore 3011 "Balances", "ExistentialDeposit", block_hash 3012 ).value # type: ignore 3013 3014 return result 3015 3016 def get_voting_power_delegators(self) -> list[Ss58Address]: 3017 result = self.query("NotDelegatingVotingPower", [], module="GovernanceModule") 3018 return result 3019 3020 def add_transfer_dao_treasury_proposal( 3021 self, 3022 key: Keypair, 3023 data: str, 3024 amount_nano: int, 3025 dest: Ss58Address, 3026 ): 3027 params = {"dest": dest, "value": amount_nano, "data": data} 3028 3029 return self.compose_call( 3030 module="GovernanceModule", 3031 fn="add_transfer_dao_treasury_proposal", 3032 params=params, 3033 key=key, 3034 ) 3035 3036 def delegate_rootnet_control(self, key: Keypair, dest: Ss58Address): 3037 params = {"origin": key, "target": dest} 3038 3039 return self.compose_call( 3040 module="SubspaceModule", 3041 fn="delegate_rootnet_control", 3042 params=params, 3043 key=key, 3044 )
MAX_REQUEST_SIZE =
9000000
@dataclass
class
Chunk:
24@dataclass 25class Chunk: 26 batch_requests: list[tuple[Any, Any]] 27 prefix_list: list[list[str]] 28 fun_params: list[tuple[Any, Any, Any, Any, str]]
class
CommuneClient:
35class CommuneClient: 36 """ 37 A client for interacting with Commune network nodes, querying storage, 38 submitting transactions, etc. 39 40 Attributes: 41 wait_for_finalization: Whether to wait for transaction finalization. 42 43 Example: 44 ```py 45 client = CommuneClient() 46 client.query(name='function_name', params=['param1', 'param2']) 47 ``` 48 49 Raises: 50 AssertionError: If the maximum connections value is less than or equal 51 to zero. 52 """ 53 54 wait_for_finalization: bool 55 _num_connections: int 56 _connection_queue: queue.Queue[SubstrateInterface] 57 url: str 58 59 def __init__( 60 self, 61 url: str, 62 num_connections: int = 1, 63 wait_for_finalization: bool = False, 64 timeout: int | None = None, 65 ): 66 """ 67 Args: 68 url: The URL of the network node to connect to. 69 num_connections: The number of websocket connections to be opened. 70 """ 71 assert num_connections > 0 72 self._num_connections = num_connections 73 self.wait_for_finalization = wait_for_finalization 74 self._connection_queue = queue.Queue(num_connections) 75 self.url = url 76 ws_options: dict[str, int] = {} 77 if timeout is not None: 78 ws_options["timeout"] = timeout 79 self.ws_options = ws_options 80 for _ in range(num_connections): 81 self._connection_queue.put( 82 SubstrateInterface(url, ws_options=ws_options) 83 ) 84 85 @property 86 def connections(self) -> int: 87 """ 88 Gets the maximum allowed number of simultaneous connections to the 89 network node. 90 """ 91 return self._num_connections 92 93 @contextmanager 94 def get_conn(self, timeout: float | None = None, init: bool = False): 95 """ 96 Context manager to get a connection from the pool. 97 98 Tries to get a connection from the pool queue. If the queue is empty, 99 it blocks for `timeout` seconds until a connection is available. If 100 `timeout` is None, it blocks indefinitely. 101 102 Args: 103 timeout: The maximum time in seconds to wait for a connection. 104 105 Yields: 106 The connection object from the pool. 107 108 Raises: 109 QueueEmptyError: If no connection is available within the timeout 110 period. 111 """ 112 conn = self._connection_queue.get(timeout=timeout) 113 if init: 114 conn.init_runtime() # type: ignore 115 try: 116 if conn.websocket and conn.websocket.connected: # type: ignore 117 yield conn 118 else: 119 conn = SubstrateInterface(self.url, ws_options=self.ws_options) 120 yield conn 121 finally: 122 self._connection_queue.put(conn) 123 124 def _get_storage_keys( 125 self, 126 storage: str, 127 queries: list[tuple[str, list[Any]]], 128 block_hash: str | None, 129 ): 130 131 send: list[tuple[str, list[Any]]] = [] 132 prefix_list: list[Any] = [] 133 134 key_idx = 0 135 with self.get_conn(init=True) as substrate: 136 for function, params in queries: 137 storage_key = StorageKey.create_from_storage_function( # type: ignore 138 storage, function, params, runtime_config=substrate.runtime_config, metadata=substrate.metadata # type: ignore 139 ) 140 141 prefix = storage_key.to_hex() 142 prefix_list.append(prefix) 143 send.append(("state_getKeys", [prefix, block_hash])) 144 key_idx += 1 145 return send, prefix_list 146 147 def _get_lists( 148 self, 149 storage_module: str, 150 queries: list[tuple[str, list[Any]]], 151 substrate: SubstrateInterface, 152 ) -> list[tuple[Any, Any, Any, Any, str]]: 153 """ 154 Generates a list of tuples containing parameters for each storage function based on the given functions and substrate interface. 155 156 Args: 157 functions (dict[str, list[query_call]]): A dictionary where keys are storage module names and values are lists of tuples. 158 Each tuple consists of a storage function name and its parameters. 159 substrate: An instance of the SubstrateInterface class used to interact with the substrate. 160 161 Returns: 162 A list of tuples in the format `(value_type, param_types, key_hashers, params, storage_function)` for each storage function in the given functions. 163 164 Example: 165 >>> _get_lists( 166 functions={'storage_module': [('storage_function', ['param1', 'param2'])]}, 167 substrate=substrate_instance 168 ) 169 [('value_type', 'param_types', 'key_hashers', ['param1', 'param2'], 'storage_function'), ...] 170 """ 171 172 function_parameters: list[tuple[Any, Any, Any, Any, str]] = [] 173 174 metadata_pallet = substrate.metadata.get_metadata_pallet( # type: ignore 175 storage_module 176 ) 177 for storage_function, params in queries: 178 storage_item = metadata_pallet.get_storage_function( # type: ignore 179 storage_function 180 ) 181 182 value_type = storage_item.get_value_type_string() # type: ignore 183 param_types = storage_item.get_params_type_string() # type: ignore 184 key_hashers = storage_item.get_param_hashers() # type: ignore 185 function_parameters.append( 186 ( 187 value_type, 188 param_types, 189 key_hashers, 190 params, 191 storage_function, 192 ) # type: ignore 193 ) 194 return function_parameters 195 196 def _send_batch( 197 self, 198 batch_payload: list[Any], 199 request_ids: list[int], 200 extract_result: bool = True, 201 ): 202 """ 203 Sends a batch of requests to the substrate and collects the results. 204 205 Args: 206 substrate: An instance of the substrate interface. 207 batch_payload: The payload of the batch request. 208 request_ids: A list of request IDs for tracking responses. 209 results: A list to store the results of the requests. 210 extract_result: Whether to extract the result from the response. 211 212 Raises: 213 NetworkQueryError: If there is an `error` in the response message. 214 215 Note: 216 No explicit return value as results are appended to the provided 'results' list. 217 """ 218 results: list[str | dict[Any, Any]] = [] 219 with self.get_conn(init=True) as substrate: 220 try: 221 222 substrate.websocket.send( # type: ignore 223 json.dumps(batch_payload) 224 ) # type: ignore 225 except NetworkQueryError: 226 pass 227 while len(results) < len(request_ids): 228 received_messages = json.loads( 229 substrate.websocket.recv() # type: ignore 230 ) # type: ignore 231 if isinstance(received_messages, dict): 232 received_messages: list[dict[Any, Any]] = [received_messages] 233 234 for message in received_messages: 235 if message.get("id") in request_ids: 236 if extract_result: 237 try: 238 results.append(message["result"]) 239 except Exception: 240 raise ( 241 RuntimeError( 242 f"Error extracting result from message: {message}" 243 ) 244 ) 245 else: 246 results.append(message) 247 if "error" in message: 248 raise NetworkQueryError(message["error"]) 249 250 return results 251 252 def _make_request_smaller( 253 self, 254 batch_request: list[tuple[T1, T2]], 255 prefix_list: list[list[str]], 256 fun_params: list[tuple[Any, Any, Any, Any, str]], 257 ) -> tuple[list[list[tuple[T1, T2]]], list[Chunk]]: 258 """ 259 Splits a batch of requests into smaller batches, each not exceeding the specified maximum size. 260 261 Args: 262 batch_request: A list of requests to be sent in a batch. 263 max_size: Maximum size of each batch in bytes. 264 265 Returns: 266 A list of smaller request batches. 267 268 Example: 269 >>> _make_request_smaller(batch_request=[('method1', 'params1'), ('method2', 'params2')], max_size=1000) 270 [[('method1', 'params1')], [('method2', 'params2')]] 271 """ 272 assert len(prefix_list) == len(fun_params) == len(batch_request) 273 274 def estimate_size(request: tuple[T1, T2]): 275 """Convert the batch request to a string and measure its length""" 276 return len(json.dumps(request)) 277 278 # Initialize variables 279 result: list[list[tuple[T1, T2]]] = [] 280 current_batch = [] 281 current_prefix_batch = [] 282 current_params_batch = [] 283 current_size = 0 284 285 chunk_list: list[Chunk] = [] 286 287 # Iterate through each request in the batch 288 for request, prefix, params in zip(batch_request, prefix_list, fun_params): 289 request_size = estimate_size(request) 290 291 # Check if adding this request exceeds the max size 292 if current_size + request_size > MAX_REQUEST_SIZE: 293 # If so, start a new batch 294 295 # Essentiatly checks that it's not the first iteration 296 if current_batch: 297 chunk = Chunk( 298 current_batch, current_prefix_batch, current_params_batch 299 ) 300 chunk_list.append(chunk) 301 result.append(current_batch) 302 303 current_batch = [request] 304 current_prefix_batch = [prefix] 305 current_params_batch = [params] 306 current_size = request_size 307 else: 308 # Otherwise, add to the current batch 309 current_batch.append(request) 310 current_size += request_size 311 current_prefix_batch.append(prefix) 312 current_params_batch.append(params) 313 314 # Add the last batch if it's not empty 315 if current_batch: 316 result.append(current_batch) 317 chunk = Chunk(current_batch, current_prefix_batch, current_params_batch) 318 chunk_list.append(chunk) 319 320 return result, chunk_list 321 322 def _are_changes_equal(self, change_a: Any, change_b: Any): 323 for (a, b), (c, d) in zip(change_a, change_b): 324 if a != c or b != d: 325 return False 326 327 def _rpc_request_batch( 328 self, batch_requests: list[tuple[str, list[Any]]], extract_result: bool = True 329 ) -> list[str]: 330 """ 331 Sends batch requests to the substrate node using multiple threads and collects the results. 332 333 Args: 334 substrate: An instance of the substrate interface. 335 batch_requests : A list of requests to be sent in batches. 336 max_size: Maximum size of each batch in bytes. 337 extract_result: Whether to extract the result from the response message. 338 339 Returns: 340 A list of results from the batch requests. 341 342 Example: 343 >>> _rpc_request_batch(substrate_instance, [('method1', ['param1']), ('method2', ['param2'])]) 344 ['result1', 'result2', ...] 345 """ 346 347 chunk_results: list[Any] = [] 348 # smaller_requests = self._make_request_smaller(batch_requests) 349 request_id = 0 350 with ThreadPoolExecutor() as executor: 351 futures: list[Future[list[str | dict[Any, Any]]]] = [] 352 for chunk in [batch_requests]: 353 request_ids: list[int] = [] 354 batch_payload: list[Any] = [] 355 for method, params in chunk: 356 request_id += 1 357 request_ids.append(request_id) 358 batch_payload.append( 359 { 360 "jsonrpc": "2.0", 361 "method": method, 362 "params": params, 363 "id": request_id, 364 } 365 ) 366 367 futures.append( 368 executor.submit( 369 self._send_batch, 370 batch_payload=batch_payload, 371 request_ids=request_ids, 372 extract_result=extract_result, 373 ) 374 ) 375 for future in futures: 376 resul = future.result() 377 chunk_results.append(resul) 378 return chunk_results 379 380 def _rpc_request_batch_chunked( 381 self, chunk_requests: list[Chunk], extract_result: bool = True 382 ): 383 """ 384 Sends batch requests to the substrate node using multiple threads and collects the results. 385 386 Args: 387 substrate: An instance of the substrate interface. 388 batch_requests : A list of requests to be sent in batches. 389 max_size: Maximum size of each batch in bytes. 390 extract_result: Whether to extract the result from the response message. 391 392 Returns: 393 A list of results from the batch requests. 394 395 Example: 396 >>> _rpc_request_batch(substrate_instance, [('method1', ['param1']), ('method2', ['param2'])]) 397 ['result1', 'result2', ...] 398 """ 399 400 def split_chunks(chunk: Chunk, chunk_info: list[Chunk], chunk_info_idx: int): 401 manhattam_chunks: list[tuple[Any, Any]] = [] 402 mutaded_chunk_info = deepcopy(chunk_info) 403 max_n_keys = 35000 404 for query in chunk.batch_requests: 405 result_keys = query[1][0] 406 keys_amount = len(result_keys) 407 if keys_amount > max_n_keys: 408 mutaded_chunk_info.pop(chunk_info_idx) 409 for i in range(0, keys_amount, max_n_keys): 410 new_chunk = deepcopy(chunk) 411 splitted_keys = result_keys[i: i + max_n_keys] 412 splitted_query = deepcopy(query) 413 splitted_query[1][0] = splitted_keys 414 new_chunk.batch_requests = [splitted_query] 415 manhattam_chunks.append(splitted_query) 416 mutaded_chunk_info.insert(chunk_info_idx, new_chunk) 417 else: 418 manhattam_chunks.append(query) 419 return manhattam_chunks, mutaded_chunk_info 420 421 assert len(chunk_requests) > 0 422 mutated_chunk_info: list[Chunk] = [] 423 chunk_results: list[Any] = [] 424 # smaller_requests = self._make_request_smaller(batch_requests) 425 request_id = 0 426 427 with ThreadPoolExecutor() as executor: 428 futures: list[Future[list[str | dict[Any, Any]]]] = [] 429 for idx, macro_chunk in enumerate(chunk_requests): 430 _, mutated_chunk_info = split_chunks(macro_chunk, chunk_requests, idx) 431 for chunk in mutated_chunk_info: 432 request_ids: list[int] = [] 433 batch_payload: list[Any] = [] 434 for method, params in chunk.batch_requests: 435 # for method, params in micro_chunk: 436 request_id += 1 437 request_ids.append(request_id) 438 batch_payload.append( 439 { 440 "jsonrpc": "2.0", 441 "method": method, 442 "params": params, 443 "id": request_id, 444 } 445 ) 446 futures.append( 447 executor.submit( 448 self._send_batch, 449 batch_payload=batch_payload, 450 request_ids=request_ids, 451 extract_result=extract_result, 452 ) 453 ) 454 for future in futures: 455 resul = future.result() 456 chunk_results.append(resul) 457 return chunk_results, mutated_chunk_info 458 459 def _decode_response( 460 self, 461 response: list[str], 462 function_parameters: list[tuple[Any, Any, Any, Any, str]], 463 prefix_list: list[Any], 464 block_hash: str, 465 ) -> dict[str, dict[Any, Any]]: 466 """ 467 Decodes a response from the substrate interface and organizes the data into a dictionary. 468 469 Args: 470 response: A list of encoded responses from a substrate query. 471 function_parameters: A list of tuples containing the parameters for each storage function. 472 last_keys: A list of the last keys used in the substrate query. 473 prefix_list: A list of prefixes used in the substrate query. 474 substrate: An instance of the SubstrateInterface class. 475 block_hash: The hash of the block to be queried. 476 477 Returns: 478 A dictionary where each key is a storage function name and the value is another dictionary. 479 This inner dictionary's key is the decoded key from the response and the value is the corresponding decoded value. 480 481 Raises: 482 ValueError: If an unsupported hash type is encountered in the `concat_hash_len` function. 483 484 Example: 485 >>> _decode_response( 486 response=[...], 487 function_parameters=[...], 488 last_keys=[...], 489 prefix_list=[...], 490 substrate=substrate_instance, 491 block_hash="0x123..." 492 ) 493 {'storage_function_name': {decoded_key: decoded_value, ...}, ...} 494 """ 495 496 def get_item_key_value(item_key: tuple[Any, ...] | Any) -> tuple[Any, ...] | Any: 497 if isinstance(item_key, tuple): 498 return tuple(k.value for k in item_key) 499 return item_key.value 500 501 def concat_hash_len(key_hasher: str) -> int: 502 """ 503 Determines the length of the hash based on the given key hasher type. 504 505 Args: 506 key_hasher: The type of key hasher. 507 508 Returns: 509 The length of the hash corresponding to the given key hasher type. 510 511 Raises: 512 ValueError: If the key hasher type is not supported. 513 514 Example: 515 >>> concat_hash_len("Blake2_128Concat") 516 16 517 """ 518 519 if key_hasher == "Blake2_128Concat": 520 return 16 521 elif key_hasher == "Twox64Concat": 522 return 8 523 elif key_hasher == "Identity": 524 return 0 525 else: 526 raise ValueError("Unsupported hash type") 527 528 assert len(response) == len(function_parameters) == len(prefix_list) 529 result_dict: dict[str, dict[Any, Any]] = {} 530 for res, fun_params_tuple, prefix in zip( 531 response, function_parameters, prefix_list 532 ): 533 if not res: 534 continue 535 res = res[0] 536 changes = res["changes"] # type: ignore 537 value_type, param_types, key_hashers, params, storage_function = ( 538 fun_params_tuple 539 ) 540 with self.get_conn(init=True) as substrate: 541 for item in changes: 542 # Determine type string 543 key_type_string: list[Any] = [] 544 for n in range(len(params), len(param_types)): 545 key_type_string.append( 546 f"[u8; {concat_hash_len(key_hashers[n])}]" 547 ) 548 key_type_string.append(param_types[n]) 549 550 item_key_obj = substrate.decode_scale( # type: ignore 551 type_string=f"({', '.join(key_type_string)})", 552 scale_bytes="0x" + item[0][len(prefix):], 553 return_scale_obj=True, 554 block_hash=block_hash, 555 ) 556 # strip key_hashers to use as item key 557 if len(param_types) - len(params) == 1: 558 item_key = item_key_obj.value_object[1] # type: ignore 559 else: 560 item_key = tuple( # type: ignore 561 item_key_obj.value_object[key + 1] # type: ignore 562 for key in range( # type: ignore 563 len(params), len(param_types) + 1, 2 564 ) 565 ) 566 567 item_value = substrate.decode_scale( # type: ignore 568 type_string=value_type, 569 scale_bytes=item[1], 570 return_scale_obj=True, 571 block_hash=block_hash, 572 ) 573 result_dict.setdefault(storage_function, {}) 574 key = get_item_key_value(item_key) # type: ignore 575 result_dict[storage_function][key] = item_value.value # type: ignore 576 577 return result_dict 578 579 def query_batch( 580 self, functions: dict[str, list[tuple[str, list[Any]]]] 581 ) -> dict[str, str]: 582 """ 583 Executes batch queries on a substrate and returns results in a dictionary format. 584 585 Args: 586 substrate: An instance of SubstrateInterface to interact with the substrate. 587 functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls (function name and parameters). 588 589 Returns: 590 A dictionary where keys are storage function names and values are the query results. 591 592 Raises: 593 Exception: If no result is found from the batch queries. 594 595 Example: 596 >>> query_batch(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) 597 {'function_name': 'query_result', ...} 598 """ 599 600 result: dict[str, str] = {} 601 if not functions: 602 raise Exception("No result") 603 with self.get_conn(init=True) as substrate: 604 for module, queries in functions.items(): 605 storage_keys: list[Any] = [] 606 for fn, params in queries: 607 storage_function = substrate.create_storage_key( # type: ignore 608 pallet=module, storage_function=fn, params=params 609 ) 610 storage_keys.append(storage_function) 611 612 block_hash = substrate.get_block_hash() 613 responses: list[Any] = substrate.query_multi( # type: ignore 614 storage_keys=storage_keys, block_hash=block_hash 615 ) 616 617 for item in responses: 618 fun = item[0] 619 query = item[1] 620 storage_fun = fun.storage_function 621 result[storage_fun] = query.value 622 623 return result 624 625 def query_batch_map( 626 self, 627 functions: dict[str, list[tuple[str, list[Any]]]], 628 block_hash: str | None = None, 629 ) -> dict[str, dict[Any, Any]]: 630 """ 631 Queries multiple storage functions using a map batch approach and returns the combined result. 632 633 Args: 634 substrate: An instance of SubstrateInterface for substrate interaction. 635 functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls. 636 637 Returns: 638 The combined result of the map batch query. 639 640 Example: 641 >>> query_batch_map(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) 642 # Returns the combined result of the map batch query 643 """ 644 multi_result: dict[str, dict[Any, Any]] = {} 645 646 def recursive_update( 647 d: dict[str, dict[T1, T2] | dict[str, Any]], 648 u: Mapping[str, dict[Any, Any] | str], 649 ) -> dict[str, dict[T1, T2]]: 650 for k, v in u.items(): 651 if isinstance(v, dict): 652 d[k] = recursive_update(d.get(k, {}), v) # type: ignore 653 else: 654 d[k] = v # type: ignore 655 return d # type: ignore 656 657 def get_page(): 658 send, prefix_list = self._get_storage_keys(storage, queries, block_hash) 659 with self.get_conn(init=True) as substrate: 660 function_parameters = self._get_lists(storage, queries, substrate) 661 responses = self._rpc_request_batch(send) 662 # assumption because send is just the storage_function keys 663 # so it should always be really small regardless of the amount of queries 664 assert len(responses) == 1 665 res = responses[0] 666 built_payload: list[tuple[str, list[Any]]] = [] 667 for result_keys in res: 668 built_payload.append( 669 ("state_queryStorageAt", [result_keys, block_hash]) 670 ) 671 _, chunks_info = self._make_request_smaller( 672 built_payload, prefix_list, function_parameters 673 ) 674 chunks_response, chunks_info = self._rpc_request_batch_chunked(chunks_info) 675 return chunks_response, chunks_info 676 677 if not block_hash: 678 with self.get_conn(init=True) as substrate: 679 block_hash = substrate.get_block_hash() 680 for storage, queries in functions.items(): 681 chunks, chunks_info = get_page() 682 # if this doesn't happen something is wrong on the code 683 # and we won't be able to decode the data properly 684 assert len(chunks) == len(chunks_info) 685 for chunk_info, response in zip(chunks_info, chunks): 686 storage_result = self._decode_response( 687 response, chunk_info.fun_params, chunk_info.prefix_list, block_hash 688 ) 689 multi_result = recursive_update(multi_result, storage_result) 690 691 return multi_result 692 693 def query( 694 self, 695 name: str, 696 params: list[Any] = [], 697 module: str = "SubspaceModule", 698 block_hash: str | None = None, 699 ) -> Any: 700 """ 701 Queries a storage function on the network. 702 703 Sends a query to the network and retrieves data from a 704 specified storage function. 705 706 Args: 707 name: The name of the storage function to query. 708 params: The parameters to pass to the storage function. 709 module: The module where the storage function is located. 710 711 Returns: 712 The result of the query from the network. 713 714 Raises: 715 NetworkQueryError: If the query fails or is invalid. 716 """ 717 718 result = self.query_batch({module: [(name, params)]}) 719 720 return result[name] 721 722 def query_map( 723 self, 724 name: str, 725 params: list[Any] = [], 726 module: str = "SubspaceModule", 727 extract_value: bool = True, 728 block_hash: str | None = None, 729 ) -> dict[Any, Any]: 730 """ 731 Queries a storage map from a network node. 732 733 Args: 734 name: The name of the storage map to query. 735 params: A list of parameters for the query. 736 module: The module in which the storage map is located. 737 738 Returns: 739 A dictionary representing the key-value pairs 740 retrieved from the storage map. 741 742 Raises: 743 QueryError: If the query to the network fails or is invalid. 744 """ 745 746 result = self.query_batch_map({module: [(name, params)]}, block_hash) 747 748 if extract_value: 749 return {k.value: v.value for k, v in result} # type: ignore 750 751 return result 752 753 def compose_call( 754 self, 755 fn: str, 756 params: dict[str, Any], 757 key: Keypair | None, 758 module: str = "SubspaceModule", 759 wait_for_inclusion: bool = True, 760 wait_for_finalization: bool | None = None, 761 sudo: bool = False, 762 unsigned: bool = False, 763 ) -> ExtrinsicReceipt: 764 """ 765 Composes and submits a call to the network node. 766 767 Composes and signs a call with the provided keypair, and submits it to 768 the network. The call can be a standard extrinsic or a sudo extrinsic if 769 elevated permissions are required. The method can optionally wait for 770 the call's inclusion in a block and/or its finalization. 771 772 Args: 773 fn: The function name to call on the network. 774 params: A dictionary of parameters for the call. 775 key: The keypair for signing the extrinsic. 776 module: The module containing the function. 777 wait_for_inclusion: Wait for the call's inclusion in a block. 778 wait_for_finalization: Wait for the transaction's finalization. 779 sudo: Execute the call as a sudo (superuser) operation. 780 781 Returns: 782 The receipt of the submitted extrinsic, if 783 `wait_for_inclusion` is True. Otherwise, returns a string 784 identifier of the extrinsic. 785 786 Raises: 787 ChainTransactionError: If the transaction fails. 788 """ 789 790 if key is None and not unsigned: 791 raise ValueError("Key must be provided for signed extrinsics.") 792 793 with self.get_conn() as substrate: 794 if wait_for_finalization is None: 795 wait_for_finalization = self.wait_for_finalization 796 797 call = substrate.compose_call( # type: ignore 798 call_module=module, call_function=fn, call_params=params 799 ) 800 if sudo: 801 call = substrate.compose_call( # type: ignore 802 call_module="Sudo", 803 call_function="sudo", 804 call_params={ 805 "call": call.value, # type: ignore 806 }, 807 ) 808 809 if not unsigned: 810 extrinsic = substrate.create_signed_extrinsic( # type: ignore 811 call=call, keypair=key # type: ignore 812 ) # type: ignore 813 else: 814 extrinsic = substrate.create_unsigned_extrinsic(call=call) # type: ignore 815 816 response = substrate.submit_extrinsic( 817 extrinsic=extrinsic, 818 wait_for_inclusion=wait_for_inclusion, 819 wait_for_finalization=wait_for_finalization, 820 ) 821 if wait_for_inclusion: 822 if not response.is_success: 823 raise ChainTransactionError( 824 response.error_message, response # type: ignore 825 ) 826 827 return response 828 829 def compose_call_multisig( 830 self, 831 fn: str, 832 params: dict[str, Any], 833 key: Keypair, 834 signatories: list[Ss58Address], 835 threshold: int, 836 module: str = "SubspaceModule", 837 wait_for_inclusion: bool = True, 838 wait_for_finalization: bool | None = None, 839 sudo: bool = False, 840 era: dict[str, int] | None = None, 841 ) -> ExtrinsicReceipt: 842 """ 843 Composes and submits a multisignature call to the network node. 844 845 This method allows the composition and submission of a call that 846 requires multiple signatures for execution, known as a multisignature 847 call. It supports specifying signatories, a threshold of signatures for 848 the call's execution, and an optional era for the call's mortality. The 849 call can be a standard extrinsic, a sudo extrinsic for elevated 850 permissions, or a multisig extrinsic if multiple signatures are 851 required. Optionally, the method can wait for the call's inclusion in a 852 block and/or its finalization. Make sure to pass all keys, 853 that are part of the multisignature. 854 855 Args: 856 fn: The function name to call on the network. params: A dictionary 857 of parameters for the call. key: The keypair for signing the 858 extrinsic. signatories: List of SS58 addresses of the signatories. 859 Include ALL KEYS that are part of the multisig. threshold: The 860 minimum number of signatories required to execute the extrinsic. 861 module: The module containing the function to call. 862 wait_for_inclusion: Whether to wait for the call's inclusion in a 863 block. wait_for_finalization: Whether to wait for the transaction's 864 finalization. sudo: Execute the call as a sudo (superuser) 865 operation. era: Specifies the call's mortality in terms of blocks in 866 the format 867 {'period': amount_blocks}. If omitted, the extrinsic is 868 immortal. 869 870 Returns: 871 The receipt of the submitted extrinsic if `wait_for_inclusion` is 872 True. Otherwise, returns a string identifier of the extrinsic. 873 874 Raises: 875 ChainTransactionError: If the transaction fails. 876 """ 877 878 # getting the call ready 879 with self.get_conn() as substrate: 880 if wait_for_finalization is None: 881 wait_for_finalization = self.wait_for_finalization 882 883 # prepares the `GenericCall` object 884 call = substrate.compose_call( # type: ignore 885 call_module=module, call_function=fn, call_params=params 886 ) 887 if sudo: 888 call = substrate.compose_call( # type: ignore 889 call_module="Sudo", 890 call_function="sudo", 891 call_params={ 892 "call": call.value, # type: ignore 893 }, 894 ) 895 896 # modify the rpc methods at runtime, to allow for correct payment 897 # fee calculation parity has a bug in this version, 898 # where the method has to be removed 899 rpc_methods = substrate.config.get("rpc_methods") # type: ignore 900 901 if "state_call" in rpc_methods: # type: ignore 902 rpc_methods.remove("state_call") # type: ignore 903 904 # create the multisig account 905 multisig_acc = substrate.generate_multisig_account( # type: ignore 906 signatories, threshold 907 ) 908 909 # send the multisig extrinsic 910 extrinsic = substrate.create_multisig_extrinsic( # type: ignore 911 call=call, # type: ignore 912 keypair=key, 913 multisig_account=multisig_acc, # type: ignore 914 era=era, # type: ignore 915 ) # type: ignore 916 917 response = substrate.submit_extrinsic( 918 extrinsic=extrinsic, 919 wait_for_inclusion=wait_for_inclusion, 920 wait_for_finalization=wait_for_finalization, 921 ) 922 923 if wait_for_inclusion: 924 if not response.is_success: 925 raise ChainTransactionError( 926 response.error_message, response # type: ignore 927 ) 928 929 return response 930 931 def transfer( 932 self, 933 key: Keypair, 934 amount: int, 935 dest: Ss58Address, 936 ) -> ExtrinsicReceipt: 937 """ 938 Transfers a specified amount of tokens from the signer's account to the 939 specified account. 940 941 Args: 942 key: The keypair associated with the sender's account. 943 amount: The amount to transfer, in nanotokens. 944 dest: The SS58 address of the recipient. 945 946 Returns: 947 A receipt of the transaction. 948 949 Raises: 950 InsufficientBalanceError: If the sender's account does not have 951 enough balance. 952 ChainTransactionError: If the transaction fails. 953 """ 954 955 params = {"dest": dest, "value": amount} 956 957 return self.compose_call( 958 module="Balances", fn="transfer_keep_alive", params=params, key=key 959 ) 960 961 def transfer_multiple( 962 self, 963 key: Keypair, 964 destinations: list[Ss58Address], 965 amounts: list[int], 966 netuid: str | int = 0, 967 ) -> ExtrinsicReceipt: 968 """ 969 Transfers specified amounts of tokens from the signer's account to 970 multiple target accounts. 971 972 The `destinations` and `amounts` lists must be of the same length. 973 974 Args: 975 key: The keypair associated with the sender's account. 976 destinations: A list of SS58 addresses of the recipients. 977 amounts: Amount to transfer to each recipient, in nanotokens. 978 netuid: The network identifier. 979 980 Returns: 981 A receipt of the transaction. 982 983 Raises: 984 InsufficientBalanceError: If the sender's account does not have 985 enough balance for all transfers. 986 ChainTransactionError: If the transaction fails. 987 """ 988 989 assert len(destinations) == len(amounts) 990 991 # extract existential deposit from amounts 992 existential_deposit = self.get_existential_deposit() 993 amounts = [a - existential_deposit for a in amounts] 994 995 params = { 996 "netuid": netuid, 997 "destinations": destinations, 998 "amounts": amounts, 999 } 1000 1001 return self.compose_call( 1002 module="SubspaceModule", fn="transfer_multiple", params=params, key=key 1003 ) 1004 1005 def stake( 1006 self, 1007 key: Keypair, 1008 amount: int, 1009 dest: Ss58Address, 1010 ) -> ExtrinsicReceipt: 1011 """ 1012 Stakes the specified amount of tokens to a module key address. 1013 1014 Args: 1015 key: The keypair associated with the staker's account. 1016 amount: The amount of tokens to stake, in nanotokens. 1017 dest: The SS58 address of the module key to stake to. 1018 netuid: The network identifier. 1019 1020 Returns: 1021 A receipt of the staking transaction. 1022 1023 Raises: 1024 InsufficientBalanceError: If the staker's account does not have 1025 enough balance. 1026 ChainTransactionError: If the transaction fails. 1027 """ 1028 1029 params = {"amount": amount, "module_key": dest} 1030 1031 return self.compose_call(fn="add_stake", params=params, key=key) 1032 1033 def unstake( 1034 self, 1035 key: Keypair, 1036 amount: int, 1037 dest: Ss58Address, 1038 ) -> ExtrinsicReceipt: 1039 """ 1040 Unstakes the specified amount of tokens from a module key address. 1041 1042 Args: 1043 key: The keypair associated with the unstaker's account. 1044 amount: The amount of tokens to unstake, in nanotokens. 1045 dest: The SS58 address of the module key to unstake from. 1046 netuid: The network identifier. 1047 1048 Returns: 1049 A receipt of the unstaking transaction. 1050 1051 Raises: 1052 InsufficientStakeError: If the staked key does not have enough 1053 staked tokens by the signer key. 1054 ChainTransactionError: If the transaction fails. 1055 """ 1056 1057 params = {"amount": amount, "module_key": dest} 1058 return self.compose_call(fn="remove_stake", params=params, key=key) 1059 1060 def update_module( 1061 self, 1062 key: Keypair, 1063 name: str, 1064 address: str, 1065 metadata: str | None = None, 1066 delegation_fee: int = 20, 1067 netuid: int = 0, 1068 ) -> ExtrinsicReceipt: 1069 """ 1070 Updates the parameters of a registered module. 1071 1072 The delegation fee must be an integer between 0 and 100. 1073 1074 Args: 1075 key: The keypair associated with the module's account. 1076 name: The new name for the module. If None, the name is not updated. 1077 address: The new address for the module. 1078 If None, the address is not updated. 1079 delegation_fee: The new delegation fee for the module, 1080 between 0 and 100. 1081 netuid: The network identifier. 1082 1083 Returns: 1084 A receipt of the module update transaction. 1085 1086 Raises: 1087 InvalidParameterError: If the provided parameters are invalid. 1088 ChainTransactionError: If the transaction fails. 1089 """ 1090 1091 assert isinstance(delegation_fee, int) 1092 1093 params = { 1094 "netuid": netuid, 1095 "name": name, 1096 "address": address, 1097 "delegation_fee": delegation_fee, 1098 "metadata": metadata, 1099 } 1100 1101 response = self.compose_call("update_module", params=params, key=key) 1102 1103 return response 1104 1105 def register_module( 1106 self, 1107 key: Keypair, 1108 name: str, 1109 address: str | None = None, 1110 subnet: str = "Rootnet", 1111 metadata: str | None = None, 1112 ) -> ExtrinsicReceipt: 1113 """ 1114 Registers a new module in the network. 1115 1116 Args: 1117 key: The keypair used for registering the module. 1118 name: The name of the module. If None, a default or previously 1119 set name is used. # How does this work? 1120 address: The address of the module. If None, a default or 1121 previously set address is used. # How does this work? 1122 subnet: The network subnet to register the module in. 1123 min_stake: The minimum stake required for the module, in nanotokens. 1124 If None, a default value is used. 1125 1126 Returns: 1127 A receipt of the registration transaction. 1128 1129 Raises: 1130 InvalidParameterError: If the provided parameters are invalid. 1131 ChainTransactionError: If the transaction fails. 1132 """ 1133 1134 key_addr = key.ss58_address 1135 1136 params = { 1137 "network_name": subnet, 1138 "address": address, 1139 "name": name, 1140 "module_key": key_addr, 1141 "metadata": metadata, 1142 } 1143 1144 response = self.compose_call("register", params=params, key=key) 1145 return response 1146 1147 def deregister_module(self, key: Keypair, netuid: int) -> ExtrinsicReceipt: 1148 """ 1149 Deregisters a module from the network. 1150 1151 Args: 1152 key: The keypair associated with the module's account. 1153 netuid: The network identifier. 1154 1155 Returns: 1156 A receipt of the module deregistration transaction. 1157 1158 Raises: 1159 ChainTransactionError: If the transaction fails. 1160 """ 1161 1162 params = {"netuid": netuid} 1163 1164 response = self.compose_call("deregister", params=params, key=key) 1165 1166 return response 1167 1168 def register_subnet(self, key: Keypair, name: str, metadata: str | None = None) -> ExtrinsicReceipt: 1169 """ 1170 Registers a new subnet in the network. 1171 1172 Args: 1173 key (Keypair): The keypair used for registering the subnet. 1174 name (str): The name of the subnet to be registered. 1175 metadata (str | None, optional): Additional metadata for the subnet. Defaults to None. 1176 1177 Returns: 1178 ExtrinsicReceipt: A receipt of the subnet registration transaction. 1179 1180 Raises: 1181 ChainTransactionError: If the transaction fails. 1182 """ 1183 1184 params = { 1185 "name": name, 1186 "metadata": metadata, 1187 } 1188 1189 response = self.compose_call("register_subnet", params=params, key=key) 1190 1191 return response 1192 1193 def vote( 1194 self, 1195 key: Keypair, 1196 uids: list[int], 1197 weights: list[int], 1198 netuid: int = 0, 1199 ) -> ExtrinsicReceipt: 1200 """ 1201 Casts votes on a list of module UIDs with corresponding weights. 1202 1203 The length of the UIDs list and the weights list should be the same. 1204 Each weight corresponds to the UID at the same index. 1205 1206 Args: 1207 key: The keypair used for signing the vote transaction. 1208 uids: A list of module UIDs to vote on. 1209 weights: A list of weights corresponding to each UID. 1210 netuid: The network identifier. 1211 1212 Returns: 1213 A receipt of the voting transaction. 1214 1215 Raises: 1216 InvalidParameterError: If the lengths of UIDs and weights lists 1217 do not match. 1218 ChainTransactionError: If the transaction fails. 1219 """ 1220 1221 assert len(uids) == len(weights) 1222 1223 params = { 1224 "uids": uids, 1225 "weights": weights, 1226 "netuid": netuid, 1227 } 1228 1229 response = self.compose_call("set_weights", params=params, key=key) 1230 1231 return response 1232 1233 def update_subnet( 1234 self, 1235 key: Keypair, 1236 params: SubnetParams, 1237 netuid: int = 0, 1238 ) -> ExtrinsicReceipt: 1239 """ 1240 Update a subnet's configuration. 1241 1242 It requires the founder key for authorization. 1243 1244 Args: 1245 key: The founder keypair of the subnet. 1246 params: The new parameters for the subnet. 1247 netuid: The network identifier. 1248 1249 Returns: 1250 A receipt of the subnet update transaction. 1251 1252 Raises: 1253 AuthorizationError: If the key is not authorized. 1254 ChainTransactionError: If the transaction fails. 1255 """ 1256 1257 general_params = dict(params) 1258 general_params["netuid"] = netuid 1259 if "metadata" not in general_params: 1260 general_params["metadata"] = None 1261 1262 response = self.compose_call( 1263 fn="update_subnet", 1264 params=general_params, 1265 key=key, 1266 ) 1267 1268 return response 1269 1270 def transfer_stake( 1271 self, 1272 key: Keypair, 1273 amount: int, 1274 from_module_key: Ss58Address, 1275 dest_module_address: Ss58Address, 1276 ) -> ExtrinsicReceipt: 1277 """ 1278 Realocate staked tokens from one staked module to another module. 1279 1280 Args: 1281 key: The keypair associated with the account that is delegating the tokens. 1282 amount: The amount of staked tokens to transfer, in nanotokens. 1283 from_module_key: The SS58 address of the module you want to transfer from (currently delegated by the key). 1284 dest_module_address: The SS58 address of the destination (newly delegated key). 1285 netuid: The network identifier. 1286 1287 Returns: 1288 A receipt of the stake transfer transaction. 1289 1290 Raises: 1291 InsufficientStakeError: If the source module key does not have 1292 enough staked tokens. ChainTransactionError: If the transaction 1293 fails. 1294 """ 1295 1296 amount = amount - self.get_existential_deposit() 1297 1298 params = { 1299 "amount": amount, 1300 "module_key": from_module_key, 1301 "new_module_key": dest_module_address, 1302 } 1303 1304 response = self.compose_call("transfer_stake", key=key, params=params) 1305 1306 return response 1307 1308 def multiunstake( 1309 self, 1310 key: Keypair, 1311 keys: list[Ss58Address], 1312 amounts: list[int], 1313 ) -> ExtrinsicReceipt: 1314 """ 1315 Unstakes tokens from multiple module keys. 1316 1317 And the lists `keys` and `amounts` must be of the same length. Each 1318 amount corresponds to the module key at the same index. 1319 1320 Args: 1321 key: The keypair associated with the unstaker's account. 1322 keys: A list of SS58 addresses of the module keys to unstake from. 1323 amounts: A list of amounts to unstake from each module key, 1324 in nanotokens. 1325 netuid: The network identifier. 1326 1327 Returns: 1328 A receipt of the multi-unstaking transaction. 1329 1330 Raises: 1331 MismatchedLengthError: If the lengths of keys and amounts lists do 1332 not match. InsufficientStakeError: If any of the module keys do not 1333 have enough staked tokens. ChainTransactionError: If the transaction 1334 fails. 1335 """ 1336 1337 assert len(keys) == len(amounts) 1338 1339 params = {"module_keys": keys, "amounts": amounts} 1340 1341 response = self.compose_call("remove_stake_multiple", params=params, key=key) 1342 1343 return response 1344 1345 def multistake( 1346 self, 1347 key: Keypair, 1348 keys: list[Ss58Address], 1349 amounts: list[int], 1350 ) -> ExtrinsicReceipt: 1351 """ 1352 Stakes tokens to multiple module keys. 1353 1354 The lengths of the `keys` and `amounts` lists must be the same. Each 1355 amount corresponds to the module key at the same index. 1356 1357 Args: 1358 key: The keypair associated with the staker's account. 1359 keys: A list of SS58 addresses of the module keys to stake to. 1360 amounts: A list of amounts to stake to each module key, 1361 in nanotokens. 1362 netuid: The network identifier. 1363 1364 Returns: 1365 A receipt of the multi-staking transaction. 1366 1367 Raises: 1368 MismatchedLengthError: If the lengths of keys and amounts lists 1369 do not match. 1370 ChainTransactionError: If the transaction fails. 1371 """ 1372 1373 assert len(keys) == len(amounts) 1374 1375 params = { 1376 "module_keys": keys, 1377 "amounts": amounts, 1378 } 1379 1380 response = self.compose_call("add_stake_multiple", params=params, key=key) 1381 1382 return response 1383 1384 def add_profit_shares( 1385 self, 1386 key: Keypair, 1387 keys: list[Ss58Address], 1388 shares: list[int], 1389 ) -> ExtrinsicReceipt: 1390 """ 1391 Allocates profit shares to multiple keys. 1392 1393 The lists `keys` and `shares` must be of the same length, 1394 with each share amount corresponding to the key at the same index. 1395 1396 Args: 1397 key: The keypair associated with the account 1398 distributing the shares. 1399 keys: A list of SS58 addresses to allocate shares to. 1400 shares: A list of share amounts to allocate to each key, 1401 in nanotokens. 1402 1403 Returns: 1404 A receipt of the profit sharing transaction. 1405 1406 Raises: 1407 MismatchedLengthError: If the lengths of keys and shares 1408 lists do not match. 1409 ChainTransactionError: If the transaction fails. 1410 """ 1411 1412 assert len(keys) == len(shares) 1413 1414 params = {"keys": keys, "shares": shares} 1415 1416 response = self.compose_call("add_profit_shares", params=params, key=key) 1417 1418 return response 1419 1420 def add_subnet_proposal( 1421 self, key: Keypair, 1422 params: dict[str, Any], 1423 ipfs: str, 1424 netuid: int = 0 1425 ) -> ExtrinsicReceipt: 1426 """ 1427 Submits a proposal for creating or modifying a subnet within the 1428 network. 1429 1430 The proposal includes various parameters like the name, founder, share 1431 allocations, and other subnet-specific settings. 1432 1433 Args: 1434 key: The keypair used for signing the proposal transaction. 1435 params: The parameters for the subnet proposal. 1436 netuid: The network identifier. 1437 1438 Returns: 1439 A receipt of the subnet proposal transaction. 1440 1441 Raises: 1442 InvalidParameterError: If the provided subnet 1443 parameters are invalid. 1444 ChainTransactionError: If the transaction fails. 1445 """ 1446 1447 general_params = dict(params) 1448 general_params["netuid"] = netuid 1449 general_params["data"] = ipfs 1450 if "metadata" not in general_params: 1451 general_params["metadata"] = None 1452 1453 # general_params["burn_config"] = json.dumps(general_params["burn_config"]) 1454 response = self.compose_call( 1455 fn="add_subnet_params_proposal", 1456 params=general_params, 1457 key=key, 1458 module="GovernanceModule", 1459 ) 1460 1461 return response 1462 1463 def add_custom_proposal( 1464 self, 1465 key: Keypair, 1466 cid: str, 1467 ) -> ExtrinsicReceipt: 1468 1469 params = {"data": cid} 1470 1471 response = self.compose_call( 1472 fn="add_global_custom_proposal", 1473 params=params, 1474 key=key, 1475 module="GovernanceModule", 1476 ) 1477 return response 1478 1479 def add_custom_subnet_proposal( 1480 self, 1481 key: Keypair, 1482 cid: str, 1483 netuid: int = 0, 1484 ) -> ExtrinsicReceipt: 1485 """ 1486 Submits a proposal for creating or modifying a custom subnet within the 1487 network. 1488 1489 The proposal includes various parameters like the name, founder, share 1490 allocations, and other subnet-specific settings. 1491 1492 Args: 1493 key: The keypair used for signing the proposal transaction. 1494 params: The parameters for the subnet proposal. 1495 netuid: The network identifier. 1496 1497 Returns: 1498 A receipt of the subnet proposal transaction. 1499 """ 1500 1501 params = { 1502 "data": cid, 1503 "netuid": netuid, 1504 } 1505 1506 response = self.compose_call( 1507 fn="add_subnet_custom_proposal", 1508 params=params, 1509 key=key, 1510 module="GovernanceModule", 1511 ) 1512 1513 return response 1514 1515 def add_global_proposal( 1516 self, 1517 key: Keypair, 1518 params: NetworkParams, 1519 cid: str | None, 1520 ) -> ExtrinsicReceipt: 1521 """ 1522 Submits a proposal for altering the global network parameters. 1523 1524 Allows for the submission of a proposal to 1525 change various global parameters 1526 of the network, such as emission rates, rate limits, and voting 1527 thresholds. It is used to 1528 suggest changes that affect the entire network's operation. 1529 1530 Args: 1531 key: The keypair used for signing the proposal transaction. 1532 params: A dictionary containing global network parameters 1533 like maximum allowed subnets, modules, 1534 transaction rate limits, and others. 1535 1536 Returns: 1537 A receipt of the global proposal transaction. 1538 1539 Raises: 1540 InvalidParameterError: If the provided network 1541 parameters are invalid. 1542 ChainTransactionError: If the transaction fails. 1543 """ 1544 general_params = cast(dict[str, Any], params) 1545 cid = cid or "" 1546 general_params["data"] = cid 1547 1548 response = self.compose_call( 1549 fn="add_global_params_proposal", 1550 params=general_params, 1551 key=key, 1552 module="GovernanceModule", 1553 ) 1554 1555 return response 1556 1557 def vote_on_proposal( 1558 self, 1559 key: Keypair, 1560 proposal_id: int, 1561 agree: bool, 1562 ) -> ExtrinsicReceipt: 1563 """ 1564 Casts a vote on a specified proposal within the network. 1565 1566 Args: 1567 key: The keypair used for signing the vote transaction. 1568 proposal_id: The unique identifier of the proposal to vote on. 1569 1570 Returns: 1571 A receipt of the voting transaction in nanotokens. 1572 1573 Raises: 1574 InvalidProposalIDError: If the provided proposal ID does not 1575 exist or is invalid. 1576 ChainTransactionError: If the transaction fails. 1577 """ 1578 1579 params = {"proposal_id": proposal_id, "agree": agree} 1580 1581 response = self.compose_call( 1582 "vote_proposal", 1583 key=key, 1584 params=params, 1585 module="GovernanceModule", 1586 ) 1587 1588 return response 1589 1590 def unvote_on_proposal( 1591 self, 1592 key: Keypair, 1593 proposal_id: int, 1594 ) -> ExtrinsicReceipt: 1595 """ 1596 Retracts a previously cast vote on a specified proposal. 1597 1598 Args: 1599 key: The keypair used for signing the unvote transaction. 1600 proposal_id: The unique identifier of the proposal to withdraw the 1601 vote from. 1602 1603 Returns: 1604 A receipt of the unvoting transaction in nanotokens. 1605 1606 Raises: 1607 InvalidProposalIDError: If the provided proposal ID does not 1608 exist or is invalid. 1609 ChainTransactionError: If the transaction fails to be processed, or 1610 if there was no prior vote to retract. 1611 """ 1612 1613 params = {"proposal_id": proposal_id} 1614 1615 response = self.compose_call( 1616 "remove_vote_proposal", 1617 key=key, 1618 params=params, 1619 module="GovernanceModule", 1620 ) 1621 1622 return response 1623 1624 def enable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1625 """ 1626 Enables vote power delegation for the signer's account. 1627 1628 Args: 1629 key: The keypair used for signing the delegation transaction. 1630 1631 Returns: 1632 A receipt of the vote power delegation transaction. 1633 1634 Raises: 1635 ChainTransactionError: If the transaction fails. 1636 """ 1637 1638 response = self.compose_call( 1639 "enable_vote_power_delegation", 1640 params={}, 1641 key=key, 1642 module="GovernanceModule", 1643 ) 1644 1645 return response 1646 1647 def disable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1648 """ 1649 Disables vote power delegation for the signer's account. 1650 1651 Args: 1652 key: The keypair used for signing the delegation transaction. 1653 1654 Returns: 1655 A receipt of the vote power delegation transaction. 1656 1657 Raises: 1658 ChainTransactionError: If the transaction fails. 1659 """ 1660 1661 response = self.compose_call( 1662 "disable_vote_power_delegation", 1663 params={}, 1664 key=key, 1665 module="GovernanceModule", 1666 ) 1667 1668 return response 1669 1670 def add_dao_application( 1671 self, key: Keypair, application_key: Ss58Address, data: str 1672 ) -> ExtrinsicReceipt: 1673 """ 1674 Submits a new application to the general subnet DAO. 1675 1676 Args: 1677 key: The keypair used for signing the application transaction. 1678 application_key: The SS58 address of the application key. 1679 data: The data associated with the application. 1680 1681 Returns: 1682 A receipt of the application transaction. 1683 1684 Raises: 1685 ChainTransactionError: If the transaction fails. 1686 """ 1687 1688 params = {"application_key": application_key, "data": data} 1689 1690 response = self.compose_call( 1691 "add_dao_application", module="GovernanceModule", key=key, 1692 params=params 1693 ) 1694 1695 return response 1696 1697 def query_map_curator_applications(self) -> dict[str, dict[str, str]]: 1698 query_result = self.query_map( 1699 "CuratorApplications", module="GovernanceModule", params=[], 1700 extract_value=False 1701 ) 1702 applications = query_result.get("CuratorApplications", {}) 1703 return applications 1704 1705 def query_map_proposals( 1706 self, extract_value: bool = False 1707 ) -> dict[int, dict[str, Any]]: 1708 """ 1709 Retrieves a mappping of proposals from the network. 1710 1711 Queries the network and returns a mapping of proposal IDs to 1712 their respective parameters. 1713 1714 Returns: 1715 A dictionary mapping proposal IDs 1716 to dictionaries of their parameters. 1717 1718 Raises: 1719 QueryError: If the query to the network fails or is invalid. 1720 """ 1721 1722 return self.query_map( 1723 "Proposals", extract_value=extract_value, module="GovernanceModule" 1724 )["Proposals"] 1725 1726 def query_map_weights( 1727 self, netuid: int = 0, extract_value: bool = False 1728 ) -> dict[int, list[tuple[int, int]]] | None: 1729 """ 1730 Retrieves a mapping of weights for keys on the network. 1731 1732 Queries the network and returns a mapping of key UIDs to 1733 their respective weights. 1734 1735 Args: 1736 netuid: The network UID from which to get the weights. 1737 1738 Returns: 1739 A dictionary mapping key UIDs to lists of their weights. 1740 1741 Raises: 1742 QueryError: If the query to the network fails or is invalid. 1743 """ 1744 1745 weights_dict = self.query_map( 1746 "Weights", 1747 [netuid], 1748 extract_value=extract_value 1749 ).get("Weights") 1750 return weights_dict 1751 1752 def query_map_key( 1753 self, 1754 netuid: int = 0, 1755 extract_value: bool = False, 1756 ) -> dict[int, Ss58Address]: 1757 """ 1758 Retrieves a map of keys from the network. 1759 1760 Fetches a mapping of key UIDs to their associated 1761 addresses on the network. 1762 The query can be targeted at a specific network UID if required. 1763 1764 Args: 1765 netuid: The network UID from which to get the keys. 1766 1767 Returns: 1768 A dictionary mapping key UIDs to their addresses. 1769 1770 Raises: 1771 QueryError: If the query to the network fails or is invalid. 1772 """ 1773 return self.query_map("Keys", [netuid], extract_value=extract_value)["Keys"] 1774 1775 def query_map_address( 1776 self, netuid: int = 0, extract_value: bool = False 1777 ) -> dict[int, str]: 1778 """ 1779 Retrieves a map of key addresses from the network. 1780 1781 Queries the network for a mapping of key UIDs to their addresses. 1782 1783 Args: 1784 netuid: The network UID from which to get the addresses. 1785 1786 Returns: 1787 A dictionary mapping key UIDs to their addresses. 1788 1789 Raises: 1790 QueryError: If the query to the network fails or is invalid. 1791 """ 1792 1793 return self.query_map("Address", [netuid], extract_value=extract_value)[ 1794 "Address" 1795 ] 1796 1797 def query_map_emission(self, extract_value: bool = False) -> dict[int, list[int]]: 1798 """ 1799 Retrieves a map of emissions for keys on the network. 1800 1801 Queries the network to get a mapping of 1802 key UIDs to their emission values. 1803 1804 Returns: 1805 A dictionary mapping key UIDs to lists of their emission values. 1806 1807 Raises: 1808 QueryError: If the query to the network fails or is invalid. 1809 """ 1810 1811 return self.query_map("Emission", extract_value=extract_value)["Emission"] 1812 1813 def query_map_pending_emission(self, extract_value: bool = False) -> int: 1814 """ 1815 Retrieves a map of pending emissions for the subnets. 1816 1817 Queries the network for a mapping of subnet UIDs to their pending emission values. 1818 1819 Returns: 1820 A dictionary mapping subnet UIDs to their pending emission values. 1821 1822 Raises: 1823 QueryError: If the query to the network fails or is invalid. 1824 """ 1825 return self.query_map("PendingEmission", extract_value=extract_value, module="SubnetEmissionModule")["PendingEmission"] 1826 1827 def query_map_subnet_emission(self, extract_value: bool = False) -> dict[int, int]: 1828 """ 1829 Retrieves a map of subnet emissions for the network. 1830 1831 Queries the network for a mapping of subnet UIDs to their emission values. 1832 1833 Returns: 1834 A dictionary mapping subnet UIDs to their emission values. 1835 1836 Raises: 1837 QueryError: If the query to the network fails or is invalid. 1838 """ 1839 1840 return self.query_map("SubnetEmission", extract_value=extract_value, module="SubnetEmissionModule")["SubnetEmission"] 1841 1842 def query_map_subnet_consensus(self, extract_value: bool = False) -> dict[int, str]: 1843 """ 1844 Retrieves a map of subnet consensus types for the network. 1845 1846 Queries the network for a mapping of subnet UIDs to their consensus types. 1847 1848 Returns: 1849 A dictionary mapping subnet UIDs to their consensus types. 1850 1851 Raises: 1852 QueryError: If the query to the network fails or is invalid. 1853 """ 1854 1855 return self.query_map("SubnetConsensusType", extract_value=extract_value, module="SubnetEmissionModule")["SubnetConsensusType"] 1856 1857 def query_map_incentive(self, extract_value: bool = False) -> dict[int, list[int]]: 1858 """ 1859 Retrieves a mapping of incentives for keys on the network. 1860 1861 Queries the network and returns a mapping of key UIDs to 1862 their respective incentive values. 1863 1864 Returns: 1865 A dictionary mapping key UIDs to lists of their incentive values. 1866 1867 Raises: 1868 QueryError: If the query to the network fails or is invalid. 1869 """ 1870 1871 return self.query_map("Incentive", extract_value=extract_value)["Incentive"] 1872 1873 def query_map_dividend(self, extract_value: bool = False) -> dict[int, list[int]]: 1874 """ 1875 Retrieves a mapping of dividends for keys on the network. 1876 1877 Queries the network for a mapping of key UIDs to 1878 their dividend values. 1879 1880 Returns: 1881 A dictionary mapping key UIDs to lists of their dividend values. 1882 1883 Raises: 1884 QueryError: If the query to the network fails or is invalid. 1885 """ 1886 1887 return self.query_map("Dividends", extract_value=extract_value)["Dividends"] 1888 1889 def query_map_regblock( 1890 self, netuid: int = 0, extract_value: bool = False 1891 ) -> dict[int, int]: 1892 """ 1893 Retrieves a mapping of registration blocks for keys on the network. 1894 1895 Queries the network for a mapping of key UIDs to 1896 the blocks where they were registered. 1897 1898 Args: 1899 netuid: The network UID from which to get the registration blocks. 1900 1901 Returns: 1902 A dictionary mapping key UIDs to their registration blocks. 1903 1904 Raises: 1905 QueryError: If the query to the network fails or is invalid. 1906 """ 1907 1908 return self.query_map( 1909 "RegistrationBlock", [netuid], extract_value=extract_value 1910 )["RegistrationBlock"] 1911 1912 def query_map_lastupdate(self, extract_value: bool = False) -> dict[int, list[int]]: 1913 """ 1914 Retrieves a mapping of the last update times for keys on the network. 1915 1916 Queries the network for a mapping of key UIDs to their last update times. 1917 1918 Returns: 1919 A dictionary mapping key UIDs to lists of their last update times. 1920 1921 Raises: 1922 QueryError: If the query to the network fails or is invalid. 1923 """ 1924 1925 return self.query_map("LastUpdate", extract_value=extract_value)["LastUpdate"] 1926 1927 def query_map_stakefrom( 1928 self, extract_value: bool = False 1929 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1930 """ 1931 Retrieves a mapping of stakes from various sources for keys on the network. 1932 1933 Queries the network to obtain a mapping of key addresses to the sources 1934 and amounts of stakes they have received. 1935 1936 Args: 1937 netuid: The network UID from which to get the stakes. 1938 1939 Returns: 1940 A dictionary mapping key addresses to lists of tuples 1941 (module_key_address, amount). 1942 1943 Raises: 1944 QueryError: If the query to the network fails or is invalid. 1945 """ 1946 1947 result = self.query_map("StakeFrom", [], extract_value=extract_value)[ 1948 "StakeFrom" 1949 ] 1950 1951 return transform_stake_dmap(result) 1952 1953 def query_map_staketo( 1954 self, extract_value: bool = False 1955 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1956 """ 1957 Retrieves a mapping of stakes to destinations for keys on the network. 1958 1959 Queries the network for a mapping of key addresses to the destinations 1960 and amounts of stakes they have made. 1961 1962 Args: 1963 netuid: The network UID from which to get the stakes. 1964 1965 Returns: 1966 A dictionary mapping key addresses to lists of tuples 1967 (module_key_address, amount). 1968 1969 Raises: 1970 QueryError: If the query to the network fails or is invalid. 1971 """ 1972 1973 result = self.query_map("StakeTo", [], extract_value=extract_value)[ 1974 "StakeTo" 1975 ] 1976 return transform_stake_dmap(result) 1977 1978 def query_map_delegationfee( 1979 self, netuid: int = 0, extract_value: bool = False 1980 ) -> dict[str, int]: 1981 """ 1982 Retrieves a mapping of delegation fees for keys on the network. 1983 1984 Queries the network to obtain a mapping of key addresses to their 1985 respective delegation fees. 1986 1987 Args: 1988 netuid: The network UID to filter the delegation fees. 1989 1990 Returns: 1991 A dictionary mapping key addresses to their delegation fees. 1992 1993 Raises: 1994 QueryError: If the query to the network fails or is invalid. 1995 """ 1996 1997 return self.query_map("DelegationFee", [netuid], extract_value=extract_value)[ 1998 "DelegationFee" 1999 ] 2000 2001 def query_map_tempo(self, extract_value: bool = False) -> dict[int, int]: 2002 """ 2003 Retrieves a mapping of tempo settings for the network. 2004 2005 Queries the network to obtain the tempo (rate of reward distributions) 2006 settings for various network subnets. 2007 2008 Returns: 2009 A dictionary mapping network UIDs to their tempo settings. 2010 2011 Raises: 2012 QueryError: If the query to the network fails or is invalid. 2013 """ 2014 2015 return self.query_map("Tempo", extract_value=extract_value)["Tempo"] 2016 2017 def query_map_immunity_period(self, extract_value: bool) -> dict[int, int]: 2018 """ 2019 Retrieves a mapping of immunity periods for the network. 2020 2021 Queries the network for the immunity period settings, 2022 which represent the time duration during which modules 2023 can not get deregistered. 2024 2025 Returns: 2026 A dictionary mapping network UIDs to their immunity period settings. 2027 2028 Raises: 2029 QueryError: If the query to the network fails or is invalid. 2030 """ 2031 2032 return self.query_map("ImmunityPeriod", extract_value=extract_value)[ 2033 "ImmunityPeriod" 2034 ] 2035 2036 def query_map_min_allowed_weights( 2037 self, extract_value: bool = False 2038 ) -> dict[int, int]: 2039 """ 2040 Retrieves a mapping of minimum allowed weights for the network. 2041 2042 Queries the network to obtain the minimum allowed weights, 2043 which are the lowest permissible weight values that can be set by 2044 validators. 2045 2046 Returns: 2047 A dictionary mapping network UIDs to 2048 their minimum allowed weight values. 2049 2050 Raises: 2051 QueryError: If the query to the network fails or is invalid. 2052 """ 2053 2054 return self.query_map("MinAllowedWeights", extract_value=extract_value)[ 2055 "MinAllowedWeights" 2056 ] 2057 2058 def query_map_max_allowed_weights( 2059 self, extract_value: bool = False 2060 ) -> dict[int, int]: 2061 """ 2062 Retrieves a mapping of maximum allowed weights for the network. 2063 2064 Queries the network for the maximum allowed weights, 2065 which are the highest permissible 2066 weight values that can be set by validators. 2067 2068 Returns: 2069 A dictionary mapping network UIDs to 2070 their maximum allowed weight values. 2071 2072 Raises: 2073 QueryError: If the query to the network fails or is invalid. 2074 """ 2075 2076 return self.query_map("MaxAllowedWeights", extract_value=extract_value)[ 2077 "MaxAllowedWeights" 2078 ] 2079 2080 def query_map_max_allowed_uids(self, extract_value: bool = False) -> dict[int, int]: 2081 """ 2082 Queries the network for the maximum number of allowed user IDs (UIDs) 2083 for each network subnet. 2084 2085 Fetches a mapping of network subnets to their respective 2086 limits on the number of user IDs that can be created or used. 2087 2088 Returns: 2089 A dictionary mapping network UIDs (unique identifiers) to their 2090 maximum allowed number of UIDs. 2091 Each entry represents a network subnet 2092 with its corresponding UID limit. 2093 2094 Raises: 2095 QueryError: If the query to the network fails or is invalid. 2096 """ 2097 2098 return self.query_map("MaxAllowedUids", extract_value=extract_value)[ 2099 "MaxAllowedUids" 2100 ] 2101 2102 def query_map_min_stake(self, extract_value: bool = False) -> dict[int, int]: 2103 """ 2104 Retrieves a mapping of minimum allowed stake on the network. 2105 2106 Queries the network to obtain the minimum number of stake, 2107 which is represented in nanotokens. 2108 2109 Returns: 2110 A dictionary mapping network UIDs to 2111 their minimum allowed stake values. 2112 2113 Raises: 2114 QueryError: If the query to the network fails or is invalid. 2115 """ 2116 2117 return self.query_map("MinStake", extract_value=extract_value)["MinStake"] 2118 2119 def query_map_max_stake(self, extract_value: bool = False) -> dict[int, int]: 2120 """ 2121 Retrieves a mapping of the maximum stake values for the network. 2122 2123 Queries the network for the maximum stake values across various s 2124 ubnets of the network. 2125 2126 Returns: 2127 A dictionary mapping network UIDs to their maximum stake values. 2128 2129 Raises: 2130 QueryError: If the query to the network fails or is invalid. 2131 """ 2132 2133 return self.query_map("MaxStake", extract_value=extract_value)["MaxStake"] 2134 2135 def query_map_founder(self, extract_value: bool = False) -> dict[int, str]: 2136 """ 2137 Retrieves a mapping of founders for the network. 2138 2139 Queries the network to obtain the founders associated with 2140 various subnets. 2141 2142 Returns: 2143 A dictionary mapping network UIDs to their respective founders. 2144 2145 Raises: 2146 QueryError: If the query to the network fails or is invalid. 2147 """ 2148 2149 return self.query_map("Founder", extract_value=extract_value)["Founder"] 2150 2151 def query_map_founder_share(self, extract_value: bool = False) -> dict[int, int]: 2152 """ 2153 Retrieves a mapping of founder shares for the network. 2154 2155 Queries the network for the share percentages 2156 allocated to founders across different subnets. 2157 2158 Returns: 2159 A dictionary mapping network UIDs to their founder share percentages. 2160 2161 Raises: 2162 QueryError: If the query to the network fails or is invalid. 2163 """ 2164 2165 return self.query_map("FounderShare", extract_value=extract_value)[ 2166 "FounderShare" 2167 ] 2168 2169 def query_map_incentive_ratio(self, extract_value: bool = False) -> dict[int, int]: 2170 """ 2171 Retrieves a mapping of incentive ratios for the network. 2172 2173 Queries the network for the incentive ratios, 2174 which are the proportions of rewards or incentives 2175 allocated in different subnets of the network. 2176 2177 Returns: 2178 A dictionary mapping network UIDs to their incentive ratios. 2179 2180 Raises: 2181 QueryError: If the query to the network fails or is invalid. 2182 """ 2183 2184 return self.query_map("IncentiveRatio", extract_value=extract_value)[ 2185 "IncentiveRatio" 2186 ] 2187 2188 def query_map_trust_ratio(self, extract_value: bool = False) -> dict[int, int]: 2189 """ 2190 Retrieves a mapping of trust ratios for the network. 2191 2192 Queries the network for trust ratios, 2193 indicative of the level of trust or credibility assigned 2194 to different subnets of the network. 2195 2196 Returns: 2197 A dictionary mapping network UIDs to their trust ratios. 2198 2199 Raises: 2200 QueryError: If the query to the network fails or is invalid. 2201 """ 2202 2203 return self.query_map("TrustRatio", extract_value=extract_value)["TrustRatio"] 2204 2205 def query_map_vote_mode_subnet(self, extract_value: bool = False) -> dict[int, str]: 2206 """ 2207 Retrieves a mapping of vote modes for subnets within the network. 2208 2209 Queries the network for the voting modes used in different 2210 subnets, which define the methodology or approach of voting within those 2211 subnets. 2212 2213 Returns: 2214 A dictionary mapping network UIDs to their vote 2215 modes for subnets. 2216 2217 Raises: 2218 QueryError: If the query to the network fails or is invalid. 2219 """ 2220 2221 return self.query_map("VoteModeSubnet", extract_value=extract_value)[ 2222 "VoteModeSubnet" 2223 ] 2224 2225 def query_map_legit_whitelist( 2226 self, extract_value: bool = False 2227 ) -> dict[Ss58Address, int]: 2228 """ 2229 Retrieves a mapping of whitelisted addresses for the network. 2230 2231 Queries the network for a mapping of whitelisted addresses 2232 and their respective legitimacy status. 2233 2234 Returns: 2235 A dictionary mapping addresses to their legitimacy status. 2236 2237 Raises: 2238 QueryError: If the query to the network fails or is invalid. 2239 """ 2240 2241 return self.query_map( 2242 "LegitWhitelist", module="GovernanceModule", extract_value=extract_value)[ 2243 "LegitWhitelist" 2244 ] 2245 2246 def query_map_subnet_names(self, extract_value: bool = False) -> dict[int, str]: 2247 """ 2248 Retrieves a mapping of subnet names within the network. 2249 2250 Queries the network for the names of various subnets, 2251 providing an overview of the different 2252 subnets within the network. 2253 2254 Returns: 2255 A dictionary mapping network UIDs to their subnet names. 2256 2257 Raises: 2258 QueryError: If the query to the network fails or is invalid. 2259 """ 2260 2261 return self.query_map("SubnetNames", extract_value=extract_value)["SubnetNames"] 2262 2263 def query_map_balances( 2264 self, extract_value: bool = False, block_hash: str | None = None 2265 ) -> dict[str, dict[str, int | dict[str, int | float]]]: 2266 """ 2267 Retrieves a mapping of account balances within the network. 2268 2269 Queries the network for the balances associated with different accounts. 2270 It provides detailed information including various types of 2271 balances for each account. 2272 2273 Returns: 2274 A dictionary mapping account addresses to their balance details. 2275 2276 Raises: 2277 QueryError: If the query to the network fails or is invalid. 2278 """ 2279 2280 return self.query_map("Account", module="System", extract_value=extract_value, block_hash=block_hash)[ 2281 "Account" 2282 ] 2283 2284 def query_map_registration_blocks( 2285 self, netuid: int = 0, extract_value: bool = False 2286 ) -> dict[int, int]: 2287 """ 2288 Retrieves a mapping of registration blocks for UIDs on the network. 2289 2290 Queries the network to find the block numbers at which various 2291 UIDs were registered. 2292 2293 Args: 2294 netuid: The network UID from which to get the registrations. 2295 2296 Returns: 2297 A dictionary mapping UIDs to their registration block numbers. 2298 2299 Raises: 2300 QueryError: If the query to the network fails or is invalid. 2301 """ 2302 2303 return self.query_map( 2304 "RegistrationBlock", [netuid], extract_value=extract_value 2305 )["RegistrationBlock"] 2306 2307 def query_map_name( 2308 self, netuid: int = 0, extract_value: bool = False 2309 ) -> dict[int, str]: 2310 """ 2311 Retrieves a mapping of names for keys on the network. 2312 2313 Queries the network for the names associated with different keys. 2314 It provides a mapping of key UIDs to their registered names. 2315 2316 Args: 2317 netuid: The network UID from which to get the names. 2318 2319 Returns: 2320 A dictionary mapping key UIDs to their names. 2321 2322 Raises: 2323 QueryError: If the query to the network fails or is invalid. 2324 """ 2325 2326 return self.query_map("Name", [netuid], extract_value=extract_value)["Name"] 2327 2328 # == QUERY FUNCTIONS == # 2329 2330 def get_immunity_period(self, netuid: int = 0) -> int: 2331 """ 2332 Queries the network for the immunity period setting. 2333 2334 The immunity period is a time duration during which a module 2335 can not be deregistered from the network. 2336 Fetches the immunity period for a specified network subnet. 2337 2338 Args: 2339 netuid: The network UID for which to query the immunity period. 2340 2341 Returns: 2342 The immunity period setting for the specified network subnet. 2343 2344 Raises: 2345 QueryError: If the query to the network fails or is invalid. 2346 """ 2347 2348 return self.query( 2349 "ImmunityPeriod", 2350 params=[netuid], 2351 ) 2352 2353 def get_max_set_weights_per_epoch(self): 2354 return self.query("MaximumSetWeightCallsPerEpoch") 2355 2356 def get_min_allowed_weights(self, netuid: int = 0) -> int: 2357 """ 2358 Queries the network for the minimum allowed weights setting. 2359 2360 Retrieves the minimum weight values that are possible to set 2361 by a validator within a specific network subnet. 2362 2363 Args: 2364 netuid: The network UID for which to query the minimum allowed 2365 weights. 2366 2367 Returns: 2368 The minimum allowed weight values for the specified network 2369 subnet. 2370 2371 Raises: 2372 QueryError: If the query to the network fails or is invalid. 2373 """ 2374 2375 return self.query( 2376 "MinAllowedWeights", 2377 params=[netuid], 2378 ) 2379 2380 def get_dao_treasury_address(self) -> Ss58Address: 2381 return self.query("DaoTreasuryAddress", module="GovernanceModule") 2382 2383 def get_max_allowed_weights(self, netuid: int = 0) -> int: 2384 """ 2385 Queries the network for the maximum allowed weights setting. 2386 2387 Retrieves the maximum weight values that are possible to set 2388 by a validator within a specific network subnet. 2389 2390 Args: 2391 netuid: The network UID for which to query the maximum allowed 2392 weights. 2393 2394 Returns: 2395 The maximum allowed weight values for the specified network 2396 subnet. 2397 2398 Raises: 2399 QueryError: If the query to the network fails or is invalid. 2400 """ 2401 2402 return self.query("MaxAllowedWeights", params=[netuid]) 2403 2404 def get_max_allowed_uids(self, netuid: int = 0) -> int: 2405 """ 2406 Queries the network for the maximum allowed UIDs setting. 2407 2408 Fetches the upper limit on the number of user IDs that can 2409 be allocated or used within a specific network subnet. 2410 2411 Args: 2412 netuid: The network UID for which to query the maximum allowed UIDs. 2413 2414 Returns: 2415 The maximum number of allowed UIDs for the specified network subnet. 2416 2417 Raises: 2418 QueryError: If the query to the network fails or is invalid. 2419 """ 2420 2421 return self.query("MaxAllowedUids", params=[netuid]) 2422 2423 def get_name(self, netuid: int = 0) -> str: 2424 """ 2425 Queries the network for the name of a specific subnet. 2426 2427 Args: 2428 netuid: The network UID for which to query the name. 2429 2430 Returns: 2431 The name of the specified network subnet. 2432 2433 Raises: 2434 QueryError: If the query to the network fails or is invalid. 2435 """ 2436 2437 return self.query("Name", params=[netuid]) 2438 2439 def get_subnet_name(self, netuid: int = 0) -> str: 2440 """ 2441 Queries the network for the name of a specific subnet. 2442 2443 Args: 2444 netuid: The network UID for which to query the name. 2445 2446 Returns: 2447 The name of the specified network subnet. 2448 2449 Raises: 2450 QueryError: If the query to the network fails or is invalid. 2451 """ 2452 2453 return self.query("SubnetNames", params=[netuid]) 2454 2455 def get_global_dao_treasury(self): 2456 return self.query("GlobalDaoTreasury", module="GovernanceModule") 2457 2458 def get_n(self, netuid: int = 0) -> int: 2459 """ 2460 Queries the network for the 'N' hyperparameter, which represents how 2461 many modules are on the network. 2462 2463 Args: 2464 netuid: The network UID for which to query the 'N' hyperparameter. 2465 2466 Returns: 2467 The value of the 'N' hyperparameter for the specified network 2468 subnet. 2469 2470 Raises: 2471 QueryError: If the query to the network fails or is invalid. 2472 """ 2473 2474 return self.query("N", params=[netuid]) 2475 2476 def get_tempo(self, netuid: int = 0) -> int: 2477 """ 2478 Queries the network for the tempo setting, measured in blocks, for the 2479 specified subnet. 2480 2481 Args: 2482 netuid: The network UID for which to query the tempo. 2483 2484 Returns: 2485 The tempo setting for the specified subnet. 2486 2487 Raises: 2488 QueryError: If the query to the network fails or is invalid. 2489 """ 2490 2491 return self.query("Tempo", params=[netuid]) 2492 2493 def get_total_free_issuance(self, block_hash: str | None = None) -> int: 2494 """ 2495 Queries the network for the total free issuance. 2496 2497 Fetches the total amount of free issuance tokens available 2498 2499 Returns: 2500 The total free issuance amount. 2501 2502 Raises: 2503 QueryError: If the query to the network fails or is invalid. 2504 """ 2505 2506 return self.query("TotalIssuance", module="Balances", block_hash=block_hash) 2507 2508 def get_total_stake(self, block_hash: str | None = None) -> int: 2509 """ 2510 Retrieves a mapping of total stakes for keys on the network. 2511 2512 Queries the network for a mapping of key UIDs to their total stake amounts. 2513 2514 Returns: 2515 A dictionary mapping key UIDs to their total stake amounts. 2516 2517 Raises: 2518 QueryError: If the query to the network fails or is invalid. 2519 """ 2520 2521 return self.query("TotalStake", block_hash=block_hash) 2522 2523 def get_registrations_per_block(self): 2524 """ 2525 Queries the network for the number of registrations per block. 2526 2527 Fetches the number of registrations that are processed per 2528 block within the network. 2529 2530 Returns: 2531 The number of registrations processed per block. 2532 2533 Raises: 2534 QueryError: If the query to the network fails or is invalid. 2535 """ 2536 2537 return self.query( 2538 "RegistrationsPerBlock", 2539 ) 2540 2541 def max_registrations_per_block(self, netuid: int = 0): 2542 """ 2543 Queries the network for the maximum number of registrations per block. 2544 2545 Retrieves the upper limit of registrations that can be processed in 2546 each block within a specific network subnet. 2547 2548 Args: 2549 netuid: The network UID for which to query. 2550 2551 Returns: 2552 The maximum number of registrations per block for 2553 the specified network subnet. 2554 2555 Raises: 2556 QueryError: If the query to the network fails or is invalid. 2557 """ 2558 2559 return self.query( 2560 "MaxRegistrationsPerBlock", 2561 params=[netuid], 2562 ) 2563 2564 def get_proposal(self, proposal_id: int = 0): 2565 """ 2566 Queries the network for a specific proposal. 2567 2568 Args: 2569 proposal_id: The ID of the proposal to query. 2570 2571 Returns: 2572 The details of the specified proposal. 2573 2574 Raises: 2575 QueryError: If the query to the network fails, is invalid, 2576 or if the proposal ID does not exist. 2577 """ 2578 2579 return self.query( 2580 "Proposals", 2581 params=[proposal_id], 2582 ) 2583 2584 def get_trust(self, netuid: int = 0): 2585 """ 2586 Queries the network for the trust setting of a specific network subnet. 2587 2588 Retrieves the trust level or score, which may represent the 2589 level of trustworthiness or reliability within a 2590 particular network subnet. 2591 2592 Args: 2593 netuid: The network UID for which to query the trust setting. 2594 2595 Returns: 2596 The trust level or score for the specified network subnet. 2597 2598 Raises: 2599 QueryError: If the query to the network fails or is invalid. 2600 """ 2601 2602 return self.query( 2603 "Trust", 2604 params=[netuid], 2605 ) 2606 2607 def get_uids(self, key: Ss58Address, netuid: int = 0) -> bool | None: 2608 """ 2609 Queries the network for module UIDs associated with a specific key. 2610 2611 Args: 2612 key: The key address for which to query UIDs. 2613 netuid: The network UID within which to search for the key. 2614 2615 Returns: 2616 A list of UIDs associated with the specified key. 2617 2618 Raises: 2619 QueryError: If the query to the network fails or is invalid. 2620 """ 2621 2622 return self.query( 2623 "Uids", 2624 params=[netuid, key], 2625 ) 2626 2627 def get_unit_emission(self) -> int: 2628 """ 2629 Queries the network for the unit emission setting. 2630 2631 Retrieves the unit emission value, which represents the 2632 emission rate or quantity for the $COMM token. 2633 2634 Returns: 2635 The unit emission value in nanos for the network. 2636 2637 Raises: 2638 QueryError: If the query to the network fails or is invalid. 2639 """ 2640 2641 return self.query("UnitEmission", module="SubnetEmissionModule") 2642 2643 def get_tx_rate_limit(self) -> int: 2644 """ 2645 Queries the network for the transaction rate limit. 2646 2647 Retrieves the rate limit for transactions within the network, 2648 which defines the maximum number of transactions that can be 2649 processed within a certain timeframe. 2650 2651 Returns: 2652 The transaction rate limit for the network. 2653 2654 Raises: 2655 QueryError: If the query to the network fails or is invalid. 2656 """ 2657 2658 return self.query( 2659 "TxRateLimit", 2660 ) 2661 2662 def get_subnet_burn(self) -> int: 2663 """Queries the network for the subnet burn value. 2664 2665 Retrieves the subnet burn value from the network, which represents 2666 the amount of tokens that are burned (permanently removed from 2667 circulation) for subnet-related operations. 2668 2669 Returns: 2670 int: The subnet burn value. 2671 2672 Raises: 2673 QueryError: If the query to the network fails or returns invalid data. 2674 """ 2675 2676 return self.query( 2677 "SubnetBurn", 2678 ) 2679 2680 def get_burn_rate(self) -> int: 2681 """ 2682 Queries the network for the burn rate setting. 2683 2684 Retrieves the burn rate, which represents the rate at 2685 which the $COMM token is permanently 2686 removed or 'burned' from circulation. 2687 2688 Returns: 2689 The burn rate for the network. 2690 2691 Raises: 2692 QueryError: If the query to the network fails or is invalid. 2693 """ 2694 2695 return self.query( 2696 "BurnRate", 2697 params=[], 2698 ) 2699 2700 def get_burn(self, netuid: int = 0) -> int: 2701 """ 2702 Queries the network for the burn setting. 2703 2704 Retrieves the burn value, which represents the amount of the 2705 $COMM token that is 'burned' or permanently removed from 2706 circulation. 2707 2708 Args: 2709 netuid: The network UID for which to query the burn value. 2710 2711 Returns: 2712 The burn value for the specified network subnet. 2713 2714 Raises: 2715 QueryError: If the query to the network fails or is invalid. 2716 """ 2717 2718 return self.query("Burn", params=[netuid]) 2719 2720 def get_min_burn(self) -> int: 2721 """ 2722 Queries the network for the minimum burn setting. 2723 2724 Retrieves the minimum burn value, indicating the lowest 2725 amount of the $COMM tokens that can be 'burned' or 2726 permanently removed from circulation. 2727 2728 Returns: 2729 The minimum burn value for the network. 2730 2731 Raises: 2732 QueryError: If the query to the network fails or is invalid. 2733 """ 2734 2735 return self.query( 2736 "BurnConfig", 2737 params=[], 2738 )["min_burn"] 2739 2740 def get_min_weight_stake(self) -> int: 2741 """ 2742 Queries the network for the minimum weight stake setting. 2743 2744 Retrieves the minimum weight stake, which represents the lowest 2745 stake weight that is allowed for certain operations or 2746 transactions within the network. 2747 2748 Returns: 2749 The minimum weight stake for the network. 2750 2751 Raises: 2752 QueryError: If the query to the network fails or is invalid. 2753 """ 2754 2755 return self.query("MinWeightStake", params=[]) 2756 2757 def get_vote_mode_global(self) -> str: 2758 """ 2759 Queries the network for the global vote mode setting. 2760 2761 Retrieves the global vote mode, which defines the overall voting 2762 methodology or approach used across the network in default. 2763 2764 Returns: 2765 The global vote mode setting for the network. 2766 2767 Raises: 2768 QueryError: If the query to the network fails or is invalid. 2769 """ 2770 2771 return self.query( 2772 "VoteModeGlobal", 2773 ) 2774 2775 def get_max_proposals(self) -> int: 2776 """ 2777 Queries the network for the maximum number of proposals allowed. 2778 2779 Retrieves the upper limit on the number of proposals that can be 2780 active or considered at any given time within the network. 2781 2782 Returns: 2783 The maximum number of proposals allowed on the network. 2784 2785 Raises: 2786 QueryError: If the query to the network fails or is invalid. 2787 """ 2788 2789 return self.query( 2790 "MaxProposals", 2791 ) 2792 2793 def get_max_registrations_per_block(self) -> int: 2794 """ 2795 Queries the network for the maximum number of registrations per block. 2796 2797 Retrieves the maximum number of registrations that can 2798 be processed in each block within the network. 2799 2800 Returns: 2801 The maximum number of registrations per block on the network. 2802 2803 Raises: 2804 QueryError: If the query to the network fails or is invalid. 2805 """ 2806 2807 return self.query( 2808 "MaxRegistrationsPerBlock", 2809 params=[], 2810 ) 2811 2812 def get_max_name_length(self) -> int: 2813 """ 2814 Queries the network for the maximum length allowed for names. 2815 2816 Retrieves the maximum character length permitted for names 2817 within the network. Such as the module names 2818 2819 Returns: 2820 The maximum length allowed for names on the network. 2821 2822 Raises: 2823 QueryError: If the query to the network fails or is invalid. 2824 """ 2825 2826 return self.query( 2827 "MaxNameLength", 2828 params=[], 2829 ) 2830 2831 def get_global_vote_threshold(self) -> int: 2832 """ 2833 Queries the network for the global vote threshold. 2834 2835 Retrieves the global vote threshold, which is the critical value or 2836 percentage required for decisions in the network's governance process. 2837 2838 Returns: 2839 The global vote threshold for the network. 2840 2841 Raises: 2842 QueryError: If the query to the network fails or is invalid. 2843 """ 2844 2845 return self.query( 2846 "GlobalVoteThreshold", 2847 ) 2848 2849 def get_max_allowed_subnets(self) -> int: 2850 """ 2851 Queries the network for the maximum number of allowed subnets. 2852 2853 Retrieves the upper limit on the number of subnets that can 2854 be created or operated within the network. 2855 2856 Returns: 2857 The maximum number of allowed subnets on the network. 2858 2859 Raises: 2860 QueryError: If the query to the network fails or is invalid. 2861 """ 2862 2863 return self.query( 2864 "MaxAllowedSubnets", 2865 params=[], 2866 ) 2867 2868 def get_max_allowed_modules(self) -> int: 2869 """ 2870 Queries the network for the maximum number of allowed modules. 2871 2872 Retrieves the upper limit on the number of modules that 2873 can be registered within the network. 2874 2875 Returns: 2876 The maximum number of allowed modules on the network. 2877 2878 Raises: 2879 QueryError: If the query to the network fails or is invalid. 2880 """ 2881 2882 return self.query( 2883 "MaxAllowedModules", 2884 params=[], 2885 ) 2886 2887 def get_min_stake(self, netuid: int = 0) -> int: 2888 """ 2889 Queries the network for the minimum stake required to register a key. 2890 2891 Retrieves the minimum amount of stake necessary for 2892 registering a key within a specific network subnet. 2893 2894 Args: 2895 netuid: The network UID for which to query the minimum stake. 2896 2897 Returns: 2898 The minimum stake required for key registration in nanos. 2899 2900 Raises: 2901 QueryError: If the query to the network fails or is invalid. 2902 """ 2903 2904 return self.query("MinStake", params=[netuid]) 2905 2906 def get_stakefrom( 2907 self, 2908 key: Ss58Address, 2909 ) -> dict[str, int]: 2910 """ 2911 Retrieves the stake amounts from all stakers to a specific staked address. 2912 2913 Queries the network for the stakes received by a particular staked address 2914 from all stakers. 2915 2916 Args: 2917 key: The address of the key receiving the stakes. 2918 2919 Returns: 2920 A dictionary mapping staker addresses to their respective stake amounts. 2921 2922 Raises: 2923 QueryError: If the query to the network fails or is invalid. 2924 """ 2925 2926 # Has to use query map in order to iterate through the storage prefix. 2927 return self.query_map("StakeFrom", [key], extract_value=False).get("StakeFrom", {}) 2928 2929 def get_staketo( 2930 self, 2931 key: Ss58Address, 2932 ) -> dict[str, int]: 2933 """ 2934 Retrieves the stake amounts provided by a specific staker to all staked addresses. 2935 2936 Queries the network for the stakes provided by a particular staker to 2937 all staked addresses. 2938 2939 Args: 2940 key: The address of the key providing the stakes. 2941 2942 Returns: 2943 A dictionary mapping staked addresses to their respective received stake amounts. 2944 2945 Raises: 2946 QueryError: If the query to the network fails or is invalid. 2947 """ 2948 2949 # Has to use query map in order to iterate through the storage prefix. 2950 return self.query_map("StakeTo", [key], extract_value=False).get("StakeTo", {}) 2951 2952 def get_balance( 2953 self, 2954 addr: Ss58Address, 2955 ) -> int: 2956 """ 2957 Retrieves the balance of a specific key. 2958 2959 Args: 2960 addr: The address of the key to query the balance for. 2961 2962 Returns: 2963 The balance of the specified key. 2964 2965 Raises: 2966 QueryError: If the query to the network fails or is invalid. 2967 """ 2968 2969 result = self.query("Account", module="System", params=[addr]) 2970 2971 return result["data"]["free"] 2972 2973 def get_block(self, block_hash: str | None = None) -> dict[Any, Any] | None: 2974 """ 2975 Retrieves information about a specific block in the network. 2976 2977 Queries the network for details about a block, such as its number, 2978 hash, and other relevant information. 2979 2980 Returns: 2981 The requested information about the block, 2982 or None if the block does not exist 2983 or the information is not available. 2984 2985 Raises: 2986 QueryError: If the query to the network fails or is invalid. 2987 """ 2988 2989 with self.get_conn() as substrate: 2990 block: dict[Any, Any] | None = substrate.get_block( # type: ignore 2991 block_hash # type: ignore 2992 ) 2993 2994 return block 2995 2996 def get_existential_deposit(self, block_hash: str | None = None) -> int: 2997 """ 2998 Retrieves the existential deposit value for the network. 2999 3000 The existential deposit is the minimum balance that must be maintained 3001 in an account to prevent it from being purged. Denotated in nano units. 3002 3003 Returns: 3004 The existential deposit value in nano units. 3005 Note: 3006 The value returned is a fixed value defined in the 3007 client and may not reflect changes in the network's configuration. 3008 """ 3009 3010 with self.get_conn() as substrate: 3011 result: int = substrate.get_constant( # type: ignore 3012 "Balances", "ExistentialDeposit", block_hash 3013 ).value # type: ignore 3014 3015 return result 3016 3017 def get_voting_power_delegators(self) -> list[Ss58Address]: 3018 result = self.query("NotDelegatingVotingPower", [], module="GovernanceModule") 3019 return result 3020 3021 def add_transfer_dao_treasury_proposal( 3022 self, 3023 key: Keypair, 3024 data: str, 3025 amount_nano: int, 3026 dest: Ss58Address, 3027 ): 3028 params = {"dest": dest, "value": amount_nano, "data": data} 3029 3030 return self.compose_call( 3031 module="GovernanceModule", 3032 fn="add_transfer_dao_treasury_proposal", 3033 params=params, 3034 key=key, 3035 ) 3036 3037 def delegate_rootnet_control(self, key: Keypair, dest: Ss58Address): 3038 params = {"origin": key, "target": dest} 3039 3040 return self.compose_call( 3041 module="SubspaceModule", 3042 fn="delegate_rootnet_control", 3043 params=params, 3044 key=key, 3045 )
A client for interacting with Commune network nodes, querying storage, submitting transactions, etc.
Attributes:
- wait_for_finalization: Whether to wait for transaction finalization.
Example:
client = CommuneClient()
client.query(name='function_name', params=['param1', 'param2'])
Raises:
- AssertionError: If the maximum connections value is less than or equal to zero.
CommuneClient( url: str, num_connections: int = 1, wait_for_finalization: bool = False, timeout: int | None = None)
59 def __init__( 60 self, 61 url: str, 62 num_connections: int = 1, 63 wait_for_finalization: bool = False, 64 timeout: int | None = None, 65 ): 66 """ 67 Args: 68 url: The URL of the network node to connect to. 69 num_connections: The number of websocket connections to be opened. 70 """ 71 assert num_connections > 0 72 self._num_connections = num_connections 73 self.wait_for_finalization = wait_for_finalization 74 self._connection_queue = queue.Queue(num_connections) 75 self.url = url 76 ws_options: dict[str, int] = {} 77 if timeout is not None: 78 ws_options["timeout"] = timeout 79 self.ws_options = ws_options 80 for _ in range(num_connections): 81 self._connection_queue.put( 82 SubstrateInterface(url, ws_options=ws_options) 83 )
Arguments:
- url: The URL of the network node to connect to.
- num_connections: The number of websocket connections to be opened.
connections: int
85 @property 86 def connections(self) -> int: 87 """ 88 Gets the maximum allowed number of simultaneous connections to the 89 network node. 90 """ 91 return self._num_connections
Gets the maximum allowed number of simultaneous connections to the network node.
@contextmanager
def
get_conn(self, timeout: float | None = None, init: bool = False):
93 @contextmanager 94 def get_conn(self, timeout: float | None = None, init: bool = False): 95 """ 96 Context manager to get a connection from the pool. 97 98 Tries to get a connection from the pool queue. If the queue is empty, 99 it blocks for `timeout` seconds until a connection is available. If 100 `timeout` is None, it blocks indefinitely. 101 102 Args: 103 timeout: The maximum time in seconds to wait for a connection. 104 105 Yields: 106 The connection object from the pool. 107 108 Raises: 109 QueueEmptyError: If no connection is available within the timeout 110 period. 111 """ 112 conn = self._connection_queue.get(timeout=timeout) 113 if init: 114 conn.init_runtime() # type: ignore 115 try: 116 if conn.websocket and conn.websocket.connected: # type: ignore 117 yield conn 118 else: 119 conn = SubstrateInterface(self.url, ws_options=self.ws_options) 120 yield conn 121 finally: 122 self._connection_queue.put(conn)
Context manager to get a connection from the pool.
Tries to get a connection from the pool queue. If the queue is empty,
it blocks for timeout seconds until a connection is available. If
timeout is None, it blocks indefinitely.
Arguments:
- timeout: The maximum time in seconds to wait for a connection.
Yields:
The connection object from the pool.
Raises:
- QueueEmptyError: If no connection is available within the timeout period.
def
query_batch( self, functions: dict[str, list[tuple[str, list[typing.Any]]]]) -> dict[str, str]:
579 def query_batch( 580 self, functions: dict[str, list[tuple[str, list[Any]]]] 581 ) -> dict[str, str]: 582 """ 583 Executes batch queries on a substrate and returns results in a dictionary format. 584 585 Args: 586 substrate: An instance of SubstrateInterface to interact with the substrate. 587 functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls (function name and parameters). 588 589 Returns: 590 A dictionary where keys are storage function names and values are the query results. 591 592 Raises: 593 Exception: If no result is found from the batch queries. 594 595 Example: 596 >>> query_batch(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) 597 {'function_name': 'query_result', ...} 598 """ 599 600 result: dict[str, str] = {} 601 if not functions: 602 raise Exception("No result") 603 with self.get_conn(init=True) as substrate: 604 for module, queries in functions.items(): 605 storage_keys: list[Any] = [] 606 for fn, params in queries: 607 storage_function = substrate.create_storage_key( # type: ignore 608 pallet=module, storage_function=fn, params=params 609 ) 610 storage_keys.append(storage_function) 611 612 block_hash = substrate.get_block_hash() 613 responses: list[Any] = substrate.query_multi( # type: ignore 614 storage_keys=storage_keys, block_hash=block_hash 615 ) 616 617 for item in responses: 618 fun = item[0] 619 query = item[1] 620 storage_fun = fun.storage_function 621 result[storage_fun] = query.value 622 623 return result
Executes batch queries on a substrate and returns results in a dictionary format.
Arguments:
- substrate: An instance of SubstrateInterface to interact with the substrate.
- functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls (function name and parameters).
Returns:
A dictionary where keys are storage function names and values are the query results.
Raises:
- Exception: If no result is found from the batch queries.
Example:
>>> query_batch(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) {'function_name': 'query_result', ...}
def
query_batch_map( self, functions: dict[str, list[tuple[str, list[typing.Any]]]], block_hash: str | None = None) -> dict[str, dict[typing.Any, typing.Any]]:
625 def query_batch_map( 626 self, 627 functions: dict[str, list[tuple[str, list[Any]]]], 628 block_hash: str | None = None, 629 ) -> dict[str, dict[Any, Any]]: 630 """ 631 Queries multiple storage functions using a map batch approach and returns the combined result. 632 633 Args: 634 substrate: An instance of SubstrateInterface for substrate interaction. 635 functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls. 636 637 Returns: 638 The combined result of the map batch query. 639 640 Example: 641 >>> query_batch_map(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) 642 # Returns the combined result of the map batch query 643 """ 644 multi_result: dict[str, dict[Any, Any]] = {} 645 646 def recursive_update( 647 d: dict[str, dict[T1, T2] | dict[str, Any]], 648 u: Mapping[str, dict[Any, Any] | str], 649 ) -> dict[str, dict[T1, T2]]: 650 for k, v in u.items(): 651 if isinstance(v, dict): 652 d[k] = recursive_update(d.get(k, {}), v) # type: ignore 653 else: 654 d[k] = v # type: ignore 655 return d # type: ignore 656 657 def get_page(): 658 send, prefix_list = self._get_storage_keys(storage, queries, block_hash) 659 with self.get_conn(init=True) as substrate: 660 function_parameters = self._get_lists(storage, queries, substrate) 661 responses = self._rpc_request_batch(send) 662 # assumption because send is just the storage_function keys 663 # so it should always be really small regardless of the amount of queries 664 assert len(responses) == 1 665 res = responses[0] 666 built_payload: list[tuple[str, list[Any]]] = [] 667 for result_keys in res: 668 built_payload.append( 669 ("state_queryStorageAt", [result_keys, block_hash]) 670 ) 671 _, chunks_info = self._make_request_smaller( 672 built_payload, prefix_list, function_parameters 673 ) 674 chunks_response, chunks_info = self._rpc_request_batch_chunked(chunks_info) 675 return chunks_response, chunks_info 676 677 if not block_hash: 678 with self.get_conn(init=True) as substrate: 679 block_hash = substrate.get_block_hash() 680 for storage, queries in functions.items(): 681 chunks, chunks_info = get_page() 682 # if this doesn't happen something is wrong on the code 683 # and we won't be able to decode the data properly 684 assert len(chunks) == len(chunks_info) 685 for chunk_info, response in zip(chunks_info, chunks): 686 storage_result = self._decode_response( 687 response, chunk_info.fun_params, chunk_info.prefix_list, block_hash 688 ) 689 multi_result = recursive_update(multi_result, storage_result) 690 691 return multi_result
Queries multiple storage functions using a map batch approach and returns the combined result.
Arguments:
- substrate: An instance of SubstrateInterface for substrate interaction.
- functions (dict[str, list[query_call]]): A dictionary mapping module names to lists of query calls.
Returns:
The combined result of the map batch query.
Example:
>>> query_batch_map(substrate_instance, {'module_name': [('function_name', ['param1', 'param2'])]}) <h1 id="returns-the-combined-result-of-the-map-batch-query">Returns the combined result of the map batch query</h1>
def
query( self, name: str, params: list[typing.Any] = [], module: str = 'SubspaceModule', block_hash: str | None = None) -> Any:
693 def query( 694 self, 695 name: str, 696 params: list[Any] = [], 697 module: str = "SubspaceModule", 698 block_hash: str | None = None, 699 ) -> Any: 700 """ 701 Queries a storage function on the network. 702 703 Sends a query to the network and retrieves data from a 704 specified storage function. 705 706 Args: 707 name: The name of the storage function to query. 708 params: The parameters to pass to the storage function. 709 module: The module where the storage function is located. 710 711 Returns: 712 The result of the query from the network. 713 714 Raises: 715 NetworkQueryError: If the query fails or is invalid. 716 """ 717 718 result = self.query_batch({module: [(name, params)]}) 719 720 return result[name]
Queries a storage function on the network.
Sends a query to the network and retrieves data from a specified storage function.
Arguments:
- name: The name of the storage function to query.
- params: The parameters to pass to the storage function.
- module: The module where the storage function is located.
Returns:
The result of the query from the network.
Raises:
- NetworkQueryError: If the query fails or is invalid.
def
query_map( self, name: str, params: list[typing.Any] = [], module: str = 'SubspaceModule', extract_value: bool = True, block_hash: str | None = None) -> dict[typing.Any, typing.Any]:
722 def query_map( 723 self, 724 name: str, 725 params: list[Any] = [], 726 module: str = "SubspaceModule", 727 extract_value: bool = True, 728 block_hash: str | None = None, 729 ) -> dict[Any, Any]: 730 """ 731 Queries a storage map from a network node. 732 733 Args: 734 name: The name of the storage map to query. 735 params: A list of parameters for the query. 736 module: The module in which the storage map is located. 737 738 Returns: 739 A dictionary representing the key-value pairs 740 retrieved from the storage map. 741 742 Raises: 743 QueryError: If the query to the network fails or is invalid. 744 """ 745 746 result = self.query_batch_map({module: [(name, params)]}, block_hash) 747 748 if extract_value: 749 return {k.value: v.value for k, v in result} # type: ignore 750 751 return result
Queries a storage map from a network node.
Arguments:
- name: The name of the storage map to query.
- params: A list of parameters for the query.
- module: The module in which the storage map is located.
Returns:
A dictionary representing the key-value pairs retrieved from the storage map.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
compose_call( self, fn: str, params: dict[str, typing.Any], key: substrateinterface.keypair.Keypair | None, module: str = 'SubspaceModule', wait_for_inclusion: bool = True, wait_for_finalization: bool | None = None, sudo: bool = False, unsigned: bool = False) -> substrateinterface.base.ExtrinsicReceipt:
753 def compose_call( 754 self, 755 fn: str, 756 params: dict[str, Any], 757 key: Keypair | None, 758 module: str = "SubspaceModule", 759 wait_for_inclusion: bool = True, 760 wait_for_finalization: bool | None = None, 761 sudo: bool = False, 762 unsigned: bool = False, 763 ) -> ExtrinsicReceipt: 764 """ 765 Composes and submits a call to the network node. 766 767 Composes and signs a call with the provided keypair, and submits it to 768 the network. The call can be a standard extrinsic or a sudo extrinsic if 769 elevated permissions are required. The method can optionally wait for 770 the call's inclusion in a block and/or its finalization. 771 772 Args: 773 fn: The function name to call on the network. 774 params: A dictionary of parameters for the call. 775 key: The keypair for signing the extrinsic. 776 module: The module containing the function. 777 wait_for_inclusion: Wait for the call's inclusion in a block. 778 wait_for_finalization: Wait for the transaction's finalization. 779 sudo: Execute the call as a sudo (superuser) operation. 780 781 Returns: 782 The receipt of the submitted extrinsic, if 783 `wait_for_inclusion` is True. Otherwise, returns a string 784 identifier of the extrinsic. 785 786 Raises: 787 ChainTransactionError: If the transaction fails. 788 """ 789 790 if key is None and not unsigned: 791 raise ValueError("Key must be provided for signed extrinsics.") 792 793 with self.get_conn() as substrate: 794 if wait_for_finalization is None: 795 wait_for_finalization = self.wait_for_finalization 796 797 call = substrate.compose_call( # type: ignore 798 call_module=module, call_function=fn, call_params=params 799 ) 800 if sudo: 801 call = substrate.compose_call( # type: ignore 802 call_module="Sudo", 803 call_function="sudo", 804 call_params={ 805 "call": call.value, # type: ignore 806 }, 807 ) 808 809 if not unsigned: 810 extrinsic = substrate.create_signed_extrinsic( # type: ignore 811 call=call, keypair=key # type: ignore 812 ) # type: ignore 813 else: 814 extrinsic = substrate.create_unsigned_extrinsic(call=call) # type: ignore 815 816 response = substrate.submit_extrinsic( 817 extrinsic=extrinsic, 818 wait_for_inclusion=wait_for_inclusion, 819 wait_for_finalization=wait_for_finalization, 820 ) 821 if wait_for_inclusion: 822 if not response.is_success: 823 raise ChainTransactionError( 824 response.error_message, response # type: ignore 825 ) 826 827 return response
Composes and submits a call to the network node.
Composes and signs a call with the provided keypair, and submits it to the network. The call can be a standard extrinsic or a sudo extrinsic if elevated permissions are required. The method can optionally wait for the call's inclusion in a block and/or its finalization.
Arguments:
- fn: The function name to call on the network.
- params: A dictionary of parameters for the call.
- key: The keypair for signing the extrinsic.
- module: The module containing the function.
- wait_for_inclusion: Wait for the call's inclusion in a block.
- wait_for_finalization: Wait for the transaction's finalization.
- sudo: Execute the call as a sudo (superuser) operation.
Returns:
The receipt of the submitted extrinsic, if
wait_for_inclusionis True. Otherwise, returns a string identifier of the extrinsic.
Raises:
- ChainTransactionError: If the transaction fails.
def
compose_call_multisig( self, fn: str, params: dict[str, typing.Any], key: substrateinterface.keypair.Keypair, signatories: list[communex.types.Ss58Address], threshold: int, module: str = 'SubspaceModule', wait_for_inclusion: bool = True, wait_for_finalization: bool | None = None, sudo: bool = False, era: dict[str, int] | None = None) -> substrateinterface.base.ExtrinsicReceipt:
829 def compose_call_multisig( 830 self, 831 fn: str, 832 params: dict[str, Any], 833 key: Keypair, 834 signatories: list[Ss58Address], 835 threshold: int, 836 module: str = "SubspaceModule", 837 wait_for_inclusion: bool = True, 838 wait_for_finalization: bool | None = None, 839 sudo: bool = False, 840 era: dict[str, int] | None = None, 841 ) -> ExtrinsicReceipt: 842 """ 843 Composes and submits a multisignature call to the network node. 844 845 This method allows the composition and submission of a call that 846 requires multiple signatures for execution, known as a multisignature 847 call. It supports specifying signatories, a threshold of signatures for 848 the call's execution, and an optional era for the call's mortality. The 849 call can be a standard extrinsic, a sudo extrinsic for elevated 850 permissions, or a multisig extrinsic if multiple signatures are 851 required. Optionally, the method can wait for the call's inclusion in a 852 block and/or its finalization. Make sure to pass all keys, 853 that are part of the multisignature. 854 855 Args: 856 fn: The function name to call on the network. params: A dictionary 857 of parameters for the call. key: The keypair for signing the 858 extrinsic. signatories: List of SS58 addresses of the signatories. 859 Include ALL KEYS that are part of the multisig. threshold: The 860 minimum number of signatories required to execute the extrinsic. 861 module: The module containing the function to call. 862 wait_for_inclusion: Whether to wait for the call's inclusion in a 863 block. wait_for_finalization: Whether to wait for the transaction's 864 finalization. sudo: Execute the call as a sudo (superuser) 865 operation. era: Specifies the call's mortality in terms of blocks in 866 the format 867 {'period': amount_blocks}. If omitted, the extrinsic is 868 immortal. 869 870 Returns: 871 The receipt of the submitted extrinsic if `wait_for_inclusion` is 872 True. Otherwise, returns a string identifier of the extrinsic. 873 874 Raises: 875 ChainTransactionError: If the transaction fails. 876 """ 877 878 # getting the call ready 879 with self.get_conn() as substrate: 880 if wait_for_finalization is None: 881 wait_for_finalization = self.wait_for_finalization 882 883 # prepares the `GenericCall` object 884 call = substrate.compose_call( # type: ignore 885 call_module=module, call_function=fn, call_params=params 886 ) 887 if sudo: 888 call = substrate.compose_call( # type: ignore 889 call_module="Sudo", 890 call_function="sudo", 891 call_params={ 892 "call": call.value, # type: ignore 893 }, 894 ) 895 896 # modify the rpc methods at runtime, to allow for correct payment 897 # fee calculation parity has a bug in this version, 898 # where the method has to be removed 899 rpc_methods = substrate.config.get("rpc_methods") # type: ignore 900 901 if "state_call" in rpc_methods: # type: ignore 902 rpc_methods.remove("state_call") # type: ignore 903 904 # create the multisig account 905 multisig_acc = substrate.generate_multisig_account( # type: ignore 906 signatories, threshold 907 ) 908 909 # send the multisig extrinsic 910 extrinsic = substrate.create_multisig_extrinsic( # type: ignore 911 call=call, # type: ignore 912 keypair=key, 913 multisig_account=multisig_acc, # type: ignore 914 era=era, # type: ignore 915 ) # type: ignore 916 917 response = substrate.submit_extrinsic( 918 extrinsic=extrinsic, 919 wait_for_inclusion=wait_for_inclusion, 920 wait_for_finalization=wait_for_finalization, 921 ) 922 923 if wait_for_inclusion: 924 if not response.is_success: 925 raise ChainTransactionError( 926 response.error_message, response # type: ignore 927 ) 928 929 return response
Composes and submits a multisignature call to the network node.
This method allows the composition and submission of a call that requires multiple signatures for execution, known as a multisignature call. It supports specifying signatories, a threshold of signatures for the call's execution, and an optional era for the call's mortality. The call can be a standard extrinsic, a sudo extrinsic for elevated permissions, or a multisig extrinsic if multiple signatures are required. Optionally, the method can wait for the call's inclusion in a block and/or its finalization. Make sure to pass all keys, that are part of the multisignature.
Arguments:
- fn: The function name to call on the network. params: A dictionary
- of parameters for the call. key: The keypair for signing the
- extrinsic. signatories: List of SS58 addresses of the signatories.
- Include ALL KEYS that are part of the multisig. threshold: The
- minimum number of signatories required to execute the extrinsic.
- module: The module containing the function to call.
- wait_for_inclusion: Whether to wait for the call's inclusion in a
- block. wait_for_finalization: Whether to wait for the transaction's
- finalization. sudo: Execute the call as a sudo (superuser)
- operation. era: Specifies the call's mortality in terms of blocks in
- the format {'period': amount_blocks}. If omitted, the extrinsic is immortal.
Returns:
The receipt of the submitted extrinsic if
wait_for_inclusionis True. Otherwise, returns a string identifier of the extrinsic.
Raises:
- ChainTransactionError: If the transaction fails.
def
transfer( self, key: substrateinterface.keypair.Keypair, amount: int, dest: communex.types.Ss58Address) -> substrateinterface.base.ExtrinsicReceipt:
931 def transfer( 932 self, 933 key: Keypair, 934 amount: int, 935 dest: Ss58Address, 936 ) -> ExtrinsicReceipt: 937 """ 938 Transfers a specified amount of tokens from the signer's account to the 939 specified account. 940 941 Args: 942 key: The keypair associated with the sender's account. 943 amount: The amount to transfer, in nanotokens. 944 dest: The SS58 address of the recipient. 945 946 Returns: 947 A receipt of the transaction. 948 949 Raises: 950 InsufficientBalanceError: If the sender's account does not have 951 enough balance. 952 ChainTransactionError: If the transaction fails. 953 """ 954 955 params = {"dest": dest, "value": amount} 956 957 return self.compose_call( 958 module="Balances", fn="transfer_keep_alive", params=params, key=key 959 )
Transfers a specified amount of tokens from the signer's account to the specified account.
Arguments:
- key: The keypair associated with the sender's account.
- amount: The amount to transfer, in nanotokens.
- dest: The SS58 address of the recipient.
Returns:
A receipt of the transaction.
Raises:
- InsufficientBalanceError: If the sender's account does not have enough balance.
- ChainTransactionError: If the transaction fails.
def
transfer_multiple( self, key: substrateinterface.keypair.Keypair, destinations: list[communex.types.Ss58Address], amounts: list[int], netuid: str | int = 0) -> substrateinterface.base.ExtrinsicReceipt:
961 def transfer_multiple( 962 self, 963 key: Keypair, 964 destinations: list[Ss58Address], 965 amounts: list[int], 966 netuid: str | int = 0, 967 ) -> ExtrinsicReceipt: 968 """ 969 Transfers specified amounts of tokens from the signer's account to 970 multiple target accounts. 971 972 The `destinations` and `amounts` lists must be of the same length. 973 974 Args: 975 key: The keypair associated with the sender's account. 976 destinations: A list of SS58 addresses of the recipients. 977 amounts: Amount to transfer to each recipient, in nanotokens. 978 netuid: The network identifier. 979 980 Returns: 981 A receipt of the transaction. 982 983 Raises: 984 InsufficientBalanceError: If the sender's account does not have 985 enough balance for all transfers. 986 ChainTransactionError: If the transaction fails. 987 """ 988 989 assert len(destinations) == len(amounts) 990 991 # extract existential deposit from amounts 992 existential_deposit = self.get_existential_deposit() 993 amounts = [a - existential_deposit for a in amounts] 994 995 params = { 996 "netuid": netuid, 997 "destinations": destinations, 998 "amounts": amounts, 999 } 1000 1001 return self.compose_call( 1002 module="SubspaceModule", fn="transfer_multiple", params=params, key=key 1003 )
Transfers specified amounts of tokens from the signer's account to multiple target accounts.
The destinations and amounts lists must be of the same length.
Arguments:
- key: The keypair associated with the sender's account.
- destinations: A list of SS58 addresses of the recipients.
- amounts: Amount to transfer to each recipient, in nanotokens.
- netuid: The network identifier.
Returns:
A receipt of the transaction.
Raises:
- InsufficientBalanceError: If the sender's account does not have enough balance for all transfers.
- ChainTransactionError: If the transaction fails.
def
stake( self, key: substrateinterface.keypair.Keypair, amount: int, dest: communex.types.Ss58Address) -> substrateinterface.base.ExtrinsicReceipt:
1005 def stake( 1006 self, 1007 key: Keypair, 1008 amount: int, 1009 dest: Ss58Address, 1010 ) -> ExtrinsicReceipt: 1011 """ 1012 Stakes the specified amount of tokens to a module key address. 1013 1014 Args: 1015 key: The keypair associated with the staker's account. 1016 amount: The amount of tokens to stake, in nanotokens. 1017 dest: The SS58 address of the module key to stake to. 1018 netuid: The network identifier. 1019 1020 Returns: 1021 A receipt of the staking transaction. 1022 1023 Raises: 1024 InsufficientBalanceError: If the staker's account does not have 1025 enough balance. 1026 ChainTransactionError: If the transaction fails. 1027 """ 1028 1029 params = {"amount": amount, "module_key": dest} 1030 1031 return self.compose_call(fn="add_stake", params=params, key=key)
Stakes the specified amount of tokens to a module key address.
Arguments:
- key: The keypair associated with the staker's account.
- amount: The amount of tokens to stake, in nanotokens.
- dest: The SS58 address of the module key to stake to.
- netuid: The network identifier.
Returns:
A receipt of the staking transaction.
Raises:
- InsufficientBalanceError: If the staker's account does not have enough balance.
- ChainTransactionError: If the transaction fails.
def
unstake( self, key: substrateinterface.keypair.Keypair, amount: int, dest: communex.types.Ss58Address) -> substrateinterface.base.ExtrinsicReceipt:
1033 def unstake( 1034 self, 1035 key: Keypair, 1036 amount: int, 1037 dest: Ss58Address, 1038 ) -> ExtrinsicReceipt: 1039 """ 1040 Unstakes the specified amount of tokens from a module key address. 1041 1042 Args: 1043 key: The keypair associated with the unstaker's account. 1044 amount: The amount of tokens to unstake, in nanotokens. 1045 dest: The SS58 address of the module key to unstake from. 1046 netuid: The network identifier. 1047 1048 Returns: 1049 A receipt of the unstaking transaction. 1050 1051 Raises: 1052 InsufficientStakeError: If the staked key does not have enough 1053 staked tokens by the signer key. 1054 ChainTransactionError: If the transaction fails. 1055 """ 1056 1057 params = {"amount": amount, "module_key": dest} 1058 return self.compose_call(fn="remove_stake", params=params, key=key)
Unstakes the specified amount of tokens from a module key address.
Arguments:
- key: The keypair associated with the unstaker's account.
- amount: The amount of tokens to unstake, in nanotokens.
- dest: The SS58 address of the module key to unstake from.
- netuid: The network identifier.
Returns:
A receipt of the unstaking transaction.
Raises:
- InsufficientStakeError: If the staked key does not have enough staked tokens by the signer key.
- ChainTransactionError: If the transaction fails.
def
update_module( self, key: substrateinterface.keypair.Keypair, name: str, address: str, metadata: str | None = None, delegation_fee: int = 20, netuid: int = 0) -> substrateinterface.base.ExtrinsicReceipt:
1060 def update_module( 1061 self, 1062 key: Keypair, 1063 name: str, 1064 address: str, 1065 metadata: str | None = None, 1066 delegation_fee: int = 20, 1067 netuid: int = 0, 1068 ) -> ExtrinsicReceipt: 1069 """ 1070 Updates the parameters of a registered module. 1071 1072 The delegation fee must be an integer between 0 and 100. 1073 1074 Args: 1075 key: The keypair associated with the module's account. 1076 name: The new name for the module. If None, the name is not updated. 1077 address: The new address for the module. 1078 If None, the address is not updated. 1079 delegation_fee: The new delegation fee for the module, 1080 between 0 and 100. 1081 netuid: The network identifier. 1082 1083 Returns: 1084 A receipt of the module update transaction. 1085 1086 Raises: 1087 InvalidParameterError: If the provided parameters are invalid. 1088 ChainTransactionError: If the transaction fails. 1089 """ 1090 1091 assert isinstance(delegation_fee, int) 1092 1093 params = { 1094 "netuid": netuid, 1095 "name": name, 1096 "address": address, 1097 "delegation_fee": delegation_fee, 1098 "metadata": metadata, 1099 } 1100 1101 response = self.compose_call("update_module", params=params, key=key) 1102 1103 return response
Updates the parameters of a registered module.
The delegation fee must be an integer between 0 and 100.
Arguments:
- key: The keypair associated with the module's account.
- name: The new name for the module. If None, the name is not updated.
- address: The new address for the module. If None, the address is not updated.
- delegation_fee: The new delegation fee for the module, between 0 and 100.
- netuid: The network identifier.
Returns:
A receipt of the module update transaction.
Raises:
- InvalidParameterError: If the provided parameters are invalid.
- ChainTransactionError: If the transaction fails.
def
register_module( self, key: substrateinterface.keypair.Keypair, name: str, address: str | None = None, subnet: str = 'Rootnet', metadata: str | None = None) -> substrateinterface.base.ExtrinsicReceipt:
1105 def register_module( 1106 self, 1107 key: Keypair, 1108 name: str, 1109 address: str | None = None, 1110 subnet: str = "Rootnet", 1111 metadata: str | None = None, 1112 ) -> ExtrinsicReceipt: 1113 """ 1114 Registers a new module in the network. 1115 1116 Args: 1117 key: The keypair used for registering the module. 1118 name: The name of the module. If None, a default or previously 1119 set name is used. # How does this work? 1120 address: The address of the module. If None, a default or 1121 previously set address is used. # How does this work? 1122 subnet: The network subnet to register the module in. 1123 min_stake: The minimum stake required for the module, in nanotokens. 1124 If None, a default value is used. 1125 1126 Returns: 1127 A receipt of the registration transaction. 1128 1129 Raises: 1130 InvalidParameterError: If the provided parameters are invalid. 1131 ChainTransactionError: If the transaction fails. 1132 """ 1133 1134 key_addr = key.ss58_address 1135 1136 params = { 1137 "network_name": subnet, 1138 "address": address, 1139 "name": name, 1140 "module_key": key_addr, 1141 "metadata": metadata, 1142 } 1143 1144 response = self.compose_call("register", params=params, key=key) 1145 return response
Registers a new module in the network.
Arguments:
- key: The keypair used for registering the module.
- name: The name of the module. If None, a default or previously set name is used. # How does this work?
- address: The address of the module. If None, a default or previously set address is used. # How does this work?
- subnet: The network subnet to register the module in.
- min_stake: The minimum stake required for the module, in nanotokens. If None, a default value is used.
Returns:
A receipt of the registration transaction.
Raises:
- InvalidParameterError: If the provided parameters are invalid.
- ChainTransactionError: If the transaction fails.
def
deregister_module( self, key: substrateinterface.keypair.Keypair, netuid: int) -> substrateinterface.base.ExtrinsicReceipt:
1147 def deregister_module(self, key: Keypair, netuid: int) -> ExtrinsicReceipt: 1148 """ 1149 Deregisters a module from the network. 1150 1151 Args: 1152 key: The keypair associated with the module's account. 1153 netuid: The network identifier. 1154 1155 Returns: 1156 A receipt of the module deregistration transaction. 1157 1158 Raises: 1159 ChainTransactionError: If the transaction fails. 1160 """ 1161 1162 params = {"netuid": netuid} 1163 1164 response = self.compose_call("deregister", params=params, key=key) 1165 1166 return response
Deregisters a module from the network.
Arguments:
- key: The keypair associated with the module's account.
- netuid: The network identifier.
Returns:
A receipt of the module deregistration transaction.
Raises:
- ChainTransactionError: If the transaction fails.
def
register_subnet( self, key: substrateinterface.keypair.Keypair, name: str, metadata: str | None = None) -> substrateinterface.base.ExtrinsicReceipt:
1168 def register_subnet(self, key: Keypair, name: str, metadata: str | None = None) -> ExtrinsicReceipt: 1169 """ 1170 Registers a new subnet in the network. 1171 1172 Args: 1173 key (Keypair): The keypair used for registering the subnet. 1174 name (str): The name of the subnet to be registered. 1175 metadata (str | None, optional): Additional metadata for the subnet. Defaults to None. 1176 1177 Returns: 1178 ExtrinsicReceipt: A receipt of the subnet registration transaction. 1179 1180 Raises: 1181 ChainTransactionError: If the transaction fails. 1182 """ 1183 1184 params = { 1185 "name": name, 1186 "metadata": metadata, 1187 } 1188 1189 response = self.compose_call("register_subnet", params=params, key=key) 1190 1191 return response
Registers a new subnet in the network.
Arguments:
- key (Keypair): The keypair used for registering the subnet.
- name (str): The name of the subnet to be registered.
- metadata (str | None, optional): Additional metadata for the subnet. Defaults to None.
Returns:
ExtrinsicReceipt: A receipt of the subnet registration transaction.
Raises:
- ChainTransactionError: If the transaction fails.
def
vote( self, key: substrateinterface.keypair.Keypair, uids: list[int], weights: list[int], netuid: int = 0) -> substrateinterface.base.ExtrinsicReceipt:
1193 def vote( 1194 self, 1195 key: Keypair, 1196 uids: list[int], 1197 weights: list[int], 1198 netuid: int = 0, 1199 ) -> ExtrinsicReceipt: 1200 """ 1201 Casts votes on a list of module UIDs with corresponding weights. 1202 1203 The length of the UIDs list and the weights list should be the same. 1204 Each weight corresponds to the UID at the same index. 1205 1206 Args: 1207 key: The keypair used for signing the vote transaction. 1208 uids: A list of module UIDs to vote on. 1209 weights: A list of weights corresponding to each UID. 1210 netuid: The network identifier. 1211 1212 Returns: 1213 A receipt of the voting transaction. 1214 1215 Raises: 1216 InvalidParameterError: If the lengths of UIDs and weights lists 1217 do not match. 1218 ChainTransactionError: If the transaction fails. 1219 """ 1220 1221 assert len(uids) == len(weights) 1222 1223 params = { 1224 "uids": uids, 1225 "weights": weights, 1226 "netuid": netuid, 1227 } 1228 1229 response = self.compose_call("set_weights", params=params, key=key) 1230 1231 return response
Casts votes on a list of module UIDs with corresponding weights.
The length of the UIDs list and the weights list should be the same. Each weight corresponds to the UID at the same index.
Arguments:
- key: The keypair used for signing the vote transaction.
- uids: A list of module UIDs to vote on.
- weights: A list of weights corresponding to each UID.
- netuid: The network identifier.
Returns:
A receipt of the voting transaction.
Raises:
- InvalidParameterError: If the lengths of UIDs and weights lists do not match.
- ChainTransactionError: If the transaction fails.
def
update_subnet( self, key: substrateinterface.keypair.Keypair, params: communex.types.SubnetParams, netuid: int = 0) -> substrateinterface.base.ExtrinsicReceipt:
1233 def update_subnet( 1234 self, 1235 key: Keypair, 1236 params: SubnetParams, 1237 netuid: int = 0, 1238 ) -> ExtrinsicReceipt: 1239 """ 1240 Update a subnet's configuration. 1241 1242 It requires the founder key for authorization. 1243 1244 Args: 1245 key: The founder keypair of the subnet. 1246 params: The new parameters for the subnet. 1247 netuid: The network identifier. 1248 1249 Returns: 1250 A receipt of the subnet update transaction. 1251 1252 Raises: 1253 AuthorizationError: If the key is not authorized. 1254 ChainTransactionError: If the transaction fails. 1255 """ 1256 1257 general_params = dict(params) 1258 general_params["netuid"] = netuid 1259 if "metadata" not in general_params: 1260 general_params["metadata"] = None 1261 1262 response = self.compose_call( 1263 fn="update_subnet", 1264 params=general_params, 1265 key=key, 1266 ) 1267 1268 return response
Update a subnet's configuration.
It requires the founder key for authorization.
Arguments:
- key: The founder keypair of the subnet.
- params: The new parameters for the subnet.
- netuid: The network identifier.
Returns:
A receipt of the subnet update transaction.
Raises:
- AuthorizationError: If the key is not authorized.
- ChainTransactionError: If the transaction fails.
def
transfer_stake( self, key: substrateinterface.keypair.Keypair, amount: int, from_module_key: communex.types.Ss58Address, dest_module_address: communex.types.Ss58Address) -> substrateinterface.base.ExtrinsicReceipt:
1270 def transfer_stake( 1271 self, 1272 key: Keypair, 1273 amount: int, 1274 from_module_key: Ss58Address, 1275 dest_module_address: Ss58Address, 1276 ) -> ExtrinsicReceipt: 1277 """ 1278 Realocate staked tokens from one staked module to another module. 1279 1280 Args: 1281 key: The keypair associated with the account that is delegating the tokens. 1282 amount: The amount of staked tokens to transfer, in nanotokens. 1283 from_module_key: The SS58 address of the module you want to transfer from (currently delegated by the key). 1284 dest_module_address: The SS58 address of the destination (newly delegated key). 1285 netuid: The network identifier. 1286 1287 Returns: 1288 A receipt of the stake transfer transaction. 1289 1290 Raises: 1291 InsufficientStakeError: If the source module key does not have 1292 enough staked tokens. ChainTransactionError: If the transaction 1293 fails. 1294 """ 1295 1296 amount = amount - self.get_existential_deposit() 1297 1298 params = { 1299 "amount": amount, 1300 "module_key": from_module_key, 1301 "new_module_key": dest_module_address, 1302 } 1303 1304 response = self.compose_call("transfer_stake", key=key, params=params) 1305 1306 return response
Realocate staked tokens from one staked module to another module.
Arguments:
- key: The keypair associated with the account that is delegating the tokens.
- amount: The amount of staked tokens to transfer, in nanotokens.
- from_module_key: The SS58 address of the module you want to transfer from (currently delegated by the key).
- dest_module_address: The SS58 address of the destination (newly delegated key).
- netuid: The network identifier.
Returns:
A receipt of the stake transfer transaction.
Raises:
- InsufficientStakeError: If the source module key does not have
- enough staked tokens. ChainTransactionError: If the transaction
- fails.
def
multiunstake( self, key: substrateinterface.keypair.Keypair, keys: list[communex.types.Ss58Address], amounts: list[int]) -> substrateinterface.base.ExtrinsicReceipt:
1308 def multiunstake( 1309 self, 1310 key: Keypair, 1311 keys: list[Ss58Address], 1312 amounts: list[int], 1313 ) -> ExtrinsicReceipt: 1314 """ 1315 Unstakes tokens from multiple module keys. 1316 1317 And the lists `keys` and `amounts` must be of the same length. Each 1318 amount corresponds to the module key at the same index. 1319 1320 Args: 1321 key: The keypair associated with the unstaker's account. 1322 keys: A list of SS58 addresses of the module keys to unstake from. 1323 amounts: A list of amounts to unstake from each module key, 1324 in nanotokens. 1325 netuid: The network identifier. 1326 1327 Returns: 1328 A receipt of the multi-unstaking transaction. 1329 1330 Raises: 1331 MismatchedLengthError: If the lengths of keys and amounts lists do 1332 not match. InsufficientStakeError: If any of the module keys do not 1333 have enough staked tokens. ChainTransactionError: If the transaction 1334 fails. 1335 """ 1336 1337 assert len(keys) == len(amounts) 1338 1339 params = {"module_keys": keys, "amounts": amounts} 1340 1341 response = self.compose_call("remove_stake_multiple", params=params, key=key) 1342 1343 return response
Unstakes tokens from multiple module keys.
And the lists keys and amounts must be of the same length. Each
amount corresponds to the module key at the same index.
Arguments:
- key: The keypair associated with the unstaker's account.
- keys: A list of SS58 addresses of the module keys to unstake from.
- amounts: A list of amounts to unstake from each module key, in nanotokens.
- netuid: The network identifier.
Returns:
A receipt of the multi-unstaking transaction.
Raises:
- MismatchedLengthError: If the lengths of keys and amounts lists do
- not match. InsufficientStakeError: If any of the module keys do not
- have enough staked tokens. ChainTransactionError: If the transaction
- fails.
def
multistake( self, key: substrateinterface.keypair.Keypair, keys: list[communex.types.Ss58Address], amounts: list[int]) -> substrateinterface.base.ExtrinsicReceipt:
1345 def multistake( 1346 self, 1347 key: Keypair, 1348 keys: list[Ss58Address], 1349 amounts: list[int], 1350 ) -> ExtrinsicReceipt: 1351 """ 1352 Stakes tokens to multiple module keys. 1353 1354 The lengths of the `keys` and `amounts` lists must be the same. Each 1355 amount corresponds to the module key at the same index. 1356 1357 Args: 1358 key: The keypair associated with the staker's account. 1359 keys: A list of SS58 addresses of the module keys to stake to. 1360 amounts: A list of amounts to stake to each module key, 1361 in nanotokens. 1362 netuid: The network identifier. 1363 1364 Returns: 1365 A receipt of the multi-staking transaction. 1366 1367 Raises: 1368 MismatchedLengthError: If the lengths of keys and amounts lists 1369 do not match. 1370 ChainTransactionError: If the transaction fails. 1371 """ 1372 1373 assert len(keys) == len(amounts) 1374 1375 params = { 1376 "module_keys": keys, 1377 "amounts": amounts, 1378 } 1379 1380 response = self.compose_call("add_stake_multiple", params=params, key=key) 1381 1382 return response
Stakes tokens to multiple module keys.
The lengths of the keys and amounts lists must be the same. Each
amount corresponds to the module key at the same index.
Arguments:
- key: The keypair associated with the staker's account.
- keys: A list of SS58 addresses of the module keys to stake to.
- amounts: A list of amounts to stake to each module key, in nanotokens.
- netuid: The network identifier.
Returns:
A receipt of the multi-staking transaction.
Raises:
- MismatchedLengthError: If the lengths of keys and amounts lists do not match.
- ChainTransactionError: If the transaction fails.
def
add_subnet_proposal( self, key: substrateinterface.keypair.Keypair, params: dict[str, typing.Any], ipfs: str, netuid: int = 0) -> substrateinterface.base.ExtrinsicReceipt:
1420 def add_subnet_proposal( 1421 self, key: Keypair, 1422 params: dict[str, Any], 1423 ipfs: str, 1424 netuid: int = 0 1425 ) -> ExtrinsicReceipt: 1426 """ 1427 Submits a proposal for creating or modifying a subnet within the 1428 network. 1429 1430 The proposal includes various parameters like the name, founder, share 1431 allocations, and other subnet-specific settings. 1432 1433 Args: 1434 key: The keypair used for signing the proposal transaction. 1435 params: The parameters for the subnet proposal. 1436 netuid: The network identifier. 1437 1438 Returns: 1439 A receipt of the subnet proposal transaction. 1440 1441 Raises: 1442 InvalidParameterError: If the provided subnet 1443 parameters are invalid. 1444 ChainTransactionError: If the transaction fails. 1445 """ 1446 1447 general_params = dict(params) 1448 general_params["netuid"] = netuid 1449 general_params["data"] = ipfs 1450 if "metadata" not in general_params: 1451 general_params["metadata"] = None 1452 1453 # general_params["burn_config"] = json.dumps(general_params["burn_config"]) 1454 response = self.compose_call( 1455 fn="add_subnet_params_proposal", 1456 params=general_params, 1457 key=key, 1458 module="GovernanceModule", 1459 ) 1460 1461 return response
Submits a proposal for creating or modifying a subnet within the network.
The proposal includes various parameters like the name, founder, share allocations, and other subnet-specific settings.
Arguments:
- key: The keypair used for signing the proposal transaction.
- params: The parameters for the subnet proposal.
- netuid: The network identifier.
Returns:
A receipt of the subnet proposal transaction.
Raises:
- InvalidParameterError: If the provided subnet parameters are invalid.
- ChainTransactionError: If the transaction fails.
def
add_custom_proposal( self, key: substrateinterface.keypair.Keypair, cid: str) -> substrateinterface.base.ExtrinsicReceipt:
1463 def add_custom_proposal( 1464 self, 1465 key: Keypair, 1466 cid: str, 1467 ) -> ExtrinsicReceipt: 1468 1469 params = {"data": cid} 1470 1471 response = self.compose_call( 1472 fn="add_global_custom_proposal", 1473 params=params, 1474 key=key, 1475 module="GovernanceModule", 1476 ) 1477 return response
def
add_custom_subnet_proposal( self, key: substrateinterface.keypair.Keypair, cid: str, netuid: int = 0) -> substrateinterface.base.ExtrinsicReceipt:
1479 def add_custom_subnet_proposal( 1480 self, 1481 key: Keypair, 1482 cid: str, 1483 netuid: int = 0, 1484 ) -> ExtrinsicReceipt: 1485 """ 1486 Submits a proposal for creating or modifying a custom subnet within the 1487 network. 1488 1489 The proposal includes various parameters like the name, founder, share 1490 allocations, and other subnet-specific settings. 1491 1492 Args: 1493 key: The keypair used for signing the proposal transaction. 1494 params: The parameters for the subnet proposal. 1495 netuid: The network identifier. 1496 1497 Returns: 1498 A receipt of the subnet proposal transaction. 1499 """ 1500 1501 params = { 1502 "data": cid, 1503 "netuid": netuid, 1504 } 1505 1506 response = self.compose_call( 1507 fn="add_subnet_custom_proposal", 1508 params=params, 1509 key=key, 1510 module="GovernanceModule", 1511 ) 1512 1513 return response
Submits a proposal for creating or modifying a custom subnet within the network.
The proposal includes various parameters like the name, founder, share allocations, and other subnet-specific settings.
Arguments:
- key: The keypair used for signing the proposal transaction.
- params: The parameters for the subnet proposal.
- netuid: The network identifier.
Returns:
A receipt of the subnet proposal transaction.
def
add_global_proposal( self, key: substrateinterface.keypair.Keypair, params: communex.types.NetworkParams, cid: str | None) -> substrateinterface.base.ExtrinsicReceipt:
1515 def add_global_proposal( 1516 self, 1517 key: Keypair, 1518 params: NetworkParams, 1519 cid: str | None, 1520 ) -> ExtrinsicReceipt: 1521 """ 1522 Submits a proposal for altering the global network parameters. 1523 1524 Allows for the submission of a proposal to 1525 change various global parameters 1526 of the network, such as emission rates, rate limits, and voting 1527 thresholds. It is used to 1528 suggest changes that affect the entire network's operation. 1529 1530 Args: 1531 key: The keypair used for signing the proposal transaction. 1532 params: A dictionary containing global network parameters 1533 like maximum allowed subnets, modules, 1534 transaction rate limits, and others. 1535 1536 Returns: 1537 A receipt of the global proposal transaction. 1538 1539 Raises: 1540 InvalidParameterError: If the provided network 1541 parameters are invalid. 1542 ChainTransactionError: If the transaction fails. 1543 """ 1544 general_params = cast(dict[str, Any], params) 1545 cid = cid or "" 1546 general_params["data"] = cid 1547 1548 response = self.compose_call( 1549 fn="add_global_params_proposal", 1550 params=general_params, 1551 key=key, 1552 module="GovernanceModule", 1553 ) 1554 1555 return response
Submits a proposal for altering the global network parameters.
Allows for the submission of a proposal to change various global parameters of the network, such as emission rates, rate limits, and voting thresholds. It is used to suggest changes that affect the entire network's operation.
Arguments:
- key: The keypair used for signing the proposal transaction.
- params: A dictionary containing global network parameters like maximum allowed subnets, modules, transaction rate limits, and others.
Returns:
A receipt of the global proposal transaction.
Raises:
- InvalidParameterError: If the provided network parameters are invalid.
- ChainTransactionError: If the transaction fails.
def
vote_on_proposal( self, key: substrateinterface.keypair.Keypair, proposal_id: int, agree: bool) -> substrateinterface.base.ExtrinsicReceipt:
1557 def vote_on_proposal( 1558 self, 1559 key: Keypair, 1560 proposal_id: int, 1561 agree: bool, 1562 ) -> ExtrinsicReceipt: 1563 """ 1564 Casts a vote on a specified proposal within the network. 1565 1566 Args: 1567 key: The keypair used for signing the vote transaction. 1568 proposal_id: The unique identifier of the proposal to vote on. 1569 1570 Returns: 1571 A receipt of the voting transaction in nanotokens. 1572 1573 Raises: 1574 InvalidProposalIDError: If the provided proposal ID does not 1575 exist or is invalid. 1576 ChainTransactionError: If the transaction fails. 1577 """ 1578 1579 params = {"proposal_id": proposal_id, "agree": agree} 1580 1581 response = self.compose_call( 1582 "vote_proposal", 1583 key=key, 1584 params=params, 1585 module="GovernanceModule", 1586 ) 1587 1588 return response
Casts a vote on a specified proposal within the network.
Arguments:
- key: The keypair used for signing the vote transaction.
- proposal_id: The unique identifier of the proposal to vote on.
Returns:
A receipt of the voting transaction in nanotokens.
Raises:
- InvalidProposalIDError: If the provided proposal ID does not exist or is invalid.
- ChainTransactionError: If the transaction fails.
def
unvote_on_proposal( self, key: substrateinterface.keypair.Keypair, proposal_id: int) -> substrateinterface.base.ExtrinsicReceipt:
1590 def unvote_on_proposal( 1591 self, 1592 key: Keypair, 1593 proposal_id: int, 1594 ) -> ExtrinsicReceipt: 1595 """ 1596 Retracts a previously cast vote on a specified proposal. 1597 1598 Args: 1599 key: The keypair used for signing the unvote transaction. 1600 proposal_id: The unique identifier of the proposal to withdraw the 1601 vote from. 1602 1603 Returns: 1604 A receipt of the unvoting transaction in nanotokens. 1605 1606 Raises: 1607 InvalidProposalIDError: If the provided proposal ID does not 1608 exist or is invalid. 1609 ChainTransactionError: If the transaction fails to be processed, or 1610 if there was no prior vote to retract. 1611 """ 1612 1613 params = {"proposal_id": proposal_id} 1614 1615 response = self.compose_call( 1616 "remove_vote_proposal", 1617 key=key, 1618 params=params, 1619 module="GovernanceModule", 1620 ) 1621 1622 return response
Retracts a previously cast vote on a specified proposal.
Arguments:
- key: The keypair used for signing the unvote transaction.
- proposal_id: The unique identifier of the proposal to withdraw the vote from.
Returns:
A receipt of the unvoting transaction in nanotokens.
Raises:
- InvalidProposalIDError: If the provided proposal ID does not exist or is invalid.
- ChainTransactionError: If the transaction fails to be processed, or if there was no prior vote to retract.
def
enable_vote_power_delegation( self, key: substrateinterface.keypair.Keypair) -> substrateinterface.base.ExtrinsicReceipt:
1624 def enable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1625 """ 1626 Enables vote power delegation for the signer's account. 1627 1628 Args: 1629 key: The keypair used for signing the delegation transaction. 1630 1631 Returns: 1632 A receipt of the vote power delegation transaction. 1633 1634 Raises: 1635 ChainTransactionError: If the transaction fails. 1636 """ 1637 1638 response = self.compose_call( 1639 "enable_vote_power_delegation", 1640 params={}, 1641 key=key, 1642 module="GovernanceModule", 1643 ) 1644 1645 return response
Enables vote power delegation for the signer's account.
Arguments:
- key: The keypair used for signing the delegation transaction.
Returns:
A receipt of the vote power delegation transaction.
Raises:
- ChainTransactionError: If the transaction fails.
def
disable_vote_power_delegation( self, key: substrateinterface.keypair.Keypair) -> substrateinterface.base.ExtrinsicReceipt:
1647 def disable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1648 """ 1649 Disables vote power delegation for the signer's account. 1650 1651 Args: 1652 key: The keypair used for signing the delegation transaction. 1653 1654 Returns: 1655 A receipt of the vote power delegation transaction. 1656 1657 Raises: 1658 ChainTransactionError: If the transaction fails. 1659 """ 1660 1661 response = self.compose_call( 1662 "disable_vote_power_delegation", 1663 params={}, 1664 key=key, 1665 module="GovernanceModule", 1666 ) 1667 1668 return response
Disables vote power delegation for the signer's account.
Arguments:
- key: The keypair used for signing the delegation transaction.
Returns:
A receipt of the vote power delegation transaction.
Raises:
- ChainTransactionError: If the transaction fails.
def
add_dao_application( self, key: substrateinterface.keypair.Keypair, application_key: communex.types.Ss58Address, data: str) -> substrateinterface.base.ExtrinsicReceipt:
1670 def add_dao_application( 1671 self, key: Keypair, application_key: Ss58Address, data: str 1672 ) -> ExtrinsicReceipt: 1673 """ 1674 Submits a new application to the general subnet DAO. 1675 1676 Args: 1677 key: The keypair used for signing the application transaction. 1678 application_key: The SS58 address of the application key. 1679 data: The data associated with the application. 1680 1681 Returns: 1682 A receipt of the application transaction. 1683 1684 Raises: 1685 ChainTransactionError: If the transaction fails. 1686 """ 1687 1688 params = {"application_key": application_key, "data": data} 1689 1690 response = self.compose_call( 1691 "add_dao_application", module="GovernanceModule", key=key, 1692 params=params 1693 ) 1694 1695 return response
Submits a new application to the general subnet DAO.
Arguments:
- key: The keypair used for signing the application transaction.
- application_key: The SS58 address of the application key.
- data: The data associated with the application.
Returns:
A receipt of the application transaction.
Raises:
- ChainTransactionError: If the transaction fails.
def
query_map_proposals(self, extract_value: bool = False) -> dict[int, dict[str, typing.Any]]:
1705 def query_map_proposals( 1706 self, extract_value: bool = False 1707 ) -> dict[int, dict[str, Any]]: 1708 """ 1709 Retrieves a mappping of proposals from the network. 1710 1711 Queries the network and returns a mapping of proposal IDs to 1712 their respective parameters. 1713 1714 Returns: 1715 A dictionary mapping proposal IDs 1716 to dictionaries of their parameters. 1717 1718 Raises: 1719 QueryError: If the query to the network fails or is invalid. 1720 """ 1721 1722 return self.query_map( 1723 "Proposals", extract_value=extract_value, module="GovernanceModule" 1724 )["Proposals"]
Retrieves a mappping of proposals from the network.
Queries the network and returns a mapping of proposal IDs to their respective parameters.
Returns:
A dictionary mapping proposal IDs to dictionaries of their parameters.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_weights( self, netuid: int = 0, extract_value: bool = False) -> dict[int, list[tuple[int, int]]] | None:
1726 def query_map_weights( 1727 self, netuid: int = 0, extract_value: bool = False 1728 ) -> dict[int, list[tuple[int, int]]] | None: 1729 """ 1730 Retrieves a mapping of weights for keys on the network. 1731 1732 Queries the network and returns a mapping of key UIDs to 1733 their respective weights. 1734 1735 Args: 1736 netuid: The network UID from which to get the weights. 1737 1738 Returns: 1739 A dictionary mapping key UIDs to lists of their weights. 1740 1741 Raises: 1742 QueryError: If the query to the network fails or is invalid. 1743 """ 1744 1745 weights_dict = self.query_map( 1746 "Weights", 1747 [netuid], 1748 extract_value=extract_value 1749 ).get("Weights") 1750 return weights_dict
Retrieves a mapping of weights for keys on the network.
Queries the network and returns a mapping of key UIDs to their respective weights.
Arguments:
- netuid: The network UID from which to get the weights.
Returns:
A dictionary mapping key UIDs to lists of their weights.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_key( self, netuid: int = 0, extract_value: bool = False) -> dict[int, communex.types.Ss58Address]:
1752 def query_map_key( 1753 self, 1754 netuid: int = 0, 1755 extract_value: bool = False, 1756 ) -> dict[int, Ss58Address]: 1757 """ 1758 Retrieves a map of keys from the network. 1759 1760 Fetches a mapping of key UIDs to their associated 1761 addresses on the network. 1762 The query can be targeted at a specific network UID if required. 1763 1764 Args: 1765 netuid: The network UID from which to get the keys. 1766 1767 Returns: 1768 A dictionary mapping key UIDs to their addresses. 1769 1770 Raises: 1771 QueryError: If the query to the network fails or is invalid. 1772 """ 1773 return self.query_map("Keys", [netuid], extract_value=extract_value)["Keys"]
Retrieves a map of keys from the network.
Fetches a mapping of key UIDs to their associated addresses on the network. The query can be targeted at a specific network UID if required.
Arguments:
- netuid: The network UID from which to get the keys.
Returns:
A dictionary mapping key UIDs to their addresses.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_address(self, netuid: int = 0, extract_value: bool = False) -> dict[int, str]:
1775 def query_map_address( 1776 self, netuid: int = 0, extract_value: bool = False 1777 ) -> dict[int, str]: 1778 """ 1779 Retrieves a map of key addresses from the network. 1780 1781 Queries the network for a mapping of key UIDs to their addresses. 1782 1783 Args: 1784 netuid: The network UID from which to get the addresses. 1785 1786 Returns: 1787 A dictionary mapping key UIDs to their addresses. 1788 1789 Raises: 1790 QueryError: If the query to the network fails or is invalid. 1791 """ 1792 1793 return self.query_map("Address", [netuid], extract_value=extract_value)[ 1794 "Address" 1795 ]
Retrieves a map of key addresses from the network.
Queries the network for a mapping of key UIDs to their addresses.
Arguments:
- netuid: The network UID from which to get the addresses.
Returns:
A dictionary mapping key UIDs to their addresses.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_emission(self, extract_value: bool = False) -> dict[int, list[int]]:
1797 def query_map_emission(self, extract_value: bool = False) -> dict[int, list[int]]: 1798 """ 1799 Retrieves a map of emissions for keys on the network. 1800 1801 Queries the network to get a mapping of 1802 key UIDs to their emission values. 1803 1804 Returns: 1805 A dictionary mapping key UIDs to lists of their emission values. 1806 1807 Raises: 1808 QueryError: If the query to the network fails or is invalid. 1809 """ 1810 1811 return self.query_map("Emission", extract_value=extract_value)["Emission"]
Retrieves a map of emissions for keys on the network.
Queries the network to get a mapping of key UIDs to their emission values.
Returns:
A dictionary mapping key UIDs to lists of their emission values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_pending_emission(self, extract_value: bool = False) -> int:
1813 def query_map_pending_emission(self, extract_value: bool = False) -> int: 1814 """ 1815 Retrieves a map of pending emissions for the subnets. 1816 1817 Queries the network for a mapping of subnet UIDs to their pending emission values. 1818 1819 Returns: 1820 A dictionary mapping subnet UIDs to their pending emission values. 1821 1822 Raises: 1823 QueryError: If the query to the network fails or is invalid. 1824 """ 1825 return self.query_map("PendingEmission", extract_value=extract_value, module="SubnetEmissionModule")["PendingEmission"]
Retrieves a map of pending emissions for the subnets.
Queries the network for a mapping of subnet UIDs to their pending emission values.
Returns:
A dictionary mapping subnet UIDs to their pending emission values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_subnet_emission(self, extract_value: bool = False) -> dict[int, int]:
1827 def query_map_subnet_emission(self, extract_value: bool = False) -> dict[int, int]: 1828 """ 1829 Retrieves a map of subnet emissions for the network. 1830 1831 Queries the network for a mapping of subnet UIDs to their emission values. 1832 1833 Returns: 1834 A dictionary mapping subnet UIDs to their emission values. 1835 1836 Raises: 1837 QueryError: If the query to the network fails or is invalid. 1838 """ 1839 1840 return self.query_map("SubnetEmission", extract_value=extract_value, module="SubnetEmissionModule")["SubnetEmission"]
Retrieves a map of subnet emissions for the network.
Queries the network for a mapping of subnet UIDs to their emission values.
Returns:
A dictionary mapping subnet UIDs to their emission values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_subnet_consensus(self, extract_value: bool = False) -> dict[int, str]:
1842 def query_map_subnet_consensus(self, extract_value: bool = False) -> dict[int, str]: 1843 """ 1844 Retrieves a map of subnet consensus types for the network. 1845 1846 Queries the network for a mapping of subnet UIDs to their consensus types. 1847 1848 Returns: 1849 A dictionary mapping subnet UIDs to their consensus types. 1850 1851 Raises: 1852 QueryError: If the query to the network fails or is invalid. 1853 """ 1854 1855 return self.query_map("SubnetConsensusType", extract_value=extract_value, module="SubnetEmissionModule")["SubnetConsensusType"]
Retrieves a map of subnet consensus types for the network.
Queries the network for a mapping of subnet UIDs to their consensus types.
Returns:
A dictionary mapping subnet UIDs to their consensus types.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_incentive(self, extract_value: bool = False) -> dict[int, list[int]]:
1857 def query_map_incentive(self, extract_value: bool = False) -> dict[int, list[int]]: 1858 """ 1859 Retrieves a mapping of incentives for keys on the network. 1860 1861 Queries the network and returns a mapping of key UIDs to 1862 their respective incentive values. 1863 1864 Returns: 1865 A dictionary mapping key UIDs to lists of their incentive values. 1866 1867 Raises: 1868 QueryError: If the query to the network fails or is invalid. 1869 """ 1870 1871 return self.query_map("Incentive", extract_value=extract_value)["Incentive"]
Retrieves a mapping of incentives for keys on the network.
Queries the network and returns a mapping of key UIDs to their respective incentive values.
Returns:
A dictionary mapping key UIDs to lists of their incentive values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_dividend(self, extract_value: bool = False) -> dict[int, list[int]]:
1873 def query_map_dividend(self, extract_value: bool = False) -> dict[int, list[int]]: 1874 """ 1875 Retrieves a mapping of dividends for keys on the network. 1876 1877 Queries the network for a mapping of key UIDs to 1878 their dividend values. 1879 1880 Returns: 1881 A dictionary mapping key UIDs to lists of their dividend values. 1882 1883 Raises: 1884 QueryError: If the query to the network fails or is invalid. 1885 """ 1886 1887 return self.query_map("Dividends", extract_value=extract_value)["Dividends"]
Retrieves a mapping of dividends for keys on the network.
Queries the network for a mapping of key UIDs to their dividend values.
Returns:
A dictionary mapping key UIDs to lists of their dividend values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_regblock(self, netuid: int = 0, extract_value: bool = False) -> dict[int, int]:
1889 def query_map_regblock( 1890 self, netuid: int = 0, extract_value: bool = False 1891 ) -> dict[int, int]: 1892 """ 1893 Retrieves a mapping of registration blocks for keys on the network. 1894 1895 Queries the network for a mapping of key UIDs to 1896 the blocks where they were registered. 1897 1898 Args: 1899 netuid: The network UID from which to get the registration blocks. 1900 1901 Returns: 1902 A dictionary mapping key UIDs to their registration blocks. 1903 1904 Raises: 1905 QueryError: If the query to the network fails or is invalid. 1906 """ 1907 1908 return self.query_map( 1909 "RegistrationBlock", [netuid], extract_value=extract_value 1910 )["RegistrationBlock"]
Retrieves a mapping of registration blocks for keys on the network.
Queries the network for a mapping of key UIDs to the blocks where they were registered.
Arguments:
- netuid: The network UID from which to get the registration blocks.
Returns:
A dictionary mapping key UIDs to their registration blocks.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_lastupdate(self, extract_value: bool = False) -> dict[int, list[int]]:
1912 def query_map_lastupdate(self, extract_value: bool = False) -> dict[int, list[int]]: 1913 """ 1914 Retrieves a mapping of the last update times for keys on the network. 1915 1916 Queries the network for a mapping of key UIDs to their last update times. 1917 1918 Returns: 1919 A dictionary mapping key UIDs to lists of their last update times. 1920 1921 Raises: 1922 QueryError: If the query to the network fails or is invalid. 1923 """ 1924 1925 return self.query_map("LastUpdate", extract_value=extract_value)["LastUpdate"]
Retrieves a mapping of the last update times for keys on the network.
Queries the network for a mapping of key UIDs to their last update times.
Returns:
A dictionary mapping key UIDs to lists of their last update times.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_stakefrom( self, extract_value: bool = False) -> dict[communex.types.Ss58Address, list[tuple[communex.types.Ss58Address, int]]]:
1927 def query_map_stakefrom( 1928 self, extract_value: bool = False 1929 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1930 """ 1931 Retrieves a mapping of stakes from various sources for keys on the network. 1932 1933 Queries the network to obtain a mapping of key addresses to the sources 1934 and amounts of stakes they have received. 1935 1936 Args: 1937 netuid: The network UID from which to get the stakes. 1938 1939 Returns: 1940 A dictionary mapping key addresses to lists of tuples 1941 (module_key_address, amount). 1942 1943 Raises: 1944 QueryError: If the query to the network fails or is invalid. 1945 """ 1946 1947 result = self.query_map("StakeFrom", [], extract_value=extract_value)[ 1948 "StakeFrom" 1949 ] 1950 1951 return transform_stake_dmap(result)
Retrieves a mapping of stakes from various sources for keys on the network.
Queries the network to obtain a mapping of key addresses to the sources and amounts of stakes they have received.
Arguments:
- netuid: The network UID from which to get the stakes.
Returns:
A dictionary mapping key addresses to lists of tuples (module_key_address, amount).
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_staketo( self, extract_value: bool = False) -> dict[communex.types.Ss58Address, list[tuple[communex.types.Ss58Address, int]]]:
1953 def query_map_staketo( 1954 self, extract_value: bool = False 1955 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1956 """ 1957 Retrieves a mapping of stakes to destinations for keys on the network. 1958 1959 Queries the network for a mapping of key addresses to the destinations 1960 and amounts of stakes they have made. 1961 1962 Args: 1963 netuid: The network UID from which to get the stakes. 1964 1965 Returns: 1966 A dictionary mapping key addresses to lists of tuples 1967 (module_key_address, amount). 1968 1969 Raises: 1970 QueryError: If the query to the network fails or is invalid. 1971 """ 1972 1973 result = self.query_map("StakeTo", [], extract_value=extract_value)[ 1974 "StakeTo" 1975 ] 1976 return transform_stake_dmap(result)
Retrieves a mapping of stakes to destinations for keys on the network.
Queries the network for a mapping of key addresses to the destinations and amounts of stakes they have made.
Arguments:
- netuid: The network UID from which to get the stakes.
Returns:
A dictionary mapping key addresses to lists of tuples (module_key_address, amount).
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_delegationfee(self, netuid: int = 0, extract_value: bool = False) -> dict[str, int]:
1978 def query_map_delegationfee( 1979 self, netuid: int = 0, extract_value: bool = False 1980 ) -> dict[str, int]: 1981 """ 1982 Retrieves a mapping of delegation fees for keys on the network. 1983 1984 Queries the network to obtain a mapping of key addresses to their 1985 respective delegation fees. 1986 1987 Args: 1988 netuid: The network UID to filter the delegation fees. 1989 1990 Returns: 1991 A dictionary mapping key addresses to their delegation fees. 1992 1993 Raises: 1994 QueryError: If the query to the network fails or is invalid. 1995 """ 1996 1997 return self.query_map("DelegationFee", [netuid], extract_value=extract_value)[ 1998 "DelegationFee" 1999 ]
Retrieves a mapping of delegation fees for keys on the network.
Queries the network to obtain a mapping of key addresses to their respective delegation fees.
Arguments:
- netuid: The network UID to filter the delegation fees.
Returns:
A dictionary mapping key addresses to their delegation fees.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_tempo(self, extract_value: bool = False) -> dict[int, int]:
2001 def query_map_tempo(self, extract_value: bool = False) -> dict[int, int]: 2002 """ 2003 Retrieves a mapping of tempo settings for the network. 2004 2005 Queries the network to obtain the tempo (rate of reward distributions) 2006 settings for various network subnets. 2007 2008 Returns: 2009 A dictionary mapping network UIDs to their tempo settings. 2010 2011 Raises: 2012 QueryError: If the query to the network fails or is invalid. 2013 """ 2014 2015 return self.query_map("Tempo", extract_value=extract_value)["Tempo"]
Retrieves a mapping of tempo settings for the network.
Queries the network to obtain the tempo (rate of reward distributions) settings for various network subnets.
Returns:
A dictionary mapping network UIDs to their tempo settings.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_immunity_period(self, extract_value: bool) -> dict[int, int]:
2017 def query_map_immunity_period(self, extract_value: bool) -> dict[int, int]: 2018 """ 2019 Retrieves a mapping of immunity periods for the network. 2020 2021 Queries the network for the immunity period settings, 2022 which represent the time duration during which modules 2023 can not get deregistered. 2024 2025 Returns: 2026 A dictionary mapping network UIDs to their immunity period settings. 2027 2028 Raises: 2029 QueryError: If the query to the network fails or is invalid. 2030 """ 2031 2032 return self.query_map("ImmunityPeriod", extract_value=extract_value)[ 2033 "ImmunityPeriod" 2034 ]
Retrieves a mapping of immunity periods for the network.
Queries the network for the immunity period settings, which represent the time duration during which modules can not get deregistered.
Returns:
A dictionary mapping network UIDs to their immunity period settings.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_min_allowed_weights(self, extract_value: bool = False) -> dict[int, int]:
2036 def query_map_min_allowed_weights( 2037 self, extract_value: bool = False 2038 ) -> dict[int, int]: 2039 """ 2040 Retrieves a mapping of minimum allowed weights for the network. 2041 2042 Queries the network to obtain the minimum allowed weights, 2043 which are the lowest permissible weight values that can be set by 2044 validators. 2045 2046 Returns: 2047 A dictionary mapping network UIDs to 2048 their minimum allowed weight values. 2049 2050 Raises: 2051 QueryError: If the query to the network fails or is invalid. 2052 """ 2053 2054 return self.query_map("MinAllowedWeights", extract_value=extract_value)[ 2055 "MinAllowedWeights" 2056 ]
Retrieves a mapping of minimum allowed weights for the network.
Queries the network to obtain the minimum allowed weights, which are the lowest permissible weight values that can be set by validators.
Returns:
A dictionary mapping network UIDs to their minimum allowed weight values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_max_allowed_weights(self, extract_value: bool = False) -> dict[int, int]:
2058 def query_map_max_allowed_weights( 2059 self, extract_value: bool = False 2060 ) -> dict[int, int]: 2061 """ 2062 Retrieves a mapping of maximum allowed weights for the network. 2063 2064 Queries the network for the maximum allowed weights, 2065 which are the highest permissible 2066 weight values that can be set by validators. 2067 2068 Returns: 2069 A dictionary mapping network UIDs to 2070 their maximum allowed weight values. 2071 2072 Raises: 2073 QueryError: If the query to the network fails or is invalid. 2074 """ 2075 2076 return self.query_map("MaxAllowedWeights", extract_value=extract_value)[ 2077 "MaxAllowedWeights" 2078 ]
Retrieves a mapping of maximum allowed weights for the network.
Queries the network for the maximum allowed weights, which are the highest permissible weight values that can be set by validators.
Returns:
A dictionary mapping network UIDs to their maximum allowed weight values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_max_allowed_uids(self, extract_value: bool = False) -> dict[int, int]:
2080 def query_map_max_allowed_uids(self, extract_value: bool = False) -> dict[int, int]: 2081 """ 2082 Queries the network for the maximum number of allowed user IDs (UIDs) 2083 for each network subnet. 2084 2085 Fetches a mapping of network subnets to their respective 2086 limits on the number of user IDs that can be created or used. 2087 2088 Returns: 2089 A dictionary mapping network UIDs (unique identifiers) to their 2090 maximum allowed number of UIDs. 2091 Each entry represents a network subnet 2092 with its corresponding UID limit. 2093 2094 Raises: 2095 QueryError: If the query to the network fails or is invalid. 2096 """ 2097 2098 return self.query_map("MaxAllowedUids", extract_value=extract_value)[ 2099 "MaxAllowedUids" 2100 ]
Queries the network for the maximum number of allowed user IDs (UIDs) for each network subnet.
Fetches a mapping of network subnets to their respective limits on the number of user IDs that can be created or used.
Returns:
A dictionary mapping network UIDs (unique identifiers) to their maximum allowed number of UIDs. Each entry represents a network subnet with its corresponding UID limit.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_min_stake(self, extract_value: bool = False) -> dict[int, int]:
2102 def query_map_min_stake(self, extract_value: bool = False) -> dict[int, int]: 2103 """ 2104 Retrieves a mapping of minimum allowed stake on the network. 2105 2106 Queries the network to obtain the minimum number of stake, 2107 which is represented in nanotokens. 2108 2109 Returns: 2110 A dictionary mapping network UIDs to 2111 their minimum allowed stake values. 2112 2113 Raises: 2114 QueryError: If the query to the network fails or is invalid. 2115 """ 2116 2117 return self.query_map("MinStake", extract_value=extract_value)["MinStake"]
Retrieves a mapping of minimum allowed stake on the network.
Queries the network to obtain the minimum number of stake, which is represented in nanotokens.
Returns:
A dictionary mapping network UIDs to their minimum allowed stake values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_max_stake(self, extract_value: bool = False) -> dict[int, int]:
2119 def query_map_max_stake(self, extract_value: bool = False) -> dict[int, int]: 2120 """ 2121 Retrieves a mapping of the maximum stake values for the network. 2122 2123 Queries the network for the maximum stake values across various s 2124 ubnets of the network. 2125 2126 Returns: 2127 A dictionary mapping network UIDs to their maximum stake values. 2128 2129 Raises: 2130 QueryError: If the query to the network fails or is invalid. 2131 """ 2132 2133 return self.query_map("MaxStake", extract_value=extract_value)["MaxStake"]
Retrieves a mapping of the maximum stake values for the network.
Queries the network for the maximum stake values across various s ubnets of the network.
Returns:
A dictionary mapping network UIDs to their maximum stake values.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_founder(self, extract_value: bool = False) -> dict[int, str]:
2135 def query_map_founder(self, extract_value: bool = False) -> dict[int, str]: 2136 """ 2137 Retrieves a mapping of founders for the network. 2138 2139 Queries the network to obtain the founders associated with 2140 various subnets. 2141 2142 Returns: 2143 A dictionary mapping network UIDs to their respective founders. 2144 2145 Raises: 2146 QueryError: If the query to the network fails or is invalid. 2147 """ 2148 2149 return self.query_map("Founder", extract_value=extract_value)["Founder"]
Retrieves a mapping of founders for the network.
Queries the network to obtain the founders associated with various subnets.
Returns:
A dictionary mapping network UIDs to their respective founders.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_incentive_ratio(self, extract_value: bool = False) -> dict[int, int]:
2169 def query_map_incentive_ratio(self, extract_value: bool = False) -> dict[int, int]: 2170 """ 2171 Retrieves a mapping of incentive ratios for the network. 2172 2173 Queries the network for the incentive ratios, 2174 which are the proportions of rewards or incentives 2175 allocated in different subnets of the network. 2176 2177 Returns: 2178 A dictionary mapping network UIDs to their incentive ratios. 2179 2180 Raises: 2181 QueryError: If the query to the network fails or is invalid. 2182 """ 2183 2184 return self.query_map("IncentiveRatio", extract_value=extract_value)[ 2185 "IncentiveRatio" 2186 ]
Retrieves a mapping of incentive ratios for the network.
Queries the network for the incentive ratios, which are the proportions of rewards or incentives allocated in different subnets of the network.
Returns:
A dictionary mapping network UIDs to their incentive ratios.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_trust_ratio(self, extract_value: bool = False) -> dict[int, int]:
2188 def query_map_trust_ratio(self, extract_value: bool = False) -> dict[int, int]: 2189 """ 2190 Retrieves a mapping of trust ratios for the network. 2191 2192 Queries the network for trust ratios, 2193 indicative of the level of trust or credibility assigned 2194 to different subnets of the network. 2195 2196 Returns: 2197 A dictionary mapping network UIDs to their trust ratios. 2198 2199 Raises: 2200 QueryError: If the query to the network fails or is invalid. 2201 """ 2202 2203 return self.query_map("TrustRatio", extract_value=extract_value)["TrustRatio"]
Retrieves a mapping of trust ratios for the network.
Queries the network for trust ratios, indicative of the level of trust or credibility assigned to different subnets of the network.
Returns:
A dictionary mapping network UIDs to their trust ratios.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_vote_mode_subnet(self, extract_value: bool = False) -> dict[int, str]:
2205 def query_map_vote_mode_subnet(self, extract_value: bool = False) -> dict[int, str]: 2206 """ 2207 Retrieves a mapping of vote modes for subnets within the network. 2208 2209 Queries the network for the voting modes used in different 2210 subnets, which define the methodology or approach of voting within those 2211 subnets. 2212 2213 Returns: 2214 A dictionary mapping network UIDs to their vote 2215 modes for subnets. 2216 2217 Raises: 2218 QueryError: If the query to the network fails or is invalid. 2219 """ 2220 2221 return self.query_map("VoteModeSubnet", extract_value=extract_value)[ 2222 "VoteModeSubnet" 2223 ]
Retrieves a mapping of vote modes for subnets within the network.
Queries the network for the voting modes used in different subnets, which define the methodology or approach of voting within those subnets.
Returns:
A dictionary mapping network UIDs to their vote modes for subnets.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_legit_whitelist( self, extract_value: bool = False) -> dict[communex.types.Ss58Address, int]:
2225 def query_map_legit_whitelist( 2226 self, extract_value: bool = False 2227 ) -> dict[Ss58Address, int]: 2228 """ 2229 Retrieves a mapping of whitelisted addresses for the network. 2230 2231 Queries the network for a mapping of whitelisted addresses 2232 and their respective legitimacy status. 2233 2234 Returns: 2235 A dictionary mapping addresses to their legitimacy status. 2236 2237 Raises: 2238 QueryError: If the query to the network fails or is invalid. 2239 """ 2240 2241 return self.query_map( 2242 "LegitWhitelist", module="GovernanceModule", extract_value=extract_value)[ 2243 "LegitWhitelist" 2244 ]
Retrieves a mapping of whitelisted addresses for the network.
Queries the network for a mapping of whitelisted addresses and their respective legitimacy status.
Returns:
A dictionary mapping addresses to their legitimacy status.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_subnet_names(self, extract_value: bool = False) -> dict[int, str]:
2246 def query_map_subnet_names(self, extract_value: bool = False) -> dict[int, str]: 2247 """ 2248 Retrieves a mapping of subnet names within the network. 2249 2250 Queries the network for the names of various subnets, 2251 providing an overview of the different 2252 subnets within the network. 2253 2254 Returns: 2255 A dictionary mapping network UIDs to their subnet names. 2256 2257 Raises: 2258 QueryError: If the query to the network fails or is invalid. 2259 """ 2260 2261 return self.query_map("SubnetNames", extract_value=extract_value)["SubnetNames"]
Retrieves a mapping of subnet names within the network.
Queries the network for the names of various subnets, providing an overview of the different subnets within the network.
Returns:
A dictionary mapping network UIDs to their subnet names.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_balances( self, extract_value: bool = False, block_hash: str | None = None) -> dict[str, dict[str, int | dict[str, int | float]]]:
2263 def query_map_balances( 2264 self, extract_value: bool = False, block_hash: str | None = None 2265 ) -> dict[str, dict[str, int | dict[str, int | float]]]: 2266 """ 2267 Retrieves a mapping of account balances within the network. 2268 2269 Queries the network for the balances associated with different accounts. 2270 It provides detailed information including various types of 2271 balances for each account. 2272 2273 Returns: 2274 A dictionary mapping account addresses to their balance details. 2275 2276 Raises: 2277 QueryError: If the query to the network fails or is invalid. 2278 """ 2279 2280 return self.query_map("Account", module="System", extract_value=extract_value, block_hash=block_hash)[ 2281 "Account" 2282 ]
Retrieves a mapping of account balances within the network.
Queries the network for the balances associated with different accounts. It provides detailed information including various types of balances for each account.
Returns:
A dictionary mapping account addresses to their balance details.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_registration_blocks(self, netuid: int = 0, extract_value: bool = False) -> dict[int, int]:
2284 def query_map_registration_blocks( 2285 self, netuid: int = 0, extract_value: bool = False 2286 ) -> dict[int, int]: 2287 """ 2288 Retrieves a mapping of registration blocks for UIDs on the network. 2289 2290 Queries the network to find the block numbers at which various 2291 UIDs were registered. 2292 2293 Args: 2294 netuid: The network UID from which to get the registrations. 2295 2296 Returns: 2297 A dictionary mapping UIDs to their registration block numbers. 2298 2299 Raises: 2300 QueryError: If the query to the network fails or is invalid. 2301 """ 2302 2303 return self.query_map( 2304 "RegistrationBlock", [netuid], extract_value=extract_value 2305 )["RegistrationBlock"]
Retrieves a mapping of registration blocks for UIDs on the network.
Queries the network to find the block numbers at which various UIDs were registered.
Arguments:
- netuid: The network UID from which to get the registrations.
Returns:
A dictionary mapping UIDs to their registration block numbers.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
query_map_name(self, netuid: int = 0, extract_value: bool = False) -> dict[int, str]:
2307 def query_map_name( 2308 self, netuid: int = 0, extract_value: bool = False 2309 ) -> dict[int, str]: 2310 """ 2311 Retrieves a mapping of names for keys on the network. 2312 2313 Queries the network for the names associated with different keys. 2314 It provides a mapping of key UIDs to their registered names. 2315 2316 Args: 2317 netuid: The network UID from which to get the names. 2318 2319 Returns: 2320 A dictionary mapping key UIDs to their names. 2321 2322 Raises: 2323 QueryError: If the query to the network fails or is invalid. 2324 """ 2325 2326 return self.query_map("Name", [netuid], extract_value=extract_value)["Name"]
Retrieves a mapping of names for keys on the network.
Queries the network for the names associated with different keys. It provides a mapping of key UIDs to their registered names.
Arguments:
- netuid: The network UID from which to get the names.
Returns:
A dictionary mapping key UIDs to their names.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_immunity_period(self, netuid: int = 0) -> int:
2330 def get_immunity_period(self, netuid: int = 0) -> int: 2331 """ 2332 Queries the network for the immunity period setting. 2333 2334 The immunity period is a time duration during which a module 2335 can not be deregistered from the network. 2336 Fetches the immunity period for a specified network subnet. 2337 2338 Args: 2339 netuid: The network UID for which to query the immunity period. 2340 2341 Returns: 2342 The immunity period setting for the specified network subnet. 2343 2344 Raises: 2345 QueryError: If the query to the network fails or is invalid. 2346 """ 2347 2348 return self.query( 2349 "ImmunityPeriod", 2350 params=[netuid], 2351 )
Queries the network for the immunity period setting.
The immunity period is a time duration during which a module can not be deregistered from the network. Fetches the immunity period for a specified network subnet.
Arguments:
- netuid: The network UID for which to query the immunity period.
Returns:
The immunity period setting for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_min_allowed_weights(self, netuid: int = 0) -> int:
2356 def get_min_allowed_weights(self, netuid: int = 0) -> int: 2357 """ 2358 Queries the network for the minimum allowed weights setting. 2359 2360 Retrieves the minimum weight values that are possible to set 2361 by a validator within a specific network subnet. 2362 2363 Args: 2364 netuid: The network UID for which to query the minimum allowed 2365 weights. 2366 2367 Returns: 2368 The minimum allowed weight values for the specified network 2369 subnet. 2370 2371 Raises: 2372 QueryError: If the query to the network fails or is invalid. 2373 """ 2374 2375 return self.query( 2376 "MinAllowedWeights", 2377 params=[netuid], 2378 )
Queries the network for the minimum allowed weights setting.
Retrieves the minimum weight values that are possible to set by a validator within a specific network subnet.
Arguments:
- netuid: The network UID for which to query the minimum allowed weights.
Returns:
The minimum allowed weight values for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_max_allowed_weights(self, netuid: int = 0) -> int:
2383 def get_max_allowed_weights(self, netuid: int = 0) -> int: 2384 """ 2385 Queries the network for the maximum allowed weights setting. 2386 2387 Retrieves the maximum weight values that are possible to set 2388 by a validator within a specific network subnet. 2389 2390 Args: 2391 netuid: The network UID for which to query the maximum allowed 2392 weights. 2393 2394 Returns: 2395 The maximum allowed weight values for the specified network 2396 subnet. 2397 2398 Raises: 2399 QueryError: If the query to the network fails or is invalid. 2400 """ 2401 2402 return self.query("MaxAllowedWeights", params=[netuid])
Queries the network for the maximum allowed weights setting.
Retrieves the maximum weight values that are possible to set by a validator within a specific network subnet.
Arguments:
- netuid: The network UID for which to query the maximum allowed weights.
Returns:
The maximum allowed weight values for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_max_allowed_uids(self, netuid: int = 0) -> int:
2404 def get_max_allowed_uids(self, netuid: int = 0) -> int: 2405 """ 2406 Queries the network for the maximum allowed UIDs setting. 2407 2408 Fetches the upper limit on the number of user IDs that can 2409 be allocated or used within a specific network subnet. 2410 2411 Args: 2412 netuid: The network UID for which to query the maximum allowed UIDs. 2413 2414 Returns: 2415 The maximum number of allowed UIDs for the specified network subnet. 2416 2417 Raises: 2418 QueryError: If the query to the network fails or is invalid. 2419 """ 2420 2421 return self.query("MaxAllowedUids", params=[netuid])
Queries the network for the maximum allowed UIDs setting.
Fetches the upper limit on the number of user IDs that can be allocated or used within a specific network subnet.
Arguments:
- netuid: The network UID for which to query the maximum allowed UIDs.
Returns:
The maximum number of allowed UIDs for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_name(self, netuid: int = 0) -> str:
2423 def get_name(self, netuid: int = 0) -> str: 2424 """ 2425 Queries the network for the name of a specific subnet. 2426 2427 Args: 2428 netuid: The network UID for which to query the name. 2429 2430 Returns: 2431 The name of the specified network subnet. 2432 2433 Raises: 2434 QueryError: If the query to the network fails or is invalid. 2435 """ 2436 2437 return self.query("Name", params=[netuid])
Queries the network for the name of a specific subnet.
Arguments:
- netuid: The network UID for which to query the name.
Returns:
The name of the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_subnet_name(self, netuid: int = 0) -> str:
2439 def get_subnet_name(self, netuid: int = 0) -> str: 2440 """ 2441 Queries the network for the name of a specific subnet. 2442 2443 Args: 2444 netuid: The network UID for which to query the name. 2445 2446 Returns: 2447 The name of the specified network subnet. 2448 2449 Raises: 2450 QueryError: If the query to the network fails or is invalid. 2451 """ 2452 2453 return self.query("SubnetNames", params=[netuid])
Queries the network for the name of a specific subnet.
Arguments:
- netuid: The network UID for which to query the name.
Returns:
The name of the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_n(self, netuid: int = 0) -> int:
2458 def get_n(self, netuid: int = 0) -> int: 2459 """ 2460 Queries the network for the 'N' hyperparameter, which represents how 2461 many modules are on the network. 2462 2463 Args: 2464 netuid: The network UID for which to query the 'N' hyperparameter. 2465 2466 Returns: 2467 The value of the 'N' hyperparameter for the specified network 2468 subnet. 2469 2470 Raises: 2471 QueryError: If the query to the network fails or is invalid. 2472 """ 2473 2474 return self.query("N", params=[netuid])
Queries the network for the 'N' hyperparameter, which represents how many modules are on the network.
Arguments:
- netuid: The network UID for which to query the 'N' hyperparameter.
Returns:
The value of the 'N' hyperparameter for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_tempo(self, netuid: int = 0) -> int:
2476 def get_tempo(self, netuid: int = 0) -> int: 2477 """ 2478 Queries the network for the tempo setting, measured in blocks, for the 2479 specified subnet. 2480 2481 Args: 2482 netuid: The network UID for which to query the tempo. 2483 2484 Returns: 2485 The tempo setting for the specified subnet. 2486 2487 Raises: 2488 QueryError: If the query to the network fails or is invalid. 2489 """ 2490 2491 return self.query("Tempo", params=[netuid])
Queries the network for the tempo setting, measured in blocks, for the specified subnet.
Arguments:
- netuid: The network UID for which to query the tempo.
Returns:
The tempo setting for the specified subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_total_free_issuance(self, block_hash: str | None = None) -> int:
2493 def get_total_free_issuance(self, block_hash: str | None = None) -> int: 2494 """ 2495 Queries the network for the total free issuance. 2496 2497 Fetches the total amount of free issuance tokens available 2498 2499 Returns: 2500 The total free issuance amount. 2501 2502 Raises: 2503 QueryError: If the query to the network fails or is invalid. 2504 """ 2505 2506 return self.query("TotalIssuance", module="Balances", block_hash=block_hash)
Queries the network for the total free issuance.
Fetches the total amount of free issuance tokens available
Returns:
The total free issuance amount.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_total_stake(self, block_hash: str | None = None) -> int:
2508 def get_total_stake(self, block_hash: str | None = None) -> int: 2509 """ 2510 Retrieves a mapping of total stakes for keys on the network. 2511 2512 Queries the network for a mapping of key UIDs to their total stake amounts. 2513 2514 Returns: 2515 A dictionary mapping key UIDs to their total stake amounts. 2516 2517 Raises: 2518 QueryError: If the query to the network fails or is invalid. 2519 """ 2520 2521 return self.query("TotalStake", block_hash=block_hash)
Retrieves a mapping of total stakes for keys on the network.
Queries the network for a mapping of key UIDs to their total stake amounts.
Returns:
A dictionary mapping key UIDs to their total stake amounts.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_registrations_per_block(self):
2523 def get_registrations_per_block(self): 2524 """ 2525 Queries the network for the number of registrations per block. 2526 2527 Fetches the number of registrations that are processed per 2528 block within the network. 2529 2530 Returns: 2531 The number of registrations processed per block. 2532 2533 Raises: 2534 QueryError: If the query to the network fails or is invalid. 2535 """ 2536 2537 return self.query( 2538 "RegistrationsPerBlock", 2539 )
Queries the network for the number of registrations per block.
Fetches the number of registrations that are processed per block within the network.
Returns:
The number of registrations processed per block.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
max_registrations_per_block(self, netuid: int = 0):
2541 def max_registrations_per_block(self, netuid: int = 0): 2542 """ 2543 Queries the network for the maximum number of registrations per block. 2544 2545 Retrieves the upper limit of registrations that can be processed in 2546 each block within a specific network subnet. 2547 2548 Args: 2549 netuid: The network UID for which to query. 2550 2551 Returns: 2552 The maximum number of registrations per block for 2553 the specified network subnet. 2554 2555 Raises: 2556 QueryError: If the query to the network fails or is invalid. 2557 """ 2558 2559 return self.query( 2560 "MaxRegistrationsPerBlock", 2561 params=[netuid], 2562 )
Queries the network for the maximum number of registrations per block.
Retrieves the upper limit of registrations that can be processed in each block within a specific network subnet.
Arguments:
- netuid: The network UID for which to query.
Returns:
The maximum number of registrations per block for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_proposal(self, proposal_id: int = 0):
2564 def get_proposal(self, proposal_id: int = 0): 2565 """ 2566 Queries the network for a specific proposal. 2567 2568 Args: 2569 proposal_id: The ID of the proposal to query. 2570 2571 Returns: 2572 The details of the specified proposal. 2573 2574 Raises: 2575 QueryError: If the query to the network fails, is invalid, 2576 or if the proposal ID does not exist. 2577 """ 2578 2579 return self.query( 2580 "Proposals", 2581 params=[proposal_id], 2582 )
Queries the network for a specific proposal.
Arguments:
- proposal_id: The ID of the proposal to query.
Returns:
The details of the specified proposal.
Raises:
- QueryError: If the query to the network fails, is invalid, or if the proposal ID does not exist.
def
get_trust(self, netuid: int = 0):
2584 def get_trust(self, netuid: int = 0): 2585 """ 2586 Queries the network for the trust setting of a specific network subnet. 2587 2588 Retrieves the trust level or score, which may represent the 2589 level of trustworthiness or reliability within a 2590 particular network subnet. 2591 2592 Args: 2593 netuid: The network UID for which to query the trust setting. 2594 2595 Returns: 2596 The trust level or score for the specified network subnet. 2597 2598 Raises: 2599 QueryError: If the query to the network fails or is invalid. 2600 """ 2601 2602 return self.query( 2603 "Trust", 2604 params=[netuid], 2605 )
Queries the network for the trust setting of a specific network subnet.
Retrieves the trust level or score, which may represent the level of trustworthiness or reliability within a particular network subnet.
Arguments:
- netuid: The network UID for which to query the trust setting.
Returns:
The trust level or score for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
2607 def get_uids(self, key: Ss58Address, netuid: int = 0) -> bool | None: 2608 """ 2609 Queries the network for module UIDs associated with a specific key. 2610 2611 Args: 2612 key: The key address for which to query UIDs. 2613 netuid: The network UID within which to search for the key. 2614 2615 Returns: 2616 A list of UIDs associated with the specified key. 2617 2618 Raises: 2619 QueryError: If the query to the network fails or is invalid. 2620 """ 2621 2622 return self.query( 2623 "Uids", 2624 params=[netuid, key], 2625 )
Queries the network for module UIDs associated with a specific key.
Arguments:
- key: The key address for which to query UIDs.
- netuid: The network UID within which to search for the key.
Returns:
A list of UIDs associated with the specified key.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_unit_emission(self) -> int:
2627 def get_unit_emission(self) -> int: 2628 """ 2629 Queries the network for the unit emission setting. 2630 2631 Retrieves the unit emission value, which represents the 2632 emission rate or quantity for the $COMM token. 2633 2634 Returns: 2635 The unit emission value in nanos for the network. 2636 2637 Raises: 2638 QueryError: If the query to the network fails or is invalid. 2639 """ 2640 2641 return self.query("UnitEmission", module="SubnetEmissionModule")
Queries the network for the unit emission setting.
Retrieves the unit emission value, which represents the emission rate or quantity for the $COMM token.
Returns:
The unit emission value in nanos for the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_tx_rate_limit(self) -> int:
2643 def get_tx_rate_limit(self) -> int: 2644 """ 2645 Queries the network for the transaction rate limit. 2646 2647 Retrieves the rate limit for transactions within the network, 2648 which defines the maximum number of transactions that can be 2649 processed within a certain timeframe. 2650 2651 Returns: 2652 The transaction rate limit for the network. 2653 2654 Raises: 2655 QueryError: If the query to the network fails or is invalid. 2656 """ 2657 2658 return self.query( 2659 "TxRateLimit", 2660 )
Queries the network for the transaction rate limit.
Retrieves the rate limit for transactions within the network, which defines the maximum number of transactions that can be processed within a certain timeframe.
Returns:
The transaction rate limit for the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_subnet_burn(self) -> int:
2662 def get_subnet_burn(self) -> int: 2663 """Queries the network for the subnet burn value. 2664 2665 Retrieves the subnet burn value from the network, which represents 2666 the amount of tokens that are burned (permanently removed from 2667 circulation) for subnet-related operations. 2668 2669 Returns: 2670 int: The subnet burn value. 2671 2672 Raises: 2673 QueryError: If the query to the network fails or returns invalid data. 2674 """ 2675 2676 return self.query( 2677 "SubnetBurn", 2678 )
Queries the network for the subnet burn value.
Retrieves the subnet burn value from the network, which represents the amount of tokens that are burned (permanently removed from circulation) for subnet-related operations.
Returns:
int: The subnet burn value.
Raises:
- QueryError: If the query to the network fails or returns invalid data.
def
get_burn_rate(self) -> int:
2680 def get_burn_rate(self) -> int: 2681 """ 2682 Queries the network for the burn rate setting. 2683 2684 Retrieves the burn rate, which represents the rate at 2685 which the $COMM token is permanently 2686 removed or 'burned' from circulation. 2687 2688 Returns: 2689 The burn rate for the network. 2690 2691 Raises: 2692 QueryError: If the query to the network fails or is invalid. 2693 """ 2694 2695 return self.query( 2696 "BurnRate", 2697 params=[], 2698 )
Queries the network for the burn rate setting.
Retrieves the burn rate, which represents the rate at which the $COMM token is permanently removed or 'burned' from circulation.
Returns:
The burn rate for the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_burn(self, netuid: int = 0) -> int:
2700 def get_burn(self, netuid: int = 0) -> int: 2701 """ 2702 Queries the network for the burn setting. 2703 2704 Retrieves the burn value, which represents the amount of the 2705 $COMM token that is 'burned' or permanently removed from 2706 circulation. 2707 2708 Args: 2709 netuid: The network UID for which to query the burn value. 2710 2711 Returns: 2712 The burn value for the specified network subnet. 2713 2714 Raises: 2715 QueryError: If the query to the network fails or is invalid. 2716 """ 2717 2718 return self.query("Burn", params=[netuid])
Queries the network for the burn setting.
Retrieves the burn value, which represents the amount of the $COMM token that is 'burned' or permanently removed from circulation.
Arguments:
- netuid: The network UID for which to query the burn value.
Returns:
The burn value for the specified network subnet.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_min_burn(self) -> int:
2720 def get_min_burn(self) -> int: 2721 """ 2722 Queries the network for the minimum burn setting. 2723 2724 Retrieves the minimum burn value, indicating the lowest 2725 amount of the $COMM tokens that can be 'burned' or 2726 permanently removed from circulation. 2727 2728 Returns: 2729 The minimum burn value for the network. 2730 2731 Raises: 2732 QueryError: If the query to the network fails or is invalid. 2733 """ 2734 2735 return self.query( 2736 "BurnConfig", 2737 params=[], 2738 )["min_burn"]
Queries the network for the minimum burn setting.
Retrieves the minimum burn value, indicating the lowest amount of the $COMM tokens that can be 'burned' or permanently removed from circulation.
Returns:
The minimum burn value for the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_min_weight_stake(self) -> int:
2740 def get_min_weight_stake(self) -> int: 2741 """ 2742 Queries the network for the minimum weight stake setting. 2743 2744 Retrieves the minimum weight stake, which represents the lowest 2745 stake weight that is allowed for certain operations or 2746 transactions within the network. 2747 2748 Returns: 2749 The minimum weight stake for the network. 2750 2751 Raises: 2752 QueryError: If the query to the network fails or is invalid. 2753 """ 2754 2755 return self.query("MinWeightStake", params=[])
Queries the network for the minimum weight stake setting.
Retrieves the minimum weight stake, which represents the lowest stake weight that is allowed for certain operations or transactions within the network.
Returns:
The minimum weight stake for the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_vote_mode_global(self) -> str:
2757 def get_vote_mode_global(self) -> str: 2758 """ 2759 Queries the network for the global vote mode setting. 2760 2761 Retrieves the global vote mode, which defines the overall voting 2762 methodology or approach used across the network in default. 2763 2764 Returns: 2765 The global vote mode setting for the network. 2766 2767 Raises: 2768 QueryError: If the query to the network fails or is invalid. 2769 """ 2770 2771 return self.query( 2772 "VoteModeGlobal", 2773 )
Queries the network for the global vote mode setting.
Retrieves the global vote mode, which defines the overall voting methodology or approach used across the network in default.
Returns:
The global vote mode setting for the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_max_proposals(self) -> int:
2775 def get_max_proposals(self) -> int: 2776 """ 2777 Queries the network for the maximum number of proposals allowed. 2778 2779 Retrieves the upper limit on the number of proposals that can be 2780 active or considered at any given time within the network. 2781 2782 Returns: 2783 The maximum number of proposals allowed on the network. 2784 2785 Raises: 2786 QueryError: If the query to the network fails or is invalid. 2787 """ 2788 2789 return self.query( 2790 "MaxProposals", 2791 )
Queries the network for the maximum number of proposals allowed.
Retrieves the upper limit on the number of proposals that can be active or considered at any given time within the network.
Returns:
The maximum number of proposals allowed on the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_max_registrations_per_block(self) -> int:
2793 def get_max_registrations_per_block(self) -> int: 2794 """ 2795 Queries the network for the maximum number of registrations per block. 2796 2797 Retrieves the maximum number of registrations that can 2798 be processed in each block within the network. 2799 2800 Returns: 2801 The maximum number of registrations per block on the network. 2802 2803 Raises: 2804 QueryError: If the query to the network fails or is invalid. 2805 """ 2806 2807 return self.query( 2808 "MaxRegistrationsPerBlock", 2809 params=[], 2810 )
Queries the network for the maximum number of registrations per block.
Retrieves the maximum number of registrations that can be processed in each block within the network.
Returns:
The maximum number of registrations per block on the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_max_name_length(self) -> int:
2812 def get_max_name_length(self) -> int: 2813 """ 2814 Queries the network for the maximum length allowed for names. 2815 2816 Retrieves the maximum character length permitted for names 2817 within the network. Such as the module names 2818 2819 Returns: 2820 The maximum length allowed for names on the network. 2821 2822 Raises: 2823 QueryError: If the query to the network fails or is invalid. 2824 """ 2825 2826 return self.query( 2827 "MaxNameLength", 2828 params=[], 2829 )
Queries the network for the maximum length allowed for names.
Retrieves the maximum character length permitted for names within the network. Such as the module names
Returns:
The maximum length allowed for names on the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_global_vote_threshold(self) -> int:
2831 def get_global_vote_threshold(self) -> int: 2832 """ 2833 Queries the network for the global vote threshold. 2834 2835 Retrieves the global vote threshold, which is the critical value or 2836 percentage required for decisions in the network's governance process. 2837 2838 Returns: 2839 The global vote threshold for the network. 2840 2841 Raises: 2842 QueryError: If the query to the network fails or is invalid. 2843 """ 2844 2845 return self.query( 2846 "GlobalVoteThreshold", 2847 )
Queries the network for the global vote threshold.
Retrieves the global vote threshold, which is the critical value or percentage required for decisions in the network's governance process.
Returns:
The global vote threshold for the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_max_allowed_subnets(self) -> int:
2849 def get_max_allowed_subnets(self) -> int: 2850 """ 2851 Queries the network for the maximum number of allowed subnets. 2852 2853 Retrieves the upper limit on the number of subnets that can 2854 be created or operated within the network. 2855 2856 Returns: 2857 The maximum number of allowed subnets on the network. 2858 2859 Raises: 2860 QueryError: If the query to the network fails or is invalid. 2861 """ 2862 2863 return self.query( 2864 "MaxAllowedSubnets", 2865 params=[], 2866 )
Queries the network for the maximum number of allowed subnets.
Retrieves the upper limit on the number of subnets that can be created or operated within the network.
Returns:
The maximum number of allowed subnets on the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_max_allowed_modules(self) -> int:
2868 def get_max_allowed_modules(self) -> int: 2869 """ 2870 Queries the network for the maximum number of allowed modules. 2871 2872 Retrieves the upper limit on the number of modules that 2873 can be registered within the network. 2874 2875 Returns: 2876 The maximum number of allowed modules on the network. 2877 2878 Raises: 2879 QueryError: If the query to the network fails or is invalid. 2880 """ 2881 2882 return self.query( 2883 "MaxAllowedModules", 2884 params=[], 2885 )
Queries the network for the maximum number of allowed modules.
Retrieves the upper limit on the number of modules that can be registered within the network.
Returns:
The maximum number of allowed modules on the network.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_min_stake(self, netuid: int = 0) -> int:
2887 def get_min_stake(self, netuid: int = 0) -> int: 2888 """ 2889 Queries the network for the minimum stake required to register a key. 2890 2891 Retrieves the minimum amount of stake necessary for 2892 registering a key within a specific network subnet. 2893 2894 Args: 2895 netuid: The network UID for which to query the minimum stake. 2896 2897 Returns: 2898 The minimum stake required for key registration in nanos. 2899 2900 Raises: 2901 QueryError: If the query to the network fails or is invalid. 2902 """ 2903 2904 return self.query("MinStake", params=[netuid])
Queries the network for the minimum stake required to register a key.
Retrieves the minimum amount of stake necessary for registering a key within a specific network subnet.
Arguments:
- netuid: The network UID for which to query the minimum stake.
Returns:
The minimum stake required for key registration in nanos.
Raises:
- QueryError: If the query to the network fails or is invalid.
2906 def get_stakefrom( 2907 self, 2908 key: Ss58Address, 2909 ) -> dict[str, int]: 2910 """ 2911 Retrieves the stake amounts from all stakers to a specific staked address. 2912 2913 Queries the network for the stakes received by a particular staked address 2914 from all stakers. 2915 2916 Args: 2917 key: The address of the key receiving the stakes. 2918 2919 Returns: 2920 A dictionary mapping staker addresses to their respective stake amounts. 2921 2922 Raises: 2923 QueryError: If the query to the network fails or is invalid. 2924 """ 2925 2926 # Has to use query map in order to iterate through the storage prefix. 2927 return self.query_map("StakeFrom", [key], extract_value=False).get("StakeFrom", {})
Retrieves the stake amounts from all stakers to a specific staked address.
Queries the network for the stakes received by a particular staked address from all stakers.
Arguments:
- key: The address of the key receiving the stakes.
Returns:
A dictionary mapping staker addresses to their respective stake amounts.
Raises:
- QueryError: If the query to the network fails or is invalid.
2929 def get_staketo( 2930 self, 2931 key: Ss58Address, 2932 ) -> dict[str, int]: 2933 """ 2934 Retrieves the stake amounts provided by a specific staker to all staked addresses. 2935 2936 Queries the network for the stakes provided by a particular staker to 2937 all staked addresses. 2938 2939 Args: 2940 key: The address of the key providing the stakes. 2941 2942 Returns: 2943 A dictionary mapping staked addresses to their respective received stake amounts. 2944 2945 Raises: 2946 QueryError: If the query to the network fails or is invalid. 2947 """ 2948 2949 # Has to use query map in order to iterate through the storage prefix. 2950 return self.query_map("StakeTo", [key], extract_value=False).get("StakeTo", {})
Retrieves the stake amounts provided by a specific staker to all staked addresses.
Queries the network for the stakes provided by a particular staker to all staked addresses.
Arguments:
- key: The address of the key providing the stakes.
Returns:
A dictionary mapping staked addresses to their respective received stake amounts.
Raises:
- QueryError: If the query to the network fails or is invalid.
2952 def get_balance( 2953 self, 2954 addr: Ss58Address, 2955 ) -> int: 2956 """ 2957 Retrieves the balance of a specific key. 2958 2959 Args: 2960 addr: The address of the key to query the balance for. 2961 2962 Returns: 2963 The balance of the specified key. 2964 2965 Raises: 2966 QueryError: If the query to the network fails or is invalid. 2967 """ 2968 2969 result = self.query("Account", module="System", params=[addr]) 2970 2971 return result["data"]["free"]
Retrieves the balance of a specific key.
Arguments:
- addr: The address of the key to query the balance for.
Returns:
The balance of the specified key.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_block( self, block_hash: str | None = None) -> dict[typing.Any, typing.Any] | None:
2973 def get_block(self, block_hash: str | None = None) -> dict[Any, Any] | None: 2974 """ 2975 Retrieves information about a specific block in the network. 2976 2977 Queries the network for details about a block, such as its number, 2978 hash, and other relevant information. 2979 2980 Returns: 2981 The requested information about the block, 2982 or None if the block does not exist 2983 or the information is not available. 2984 2985 Raises: 2986 QueryError: If the query to the network fails or is invalid. 2987 """ 2988 2989 with self.get_conn() as substrate: 2990 block: dict[Any, Any] | None = substrate.get_block( # type: ignore 2991 block_hash # type: ignore 2992 ) 2993 2994 return block
Retrieves information about a specific block in the network.
Queries the network for details about a block, such as its number, hash, and other relevant information.
Returns:
The requested information about the block, or None if the block does not exist or the information is not available.
Raises:
- QueryError: If the query to the network fails or is invalid.
def
get_existential_deposit(self, block_hash: str | None = None) -> int:
2996 def get_existential_deposit(self, block_hash: str | None = None) -> int: 2997 """ 2998 Retrieves the existential deposit value for the network. 2999 3000 The existential deposit is the minimum balance that must be maintained 3001 in an account to prevent it from being purged. Denotated in nano units. 3002 3003 Returns: 3004 The existential deposit value in nano units. 3005 Note: 3006 The value returned is a fixed value defined in the 3007 client and may not reflect changes in the network's configuration. 3008 """ 3009 3010 with self.get_conn() as substrate: 3011 result: int = substrate.get_constant( # type: ignore 3012 "Balances", "ExistentialDeposit", block_hash 3013 ).value # type: ignore 3014 3015 return result
Retrieves the existential deposit value for the network.
The existential deposit is the minimum balance that must be maintained in an account to prevent it from being purged. Denotated in nano units.
Returns:
The existential deposit value in nano units.
Note:
The value returned is a fixed value defined in the client and may not reflect changes in the network's configuration.
def
add_transfer_dao_treasury_proposal( self, key: substrateinterface.keypair.Keypair, data: str, amount_nano: int, dest: communex.types.Ss58Address):
3021 def add_transfer_dao_treasury_proposal( 3022 self, 3023 key: Keypair, 3024 data: str, 3025 amount_nano: int, 3026 dest: Ss58Address, 3027 ): 3028 params = {"dest": dest, "value": amount_nano, "data": data} 3029 3030 return self.compose_call( 3031 module="GovernanceModule", 3032 fn="add_transfer_dao_treasury_proposal", 3033 params=params, 3034 key=key, 3035 )
def
delegate_rootnet_control( self, key: substrateinterface.keypair.Keypair, dest: communex.types.Ss58Address):