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 params = { 1092 "netuid": netuid, 1093 "name": name, 1094 "address": address, 1095 "delegation_fee": delegation_fee, 1096 "metadata": metadata, 1097 } 1098 1099 response = self.compose_call("update_module", params=params, key=key) 1100 1101 return response 1102 1103 def register_module( 1104 self, 1105 key: Keypair, 1106 name: str, 1107 address: str | None = None, 1108 subnet: str = "Rootnet", 1109 metadata: str | None = None, 1110 ) -> ExtrinsicReceipt: 1111 """ 1112 Registers a new module in the network. 1113 1114 Args: 1115 key: The keypair used for registering the module. 1116 name: The name of the module. If None, a default or previously 1117 set name is used. # How does this work? 1118 address: The address of the module. If None, a default or 1119 previously set address is used. # How does this work? 1120 subnet: The network subnet to register the module in. 1121 min_stake: The minimum stake required for the module, in nanotokens. 1122 If None, a default value is used. 1123 1124 Returns: 1125 A receipt of the registration transaction. 1126 1127 Raises: 1128 InvalidParameterError: If the provided parameters are invalid. 1129 ChainTransactionError: If the transaction fails. 1130 """ 1131 1132 key_addr = key.ss58_address 1133 1134 params = { 1135 "network_name": subnet, 1136 "address": address, 1137 "name": name, 1138 "module_key": key_addr, 1139 "metadata": metadata, 1140 } 1141 1142 response = self.compose_call("register", params=params, key=key) 1143 return response 1144 1145 def deregister_module(self, key: Keypair, netuid: int) -> ExtrinsicReceipt: 1146 """ 1147 Deregisters a module from the network. 1148 1149 Args: 1150 key: The keypair associated with the module's account. 1151 netuid: The network identifier. 1152 1153 Returns: 1154 A receipt of the module deregistration transaction. 1155 1156 Raises: 1157 ChainTransactionError: If the transaction fails. 1158 """ 1159 1160 params = {"netuid": netuid} 1161 1162 response = self.compose_call("deregister", params=params, key=key) 1163 1164 return response 1165 1166 def register_subnet(self, key: Keypair, name: str, metadata: str | None = None) -> ExtrinsicReceipt: 1167 """ 1168 Registers a new subnet in the network. 1169 1170 Args: 1171 key (Keypair): The keypair used for registering the subnet. 1172 name (str): The name of the subnet to be registered. 1173 metadata (str | None, optional): Additional metadata for the subnet. Defaults to None. 1174 1175 Returns: 1176 ExtrinsicReceipt: A receipt of the subnet registration transaction. 1177 1178 Raises: 1179 ChainTransactionError: If the transaction fails. 1180 """ 1181 1182 params = { 1183 "name": name, 1184 "metadata": metadata, 1185 } 1186 1187 response = self.compose_call("register_subnet", params=params, key=key) 1188 1189 return response 1190 1191 def vote( 1192 self, 1193 key: Keypair, 1194 uids: list[int], 1195 weights: list[int], 1196 netuid: int = 0, 1197 ) -> ExtrinsicReceipt: 1198 """ 1199 Casts votes on a list of module UIDs with corresponding weights. 1200 1201 The length of the UIDs list and the weights list should be the same. 1202 Each weight corresponds to the UID at the same index. 1203 1204 Args: 1205 key: The keypair used for signing the vote transaction. 1206 uids: A list of module UIDs to vote on. 1207 weights: A list of weights corresponding to each UID. 1208 netuid: The network identifier. 1209 1210 Returns: 1211 A receipt of the voting transaction. 1212 1213 Raises: 1214 InvalidParameterError: If the lengths of UIDs and weights lists 1215 do not match. 1216 ChainTransactionError: If the transaction fails. 1217 """ 1218 1219 assert len(uids) == len(weights) 1220 1221 params = { 1222 "uids": uids, 1223 "weights": weights, 1224 "netuid": netuid, 1225 } 1226 1227 response = self.compose_call("set_weights", params=params, key=key) 1228 1229 return response 1230 1231 def update_subnet( 1232 self, 1233 key: Keypair, 1234 params: SubnetParams, 1235 netuid: int = 0, 1236 ) -> ExtrinsicReceipt: 1237 """ 1238 Update a subnet's configuration. 1239 1240 It requires the founder key for authorization. 1241 1242 Args: 1243 key: The founder keypair of the subnet. 1244 params: The new parameters for the subnet. 1245 netuid: The network identifier. 1246 1247 Returns: 1248 A receipt of the subnet update transaction. 1249 1250 Raises: 1251 AuthorizationError: If the key is not authorized. 1252 ChainTransactionError: If the transaction fails. 1253 """ 1254 1255 general_params = dict(params) 1256 general_params["netuid"] = netuid 1257 if general_params.get("subnet_metadata") is None: 1258 general_params["metadata"] = None 1259 else: 1260 general_params["metadata"] = general_params["subnet_metadata"] 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 )
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]]
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 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 general_params.get("subnet_metadata") is None: 1259 general_params["metadata"] = None 1260 else: 1261 general_params["metadata"] = general_params["subnet_metadata"] 1262 1263 response = self.compose_call( 1264 fn="update_subnet", 1265 params=general_params, 1266 key=key, 1267 ) 1268 1269 return response 1270 1271 def transfer_stake( 1272 self, 1273 key: Keypair, 1274 amount: int, 1275 from_module_key: Ss58Address, 1276 dest_module_address: Ss58Address, 1277 ) -> ExtrinsicReceipt: 1278 """ 1279 Realocate staked tokens from one staked module to another module. 1280 1281 Args: 1282 key: The keypair associated with the account that is delegating the tokens. 1283 amount: The amount of staked tokens to transfer, in nanotokens. 1284 from_module_key: The SS58 address of the module you want to transfer from (currently delegated by the key). 1285 dest_module_address: The SS58 address of the destination (newly delegated key). 1286 netuid: The network identifier. 1287 1288 Returns: 1289 A receipt of the stake transfer transaction. 1290 1291 Raises: 1292 InsufficientStakeError: If the source module key does not have 1293 enough staked tokens. ChainTransactionError: If the transaction 1294 fails. 1295 """ 1296 1297 amount = amount - self.get_existential_deposit() 1298 1299 params = { 1300 "amount": amount, 1301 "module_key": from_module_key, 1302 "new_module_key": dest_module_address, 1303 } 1304 1305 response = self.compose_call("transfer_stake", key=key, params=params) 1306 1307 return response 1308 1309 def multiunstake( 1310 self, 1311 key: Keypair, 1312 keys: list[Ss58Address], 1313 amounts: list[int], 1314 ) -> ExtrinsicReceipt: 1315 """ 1316 Unstakes tokens from multiple module keys. 1317 1318 And the lists `keys` and `amounts` must be of the same length. Each 1319 amount corresponds to the module key at the same index. 1320 1321 Args: 1322 key: The keypair associated with the unstaker's account. 1323 keys: A list of SS58 addresses of the module keys to unstake from. 1324 amounts: A list of amounts to unstake from each module key, 1325 in nanotokens. 1326 netuid: The network identifier. 1327 1328 Returns: 1329 A receipt of the multi-unstaking transaction. 1330 1331 Raises: 1332 MismatchedLengthError: If the lengths of keys and amounts lists do 1333 not match. InsufficientStakeError: If any of the module keys do not 1334 have enough staked tokens. ChainTransactionError: If the transaction 1335 fails. 1336 """ 1337 1338 assert len(keys) == len(amounts) 1339 1340 params = {"module_keys": keys, "amounts": amounts} 1341 1342 response = self.compose_call("remove_stake_multiple", params=params, key=key) 1343 1344 return response 1345 1346 def multistake( 1347 self, 1348 key: Keypair, 1349 keys: list[Ss58Address], 1350 amounts: list[int], 1351 ) -> ExtrinsicReceipt: 1352 """ 1353 Stakes tokens to multiple module keys. 1354 1355 The lengths of the `keys` and `amounts` lists must be the same. Each 1356 amount corresponds to the module key at the same index. 1357 1358 Args: 1359 key: The keypair associated with the staker's account. 1360 keys: A list of SS58 addresses of the module keys to stake to. 1361 amounts: A list of amounts to stake to each module key, 1362 in nanotokens. 1363 netuid: The network identifier. 1364 1365 Returns: 1366 A receipt of the multi-staking transaction. 1367 1368 Raises: 1369 MismatchedLengthError: If the lengths of keys and amounts lists 1370 do not match. 1371 ChainTransactionError: If the transaction fails. 1372 """ 1373 1374 assert len(keys) == len(amounts) 1375 1376 params = { 1377 "module_keys": keys, 1378 "amounts": amounts, 1379 } 1380 1381 response = self.compose_call("add_stake_multiple", params=params, key=key) 1382 1383 return response 1384 1385 def add_profit_shares( 1386 self, 1387 key: Keypair, 1388 keys: list[Ss58Address], 1389 shares: list[int], 1390 ) -> ExtrinsicReceipt: 1391 """ 1392 Allocates profit shares to multiple keys. 1393 1394 The lists `keys` and `shares` must be of the same length, 1395 with each share amount corresponding to the key at the same index. 1396 1397 Args: 1398 key: The keypair associated with the account 1399 distributing the shares. 1400 keys: A list of SS58 addresses to allocate shares to. 1401 shares: A list of share amounts to allocate to each key, 1402 in nanotokens. 1403 1404 Returns: 1405 A receipt of the profit sharing transaction. 1406 1407 Raises: 1408 MismatchedLengthError: If the lengths of keys and shares 1409 lists do not match. 1410 ChainTransactionError: If the transaction fails. 1411 """ 1412 1413 assert len(keys) == len(shares) 1414 1415 params = {"keys": keys, "shares": shares} 1416 1417 response = self.compose_call("add_profit_shares", params=params, key=key) 1418 1419 return response 1420 1421 def add_subnet_proposal( 1422 self, key: Keypair, 1423 params: dict[str, Any], 1424 ipfs: str, 1425 netuid: int = 0 1426 ) -> ExtrinsicReceipt: 1427 """ 1428 Submits a proposal for creating or modifying a subnet within the 1429 network. 1430 1431 The proposal includes various parameters like the name, founder, share 1432 allocations, and other subnet-specific settings. 1433 1434 Args: 1435 key: The keypair used for signing the proposal transaction. 1436 params: The parameters for the subnet proposal. 1437 netuid: The network identifier. 1438 1439 Returns: 1440 A receipt of the subnet proposal transaction. 1441 1442 Raises: 1443 InvalidParameterError: If the provided subnet 1444 parameters are invalid. 1445 ChainTransactionError: If the transaction fails. 1446 """ 1447 1448 general_params = dict(params) 1449 general_params["netuid"] = netuid 1450 general_params["data"] = ipfs 1451 if "metadata" not in general_params: 1452 general_params["metadata"] = None 1453 1454 # general_params["burn_config"] = json.dumps(general_params["burn_config"]) 1455 response = self.compose_call( 1456 fn="add_subnet_params_proposal", 1457 params=general_params, 1458 key=key, 1459 module="GovernanceModule", 1460 ) 1461 1462 return response 1463 1464 def add_custom_proposal( 1465 self, 1466 key: Keypair, 1467 cid: str, 1468 ) -> ExtrinsicReceipt: 1469 1470 params = {"data": cid} 1471 1472 response = self.compose_call( 1473 fn="add_global_custom_proposal", 1474 params=params, 1475 key=key, 1476 module="GovernanceModule", 1477 ) 1478 return response 1479 1480 def add_custom_subnet_proposal( 1481 self, 1482 key: Keypair, 1483 cid: str, 1484 netuid: int = 0, 1485 ) -> ExtrinsicReceipt: 1486 """ 1487 Submits a proposal for creating or modifying a custom subnet within the 1488 network. 1489 1490 The proposal includes various parameters like the name, founder, share 1491 allocations, and other subnet-specific settings. 1492 1493 Args: 1494 key: The keypair used for signing the proposal transaction. 1495 params: The parameters for the subnet proposal. 1496 netuid: The network identifier. 1497 1498 Returns: 1499 A receipt of the subnet proposal transaction. 1500 """ 1501 1502 params = { 1503 "data": cid, 1504 "netuid": netuid, 1505 } 1506 1507 response = self.compose_call( 1508 fn="add_subnet_custom_proposal", 1509 params=params, 1510 key=key, 1511 module="GovernanceModule", 1512 ) 1513 1514 return response 1515 1516 def add_global_proposal( 1517 self, 1518 key: Keypair, 1519 params: NetworkParams, 1520 cid: str | None, 1521 ) -> ExtrinsicReceipt: 1522 """ 1523 Submits a proposal for altering the global network parameters. 1524 1525 Allows for the submission of a proposal to 1526 change various global parameters 1527 of the network, such as emission rates, rate limits, and voting 1528 thresholds. It is used to 1529 suggest changes that affect the entire network's operation. 1530 1531 Args: 1532 key: The keypair used for signing the proposal transaction. 1533 params: A dictionary containing global network parameters 1534 like maximum allowed subnets, modules, 1535 transaction rate limits, and others. 1536 1537 Returns: 1538 A receipt of the global proposal transaction. 1539 1540 Raises: 1541 InvalidParameterError: If the provided network 1542 parameters are invalid. 1543 ChainTransactionError: If the transaction fails. 1544 """ 1545 general_params = cast(dict[str, Any], params) 1546 cid = cid or "" 1547 general_params["data"] = cid 1548 1549 response = self.compose_call( 1550 fn="add_global_params_proposal", 1551 params=general_params, 1552 key=key, 1553 module="GovernanceModule", 1554 ) 1555 1556 return response 1557 1558 def vote_on_proposal( 1559 self, 1560 key: Keypair, 1561 proposal_id: int, 1562 agree: bool, 1563 ) -> ExtrinsicReceipt: 1564 """ 1565 Casts a vote on a specified proposal within the network. 1566 1567 Args: 1568 key: The keypair used for signing the vote transaction. 1569 proposal_id: The unique identifier of the proposal to vote on. 1570 1571 Returns: 1572 A receipt of the voting transaction in nanotokens. 1573 1574 Raises: 1575 InvalidProposalIDError: If the provided proposal ID does not 1576 exist or is invalid. 1577 ChainTransactionError: If the transaction fails. 1578 """ 1579 1580 params = {"proposal_id": proposal_id, "agree": agree} 1581 1582 response = self.compose_call( 1583 "vote_proposal", 1584 key=key, 1585 params=params, 1586 module="GovernanceModule", 1587 ) 1588 1589 return response 1590 1591 def unvote_on_proposal( 1592 self, 1593 key: Keypair, 1594 proposal_id: int, 1595 ) -> ExtrinsicReceipt: 1596 """ 1597 Retracts a previously cast vote on a specified proposal. 1598 1599 Args: 1600 key: The keypair used for signing the unvote transaction. 1601 proposal_id: The unique identifier of the proposal to withdraw the 1602 vote from. 1603 1604 Returns: 1605 A receipt of the unvoting transaction in nanotokens. 1606 1607 Raises: 1608 InvalidProposalIDError: If the provided proposal ID does not 1609 exist or is invalid. 1610 ChainTransactionError: If the transaction fails to be processed, or 1611 if there was no prior vote to retract. 1612 """ 1613 1614 params = {"proposal_id": proposal_id} 1615 1616 response = self.compose_call( 1617 "remove_vote_proposal", 1618 key=key, 1619 params=params, 1620 module="GovernanceModule", 1621 ) 1622 1623 return response 1624 1625 def enable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1626 """ 1627 Enables vote power delegation for the signer's account. 1628 1629 Args: 1630 key: The keypair used for signing the delegation transaction. 1631 1632 Returns: 1633 A receipt of the vote power delegation transaction. 1634 1635 Raises: 1636 ChainTransactionError: If the transaction fails. 1637 """ 1638 1639 response = self.compose_call( 1640 "enable_vote_power_delegation", 1641 params={}, 1642 key=key, 1643 module="GovernanceModule", 1644 ) 1645 1646 return response 1647 1648 def disable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1649 """ 1650 Disables vote power delegation for the signer's account. 1651 1652 Args: 1653 key: The keypair used for signing the delegation transaction. 1654 1655 Returns: 1656 A receipt of the vote power delegation transaction. 1657 1658 Raises: 1659 ChainTransactionError: If the transaction fails. 1660 """ 1661 1662 response = self.compose_call( 1663 "disable_vote_power_delegation", 1664 params={}, 1665 key=key, 1666 module="GovernanceModule", 1667 ) 1668 1669 return response 1670 1671 def add_dao_application( 1672 self, key: Keypair, application_key: Ss58Address, data: str 1673 ) -> ExtrinsicReceipt: 1674 """ 1675 Submits a new application to the general subnet DAO. 1676 1677 Args: 1678 key: The keypair used for signing the application transaction. 1679 application_key: The SS58 address of the application key. 1680 data: The data associated with the application. 1681 1682 Returns: 1683 A receipt of the application transaction. 1684 1685 Raises: 1686 ChainTransactionError: If the transaction fails. 1687 """ 1688 1689 params = {"application_key": application_key, "data": data} 1690 1691 response = self.compose_call( 1692 "add_dao_application", module="GovernanceModule", key=key, 1693 params=params 1694 ) 1695 1696 return response 1697 1698 def query_map_curator_applications(self) -> dict[str, dict[str, str]]: 1699 query_result = self.query_map( 1700 "CuratorApplications", module="GovernanceModule", params=[], 1701 extract_value=False 1702 ) 1703 applications = query_result.get("CuratorApplications", {}) 1704 return applications 1705 1706 def query_map_proposals( 1707 self, extract_value: bool = False 1708 ) -> dict[int, dict[str, Any]]: 1709 """ 1710 Retrieves a mappping of proposals from the network. 1711 1712 Queries the network and returns a mapping of proposal IDs to 1713 their respective parameters. 1714 1715 Returns: 1716 A dictionary mapping proposal IDs 1717 to dictionaries of their parameters. 1718 1719 Raises: 1720 QueryError: If the query to the network fails or is invalid. 1721 """ 1722 1723 return self.query_map( 1724 "Proposals", extract_value=extract_value, module="GovernanceModule" 1725 )["Proposals"] 1726 1727 def query_map_weights( 1728 self, netuid: int = 0, extract_value: bool = False 1729 ) -> dict[int, list[tuple[int, int]]] | None: 1730 """ 1731 Retrieves a mapping of weights for keys on the network. 1732 1733 Queries the network and returns a mapping of key UIDs to 1734 their respective weights. 1735 1736 Args: 1737 netuid: The network UID from which to get the weights. 1738 1739 Returns: 1740 A dictionary mapping key UIDs to lists of their weights. 1741 1742 Raises: 1743 QueryError: If the query to the network fails or is invalid. 1744 """ 1745 1746 weights_dict = self.query_map( 1747 "Weights", 1748 [netuid], 1749 extract_value=extract_value 1750 ).get("Weights") 1751 return weights_dict 1752 1753 def query_map_key( 1754 self, 1755 netuid: int = 0, 1756 extract_value: bool = False, 1757 ) -> dict[int, Ss58Address]: 1758 """ 1759 Retrieves a map of keys from the network. 1760 1761 Fetches a mapping of key UIDs to their associated 1762 addresses on the network. 1763 The query can be targeted at a specific network UID if required. 1764 1765 Args: 1766 netuid: The network UID from which to get the keys. 1767 1768 Returns: 1769 A dictionary mapping key UIDs to their addresses. 1770 1771 Raises: 1772 QueryError: If the query to the network fails or is invalid. 1773 """ 1774 return self.query_map("Keys", [netuid], extract_value=extract_value)["Keys"] 1775 1776 def query_map_address( 1777 self, netuid: int = 0, extract_value: bool = False 1778 ) -> dict[int, str]: 1779 """ 1780 Retrieves a map of key addresses from the network. 1781 1782 Queries the network for a mapping of key UIDs to their addresses. 1783 1784 Args: 1785 netuid: The network UID from which to get the addresses. 1786 1787 Returns: 1788 A dictionary mapping key UIDs to their addresses. 1789 1790 Raises: 1791 QueryError: If the query to the network fails or is invalid. 1792 """ 1793 1794 return self.query_map("Address", [netuid], extract_value=extract_value)[ 1795 "Address" 1796 ] 1797 1798 def query_map_emission(self, extract_value: bool = False) -> dict[int, list[int]]: 1799 """ 1800 Retrieves a map of emissions for keys on the network. 1801 1802 Queries the network to get a mapping of 1803 key UIDs to their emission values. 1804 1805 Returns: 1806 A dictionary mapping key UIDs to lists of their emission values. 1807 1808 Raises: 1809 QueryError: If the query to the network fails or is invalid. 1810 """ 1811 1812 return self.query_map("Emission", extract_value=extract_value)["Emission"] 1813 1814 def query_map_pending_emission(self, extract_value: bool = False) -> int: 1815 """ 1816 Retrieves a map of pending emissions for the subnets. 1817 1818 Queries the network for a mapping of subnet UIDs to their pending emission values. 1819 1820 Returns: 1821 A dictionary mapping subnet UIDs to their pending emission values. 1822 1823 Raises: 1824 QueryError: If the query to the network fails or is invalid. 1825 """ 1826 return self.query_map("PendingEmission", extract_value=extract_value, module="SubnetEmissionModule")["PendingEmission"] 1827 1828 def query_map_subnet_emission(self, extract_value: bool = False) -> dict[int, int]: 1829 """ 1830 Retrieves a map of subnet emissions for the network. 1831 1832 Queries the network for a mapping of subnet UIDs to their emission values. 1833 1834 Returns: 1835 A dictionary mapping subnet UIDs to their emission values. 1836 1837 Raises: 1838 QueryError: If the query to the network fails or is invalid. 1839 """ 1840 1841 return self.query_map("SubnetEmission", extract_value=extract_value, module="SubnetEmissionModule")["SubnetEmission"] 1842 1843 def query_map_subnet_consensus(self, extract_value: bool = False) -> dict[int, str]: 1844 """ 1845 Retrieves a map of subnet consensus types for the network. 1846 1847 Queries the network for a mapping of subnet UIDs to their consensus types. 1848 1849 Returns: 1850 A dictionary mapping subnet UIDs to their consensus types. 1851 1852 Raises: 1853 QueryError: If the query to the network fails or is invalid. 1854 """ 1855 1856 return self.query_map("SubnetConsensusType", extract_value=extract_value, module="SubnetEmissionModule")["SubnetConsensusType"] 1857 1858 def query_map_incentive(self, extract_value: bool = False) -> dict[int, list[int]]: 1859 """ 1860 Retrieves a mapping of incentives for keys on the network. 1861 1862 Queries the network and returns a mapping of key UIDs to 1863 their respective incentive values. 1864 1865 Returns: 1866 A dictionary mapping key UIDs to lists of their incentive values. 1867 1868 Raises: 1869 QueryError: If the query to the network fails or is invalid. 1870 """ 1871 1872 return self.query_map("Incentive", extract_value=extract_value)["Incentive"] 1873 1874 def query_map_dividend(self, extract_value: bool = False) -> dict[int, list[int]]: 1875 """ 1876 Retrieves a mapping of dividends for keys on the network. 1877 1878 Queries the network for a mapping of key UIDs to 1879 their dividend values. 1880 1881 Returns: 1882 A dictionary mapping key UIDs to lists of their dividend values. 1883 1884 Raises: 1885 QueryError: If the query to the network fails or is invalid. 1886 """ 1887 1888 return self.query_map("Dividends", extract_value=extract_value)["Dividends"] 1889 1890 def query_map_regblock( 1891 self, netuid: int = 0, extract_value: bool = False 1892 ) -> dict[int, int]: 1893 """ 1894 Retrieves a mapping of registration blocks for keys on the network. 1895 1896 Queries the network for a mapping of key UIDs to 1897 the blocks where they were registered. 1898 1899 Args: 1900 netuid: The network UID from which to get the registration blocks. 1901 1902 Returns: 1903 A dictionary mapping key UIDs to their registration blocks. 1904 1905 Raises: 1906 QueryError: If the query to the network fails or is invalid. 1907 """ 1908 1909 return self.query_map( 1910 "RegistrationBlock", [netuid], extract_value=extract_value 1911 )["RegistrationBlock"] 1912 1913 def query_map_lastupdate(self, extract_value: bool = False) -> dict[int, list[int]]: 1914 """ 1915 Retrieves a mapping of the last update times for keys on the network. 1916 1917 Queries the network for a mapping of key UIDs to their last update times. 1918 1919 Returns: 1920 A dictionary mapping key UIDs to lists of their last update times. 1921 1922 Raises: 1923 QueryError: If the query to the network fails or is invalid. 1924 """ 1925 1926 return self.query_map("LastUpdate", extract_value=extract_value)["LastUpdate"] 1927 1928 def query_map_stakefrom( 1929 self, extract_value: bool = False 1930 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1931 """ 1932 Retrieves a mapping of stakes from various sources for keys on the network. 1933 1934 Queries the network to obtain a mapping of key addresses to the sources 1935 and amounts of stakes they have received. 1936 1937 Args: 1938 netuid: The network UID from which to get the stakes. 1939 1940 Returns: 1941 A dictionary mapping key addresses to lists of tuples 1942 (module_key_address, amount). 1943 1944 Raises: 1945 QueryError: If the query to the network fails or is invalid. 1946 """ 1947 1948 result = self.query_map("StakeFrom", [], extract_value=extract_value)[ 1949 "StakeFrom" 1950 ] 1951 1952 return transform_stake_dmap(result) 1953 1954 def query_map_staketo( 1955 self, extract_value: bool = False 1956 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1957 """ 1958 Retrieves a mapping of stakes to destinations for keys on the network. 1959 1960 Queries the network for a mapping of key addresses to the destinations 1961 and amounts of stakes they have made. 1962 1963 Args: 1964 netuid: The network UID from which to get the stakes. 1965 1966 Returns: 1967 A dictionary mapping key addresses to lists of tuples 1968 (module_key_address, amount). 1969 1970 Raises: 1971 QueryError: If the query to the network fails or is invalid. 1972 """ 1973 1974 result = self.query_map("StakeTo", [], extract_value=extract_value)[ 1975 "StakeTo" 1976 ] 1977 return transform_stake_dmap(result) 1978 1979 def query_map_delegationfee( 1980 self, netuid: int = 0, extract_value: bool = False 1981 ) -> dict[str, int]: 1982 """ 1983 Retrieves a mapping of delegation fees for keys on the network. 1984 1985 Queries the network to obtain a mapping of key addresses to their 1986 respective delegation fees. 1987 1988 Args: 1989 netuid: The network UID to filter the delegation fees. 1990 1991 Returns: 1992 A dictionary mapping key addresses to their delegation fees. 1993 1994 Raises: 1995 QueryError: If the query to the network fails or is invalid. 1996 """ 1997 1998 return self.query_map("DelegationFee", [netuid], extract_value=extract_value)[ 1999 "DelegationFee" 2000 ] 2001 2002 def query_map_tempo(self, extract_value: bool = False) -> dict[int, int]: 2003 """ 2004 Retrieves a mapping of tempo settings for the network. 2005 2006 Queries the network to obtain the tempo (rate of reward distributions) 2007 settings for various network subnets. 2008 2009 Returns: 2010 A dictionary mapping network UIDs to their tempo settings. 2011 2012 Raises: 2013 QueryError: If the query to the network fails or is invalid. 2014 """ 2015 2016 return self.query_map("Tempo", extract_value=extract_value)["Tempo"] 2017 2018 def query_map_immunity_period(self, extract_value: bool) -> dict[int, int]: 2019 """ 2020 Retrieves a mapping of immunity periods for the network. 2021 2022 Queries the network for the immunity period settings, 2023 which represent the time duration during which modules 2024 can not get deregistered. 2025 2026 Returns: 2027 A dictionary mapping network UIDs to their immunity period settings. 2028 2029 Raises: 2030 QueryError: If the query to the network fails or is invalid. 2031 """ 2032 2033 return self.query_map("ImmunityPeriod", extract_value=extract_value)[ 2034 "ImmunityPeriod" 2035 ] 2036 2037 def query_map_min_allowed_weights( 2038 self, extract_value: bool = False 2039 ) -> dict[int, int]: 2040 """ 2041 Retrieves a mapping of minimum allowed weights for the network. 2042 2043 Queries the network to obtain the minimum allowed weights, 2044 which are the lowest permissible weight values that can be set by 2045 validators. 2046 2047 Returns: 2048 A dictionary mapping network UIDs to 2049 their minimum allowed weight values. 2050 2051 Raises: 2052 QueryError: If the query to the network fails or is invalid. 2053 """ 2054 2055 return self.query_map("MinAllowedWeights", extract_value=extract_value)[ 2056 "MinAllowedWeights" 2057 ] 2058 2059 def query_map_max_allowed_weights( 2060 self, extract_value: bool = False 2061 ) -> dict[int, int]: 2062 """ 2063 Retrieves a mapping of maximum allowed weights for the network. 2064 2065 Queries the network for the maximum allowed weights, 2066 which are the highest permissible 2067 weight values that can be set by validators. 2068 2069 Returns: 2070 A dictionary mapping network UIDs to 2071 their maximum allowed weight values. 2072 2073 Raises: 2074 QueryError: If the query to the network fails or is invalid. 2075 """ 2076 2077 return self.query_map("MaxAllowedWeights", extract_value=extract_value)[ 2078 "MaxAllowedWeights" 2079 ] 2080 2081 def query_map_max_allowed_uids(self, extract_value: bool = False) -> dict[int, int]: 2082 """ 2083 Queries the network for the maximum number of allowed user IDs (UIDs) 2084 for each network subnet. 2085 2086 Fetches a mapping of network subnets to their respective 2087 limits on the number of user IDs that can be created or used. 2088 2089 Returns: 2090 A dictionary mapping network UIDs (unique identifiers) to their 2091 maximum allowed number of UIDs. 2092 Each entry represents a network subnet 2093 with its corresponding UID limit. 2094 2095 Raises: 2096 QueryError: If the query to the network fails or is invalid. 2097 """ 2098 2099 return self.query_map("MaxAllowedUids", extract_value=extract_value)[ 2100 "MaxAllowedUids" 2101 ] 2102 2103 def query_map_min_stake(self, extract_value: bool = False) -> dict[int, int]: 2104 """ 2105 Retrieves a mapping of minimum allowed stake on the network. 2106 2107 Queries the network to obtain the minimum number of stake, 2108 which is represented in nanotokens. 2109 2110 Returns: 2111 A dictionary mapping network UIDs to 2112 their minimum allowed stake values. 2113 2114 Raises: 2115 QueryError: If the query to the network fails or is invalid. 2116 """ 2117 2118 return self.query_map("MinStake", extract_value=extract_value)["MinStake"] 2119 2120 def query_map_max_stake(self, extract_value: bool = False) -> dict[int, int]: 2121 """ 2122 Retrieves a mapping of the maximum stake values for the network. 2123 2124 Queries the network for the maximum stake values across various s 2125 ubnets of the network. 2126 2127 Returns: 2128 A dictionary mapping network UIDs to their maximum stake values. 2129 2130 Raises: 2131 QueryError: If the query to the network fails or is invalid. 2132 """ 2133 2134 return self.query_map("MaxStake", extract_value=extract_value)["MaxStake"] 2135 2136 def query_map_founder(self, extract_value: bool = False) -> dict[int, str]: 2137 """ 2138 Retrieves a mapping of founders for the network. 2139 2140 Queries the network to obtain the founders associated with 2141 various subnets. 2142 2143 Returns: 2144 A dictionary mapping network UIDs to their respective founders. 2145 2146 Raises: 2147 QueryError: If the query to the network fails or is invalid. 2148 """ 2149 2150 return self.query_map("Founder", extract_value=extract_value)["Founder"] 2151 2152 def query_map_founder_share(self, extract_value: bool = False) -> dict[int, int]: 2153 """ 2154 Retrieves a mapping of founder shares for the network. 2155 2156 Queries the network for the share percentages 2157 allocated to founders across different subnets. 2158 2159 Returns: 2160 A dictionary mapping network UIDs to their founder share percentages. 2161 2162 Raises: 2163 QueryError: If the query to the network fails or is invalid. 2164 """ 2165 2166 return self.query_map("FounderShare", extract_value=extract_value)[ 2167 "FounderShare" 2168 ] 2169 2170 def query_map_incentive_ratio(self, extract_value: bool = False) -> dict[int, int]: 2171 """ 2172 Retrieves a mapping of incentive ratios for the network. 2173 2174 Queries the network for the incentive ratios, 2175 which are the proportions of rewards or incentives 2176 allocated in different subnets of the network. 2177 2178 Returns: 2179 A dictionary mapping network UIDs to their incentive ratios. 2180 2181 Raises: 2182 QueryError: If the query to the network fails or is invalid. 2183 """ 2184 2185 return self.query_map("IncentiveRatio", extract_value=extract_value)[ 2186 "IncentiveRatio" 2187 ] 2188 2189 def query_map_trust_ratio(self, extract_value: bool = False) -> dict[int, int]: 2190 """ 2191 Retrieves a mapping of trust ratios for the network. 2192 2193 Queries the network for trust ratios, 2194 indicative of the level of trust or credibility assigned 2195 to different subnets of the network. 2196 2197 Returns: 2198 A dictionary mapping network UIDs to their trust ratios. 2199 2200 Raises: 2201 QueryError: If the query to the network fails or is invalid. 2202 """ 2203 2204 return self.query_map("TrustRatio", extract_value=extract_value)["TrustRatio"] 2205 2206 def query_map_vote_mode_subnet(self, extract_value: bool = False) -> dict[int, str]: 2207 """ 2208 Retrieves a mapping of vote modes for subnets within the network. 2209 2210 Queries the network for the voting modes used in different 2211 subnets, which define the methodology or approach of voting within those 2212 subnets. 2213 2214 Returns: 2215 A dictionary mapping network UIDs to their vote 2216 modes for subnets. 2217 2218 Raises: 2219 QueryError: If the query to the network fails or is invalid. 2220 """ 2221 2222 return self.query_map("VoteModeSubnet", extract_value=extract_value)[ 2223 "VoteModeSubnet" 2224 ] 2225 2226 def query_map_legit_whitelist( 2227 self, extract_value: bool = False 2228 ) -> dict[Ss58Address, int]: 2229 """ 2230 Retrieves a mapping of whitelisted addresses for the network. 2231 2232 Queries the network for a mapping of whitelisted addresses 2233 and their respective legitimacy status. 2234 2235 Returns: 2236 A dictionary mapping addresses to their legitimacy status. 2237 2238 Raises: 2239 QueryError: If the query to the network fails or is invalid. 2240 """ 2241 2242 return self.query_map( 2243 "LegitWhitelist", module="GovernanceModule", extract_value=extract_value)[ 2244 "LegitWhitelist" 2245 ] 2246 2247 def query_map_subnet_names(self, extract_value: bool = False) -> dict[int, str]: 2248 """ 2249 Retrieves a mapping of subnet names within the network. 2250 2251 Queries the network for the names of various subnets, 2252 providing an overview of the different 2253 subnets within the network. 2254 2255 Returns: 2256 A dictionary mapping network UIDs to their subnet names. 2257 2258 Raises: 2259 QueryError: If the query to the network fails or is invalid. 2260 """ 2261 2262 return self.query_map("SubnetNames", extract_value=extract_value)["SubnetNames"] 2263 2264 def query_map_balances( 2265 self, extract_value: bool = False, block_hash: str | None = None 2266 ) -> dict[str, dict[str, int | dict[str, int | float]]]: 2267 """ 2268 Retrieves a mapping of account balances within the network. 2269 2270 Queries the network for the balances associated with different accounts. 2271 It provides detailed information including various types of 2272 balances for each account. 2273 2274 Returns: 2275 A dictionary mapping account addresses to their balance details. 2276 2277 Raises: 2278 QueryError: If the query to the network fails or is invalid. 2279 """ 2280 2281 return self.query_map("Account", module="System", extract_value=extract_value, block_hash=block_hash)[ 2282 "Account" 2283 ] 2284 2285 def query_map_registration_blocks( 2286 self, netuid: int = 0, extract_value: bool = False 2287 ) -> dict[int, int]: 2288 """ 2289 Retrieves a mapping of registration blocks for UIDs on the network. 2290 2291 Queries the network to find the block numbers at which various 2292 UIDs were registered. 2293 2294 Args: 2295 netuid: The network UID from which to get the registrations. 2296 2297 Returns: 2298 A dictionary mapping UIDs to their registration block numbers. 2299 2300 Raises: 2301 QueryError: If the query to the network fails or is invalid. 2302 """ 2303 2304 return self.query_map( 2305 "RegistrationBlock", [netuid], extract_value=extract_value 2306 )["RegistrationBlock"] 2307 2308 def query_map_name( 2309 self, netuid: int = 0, extract_value: bool = False 2310 ) -> dict[int, str]: 2311 """ 2312 Retrieves a mapping of names for keys on the network. 2313 2314 Queries the network for the names associated with different keys. 2315 It provides a mapping of key UIDs to their registered names. 2316 2317 Args: 2318 netuid: The network UID from which to get the names. 2319 2320 Returns: 2321 A dictionary mapping key UIDs to their names. 2322 2323 Raises: 2324 QueryError: If the query to the network fails or is invalid. 2325 """ 2326 2327 return self.query_map("Name", [netuid], extract_value=extract_value)["Name"] 2328 2329 # == QUERY FUNCTIONS == # 2330 2331 def get_immunity_period(self, netuid: int = 0) -> int: 2332 """ 2333 Queries the network for the immunity period setting. 2334 2335 The immunity period is a time duration during which a module 2336 can not be deregistered from the network. 2337 Fetches the immunity period for a specified network subnet. 2338 2339 Args: 2340 netuid: The network UID for which to query the immunity period. 2341 2342 Returns: 2343 The immunity period setting for the specified network subnet. 2344 2345 Raises: 2346 QueryError: If the query to the network fails or is invalid. 2347 """ 2348 2349 return self.query( 2350 "ImmunityPeriod", 2351 params=[netuid], 2352 ) 2353 2354 def get_max_set_weights_per_epoch(self): 2355 return self.query("MaximumSetWeightCallsPerEpoch") 2356 2357 def get_min_allowed_weights(self, netuid: int = 0) -> int: 2358 """ 2359 Queries the network for the minimum allowed weights setting. 2360 2361 Retrieves the minimum weight values that are possible to set 2362 by a validator within a specific network subnet. 2363 2364 Args: 2365 netuid: The network UID for which to query the minimum allowed 2366 weights. 2367 2368 Returns: 2369 The minimum allowed weight values for the specified network 2370 subnet. 2371 2372 Raises: 2373 QueryError: If the query to the network fails or is invalid. 2374 """ 2375 2376 return self.query( 2377 "MinAllowedWeights", 2378 params=[netuid], 2379 ) 2380 2381 def get_dao_treasury_address(self) -> Ss58Address: 2382 return self.query("DaoTreasuryAddress", module="GovernanceModule") 2383 2384 def get_max_allowed_weights(self, netuid: int = 0) -> int: 2385 """ 2386 Queries the network for the maximum allowed weights setting. 2387 2388 Retrieves the maximum weight values that are possible to set 2389 by a validator within a specific network subnet. 2390 2391 Args: 2392 netuid: The network UID for which to query the maximum allowed 2393 weights. 2394 2395 Returns: 2396 The maximum allowed weight values for the specified network 2397 subnet. 2398 2399 Raises: 2400 QueryError: If the query to the network fails or is invalid. 2401 """ 2402 2403 return self.query("MaxAllowedWeights", params=[netuid]) 2404 2405 def get_max_allowed_uids(self, netuid: int = 0) -> int: 2406 """ 2407 Queries the network for the maximum allowed UIDs setting. 2408 2409 Fetches the upper limit on the number of user IDs that can 2410 be allocated or used within a specific network subnet. 2411 2412 Args: 2413 netuid: The network UID for which to query the maximum allowed UIDs. 2414 2415 Returns: 2416 The maximum number of allowed UIDs for the specified network subnet. 2417 2418 Raises: 2419 QueryError: If the query to the network fails or is invalid. 2420 """ 2421 2422 return self.query("MaxAllowedUids", params=[netuid]) 2423 2424 def get_name(self, netuid: int = 0) -> str: 2425 """ 2426 Queries the network for the name of a specific subnet. 2427 2428 Args: 2429 netuid: The network UID for which to query the name. 2430 2431 Returns: 2432 The name of the specified network subnet. 2433 2434 Raises: 2435 QueryError: If the query to the network fails or is invalid. 2436 """ 2437 2438 return self.query("Name", params=[netuid]) 2439 2440 def get_subnet_name(self, netuid: int = 0) -> str: 2441 """ 2442 Queries the network for the name of a specific subnet. 2443 2444 Args: 2445 netuid: The network UID for which to query the name. 2446 2447 Returns: 2448 The name of the specified network subnet. 2449 2450 Raises: 2451 QueryError: If the query to the network fails or is invalid. 2452 """ 2453 2454 return self.query("SubnetNames", params=[netuid]) 2455 2456 def get_global_dao_treasury(self): 2457 return self.query("GlobalDaoTreasury", module="GovernanceModule") 2458 2459 def get_n(self, netuid: int = 0) -> int: 2460 """ 2461 Queries the network for the 'N' hyperparameter, which represents how 2462 many modules are on the network. 2463 2464 Args: 2465 netuid: The network UID for which to query the 'N' hyperparameter. 2466 2467 Returns: 2468 The value of the 'N' hyperparameter for the specified network 2469 subnet. 2470 2471 Raises: 2472 QueryError: If the query to the network fails or is invalid. 2473 """ 2474 2475 return self.query("N", params=[netuid]) 2476 2477 def get_tempo(self, netuid: int = 0) -> int: 2478 """ 2479 Queries the network for the tempo setting, measured in blocks, for the 2480 specified subnet. 2481 2482 Args: 2483 netuid: The network UID for which to query the tempo. 2484 2485 Returns: 2486 The tempo setting for the specified subnet. 2487 2488 Raises: 2489 QueryError: If the query to the network fails or is invalid. 2490 """ 2491 2492 return self.query("Tempo", params=[netuid]) 2493 2494 def get_total_free_issuance(self, block_hash: str | None = None) -> int: 2495 """ 2496 Queries the network for the total free issuance. 2497 2498 Fetches the total amount of free issuance tokens available 2499 2500 Returns: 2501 The total free issuance amount. 2502 2503 Raises: 2504 QueryError: If the query to the network fails or is invalid. 2505 """ 2506 2507 return self.query("TotalIssuance", module="Balances", block_hash=block_hash) 2508 2509 def get_total_stake(self, block_hash: str | None = None) -> int: 2510 """ 2511 Retrieves a mapping of total stakes for keys on the network. 2512 2513 Queries the network for a mapping of key UIDs to their total stake amounts. 2514 2515 Returns: 2516 A dictionary mapping key UIDs to their total stake amounts. 2517 2518 Raises: 2519 QueryError: If the query to the network fails or is invalid. 2520 """ 2521 2522 return self.query("TotalStake", block_hash=block_hash) 2523 2524 def get_registrations_per_block(self): 2525 """ 2526 Queries the network for the number of registrations per block. 2527 2528 Fetches the number of registrations that are processed per 2529 block within the network. 2530 2531 Returns: 2532 The number of registrations processed per block. 2533 2534 Raises: 2535 QueryError: If the query to the network fails or is invalid. 2536 """ 2537 2538 return self.query( 2539 "RegistrationsPerBlock", 2540 ) 2541 2542 def max_registrations_per_block(self, netuid: int = 0): 2543 """ 2544 Queries the network for the maximum number of registrations per block. 2545 2546 Retrieves the upper limit of registrations that can be processed in 2547 each block within a specific network subnet. 2548 2549 Args: 2550 netuid: The network UID for which to query. 2551 2552 Returns: 2553 The maximum number of registrations per block for 2554 the specified network subnet. 2555 2556 Raises: 2557 QueryError: If the query to the network fails or is invalid. 2558 """ 2559 2560 return self.query( 2561 "MaxRegistrationsPerBlock", 2562 params=[netuid], 2563 ) 2564 2565 def get_proposal(self, proposal_id: int = 0): 2566 """ 2567 Queries the network for a specific proposal. 2568 2569 Args: 2570 proposal_id: The ID of the proposal to query. 2571 2572 Returns: 2573 The details of the specified proposal. 2574 2575 Raises: 2576 QueryError: If the query to the network fails, is invalid, 2577 or if the proposal ID does not exist. 2578 """ 2579 2580 return self.query( 2581 "Proposals", 2582 params=[proposal_id], 2583 ) 2584 2585 def get_trust(self, netuid: int = 0): 2586 """ 2587 Queries the network for the trust setting of a specific network subnet. 2588 2589 Retrieves the trust level or score, which may represent the 2590 level of trustworthiness or reliability within a 2591 particular network subnet. 2592 2593 Args: 2594 netuid: The network UID for which to query the trust setting. 2595 2596 Returns: 2597 The trust level or score for the specified network subnet. 2598 2599 Raises: 2600 QueryError: If the query to the network fails or is invalid. 2601 """ 2602 2603 return self.query( 2604 "Trust", 2605 params=[netuid], 2606 ) 2607 2608 def get_uids(self, key: Ss58Address, netuid: int = 0) -> bool | None: 2609 """ 2610 Queries the network for module UIDs associated with a specific key. 2611 2612 Args: 2613 key: The key address for which to query UIDs. 2614 netuid: The network UID within which to search for the key. 2615 2616 Returns: 2617 A list of UIDs associated with the specified key. 2618 2619 Raises: 2620 QueryError: If the query to the network fails or is invalid. 2621 """ 2622 2623 return self.query( 2624 "Uids", 2625 params=[netuid, key], 2626 ) 2627 2628 def get_unit_emission(self) -> int: 2629 """ 2630 Queries the network for the unit emission setting. 2631 2632 Retrieves the unit emission value, which represents the 2633 emission rate or quantity for the $COMM token. 2634 2635 Returns: 2636 The unit emission value in nanos for the network. 2637 2638 Raises: 2639 QueryError: If the query to the network fails or is invalid. 2640 """ 2641 2642 return self.query("UnitEmission", module="SubnetEmissionModule") 2643 2644 def get_tx_rate_limit(self) -> int: 2645 """ 2646 Queries the network for the transaction rate limit. 2647 2648 Retrieves the rate limit for transactions within the network, 2649 which defines the maximum number of transactions that can be 2650 processed within a certain timeframe. 2651 2652 Returns: 2653 The transaction rate limit for the network. 2654 2655 Raises: 2656 QueryError: If the query to the network fails or is invalid. 2657 """ 2658 2659 return self.query( 2660 "TxRateLimit", 2661 ) 2662 2663 def get_subnet_burn(self) -> int: 2664 """Queries the network for the subnet burn value. 2665 2666 Retrieves the subnet burn value from the network, which represents 2667 the amount of tokens that are burned (permanently removed from 2668 circulation) for subnet-related operations. 2669 2670 Returns: 2671 int: The subnet burn value. 2672 2673 Raises: 2674 QueryError: If the query to the network fails or returns invalid data. 2675 """ 2676 2677 return self.query( 2678 "SubnetBurn", 2679 ) 2680 2681 def get_burn_rate(self) -> int: 2682 """ 2683 Queries the network for the burn rate setting. 2684 2685 Retrieves the burn rate, which represents the rate at 2686 which the $COMM token is permanently 2687 removed or 'burned' from circulation. 2688 2689 Returns: 2690 The burn rate for the network. 2691 2692 Raises: 2693 QueryError: If the query to the network fails or is invalid. 2694 """ 2695 2696 return self.query( 2697 "BurnRate", 2698 params=[], 2699 ) 2700 2701 def get_burn(self, netuid: int = 0) -> int: 2702 """ 2703 Queries the network for the burn setting. 2704 2705 Retrieves the burn value, which represents the amount of the 2706 $COMM token that is 'burned' or permanently removed from 2707 circulation. 2708 2709 Args: 2710 netuid: The network UID for which to query the burn value. 2711 2712 Returns: 2713 The burn value for the specified network subnet. 2714 2715 Raises: 2716 QueryError: If the query to the network fails or is invalid. 2717 """ 2718 2719 return self.query("Burn", params=[netuid]) 2720 2721 def get_min_burn(self) -> int: 2722 """ 2723 Queries the network for the minimum burn setting. 2724 2725 Retrieves the minimum burn value, indicating the lowest 2726 amount of the $COMM tokens that can be 'burned' or 2727 permanently removed from circulation. 2728 2729 Returns: 2730 The minimum burn value for the network. 2731 2732 Raises: 2733 QueryError: If the query to the network fails or is invalid. 2734 """ 2735 2736 return self.query( 2737 "BurnConfig", 2738 params=[], 2739 )["min_burn"] 2740 2741 def get_min_weight_stake(self) -> int: 2742 """ 2743 Queries the network for the minimum weight stake setting. 2744 2745 Retrieves the minimum weight stake, which represents the lowest 2746 stake weight that is allowed for certain operations or 2747 transactions within the network. 2748 2749 Returns: 2750 The minimum weight stake for the network. 2751 2752 Raises: 2753 QueryError: If the query to the network fails or is invalid. 2754 """ 2755 2756 return self.query("MinWeightStake", params=[]) 2757 2758 def get_vote_mode_global(self) -> str: 2759 """ 2760 Queries the network for the global vote mode setting. 2761 2762 Retrieves the global vote mode, which defines the overall voting 2763 methodology or approach used across the network in default. 2764 2765 Returns: 2766 The global vote mode setting for the network. 2767 2768 Raises: 2769 QueryError: If the query to the network fails or is invalid. 2770 """ 2771 2772 return self.query( 2773 "VoteModeGlobal", 2774 ) 2775 2776 def get_max_proposals(self) -> int: 2777 """ 2778 Queries the network for the maximum number of proposals allowed. 2779 2780 Retrieves the upper limit on the number of proposals that can be 2781 active or considered at any given time within the network. 2782 2783 Returns: 2784 The maximum number of proposals allowed on the network. 2785 2786 Raises: 2787 QueryError: If the query to the network fails or is invalid. 2788 """ 2789 2790 return self.query( 2791 "MaxProposals", 2792 ) 2793 2794 def get_max_registrations_per_block(self) -> int: 2795 """ 2796 Queries the network for the maximum number of registrations per block. 2797 2798 Retrieves the maximum number of registrations that can 2799 be processed in each block within the network. 2800 2801 Returns: 2802 The maximum number of registrations per block on the network. 2803 2804 Raises: 2805 QueryError: If the query to the network fails or is invalid. 2806 """ 2807 2808 return self.query( 2809 "MaxRegistrationsPerBlock", 2810 params=[], 2811 ) 2812 2813 def get_max_name_length(self) -> int: 2814 """ 2815 Queries the network for the maximum length allowed for names. 2816 2817 Retrieves the maximum character length permitted for names 2818 within the network. Such as the module names 2819 2820 Returns: 2821 The maximum length allowed for names on the network. 2822 2823 Raises: 2824 QueryError: If the query to the network fails or is invalid. 2825 """ 2826 2827 return self.query( 2828 "MaxNameLength", 2829 params=[], 2830 ) 2831 2832 def get_global_vote_threshold(self) -> int: 2833 """ 2834 Queries the network for the global vote threshold. 2835 2836 Retrieves the global vote threshold, which is the critical value or 2837 percentage required for decisions in the network's governance process. 2838 2839 Returns: 2840 The global vote threshold for the network. 2841 2842 Raises: 2843 QueryError: If the query to the network fails or is invalid. 2844 """ 2845 2846 return self.query( 2847 "GlobalVoteThreshold", 2848 ) 2849 2850 def get_max_allowed_subnets(self) -> int: 2851 """ 2852 Queries the network for the maximum number of allowed subnets. 2853 2854 Retrieves the upper limit on the number of subnets that can 2855 be created or operated within the network. 2856 2857 Returns: 2858 The maximum number of allowed subnets on the network. 2859 2860 Raises: 2861 QueryError: If the query to the network fails or is invalid. 2862 """ 2863 2864 return self.query( 2865 "MaxAllowedSubnets", 2866 params=[], 2867 ) 2868 2869 def get_max_allowed_modules(self) -> int: 2870 """ 2871 Queries the network for the maximum number of allowed modules. 2872 2873 Retrieves the upper limit on the number of modules that 2874 can be registered within the network. 2875 2876 Returns: 2877 The maximum number of allowed modules on the network. 2878 2879 Raises: 2880 QueryError: If the query to the network fails or is invalid. 2881 """ 2882 2883 return self.query( 2884 "MaxAllowedModules", 2885 params=[], 2886 ) 2887 2888 def get_min_stake(self, netuid: int = 0) -> int: 2889 """ 2890 Queries the network for the minimum stake required to register a key. 2891 2892 Retrieves the minimum amount of stake necessary for 2893 registering a key within a specific network subnet. 2894 2895 Args: 2896 netuid: The network UID for which to query the minimum stake. 2897 2898 Returns: 2899 The minimum stake required for key registration in nanos. 2900 2901 Raises: 2902 QueryError: If the query to the network fails or is invalid. 2903 """ 2904 2905 return self.query("MinStake", params=[netuid]) 2906 2907 def get_stakefrom( 2908 self, 2909 key: Ss58Address, 2910 ) -> dict[str, int]: 2911 """ 2912 Retrieves the stake amounts from all stakers to a specific staked address. 2913 2914 Queries the network for the stakes received by a particular staked address 2915 from all stakers. 2916 2917 Args: 2918 key: The address of the key receiving the stakes. 2919 2920 Returns: 2921 A dictionary mapping staker addresses to their respective stake amounts. 2922 2923 Raises: 2924 QueryError: If the query to the network fails or is invalid. 2925 """ 2926 2927 # Has to use query map in order to iterate through the storage prefix. 2928 return self.query_map("StakeFrom", [key], extract_value=False).get("StakeFrom", {}) 2929 2930 def get_staketo( 2931 self, 2932 key: Ss58Address, 2933 ) -> dict[str, int]: 2934 """ 2935 Retrieves the stake amounts provided by a specific staker to all staked addresses. 2936 2937 Queries the network for the stakes provided by a particular staker to 2938 all staked addresses. 2939 2940 Args: 2941 key: The address of the key providing the stakes. 2942 2943 Returns: 2944 A dictionary mapping staked addresses to their respective received stake amounts. 2945 2946 Raises: 2947 QueryError: If the query to the network fails or is invalid. 2948 """ 2949 2950 # Has to use query map in order to iterate through the storage prefix. 2951 return self.query_map("StakeTo", [key], extract_value=False).get("StakeTo", {}) 2952 2953 def get_balance( 2954 self, 2955 addr: Ss58Address, 2956 ) -> int: 2957 """ 2958 Retrieves the balance of a specific key. 2959 2960 Args: 2961 addr: The address of the key to query the balance for. 2962 2963 Returns: 2964 The balance of the specified key. 2965 2966 Raises: 2967 QueryError: If the query to the network fails or is invalid. 2968 """ 2969 2970 result = self.query("Account", module="System", params=[addr]) 2971 2972 return result["data"]["free"] 2973 2974 def get_block(self, block_hash: str | None = None) -> dict[Any, Any] | None: 2975 """ 2976 Retrieves information about a specific block in the network. 2977 2978 Queries the network for details about a block, such as its number, 2979 hash, and other relevant information. 2980 2981 Returns: 2982 The requested information about the block, 2983 or None if the block does not exist 2984 or the information is not available. 2985 2986 Raises: 2987 QueryError: If the query to the network fails or is invalid. 2988 """ 2989 2990 with self.get_conn() as substrate: 2991 block: dict[Any, Any] | None = substrate.get_block( # type: ignore 2992 block_hash # type: ignore 2993 ) 2994 2995 return block 2996 2997 def get_existential_deposit(self, block_hash: str | None = None) -> int: 2998 """ 2999 Retrieves the existential deposit value for the network. 3000 3001 The existential deposit is the minimum balance that must be maintained 3002 in an account to prevent it from being purged. Denotated in nano units. 3003 3004 Returns: 3005 The existential deposit value in nano units. 3006 Note: 3007 The value returned is a fixed value defined in the 3008 client and may not reflect changes in the network's configuration. 3009 """ 3010 3011 with self.get_conn() as substrate: 3012 result: int = substrate.get_constant( # type: ignore 3013 "Balances", "ExistentialDeposit", block_hash 3014 ).value # type: ignore 3015 3016 return result 3017 3018 def get_voting_power_delegators(self) -> list[Ss58Address]: 3019 result = self.query("NotDelegatingVotingPower", [], module="GovernanceModule") 3020 return result 3021 3022 def add_transfer_dao_treasury_proposal( 3023 self, 3024 key: Keypair, 3025 data: str, 3026 amount_nano: int, 3027 dest: Ss58Address, 3028 ): 3029 params = {"dest": dest, "value": amount_nano, "data": data} 3030 3031 return self.compose_call( 3032 module="GovernanceModule", 3033 fn="add_transfer_dao_treasury_proposal", 3034 params=params, 3035 key=key, 3036 ) 3037 3038 def delegate_rootnet_control(self, key: Keypair, dest: Ss58Address): 3039 params = {"origin": key, "target": dest} 3040 3041 return self.compose_call( 3042 module="SubspaceModule", 3043 fn="delegate_rootnet_control", 3044 params=params, 3045 key=key, 3046 )
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.
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.
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.
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.
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', ...}
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>
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.
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.
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_inclusion
is True. Otherwise, returns a string identifier of the extrinsic.
Raises:
- ChainTransactionError: If the transaction fails.
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_inclusion
is True. Otherwise, returns a string identifier of the extrinsic.
Raises:
- ChainTransactionError: If the transaction fails.
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.
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.
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.
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.
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 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
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.
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
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.
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
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.
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
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.
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
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.
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 general_params.get("subnet_metadata") is None: 1259 general_params["metadata"] = None 1260 else: 1261 general_params["metadata"] = general_params["subnet_metadata"] 1262 1263 response = self.compose_call( 1264 fn="update_subnet", 1265 params=general_params, 1266 key=key, 1267 ) 1268 1269 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.
1271 def transfer_stake( 1272 self, 1273 key: Keypair, 1274 amount: int, 1275 from_module_key: Ss58Address, 1276 dest_module_address: Ss58Address, 1277 ) -> ExtrinsicReceipt: 1278 """ 1279 Realocate staked tokens from one staked module to another module. 1280 1281 Args: 1282 key: The keypair associated with the account that is delegating the tokens. 1283 amount: The amount of staked tokens to transfer, in nanotokens. 1284 from_module_key: The SS58 address of the module you want to transfer from (currently delegated by the key). 1285 dest_module_address: The SS58 address of the destination (newly delegated key). 1286 netuid: The network identifier. 1287 1288 Returns: 1289 A receipt of the stake transfer transaction. 1290 1291 Raises: 1292 InsufficientStakeError: If the source module key does not have 1293 enough staked tokens. ChainTransactionError: If the transaction 1294 fails. 1295 """ 1296 1297 amount = amount - self.get_existential_deposit() 1298 1299 params = { 1300 "amount": amount, 1301 "module_key": from_module_key, 1302 "new_module_key": dest_module_address, 1303 } 1304 1305 response = self.compose_call("transfer_stake", key=key, params=params) 1306 1307 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.
1309 def multiunstake( 1310 self, 1311 key: Keypair, 1312 keys: list[Ss58Address], 1313 amounts: list[int], 1314 ) -> ExtrinsicReceipt: 1315 """ 1316 Unstakes tokens from multiple module keys. 1317 1318 And the lists `keys` and `amounts` must be of the same length. Each 1319 amount corresponds to the module key at the same index. 1320 1321 Args: 1322 key: The keypair associated with the unstaker's account. 1323 keys: A list of SS58 addresses of the module keys to unstake from. 1324 amounts: A list of amounts to unstake from each module key, 1325 in nanotokens. 1326 netuid: The network identifier. 1327 1328 Returns: 1329 A receipt of the multi-unstaking transaction. 1330 1331 Raises: 1332 MismatchedLengthError: If the lengths of keys and amounts lists do 1333 not match. InsufficientStakeError: If any of the module keys do not 1334 have enough staked tokens. ChainTransactionError: If the transaction 1335 fails. 1336 """ 1337 1338 assert len(keys) == len(amounts) 1339 1340 params = {"module_keys": keys, "amounts": amounts} 1341 1342 response = self.compose_call("remove_stake_multiple", params=params, key=key) 1343 1344 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.
1346 def multistake( 1347 self, 1348 key: Keypair, 1349 keys: list[Ss58Address], 1350 amounts: list[int], 1351 ) -> ExtrinsicReceipt: 1352 """ 1353 Stakes tokens to multiple module keys. 1354 1355 The lengths of the `keys` and `amounts` lists must be the same. Each 1356 amount corresponds to the module key at the same index. 1357 1358 Args: 1359 key: The keypair associated with the staker's account. 1360 keys: A list of SS58 addresses of the module keys to stake to. 1361 amounts: A list of amounts to stake to each module key, 1362 in nanotokens. 1363 netuid: The network identifier. 1364 1365 Returns: 1366 A receipt of the multi-staking transaction. 1367 1368 Raises: 1369 MismatchedLengthError: If the lengths of keys and amounts lists 1370 do not match. 1371 ChainTransactionError: If the transaction fails. 1372 """ 1373 1374 assert len(keys) == len(amounts) 1375 1376 params = { 1377 "module_keys": keys, 1378 "amounts": amounts, 1379 } 1380 1381 response = self.compose_call("add_stake_multiple", params=params, key=key) 1382 1383 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.
1421 def add_subnet_proposal( 1422 self, key: Keypair, 1423 params: dict[str, Any], 1424 ipfs: str, 1425 netuid: int = 0 1426 ) -> ExtrinsicReceipt: 1427 """ 1428 Submits a proposal for creating or modifying a subnet within the 1429 network. 1430 1431 The proposal includes various parameters like the name, founder, share 1432 allocations, and other subnet-specific settings. 1433 1434 Args: 1435 key: The keypair used for signing the proposal transaction. 1436 params: The parameters for the subnet proposal. 1437 netuid: The network identifier. 1438 1439 Returns: 1440 A receipt of the subnet proposal transaction. 1441 1442 Raises: 1443 InvalidParameterError: If the provided subnet 1444 parameters are invalid. 1445 ChainTransactionError: If the transaction fails. 1446 """ 1447 1448 general_params = dict(params) 1449 general_params["netuid"] = netuid 1450 general_params["data"] = ipfs 1451 if "metadata" not in general_params: 1452 general_params["metadata"] = None 1453 1454 # general_params["burn_config"] = json.dumps(general_params["burn_config"]) 1455 response = self.compose_call( 1456 fn="add_subnet_params_proposal", 1457 params=general_params, 1458 key=key, 1459 module="GovernanceModule", 1460 ) 1461 1462 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.
1464 def add_custom_proposal( 1465 self, 1466 key: Keypair, 1467 cid: str, 1468 ) -> ExtrinsicReceipt: 1469 1470 params = {"data": cid} 1471 1472 response = self.compose_call( 1473 fn="add_global_custom_proposal", 1474 params=params, 1475 key=key, 1476 module="GovernanceModule", 1477 ) 1478 return response
1480 def add_custom_subnet_proposal( 1481 self, 1482 key: Keypair, 1483 cid: str, 1484 netuid: int = 0, 1485 ) -> ExtrinsicReceipt: 1486 """ 1487 Submits a proposal for creating or modifying a custom subnet within the 1488 network. 1489 1490 The proposal includes various parameters like the name, founder, share 1491 allocations, and other subnet-specific settings. 1492 1493 Args: 1494 key: The keypair used for signing the proposal transaction. 1495 params: The parameters for the subnet proposal. 1496 netuid: The network identifier. 1497 1498 Returns: 1499 A receipt of the subnet proposal transaction. 1500 """ 1501 1502 params = { 1503 "data": cid, 1504 "netuid": netuid, 1505 } 1506 1507 response = self.compose_call( 1508 fn="add_subnet_custom_proposal", 1509 params=params, 1510 key=key, 1511 module="GovernanceModule", 1512 ) 1513 1514 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.
1516 def add_global_proposal( 1517 self, 1518 key: Keypair, 1519 params: NetworkParams, 1520 cid: str | None, 1521 ) -> ExtrinsicReceipt: 1522 """ 1523 Submits a proposal for altering the global network parameters. 1524 1525 Allows for the submission of a proposal to 1526 change various global parameters 1527 of the network, such as emission rates, rate limits, and voting 1528 thresholds. It is used to 1529 suggest changes that affect the entire network's operation. 1530 1531 Args: 1532 key: The keypair used for signing the proposal transaction. 1533 params: A dictionary containing global network parameters 1534 like maximum allowed subnets, modules, 1535 transaction rate limits, and others. 1536 1537 Returns: 1538 A receipt of the global proposal transaction. 1539 1540 Raises: 1541 InvalidParameterError: If the provided network 1542 parameters are invalid. 1543 ChainTransactionError: If the transaction fails. 1544 """ 1545 general_params = cast(dict[str, Any], params) 1546 cid = cid or "" 1547 general_params["data"] = cid 1548 1549 response = self.compose_call( 1550 fn="add_global_params_proposal", 1551 params=general_params, 1552 key=key, 1553 module="GovernanceModule", 1554 ) 1555 1556 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.
1558 def vote_on_proposal( 1559 self, 1560 key: Keypair, 1561 proposal_id: int, 1562 agree: bool, 1563 ) -> ExtrinsicReceipt: 1564 """ 1565 Casts a vote on a specified proposal within the network. 1566 1567 Args: 1568 key: The keypair used for signing the vote transaction. 1569 proposal_id: The unique identifier of the proposal to vote on. 1570 1571 Returns: 1572 A receipt of the voting transaction in nanotokens. 1573 1574 Raises: 1575 InvalidProposalIDError: If the provided proposal ID does not 1576 exist or is invalid. 1577 ChainTransactionError: If the transaction fails. 1578 """ 1579 1580 params = {"proposal_id": proposal_id, "agree": agree} 1581 1582 response = self.compose_call( 1583 "vote_proposal", 1584 key=key, 1585 params=params, 1586 module="GovernanceModule", 1587 ) 1588 1589 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.
1591 def unvote_on_proposal( 1592 self, 1593 key: Keypair, 1594 proposal_id: int, 1595 ) -> ExtrinsicReceipt: 1596 """ 1597 Retracts a previously cast vote on a specified proposal. 1598 1599 Args: 1600 key: The keypair used for signing the unvote transaction. 1601 proposal_id: The unique identifier of the proposal to withdraw the 1602 vote from. 1603 1604 Returns: 1605 A receipt of the unvoting transaction in nanotokens. 1606 1607 Raises: 1608 InvalidProposalIDError: If the provided proposal ID does not 1609 exist or is invalid. 1610 ChainTransactionError: If the transaction fails to be processed, or 1611 if there was no prior vote to retract. 1612 """ 1613 1614 params = {"proposal_id": proposal_id} 1615 1616 response = self.compose_call( 1617 "remove_vote_proposal", 1618 key=key, 1619 params=params, 1620 module="GovernanceModule", 1621 ) 1622 1623 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.
1625 def enable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1626 """ 1627 Enables vote power delegation for the signer's account. 1628 1629 Args: 1630 key: The keypair used for signing the delegation transaction. 1631 1632 Returns: 1633 A receipt of the vote power delegation transaction. 1634 1635 Raises: 1636 ChainTransactionError: If the transaction fails. 1637 """ 1638 1639 response = self.compose_call( 1640 "enable_vote_power_delegation", 1641 params={}, 1642 key=key, 1643 module="GovernanceModule", 1644 ) 1645 1646 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.
1648 def disable_vote_power_delegation(self, key: Keypair) -> ExtrinsicReceipt: 1649 """ 1650 Disables vote power delegation for the signer's account. 1651 1652 Args: 1653 key: The keypair used for signing the delegation transaction. 1654 1655 Returns: 1656 A receipt of the vote power delegation transaction. 1657 1658 Raises: 1659 ChainTransactionError: If the transaction fails. 1660 """ 1661 1662 response = self.compose_call( 1663 "disable_vote_power_delegation", 1664 params={}, 1665 key=key, 1666 module="GovernanceModule", 1667 ) 1668 1669 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.
1671 def add_dao_application( 1672 self, key: Keypair, application_key: Ss58Address, data: str 1673 ) -> ExtrinsicReceipt: 1674 """ 1675 Submits a new application to the general subnet DAO. 1676 1677 Args: 1678 key: The keypair used for signing the application transaction. 1679 application_key: The SS58 address of the application key. 1680 data: The data associated with the application. 1681 1682 Returns: 1683 A receipt of the application transaction. 1684 1685 Raises: 1686 ChainTransactionError: If the transaction fails. 1687 """ 1688 1689 params = {"application_key": application_key, "data": data} 1690 1691 response = self.compose_call( 1692 "add_dao_application", module="GovernanceModule", key=key, 1693 params=params 1694 ) 1695 1696 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.
1706 def query_map_proposals( 1707 self, extract_value: bool = False 1708 ) -> dict[int, dict[str, Any]]: 1709 """ 1710 Retrieves a mappping of proposals from the network. 1711 1712 Queries the network and returns a mapping of proposal IDs to 1713 their respective parameters. 1714 1715 Returns: 1716 A dictionary mapping proposal IDs 1717 to dictionaries of their parameters. 1718 1719 Raises: 1720 QueryError: If the query to the network fails or is invalid. 1721 """ 1722 1723 return self.query_map( 1724 "Proposals", extract_value=extract_value, module="GovernanceModule" 1725 )["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.
1727 def query_map_weights( 1728 self, netuid: int = 0, extract_value: bool = False 1729 ) -> dict[int, list[tuple[int, int]]] | None: 1730 """ 1731 Retrieves a mapping of weights for keys on the network. 1732 1733 Queries the network and returns a mapping of key UIDs to 1734 their respective weights. 1735 1736 Args: 1737 netuid: The network UID from which to get the weights. 1738 1739 Returns: 1740 A dictionary mapping key UIDs to lists of their weights. 1741 1742 Raises: 1743 QueryError: If the query to the network fails or is invalid. 1744 """ 1745 1746 weights_dict = self.query_map( 1747 "Weights", 1748 [netuid], 1749 extract_value=extract_value 1750 ).get("Weights") 1751 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.
1753 def query_map_key( 1754 self, 1755 netuid: int = 0, 1756 extract_value: bool = False, 1757 ) -> dict[int, Ss58Address]: 1758 """ 1759 Retrieves a map of keys from the network. 1760 1761 Fetches a mapping of key UIDs to their associated 1762 addresses on the network. 1763 The query can be targeted at a specific network UID if required. 1764 1765 Args: 1766 netuid: The network UID from which to get the keys. 1767 1768 Returns: 1769 A dictionary mapping key UIDs to their addresses. 1770 1771 Raises: 1772 QueryError: If the query to the network fails or is invalid. 1773 """ 1774 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.
1776 def query_map_address( 1777 self, netuid: int = 0, extract_value: bool = False 1778 ) -> dict[int, str]: 1779 """ 1780 Retrieves a map of key addresses from the network. 1781 1782 Queries the network for a mapping of key UIDs to their addresses. 1783 1784 Args: 1785 netuid: The network UID from which to get the addresses. 1786 1787 Returns: 1788 A dictionary mapping key UIDs to their addresses. 1789 1790 Raises: 1791 QueryError: If the query to the network fails or is invalid. 1792 """ 1793 1794 return self.query_map("Address", [netuid], extract_value=extract_value)[ 1795 "Address" 1796 ]
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.
1798 def query_map_emission(self, extract_value: bool = False) -> dict[int, list[int]]: 1799 """ 1800 Retrieves a map of emissions for keys on the network. 1801 1802 Queries the network to get a mapping of 1803 key UIDs to their emission values. 1804 1805 Returns: 1806 A dictionary mapping key UIDs to lists of their emission values. 1807 1808 Raises: 1809 QueryError: If the query to the network fails or is invalid. 1810 """ 1811 1812 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.
1814 def query_map_pending_emission(self, extract_value: bool = False) -> int: 1815 """ 1816 Retrieves a map of pending emissions for the subnets. 1817 1818 Queries the network for a mapping of subnet UIDs to their pending emission values. 1819 1820 Returns: 1821 A dictionary mapping subnet UIDs to their pending emission values. 1822 1823 Raises: 1824 QueryError: If the query to the network fails or is invalid. 1825 """ 1826 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.
1828 def query_map_subnet_emission(self, extract_value: bool = False) -> dict[int, int]: 1829 """ 1830 Retrieves a map of subnet emissions for the network. 1831 1832 Queries the network for a mapping of subnet UIDs to their emission values. 1833 1834 Returns: 1835 A dictionary mapping subnet UIDs to their emission values. 1836 1837 Raises: 1838 QueryError: If the query to the network fails or is invalid. 1839 """ 1840 1841 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.
1843 def query_map_subnet_consensus(self, extract_value: bool = False) -> dict[int, str]: 1844 """ 1845 Retrieves a map of subnet consensus types for the network. 1846 1847 Queries the network for a mapping of subnet UIDs to their consensus types. 1848 1849 Returns: 1850 A dictionary mapping subnet UIDs to their consensus types. 1851 1852 Raises: 1853 QueryError: If the query to the network fails or is invalid. 1854 """ 1855 1856 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.
1858 def query_map_incentive(self, extract_value: bool = False) -> dict[int, list[int]]: 1859 """ 1860 Retrieves a mapping of incentives for keys on the network. 1861 1862 Queries the network and returns a mapping of key UIDs to 1863 their respective incentive values. 1864 1865 Returns: 1866 A dictionary mapping key UIDs to lists of their incentive values. 1867 1868 Raises: 1869 QueryError: If the query to the network fails or is invalid. 1870 """ 1871 1872 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.
1874 def query_map_dividend(self, extract_value: bool = False) -> dict[int, list[int]]: 1875 """ 1876 Retrieves a mapping of dividends for keys on the network. 1877 1878 Queries the network for a mapping of key UIDs to 1879 their dividend values. 1880 1881 Returns: 1882 A dictionary mapping key UIDs to lists of their dividend values. 1883 1884 Raises: 1885 QueryError: If the query to the network fails or is invalid. 1886 """ 1887 1888 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.
1890 def query_map_regblock( 1891 self, netuid: int = 0, extract_value: bool = False 1892 ) -> dict[int, int]: 1893 """ 1894 Retrieves a mapping of registration blocks for keys on the network. 1895 1896 Queries the network for a mapping of key UIDs to 1897 the blocks where they were registered. 1898 1899 Args: 1900 netuid: The network UID from which to get the registration blocks. 1901 1902 Returns: 1903 A dictionary mapping key UIDs to their registration blocks. 1904 1905 Raises: 1906 QueryError: If the query to the network fails or is invalid. 1907 """ 1908 1909 return self.query_map( 1910 "RegistrationBlock", [netuid], extract_value=extract_value 1911 )["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.
1913 def query_map_lastupdate(self, extract_value: bool = False) -> dict[int, list[int]]: 1914 """ 1915 Retrieves a mapping of the last update times for keys on the network. 1916 1917 Queries the network for a mapping of key UIDs to their last update times. 1918 1919 Returns: 1920 A dictionary mapping key UIDs to lists of their last update times. 1921 1922 Raises: 1923 QueryError: If the query to the network fails or is invalid. 1924 """ 1925 1926 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.
1928 def query_map_stakefrom( 1929 self, extract_value: bool = False 1930 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1931 """ 1932 Retrieves a mapping of stakes from various sources for keys on the network. 1933 1934 Queries the network to obtain a mapping of key addresses to the sources 1935 and amounts of stakes they have received. 1936 1937 Args: 1938 netuid: The network UID from which to get the stakes. 1939 1940 Returns: 1941 A dictionary mapping key addresses to lists of tuples 1942 (module_key_address, amount). 1943 1944 Raises: 1945 QueryError: If the query to the network fails or is invalid. 1946 """ 1947 1948 result = self.query_map("StakeFrom", [], extract_value=extract_value)[ 1949 "StakeFrom" 1950 ] 1951 1952 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.
1954 def query_map_staketo( 1955 self, extract_value: bool = False 1956 ) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]: 1957 """ 1958 Retrieves a mapping of stakes to destinations for keys on the network. 1959 1960 Queries the network for a mapping of key addresses to the destinations 1961 and amounts of stakes they have made. 1962 1963 Args: 1964 netuid: The network UID from which to get the stakes. 1965 1966 Returns: 1967 A dictionary mapping key addresses to lists of tuples 1968 (module_key_address, amount). 1969 1970 Raises: 1971 QueryError: If the query to the network fails or is invalid. 1972 """ 1973 1974 result = self.query_map("StakeTo", [], extract_value=extract_value)[ 1975 "StakeTo" 1976 ] 1977 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.
1979 def query_map_delegationfee( 1980 self, netuid: int = 0, extract_value: bool = False 1981 ) -> dict[str, int]: 1982 """ 1983 Retrieves a mapping of delegation fees for keys on the network. 1984 1985 Queries the network to obtain a mapping of key addresses to their 1986 respective delegation fees. 1987 1988 Args: 1989 netuid: The network UID to filter the delegation fees. 1990 1991 Returns: 1992 A dictionary mapping key addresses to their delegation fees. 1993 1994 Raises: 1995 QueryError: If the query to the network fails or is invalid. 1996 """ 1997 1998 return self.query_map("DelegationFee", [netuid], extract_value=extract_value)[ 1999 "DelegationFee" 2000 ]
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.
2002 def query_map_tempo(self, extract_value: bool = False) -> dict[int, int]: 2003 """ 2004 Retrieves a mapping of tempo settings for the network. 2005 2006 Queries the network to obtain the tempo (rate of reward distributions) 2007 settings for various network subnets. 2008 2009 Returns: 2010 A dictionary mapping network UIDs to their tempo settings. 2011 2012 Raises: 2013 QueryError: If the query to the network fails or is invalid. 2014 """ 2015 2016 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.
2018 def query_map_immunity_period(self, extract_value: bool) -> dict[int, int]: 2019 """ 2020 Retrieves a mapping of immunity periods for the network. 2021 2022 Queries the network for the immunity period settings, 2023 which represent the time duration during which modules 2024 can not get deregistered. 2025 2026 Returns: 2027 A dictionary mapping network UIDs to their immunity period settings. 2028 2029 Raises: 2030 QueryError: If the query to the network fails or is invalid. 2031 """ 2032 2033 return self.query_map("ImmunityPeriod", extract_value=extract_value)[ 2034 "ImmunityPeriod" 2035 ]
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.
2037 def query_map_min_allowed_weights( 2038 self, extract_value: bool = False 2039 ) -> dict[int, int]: 2040 """ 2041 Retrieves a mapping of minimum allowed weights for the network. 2042 2043 Queries the network to obtain the minimum allowed weights, 2044 which are the lowest permissible weight values that can be set by 2045 validators. 2046 2047 Returns: 2048 A dictionary mapping network UIDs to 2049 their minimum allowed weight values. 2050 2051 Raises: 2052 QueryError: If the query to the network fails or is invalid. 2053 """ 2054 2055 return self.query_map("MinAllowedWeights", extract_value=extract_value)[ 2056 "MinAllowedWeights" 2057 ]
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.
2059 def query_map_max_allowed_weights( 2060 self, extract_value: bool = False 2061 ) -> dict[int, int]: 2062 """ 2063 Retrieves a mapping of maximum allowed weights for the network. 2064 2065 Queries the network for the maximum allowed weights, 2066 which are the highest permissible 2067 weight values that can be set by validators. 2068 2069 Returns: 2070 A dictionary mapping network UIDs to 2071 their maximum allowed weight values. 2072 2073 Raises: 2074 QueryError: If the query to the network fails or is invalid. 2075 """ 2076 2077 return self.query_map("MaxAllowedWeights", extract_value=extract_value)[ 2078 "MaxAllowedWeights" 2079 ]
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.
2081 def query_map_max_allowed_uids(self, extract_value: bool = False) -> dict[int, int]: 2082 """ 2083 Queries the network for the maximum number of allowed user IDs (UIDs) 2084 for each network subnet. 2085 2086 Fetches a mapping of network subnets to their respective 2087 limits on the number of user IDs that can be created or used. 2088 2089 Returns: 2090 A dictionary mapping network UIDs (unique identifiers) to their 2091 maximum allowed number of UIDs. 2092 Each entry represents a network subnet 2093 with its corresponding UID limit. 2094 2095 Raises: 2096 QueryError: If the query to the network fails or is invalid. 2097 """ 2098 2099 return self.query_map("MaxAllowedUids", extract_value=extract_value)[ 2100 "MaxAllowedUids" 2101 ]
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.
2103 def query_map_min_stake(self, extract_value: bool = False) -> dict[int, int]: 2104 """ 2105 Retrieves a mapping of minimum allowed stake on the network. 2106 2107 Queries the network to obtain the minimum number of stake, 2108 which is represented in nanotokens. 2109 2110 Returns: 2111 A dictionary mapping network UIDs to 2112 their minimum allowed stake values. 2113 2114 Raises: 2115 QueryError: If the query to the network fails or is invalid. 2116 """ 2117 2118 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.
2120 def query_map_max_stake(self, extract_value: bool = False) -> dict[int, int]: 2121 """ 2122 Retrieves a mapping of the maximum stake values for the network. 2123 2124 Queries the network for the maximum stake values across various s 2125 ubnets of the network. 2126 2127 Returns: 2128 A dictionary mapping network UIDs to their maximum stake values. 2129 2130 Raises: 2131 QueryError: If the query to the network fails or is invalid. 2132 """ 2133 2134 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.
2136 def query_map_founder(self, extract_value: bool = False) -> dict[int, str]: 2137 """ 2138 Retrieves a mapping of founders for the network. 2139 2140 Queries the network to obtain the founders associated with 2141 various subnets. 2142 2143 Returns: 2144 A dictionary mapping network UIDs to their respective founders. 2145 2146 Raises: 2147 QueryError: If the query to the network fails or is invalid. 2148 """ 2149 2150 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.
2170 def query_map_incentive_ratio(self, extract_value: bool = False) -> dict[int, int]: 2171 """ 2172 Retrieves a mapping of incentive ratios for the network. 2173 2174 Queries the network for the incentive ratios, 2175 which are the proportions of rewards or incentives 2176 allocated in different subnets of the network. 2177 2178 Returns: 2179 A dictionary mapping network UIDs to their incentive ratios. 2180 2181 Raises: 2182 QueryError: If the query to the network fails or is invalid. 2183 """ 2184 2185 return self.query_map("IncentiveRatio", extract_value=extract_value)[ 2186 "IncentiveRatio" 2187 ]
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.
2189 def query_map_trust_ratio(self, extract_value: bool = False) -> dict[int, int]: 2190 """ 2191 Retrieves a mapping of trust ratios for the network. 2192 2193 Queries the network for trust ratios, 2194 indicative of the level of trust or credibility assigned 2195 to different subnets of the network. 2196 2197 Returns: 2198 A dictionary mapping network UIDs to their trust ratios. 2199 2200 Raises: 2201 QueryError: If the query to the network fails or is invalid. 2202 """ 2203 2204 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.
2206 def query_map_vote_mode_subnet(self, extract_value: bool = False) -> dict[int, str]: 2207 """ 2208 Retrieves a mapping of vote modes for subnets within the network. 2209 2210 Queries the network for the voting modes used in different 2211 subnets, which define the methodology or approach of voting within those 2212 subnets. 2213 2214 Returns: 2215 A dictionary mapping network UIDs to their vote 2216 modes for subnets. 2217 2218 Raises: 2219 QueryError: If the query to the network fails or is invalid. 2220 """ 2221 2222 return self.query_map("VoteModeSubnet", extract_value=extract_value)[ 2223 "VoteModeSubnet" 2224 ]
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.
2226 def query_map_legit_whitelist( 2227 self, extract_value: bool = False 2228 ) -> dict[Ss58Address, int]: 2229 """ 2230 Retrieves a mapping of whitelisted addresses for the network. 2231 2232 Queries the network for a mapping of whitelisted addresses 2233 and their respective legitimacy status. 2234 2235 Returns: 2236 A dictionary mapping addresses to their legitimacy status. 2237 2238 Raises: 2239 QueryError: If the query to the network fails or is invalid. 2240 """ 2241 2242 return self.query_map( 2243 "LegitWhitelist", module="GovernanceModule", extract_value=extract_value)[ 2244 "LegitWhitelist" 2245 ]
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.
2247 def query_map_subnet_names(self, extract_value: bool = False) -> dict[int, str]: 2248 """ 2249 Retrieves a mapping of subnet names within the network. 2250 2251 Queries the network for the names of various subnets, 2252 providing an overview of the different 2253 subnets within the network. 2254 2255 Returns: 2256 A dictionary mapping network UIDs to their subnet names. 2257 2258 Raises: 2259 QueryError: If the query to the network fails or is invalid. 2260 """ 2261 2262 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.
2264 def query_map_balances( 2265 self, extract_value: bool = False, block_hash: str | None = None 2266 ) -> dict[str, dict[str, int | dict[str, int | float]]]: 2267 """ 2268 Retrieves a mapping of account balances within the network. 2269 2270 Queries the network for the balances associated with different accounts. 2271 It provides detailed information including various types of 2272 balances for each account. 2273 2274 Returns: 2275 A dictionary mapping account addresses to their balance details. 2276 2277 Raises: 2278 QueryError: If the query to the network fails or is invalid. 2279 """ 2280 2281 return self.query_map("Account", module="System", extract_value=extract_value, block_hash=block_hash)[ 2282 "Account" 2283 ]
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.
2285 def query_map_registration_blocks( 2286 self, netuid: int = 0, extract_value: bool = False 2287 ) -> dict[int, int]: 2288 """ 2289 Retrieves a mapping of registration blocks for UIDs on the network. 2290 2291 Queries the network to find the block numbers at which various 2292 UIDs were registered. 2293 2294 Args: 2295 netuid: The network UID from which to get the registrations. 2296 2297 Returns: 2298 A dictionary mapping UIDs to their registration block numbers. 2299 2300 Raises: 2301 QueryError: If the query to the network fails or is invalid. 2302 """ 2303 2304 return self.query_map( 2305 "RegistrationBlock", [netuid], extract_value=extract_value 2306 )["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.
2308 def query_map_name( 2309 self, netuid: int = 0, extract_value: bool = False 2310 ) -> dict[int, str]: 2311 """ 2312 Retrieves a mapping of names for keys on the network. 2313 2314 Queries the network for the names associated with different keys. 2315 It provides a mapping of key UIDs to their registered names. 2316 2317 Args: 2318 netuid: The network UID from which to get the names. 2319 2320 Returns: 2321 A dictionary mapping key UIDs to their names. 2322 2323 Raises: 2324 QueryError: If the query to the network fails or is invalid. 2325 """ 2326 2327 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.
2331 def get_immunity_period(self, netuid: int = 0) -> int: 2332 """ 2333 Queries the network for the immunity period setting. 2334 2335 The immunity period is a time duration during which a module 2336 can not be deregistered from the network. 2337 Fetches the immunity period for a specified network subnet. 2338 2339 Args: 2340 netuid: The network UID for which to query the immunity period. 2341 2342 Returns: 2343 The immunity period setting for the specified network subnet. 2344 2345 Raises: 2346 QueryError: If the query to the network fails or is invalid. 2347 """ 2348 2349 return self.query( 2350 "ImmunityPeriod", 2351 params=[netuid], 2352 )
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.
2357 def get_min_allowed_weights(self, netuid: int = 0) -> int: 2358 """ 2359 Queries the network for the minimum allowed weights setting. 2360 2361 Retrieves the minimum weight values that are possible to set 2362 by a validator within a specific network subnet. 2363 2364 Args: 2365 netuid: The network UID for which to query the minimum allowed 2366 weights. 2367 2368 Returns: 2369 The minimum allowed weight values for the specified network 2370 subnet. 2371 2372 Raises: 2373 QueryError: If the query to the network fails or is invalid. 2374 """ 2375 2376 return self.query( 2377 "MinAllowedWeights", 2378 params=[netuid], 2379 )
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.
2384 def get_max_allowed_weights(self, netuid: int = 0) -> int: 2385 """ 2386 Queries the network for the maximum allowed weights setting. 2387 2388 Retrieves the maximum weight values that are possible to set 2389 by a validator within a specific network subnet. 2390 2391 Args: 2392 netuid: The network UID for which to query the maximum allowed 2393 weights. 2394 2395 Returns: 2396 The maximum allowed weight values for the specified network 2397 subnet. 2398 2399 Raises: 2400 QueryError: If the query to the network fails or is invalid. 2401 """ 2402 2403 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.
2405 def get_max_allowed_uids(self, netuid: int = 0) -> int: 2406 """ 2407 Queries the network for the maximum allowed UIDs setting. 2408 2409 Fetches the upper limit on the number of user IDs that can 2410 be allocated or used within a specific network subnet. 2411 2412 Args: 2413 netuid: The network UID for which to query the maximum allowed UIDs. 2414 2415 Returns: 2416 The maximum number of allowed UIDs for the specified network subnet. 2417 2418 Raises: 2419 QueryError: If the query to the network fails or is invalid. 2420 """ 2421 2422 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.
2424 def get_name(self, netuid: int = 0) -> str: 2425 """ 2426 Queries the network for the name of a specific subnet. 2427 2428 Args: 2429 netuid: The network UID for which to query the name. 2430 2431 Returns: 2432 The name of the specified network subnet. 2433 2434 Raises: 2435 QueryError: If the query to the network fails or is invalid. 2436 """ 2437 2438 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.
2440 def get_subnet_name(self, netuid: int = 0) -> str: 2441 """ 2442 Queries the network for the name of a specific subnet. 2443 2444 Args: 2445 netuid: The network UID for which to query the name. 2446 2447 Returns: 2448 The name of the specified network subnet. 2449 2450 Raises: 2451 QueryError: If the query to the network fails or is invalid. 2452 """ 2453 2454 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.
2459 def get_n(self, netuid: int = 0) -> int: 2460 """ 2461 Queries the network for the 'N' hyperparameter, which represents how 2462 many modules are on the network. 2463 2464 Args: 2465 netuid: The network UID for which to query the 'N' hyperparameter. 2466 2467 Returns: 2468 The value of the 'N' hyperparameter for the specified network 2469 subnet. 2470 2471 Raises: 2472 QueryError: If the query to the network fails or is invalid. 2473 """ 2474 2475 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.
2477 def get_tempo(self, netuid: int = 0) -> int: 2478 """ 2479 Queries the network for the tempo setting, measured in blocks, for the 2480 specified subnet. 2481 2482 Args: 2483 netuid: The network UID for which to query the tempo. 2484 2485 Returns: 2486 The tempo setting for the specified subnet. 2487 2488 Raises: 2489 QueryError: If the query to the network fails or is invalid. 2490 """ 2491 2492 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.
2494 def get_total_free_issuance(self, block_hash: str | None = None) -> int: 2495 """ 2496 Queries the network for the total free issuance. 2497 2498 Fetches the total amount of free issuance tokens available 2499 2500 Returns: 2501 The total free issuance amount. 2502 2503 Raises: 2504 QueryError: If the query to the network fails or is invalid. 2505 """ 2506 2507 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.
2509 def get_total_stake(self, block_hash: str | None = None) -> int: 2510 """ 2511 Retrieves a mapping of total stakes for keys on the network. 2512 2513 Queries the network for a mapping of key UIDs to their total stake amounts. 2514 2515 Returns: 2516 A dictionary mapping key UIDs to their total stake amounts. 2517 2518 Raises: 2519 QueryError: If the query to the network fails or is invalid. 2520 """ 2521 2522 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.
2524 def get_registrations_per_block(self): 2525 """ 2526 Queries the network for the number of registrations per block. 2527 2528 Fetches the number of registrations that are processed per 2529 block within the network. 2530 2531 Returns: 2532 The number of registrations processed per block. 2533 2534 Raises: 2535 QueryError: If the query to the network fails or is invalid. 2536 """ 2537 2538 return self.query( 2539 "RegistrationsPerBlock", 2540 )
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.
2542 def max_registrations_per_block(self, netuid: int = 0): 2543 """ 2544 Queries the network for the maximum number of registrations per block. 2545 2546 Retrieves the upper limit of registrations that can be processed in 2547 each block within a specific network subnet. 2548 2549 Args: 2550 netuid: The network UID for which to query. 2551 2552 Returns: 2553 The maximum number of registrations per block for 2554 the specified network subnet. 2555 2556 Raises: 2557 QueryError: If the query to the network fails or is invalid. 2558 """ 2559 2560 return self.query( 2561 "MaxRegistrationsPerBlock", 2562 params=[netuid], 2563 )
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.
2565 def get_proposal(self, proposal_id: int = 0): 2566 """ 2567 Queries the network for a specific proposal. 2568 2569 Args: 2570 proposal_id: The ID of the proposal to query. 2571 2572 Returns: 2573 The details of the specified proposal. 2574 2575 Raises: 2576 QueryError: If the query to the network fails, is invalid, 2577 or if the proposal ID does not exist. 2578 """ 2579 2580 return self.query( 2581 "Proposals", 2582 params=[proposal_id], 2583 )
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.
2585 def get_trust(self, netuid: int = 0): 2586 """ 2587 Queries the network for the trust setting of a specific network subnet. 2588 2589 Retrieves the trust level or score, which may represent the 2590 level of trustworthiness or reliability within a 2591 particular network subnet. 2592 2593 Args: 2594 netuid: The network UID for which to query the trust setting. 2595 2596 Returns: 2597 The trust level or score for the specified network subnet. 2598 2599 Raises: 2600 QueryError: If the query to the network fails or is invalid. 2601 """ 2602 2603 return self.query( 2604 "Trust", 2605 params=[netuid], 2606 )
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.
2608 def get_uids(self, key: Ss58Address, netuid: int = 0) -> bool | None: 2609 """ 2610 Queries the network for module UIDs associated with a specific key. 2611 2612 Args: 2613 key: The key address for which to query UIDs. 2614 netuid: The network UID within which to search for the key. 2615 2616 Returns: 2617 A list of UIDs associated with the specified key. 2618 2619 Raises: 2620 QueryError: If the query to the network fails or is invalid. 2621 """ 2622 2623 return self.query( 2624 "Uids", 2625 params=[netuid, key], 2626 )
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.
2628 def get_unit_emission(self) -> int: 2629 """ 2630 Queries the network for the unit emission setting. 2631 2632 Retrieves the unit emission value, which represents the 2633 emission rate or quantity for the $COMM token. 2634 2635 Returns: 2636 The unit emission value in nanos for the network. 2637 2638 Raises: 2639 QueryError: If the query to the network fails or is invalid. 2640 """ 2641 2642 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.
2644 def get_tx_rate_limit(self) -> int: 2645 """ 2646 Queries the network for the transaction rate limit. 2647 2648 Retrieves the rate limit for transactions within the network, 2649 which defines the maximum number of transactions that can be 2650 processed within a certain timeframe. 2651 2652 Returns: 2653 The transaction rate limit for the network. 2654 2655 Raises: 2656 QueryError: If the query to the network fails or is invalid. 2657 """ 2658 2659 return self.query( 2660 "TxRateLimit", 2661 )
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.
2663 def get_subnet_burn(self) -> int: 2664 """Queries the network for the subnet burn value. 2665 2666 Retrieves the subnet burn value from the network, which represents 2667 the amount of tokens that are burned (permanently removed from 2668 circulation) for subnet-related operations. 2669 2670 Returns: 2671 int: The subnet burn value. 2672 2673 Raises: 2674 QueryError: If the query to the network fails or returns invalid data. 2675 """ 2676 2677 return self.query( 2678 "SubnetBurn", 2679 )
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.
2681 def get_burn_rate(self) -> int: 2682 """ 2683 Queries the network for the burn rate setting. 2684 2685 Retrieves the burn rate, which represents the rate at 2686 which the $COMM token is permanently 2687 removed or 'burned' from circulation. 2688 2689 Returns: 2690 The burn rate for the network. 2691 2692 Raises: 2693 QueryError: If the query to the network fails or is invalid. 2694 """ 2695 2696 return self.query( 2697 "BurnRate", 2698 params=[], 2699 )
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.
2701 def get_burn(self, netuid: int = 0) -> int: 2702 """ 2703 Queries the network for the burn setting. 2704 2705 Retrieves the burn value, which represents the amount of the 2706 $COMM token that is 'burned' or permanently removed from 2707 circulation. 2708 2709 Args: 2710 netuid: The network UID for which to query the burn value. 2711 2712 Returns: 2713 The burn value for the specified network subnet. 2714 2715 Raises: 2716 QueryError: If the query to the network fails or is invalid. 2717 """ 2718 2719 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.
2721 def get_min_burn(self) -> int: 2722 """ 2723 Queries the network for the minimum burn setting. 2724 2725 Retrieves the minimum burn value, indicating the lowest 2726 amount of the $COMM tokens that can be 'burned' or 2727 permanently removed from circulation. 2728 2729 Returns: 2730 The minimum burn value for the network. 2731 2732 Raises: 2733 QueryError: If the query to the network fails or is invalid. 2734 """ 2735 2736 return self.query( 2737 "BurnConfig", 2738 params=[], 2739 )["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.
2741 def get_min_weight_stake(self) -> int: 2742 """ 2743 Queries the network for the minimum weight stake setting. 2744 2745 Retrieves the minimum weight stake, which represents the lowest 2746 stake weight that is allowed for certain operations or 2747 transactions within the network. 2748 2749 Returns: 2750 The minimum weight stake for the network. 2751 2752 Raises: 2753 QueryError: If the query to the network fails or is invalid. 2754 """ 2755 2756 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.
2758 def get_vote_mode_global(self) -> str: 2759 """ 2760 Queries the network for the global vote mode setting. 2761 2762 Retrieves the global vote mode, which defines the overall voting 2763 methodology or approach used across the network in default. 2764 2765 Returns: 2766 The global vote mode setting for the network. 2767 2768 Raises: 2769 QueryError: If the query to the network fails or is invalid. 2770 """ 2771 2772 return self.query( 2773 "VoteModeGlobal", 2774 )
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.
2776 def get_max_proposals(self) -> int: 2777 """ 2778 Queries the network for the maximum number of proposals allowed. 2779 2780 Retrieves the upper limit on the number of proposals that can be 2781 active or considered at any given time within the network. 2782 2783 Returns: 2784 The maximum number of proposals allowed on the network. 2785 2786 Raises: 2787 QueryError: If the query to the network fails or is invalid. 2788 """ 2789 2790 return self.query( 2791 "MaxProposals", 2792 )
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.
2794 def get_max_registrations_per_block(self) -> int: 2795 """ 2796 Queries the network for the maximum number of registrations per block. 2797 2798 Retrieves the maximum number of registrations that can 2799 be processed in each block within the network. 2800 2801 Returns: 2802 The maximum number of registrations per block on the network. 2803 2804 Raises: 2805 QueryError: If the query to the network fails or is invalid. 2806 """ 2807 2808 return self.query( 2809 "MaxRegistrationsPerBlock", 2810 params=[], 2811 )
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.
2813 def get_max_name_length(self) -> int: 2814 """ 2815 Queries the network for the maximum length allowed for names. 2816 2817 Retrieves the maximum character length permitted for names 2818 within the network. Such as the module names 2819 2820 Returns: 2821 The maximum length allowed for names on the network. 2822 2823 Raises: 2824 QueryError: If the query to the network fails or is invalid. 2825 """ 2826 2827 return self.query( 2828 "MaxNameLength", 2829 params=[], 2830 )
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.
2832 def get_global_vote_threshold(self) -> int: 2833 """ 2834 Queries the network for the global vote threshold. 2835 2836 Retrieves the global vote threshold, which is the critical value or 2837 percentage required for decisions in the network's governance process. 2838 2839 Returns: 2840 The global vote threshold for the network. 2841 2842 Raises: 2843 QueryError: If the query to the network fails or is invalid. 2844 """ 2845 2846 return self.query( 2847 "GlobalVoteThreshold", 2848 )
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.
2850 def get_max_allowed_subnets(self) -> int: 2851 """ 2852 Queries the network for the maximum number of allowed subnets. 2853 2854 Retrieves the upper limit on the number of subnets that can 2855 be created or operated within the network. 2856 2857 Returns: 2858 The maximum number of allowed subnets on the network. 2859 2860 Raises: 2861 QueryError: If the query to the network fails or is invalid. 2862 """ 2863 2864 return self.query( 2865 "MaxAllowedSubnets", 2866 params=[], 2867 )
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.
2869 def get_max_allowed_modules(self) -> int: 2870 """ 2871 Queries the network for the maximum number of allowed modules. 2872 2873 Retrieves the upper limit on the number of modules that 2874 can be registered within the network. 2875 2876 Returns: 2877 The maximum number of allowed modules on the network. 2878 2879 Raises: 2880 QueryError: If the query to the network fails or is invalid. 2881 """ 2882 2883 return self.query( 2884 "MaxAllowedModules", 2885 params=[], 2886 )
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.
2888 def get_min_stake(self, netuid: int = 0) -> int: 2889 """ 2890 Queries the network for the minimum stake required to register a key. 2891 2892 Retrieves the minimum amount of stake necessary for 2893 registering a key within a specific network subnet. 2894 2895 Args: 2896 netuid: The network UID for which to query the minimum stake. 2897 2898 Returns: 2899 The minimum stake required for key registration in nanos. 2900 2901 Raises: 2902 QueryError: If the query to the network fails or is invalid. 2903 """ 2904 2905 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.
2907 def get_stakefrom( 2908 self, 2909 key: Ss58Address, 2910 ) -> dict[str, int]: 2911 """ 2912 Retrieves the stake amounts from all stakers to a specific staked address. 2913 2914 Queries the network for the stakes received by a particular staked address 2915 from all stakers. 2916 2917 Args: 2918 key: The address of the key receiving the stakes. 2919 2920 Returns: 2921 A dictionary mapping staker addresses to their respective stake amounts. 2922 2923 Raises: 2924 QueryError: If the query to the network fails or is invalid. 2925 """ 2926 2927 # Has to use query map in order to iterate through the storage prefix. 2928 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.
2930 def get_staketo( 2931 self, 2932 key: Ss58Address, 2933 ) -> dict[str, int]: 2934 """ 2935 Retrieves the stake amounts provided by a specific staker to all staked addresses. 2936 2937 Queries the network for the stakes provided by a particular staker to 2938 all staked addresses. 2939 2940 Args: 2941 key: The address of the key providing the stakes. 2942 2943 Returns: 2944 A dictionary mapping staked addresses to their respective received stake amounts. 2945 2946 Raises: 2947 QueryError: If the query to the network fails or is invalid. 2948 """ 2949 2950 # Has to use query map in order to iterate through the storage prefix. 2951 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.
2953 def get_balance( 2954 self, 2955 addr: Ss58Address, 2956 ) -> int: 2957 """ 2958 Retrieves the balance of a specific key. 2959 2960 Args: 2961 addr: The address of the key to query the balance for. 2962 2963 Returns: 2964 The balance of the specified key. 2965 2966 Raises: 2967 QueryError: If the query to the network fails or is invalid. 2968 """ 2969 2970 result = self.query("Account", module="System", params=[addr]) 2971 2972 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.
2974 def get_block(self, block_hash: str | None = None) -> dict[Any, Any] | None: 2975 """ 2976 Retrieves information about a specific block in the network. 2977 2978 Queries the network for details about a block, such as its number, 2979 hash, and other relevant information. 2980 2981 Returns: 2982 The requested information about the block, 2983 or None if the block does not exist 2984 or the information is not available. 2985 2986 Raises: 2987 QueryError: If the query to the network fails or is invalid. 2988 """ 2989 2990 with self.get_conn() as substrate: 2991 block: dict[Any, Any] | None = substrate.get_block( # type: ignore 2992 block_hash # type: ignore 2993 ) 2994 2995 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.
2997 def get_existential_deposit(self, block_hash: str | None = None) -> int: 2998 """ 2999 Retrieves the existential deposit value for the network. 3000 3001 The existential deposit is the minimum balance that must be maintained 3002 in an account to prevent it from being purged. Denotated in nano units. 3003 3004 Returns: 3005 The existential deposit value in nano units. 3006 Note: 3007 The value returned is a fixed value defined in the 3008 client and may not reflect changes in the network's configuration. 3009 """ 3010 3011 with self.get_conn() as substrate: 3012 result: int = substrate.get_constant( # type: ignore 3013 "Balances", "ExistentialDeposit", block_hash 3014 ).value # type: ignore 3015 3016 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.
3022 def add_transfer_dao_treasury_proposal( 3023 self, 3024 key: Keypair, 3025 data: str, 3026 amount_nano: int, 3027 dest: Ss58Address, 3028 ): 3029 params = {"dest": dest, "value": amount_nano, "data": data} 3030 3031 return self.compose_call( 3032 module="GovernanceModule", 3033 fn="add_transfer_dao_treasury_proposal", 3034 params=params, 3035 key=key, 3036 )