-
Notifications
You must be signed in to change notification settings - Fork 361
/
Copy pathsubtensor.py
2961 lines (2542 loc) · 117 KB
/
subtensor.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import copy
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Iterable, Optional, Union, cast
import numpy as np
import requests
import scalecodec
from async_substrate_interface.errors import SubstrateRequestException
from async_substrate_interface.types import ScaleObj
from async_substrate_interface.sync_substrate import SubstrateInterface
from async_substrate_interface.utils import json
from numpy.typing import NDArray
from bittensor.core.async_subtensor import ProposalVoteData
from bittensor.core.axon import Axon
from bittensor.core.chain_data import (
DelegateInfo,
DynamicInfo,
MetagraphInfo,
NeuronInfo,
NeuronInfoLite,
StakeInfo,
SubnetHyperparameters,
WeightCommitInfo,
SubnetIdentity,
SubnetInfo,
decode_account_id,
)
from bittensor.core.chain_data.utils import decode_metadata
from bittensor.core.config import Config
from bittensor.core.extrinsics.commit_reveal import commit_reveal_v3_extrinsic
from bittensor.core.extrinsics.commit_weights import (
commit_weights_extrinsic,
reveal_weights_extrinsic,
)
from bittensor.core.extrinsics.move_stake import (
transfer_stake_extrinsic,
swap_stake_extrinsic,
move_stake_extrinsic,
)
from bittensor.core.extrinsics.registration import (
burned_register_extrinsic,
register_extrinsic,
register_subnet_extrinsic,
set_subnet_identity_extrinsic,
)
from bittensor.core.extrinsics.root import (
root_register_extrinsic,
set_root_weights_extrinsic,
)
from bittensor.core.extrinsics.serving import (
publish_metadata,
get_metadata,
serve_axon_extrinsic,
)
from bittensor.core.extrinsics.set_weights import set_weights_extrinsic
from bittensor.core.extrinsics.staking import (
add_stake_extrinsic,
add_stake_multiple_extrinsic,
)
from bittensor.core.extrinsics.transfer import transfer_extrinsic
from bittensor.core.extrinsics.unstaking import (
unstake_extrinsic,
unstake_multiple_extrinsic,
)
from bittensor.core.metagraph import Metagraph
from bittensor.core.settings import (
version_as_int,
SS58_FORMAT,
TYPE_REGISTRY,
DELEGATES_DETAILS_URL,
)
from bittensor.core.types import ParamWithTypes, SubtensorMixin
from bittensor.utils import (
torch,
format_error_message,
decode_hex_identity_dict,
u16_normalized_float,
_decode_hex_identity_dict,
Certificate,
u64_normalized_float,
)
from bittensor.utils.balance import (
Balance,
fixed_to_float,
FixedPoint,
check_and_convert_to_balance,
)
from bittensor.utils.btlogging import logging
from bittensor.utils.delegates_details import DelegatesDetails
from bittensor.utils.weight_utils import generate_weight_hash
if TYPE_CHECKING:
from bittensor_wallet import Wallet
from async_substrate_interface.sync_substrate import QueryMapResult
from scalecodec.types import GenericCall
class Subtensor(SubtensorMixin):
"""Thin layer for interacting with Substrate Interface. Mostly a collection of frequently-used calls."""
def __init__(
self,
network: Optional[str] = None,
config: Optional["Config"] = None,
_mock: bool = False,
log_verbose: bool = False,
):
"""
Initializes an instance of the Subtensor class.
Arguments:
network (str): The network name or type to connect to.
config (Optional[Config]): Configuration object for the AsyncSubtensor instance.
_mock: Whether this is a mock instance. Mainly just for use in testing.
log_verbose (bool): Enables or disables verbose logging.
Raises:
Any exceptions raised during the setup, configuration, or connection process.
"""
if config is None:
config = self.config()
self._config = copy.deepcopy(config)
self.chain_endpoint, self.network = self.setup_config(network, self._config)
self._mock = _mock
self.log_verbose = log_verbose
self._check_and_log_network_settings()
logging.debug(
f"Connecting to network: [blue]{self.network}[/blue], "
f"chain_endpoint: [blue]{self.chain_endpoint}[/blue]> ..."
)
self.substrate = SubstrateInterface(
url=self.chain_endpoint,
ss58_format=SS58_FORMAT,
type_registry=TYPE_REGISTRY,
use_remote_preset=True,
chain_name="Bittensor",
_mock=_mock,
)
if self.log_verbose:
logging.info(
f"Connected to {self.network} network and {self.chain_endpoint}."
)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
def close(self):
"""
Closes the websocket connection
"""
self.substrate.close()
# Subtensor queries ===========================================================================================
def query_constant(
self, module_name: str, constant_name: str, block: Optional[int] = None
) -> Optional["ScaleObj"]:
"""
Retrieves a constant from the specified module on the Bittensor blockchain. This function is used to access
fixed parameters or values defined within the blockchain's modules, which are essential for understanding
the network's configuration and rules.
Args:
module_name: The name of the module containing the constant.
constant_name: The name of the constant to retrieve.
block: The blockchain block number at which to query the constant.
Returns:
Optional[async_substrate_interface.types.ScaleObj]: The value of the constant if found, `None` otherwise.
Constants queried through this function can include critical network parameters such as inflation rates,
consensus rules, or validation thresholds, providing a deeper understanding of the Bittensor network's
operational parameters.
"""
return self.substrate.get_constant(
module_name=module_name,
constant_name=constant_name,
block_hash=self.determine_block_hash(block),
)
def query_map(
self,
module: str,
name: str,
block: Optional[int] = None,
params: Optional[list] = None,
) -> "QueryMapResult":
"""
Queries map storage from any module on the Bittensor blockchain. This function retrieves data structures that
represent key-value mappings, essential for accessing complex and structured data within the blockchain
modules.
Args:
module: The name of the module from which to query the map storage.
name: The specific storage function within the module to query.
block: The blockchain block number at which to perform the query.
params: Parameters to be passed to the query.
Returns:
result: A data structure representing the map storage if found, `None` otherwise.
This function is particularly useful for retrieving detailed and structured data from various blockchain
modules, offering insights into the network's state and the relationships between its different components.
"""
result = self.substrate.query_map(
module=module,
storage_function=name,
params=params,
block_hash=self.determine_block_hash(block=block),
)
return result
def query_map_subtensor(
self, name: str, block: Optional[int] = None, params: Optional[list] = None
) -> "QueryMapResult":
"""
Queries map storage from the Subtensor module on the Bittensor blockchain. This function is designed to retrieve
a map-like data structure, which can include various neuron-specific details or network-wide attributes.
Args:
name: The name of the map storage function to query.
block: The blockchain block number at which to perform the query.
params: A list of parameters to pass to the query function.
Returns:
An object containing the map-like data structure, or `None` if not found.
This function is particularly useful for analyzing and understanding complex network structures and
relationships within the Bittensor ecosystem, such as interneuronal connections and stake distributions.
"""
return self.substrate.query_map(
module="SubtensorModule",
storage_function=name,
params=params,
block_hash=self.determine_block_hash(block),
)
def query_module(
self,
module: str,
name: str,
block: Optional[int] = None,
params: Optional[list] = None,
) -> Optional[Union["ScaleObj", Any, FixedPoint]]:
"""
Queries any module storage on the Bittensor blockchain with the specified parameters and block number. This
function is a generic query interface that allows for flexible and diverse data retrieval from various
blockchain modules.
Args:
module (str): The name of the module from which to query data.
name (str): The name of the storage function within the module.
block (Optional[int]): The blockchain block number at which to perform the query.
params (Optional[list[object]]): A list of parameters to pass to the query function.
Returns:
An object containing the requested data if found, `None` otherwise.
This versatile query function is key to accessing a wide range of data and insights from different parts of the
Bittensor blockchain, enhancing the understanding and analysis of the network's state and dynamics.
"""
return self.substrate.query(
module=module,
storage_function=name,
params=params,
block_hash=self.determine_block_hash(block),
)
def query_runtime_api(
self,
runtime_api: str,
method: str,
params: Optional[Union[list[Any], dict[str, Any]]] = None,
block: Optional[int] = None,
) -> Any:
"""
Queries the runtime API of the Bittensor blockchain, providing a way to interact with the underlying runtime and
retrieve data encoded in Scale Bytes format. This function is essential for advanced users who need to
interact with specific runtime methods and decode complex data types.
Args:
runtime_api: The name of the runtime API to query.
method: The specific method within the runtime API to call.
params: The parameters to pass to the method call.
block: the block number for this query.
Returns:
The Scale Bytes encoded result from the runtime API call, or `None` if the call fails.
This function enables access to the deeper layers of the Bittensor blockchain, allowing for detailed and
specific interactions with the network's runtime environment.
"""
block_hash = self.determine_block_hash(block)
result = self.substrate.runtime_call(runtime_api, method, params, block_hash)
return result.value
def query_subtensor(
self, name: str, block: Optional[int] = None, params: Optional[list] = None
) -> Optional[Union["ScaleObj", Any]]:
"""
Queries named storage from the Subtensor module on the Bittensor blockchain. This function is used to retrieve
specific data or parameters from the blockchain, such as stake, rank, or other neuron-specific attributes.
Args:
name: The name of the storage function to query.
block: The blockchain block number at which to perform the query.
params: A list of parameters to pass to the query function.
Returns:
query_response: An object containing the requested data.
This query function is essential for accessing detailed information about the network and its neurons, providing
valuable insights into the state and dynamics of the Bittensor ecosystem.
"""
return self.substrate.query(
module="SubtensorModule",
storage_function=name,
params=params,
block_hash=self.determine_block_hash(block),
)
def state_call(
self, method: str, data: str, block: Optional[int] = None
) -> dict[Any, Any]:
"""
Makes a state call to the Bittensor blockchain, allowing for direct queries of the blockchain's state. This
function is typically used for advanced queries that require specific method calls and data inputs.
Args:
method: The method name for the state call.
data: The data to be passed to the method.
block: The blockchain block number at which to perform the state call.
Returns:
result (dict[Any, Any]): The result of the rpc call.
The state call function provides a more direct and flexible way of querying blockchain data, useful for specific
use cases where standard queries are insufficient.
"""
block_hash = self.determine_block_hash(block)
return self.substrate.rpc_request(
method="state_call",
params=[method, data, block_hash] if block_hash else [method, data],
)
# Common subtensor calls ===========================================================================================
@property
def block(self) -> int:
return self.get_current_block()
def all_subnets(self, block: Optional[int] = None) -> Optional[list["DynamicInfo"]]:
"""
Retrieves the subnet information for all subnets in the network.
Args:
block (Optional[int]): The block number to query the subnet information from.
Returns:
Optional[DynamicInfo]: A list of DynamicInfo objects, each containing detailed information about a subnet.
"""
block_hash = self.determine_block_hash(block)
query = self.substrate.runtime_call(
"SubnetInfoRuntimeApi",
"get_all_dynamic_info",
block_hash=block_hash,
)
return DynamicInfo.list_from_dicts(query.decode())
def blocks_since_last_update(self, netuid: int, uid: int) -> Optional[int]:
"""
Returns the number of blocks since the last update for a specific UID in the subnetwork.
Arguments:
netuid (int): The unique identifier of the subnetwork.
uid (int): The unique identifier of the neuron.
Returns:
Optional[int]: The number of blocks since the last update, or ``None`` if the subnetwork or UID does not
exist.
"""
call = self.get_hyperparameter(param_name="LastUpdate", netuid=netuid)
return None if call is None else (self.get_current_block() - int(call[uid]))
def bonds(
self, netuid: int, block: Optional[int] = None
) -> list[tuple[int, list[tuple[int, int]]]]:
"""
Retrieves the bond distribution set by neurons within a specific subnet of the Bittensor network.
Bonds represent the investments or commitments made by neurons in one another, indicating a level of trust
and perceived value. This bonding mechanism is integral to the network's market-based approach to
measuring and rewarding machine intelligence.
Args:
netuid: The network UID of the subnet to query.
block: the block number for this query.
Returns:
List of tuples mapping each neuron's UID to its bonds with other neurons.
Understanding bond distributions is crucial for analyzing the trust dynamics and market behavior within the
subnet. It reflects how neurons recognize and invest in each other's intelligence and contributions,
supporting diverse and niche systems within the Bittensor ecosystem.
"""
b_map_encoded = self.substrate.query_map(
module="SubtensorModule",
storage_function="Bonds",
params=[netuid],
block_hash=self.determine_block_hash(block),
)
b_map = []
for uid, b in b_map_encoded:
if b.value is not None:
b_map.append((uid, b.value))
return b_map
def commit(self, wallet, netuid: int, data: str) -> bool:
"""
Commits arbitrary data to the Bittensor network by publishing metadata.
Arguments:
wallet (bittensor_wallet.Wallet): The wallet associated with the neuron committing the data.
netuid (int): The unique identifier of the subnetwork.
data (str): The data to be committed to the network.
"""
return publish_metadata(
subtensor=self,
wallet=wallet,
netuid=netuid,
data_type=f"Raw{len(data)}",
data=data.encode(),
)
# add explicit alias
set_commitment = commit
def commit_reveal_enabled(
self, netuid: int, block: Optional[int] = None
) -> Optional[bool]:
"""
Check if commit-reveal mechanism is enabled for a given network at a specific block.
Arguments:
netuid: The network identifier for which to check the commit-reveal mechanism.
block: The block number to query.
Returns:
Returns the integer value of the hyperparameter if available; otherwise, returns None.
"""
call = self.get_hyperparameter(
param_name="CommitRevealWeightsEnabled", block=block, netuid=netuid
)
return True if call is True else False
def difficulty(self, netuid: int, block: Optional[int] = None) -> Optional[int]:
"""
Retrieves the 'Difficulty' hyperparameter for a specified subnet in the Bittensor network.
This parameter is instrumental in determining the computational challenge required for neurons to participate in
consensus and validation processes.
Arguments:
netuid: The unique identifier of the subnet.
block: The blockchain block number for the query.
Returns:
Optional[int]: The value of the 'Difficulty' hyperparameter if the subnet exists, ``None`` otherwise.
The 'Difficulty' parameter directly impacts the network's security and integrity by setting the computational
effort required for validating transactions and participating in the network's consensus mechanism.
"""
call = self.get_hyperparameter(
param_name="Difficulty", netuid=netuid, block=block
)
if call is None:
return None
return int(call)
def does_hotkey_exist(self, hotkey_ss58: str, block: Optional[int] = None) -> bool:
"""
Returns true if the hotkey is known by the chain and there are accounts.
Args:
hotkey_ss58: The SS58 address of the hotkey.
block: the block number for this query.
Returns:
`True` if the hotkey is known by the chain and there are accounts, `False` otherwise.
"""
result = self.substrate.query(
module="SubtensorModule",
storage_function="Owner",
params=[hotkey_ss58],
block_hash=self.determine_block_hash(block),
)
return_val = (
False
if result is None
else result != "5C4hrfjw9DjXZTzV3MwzrrAr9P1MJhSrvWGWqi1eSuyUpnhM"
)
return return_val
def get_all_subnets_info(self, block: Optional[int] = None) -> list["SubnetInfo"]:
"""
Retrieves detailed information about all subnets within the Bittensor network. This function provides
comprehensive data on each subnet, including its characteristics and operational parameters.
Arguments:
block: The blockchain block number for the query.
Returns:
list[SubnetInfo]: A list of SubnetInfo objects, each containing detailed information about a subnet.
Gaining insights into the subnets' details assists in understanding the network's composition, the roles of
different subnets, and their unique features.
"""
result = self.query_runtime_api(
runtime_api="SubnetInfoRuntimeApi",
method="get_subnets_info_v2",
params=[],
block=block,
)
if not result:
return []
else:
return SubnetInfo.list_from_dicts(result)
def get_balance(self, address: str, block: Optional[int] = None) -> Balance:
"""
Retrieves the balance for given coldkey.
Arguments:
address (str): coldkey address.
block (Optional[int]): The blockchain block number for the query.
Returns:
Balance object.
"""
balance = self.substrate.query(
module="System",
storage_function="Account",
params=[address],
block_hash=self.determine_block_hash(block),
)
return Balance(balance["data"]["free"])
def get_balances(
self,
*addresses: str,
block: Optional[int] = None,
) -> dict[str, Balance]:
"""
Retrieves the balance for given coldkey(s)
Arguments:
addresses (str): coldkey addresses(s).
block (Optional[int]): The blockchain block number for the query.
Returns:
Dict of {address: Balance objects}.
"""
if not (block_hash := self.determine_block_hash(block)):
block_hash = self.substrate.get_chain_head()
calls = [
(
self.substrate.create_storage_key(
"System", "Account", [address], block_hash=block_hash
)
)
for address in addresses
]
batch_call = self.substrate.query_multi(calls, block_hash=block_hash)
results = {}
for item in batch_call:
value = item[1] or {"data": {"free": 0}}
results.update({item[0].params[0]: Balance(value["data"]["free"])})
return results
def get_current_block(self) -> int:
"""
Returns the current block number on the Bittensor blockchain. This function provides the latest block number,
indicating the most recent state of the blockchain.
Returns:
int: The current chain block number.
Knowing the current block number is essential for querying real-time data and performing time-sensitive
operations on the blockchain. It serves as a reference point for network activities and data
synchronization.
"""
return self.substrate.get_block_number(None)
@lru_cache(maxsize=128)
def _get_block_hash(self, block_id: int):
return self.substrate.get_block_hash(block_id)
def get_block_hash(self, block: Optional[int] = None) -> str:
"""
Retrieves the hash of a specific block on the Bittensor blockchain. The block hash is a unique identifier
representing the cryptographic hash of the block's content, ensuring its integrity and immutability.
Arguments:
block (int): The block number for which the hash is to be retrieved.
Returns:
str: The cryptographic hash of the specified block.
The block hash is a fundamental aspect of blockchain technology, providing a secure reference to each block's
data. It is crucial for verifying transactions, ensuring data consistency, and maintaining the
trustworthiness of the blockchain.
"""
if block:
return self._get_block_hash(block)
else:
return self.substrate.get_chain_head()
def determine_block_hash(self, block: Optional[int]) -> Optional[str]:
if block is None:
return None
else:
return self.get_block_hash(block=block)
def encode_params(
self,
call_definition: dict[str, list["ParamWithTypes"]],
params: Union[list[Any], dict[str, Any]],
) -> str:
"""Returns a hex encoded string of the params using their types."""
param_data = scalecodec.ScaleBytes(b"")
for i, param in enumerate(call_definition["params"]):
scale_obj = self.substrate.create_scale_object(param["type"])
if isinstance(params, list):
param_data += scale_obj.encode(params[i])
else:
if param["name"] not in params:
raise ValueError(f"Missing param {param['name']} in params dict.")
param_data += scale_obj.encode(params[param["name"]])
return param_data.to_hex()
def get_hyperparameter(
self, param_name: str, netuid: int, block: Optional[int] = None
) -> Optional[Any]:
"""
Retrieves a specified hyperparameter for a specific subnet.
Arguments:
param_name (str): The name of the hyperparameter to retrieve.
netuid (int): The unique identifier of the subnet.
block: the block number at which to retrieve the hyperparameter.
Returns:
The value of the specified hyperparameter if the subnet exists, or None
"""
block_hash = self.determine_block_hash(block)
if not self.subnet_exists(netuid, block=block):
logging.error(f"subnet {netuid} does not exist")
return None
result = self.substrate.query(
module="SubtensorModule",
storage_function=param_name,
params=[netuid],
block_hash=block_hash,
)
return getattr(result, "value", result)
def get_children(
self, hotkey: str, netuid: int, block: Optional[int] = None
) -> tuple[bool, list[tuple[float, str]], str]:
"""
This method retrieves the children of a given hotkey and netuid. It queries the SubtensorModule's ChildKeys
storage function to get the children and formats them before returning as a tuple.
Arguments:
hotkey (str): The hotkey value.
netuid (int): The netuid value.
block (Optional[int]): The block number for which the children are to be retrieved.
Returns:
A tuple containing a boolean indicating success or failure, a list of formatted children, and an error
message (if applicable)
"""
try:
children = self.substrate.query(
module="SubtensorModule",
storage_function="ChildKeys",
params=[hotkey, netuid],
block_hash=self.determine_block_hash(block),
)
if children:
formatted_children = []
for proportion, child in children.value:
# Convert U64 to int
formatted_child = decode_account_id(child[0])
normalized_proportion = u64_normalized_float(proportion)
formatted_children.append((normalized_proportion, formatted_child))
return True, formatted_children, ""
else:
return True, [], ""
except SubstrateRequestException as e:
return False, [], format_error_message(e)
def get_commitment(self, netuid: int, uid: int, block: Optional[int] = None) -> str:
"""
Retrieves the on-chain commitment for a specific neuron in the Bittensor network.
Arguments:
netuid (int): The unique identifier of the subnetwork.
uid (int): The unique identifier of the neuron.
block (Optional[int]): The block number to retrieve the commitment from. If None, the latest block is used.
Default is ``None``.
Returns:
str: The commitment data as a string.
"""
metagraph = self.metagraph(netuid)
try:
hotkey = metagraph.hotkeys[uid] # type: ignore
except IndexError:
logging.error(
"Your uid is not in the hotkeys. Please double-check your UID."
)
return ""
metadata = cast(dict, get_metadata(self, netuid, hotkey, block))
try:
return decode_metadata(metadata)
except TypeError:
return ""
def get_all_commitments(
self, netuid: int, block: Optional[int] = None
) -> dict[str, str]:
query = self.query_map(
module="Commitments",
name="CommitmentOf",
params=[netuid],
block=block,
)
result = {}
for id_, value in query:
result[decode_account_id(id_[0])] = decode_account_id(value)
return result
def get_current_weight_commit_info(
self, netuid: int, block: Optional[int] = None
) -> list:
"""
Retrieves CRV3 weight commit information for a specific subnet.
Arguments:
netuid (int): The unique identifier of the subnet.
block (Optional[int]): The blockchain block number for the query. Default is ``None``.
Returns:
list: A list of commit details, where each entry is a dictionary with keys 'who', 'serialized_commit', and
'reveal_round', or an empty list if no data is found.
"""
result = self.substrate.query_map(
module="SubtensorModule",
storage_function="CRV3WeightCommits",
params=[netuid],
block_hash=self.determine_block_hash(block),
)
commits = result.records[0][1] if result.records else []
return [WeightCommitInfo.from_vec_u8(commit) for commit in commits]
def get_delegate_by_hotkey(
self, hotkey_ss58: str, block: Optional[int] = None
) -> Optional["DelegateInfo"]:
"""
Retrieves detailed information about a delegate neuron based on its hotkey. This function provides a
comprehensive view of the delegate's status, including its stakes, nominators, and reward distribution.
Arguments:
hotkey_ss58 (str): The ``SS58`` address of the delegate's hotkey.
block (Optional[int]): The blockchain block number for the query.
Returns:
Optional[DelegateInfo]: Detailed information about the delegate neuron, ``None`` if not found.
This function is essential for understanding the roles and influence of delegate neurons within the Bittensor
network's consensus and governance structures.
"""
result = self.query_runtime_api(
runtime_api="DelegateInfoRuntimeApi",
method="get_delegate",
params=[hotkey_ss58],
block=block,
)
if not result:
return None
return DelegateInfo.from_dict(result)
def get_delegate_identities(
self, block: Optional[int] = None
) -> dict[str, "DelegatesDetails"]:
"""
Fetches delegates identities from the chain and GitHub. Preference is given to chain data, and missing info is
filled-in by the info from GitHub. At some point, we want to totally move away from fetching this info from
GitHub, but chain data is still limited in that regard.
Arguments:
block (Optional[int]): The blockchain block number for the query.
Returns:
Dict {ss58: DelegatesDetails, ...}
"""
block_hash = self.determine_block_hash(block)
response = requests.get(DELEGATES_DETAILS_URL)
identities_info = self.substrate.query_map(
module="Registry", storage_function="IdentityOf", block_hash=block_hash
)
all_delegates_details = {}
for ss58_address, identity in identities_info:
all_delegates_details.update(
{
decode_account_id(
ss58_address[0]
): DelegatesDetails.from_chain_data(
decode_hex_identity_dict(identity.value["info"])
)
}
)
if response.ok:
all_delegates: dict[str, Any] = json.loads(response.content)
for delegate_hotkey, delegate_details in all_delegates.items():
delegate_info = all_delegates_details.setdefault(
delegate_hotkey,
DelegatesDetails(
display=delegate_details.get("name", ""),
web=delegate_details.get("url", ""),
additional=delegate_details.get("description", ""),
pgp_fingerprint=delegate_details.get("fingerprint", ""),
),
)
delegate_info.display = delegate_info.display or delegate_details.get(
"name", ""
)
delegate_info.web = delegate_info.web or delegate_details.get("url", "")
delegate_info.additional = (
delegate_info.additional or delegate_details.get("description", "")
)
delegate_info.pgp_fingerprint = (
delegate_info.pgp_fingerprint
or delegate_details.get("fingerprint", "")
)
return all_delegates_details
def get_delegate_take(
self, hotkey_ss58: str, block: Optional[int] = None
) -> Optional[float]:
"""
Retrieves the delegate 'take' percentage for a neuron identified by its hotkey. The 'take' represents the
percentage of rewards that the delegate claims from its nominators' stakes.
Arguments:
hotkey_ss58 (str): The ``SS58`` address of the neuron's hotkey.
block (Optional[int]): The blockchain block number for the query.
Returns:
Optional[float]: The delegate take percentage, None if not available.
The delegate take is a critical parameter in the network's incentive structure, influencing the distribution of
rewards among neurons and their nominators.
"""
result = self.query_subtensor(
name="Delegates",
block=block,
params=[hotkey_ss58],
)
return (
None
if result is None
else u16_normalized_float(getattr(result, "value", 0))
)
def get_delegated(
self, coldkey_ss58: str, block: Optional[int] = None
) -> list[tuple["DelegateInfo", Balance]]:
"""
Retrieves a list of delegates and their associated stakes for a given coldkey. This function identifies the
delegates that a specific account has staked tokens on.
Arguments:
coldkey_ss58 (str): The `SS58` address of the account's coldkey.
block (Optional[int]): The blockchain block number for the query.
Returns:
A list of tuples, each containing a delegate's information and staked amount.
This function is important for account holders to understand their stake allocations and their involvement in
the network's delegation and consensus mechanisms.
"""
result = self.query_runtime_api(
runtime_api="DelegateInfoRuntimeApi",
method="get_delegated",
params=[coldkey_ss58],
block=block,
)
if not result:
return []
return DelegateInfo.delegated_list_from_dicts(result)
def get_delegates(self, block: Optional[int] = None) -> list["DelegateInfo"]:
"""
Fetches all delegates on the chain
Arguments:
block (Optional[int]): The blockchain block number for the query.
Returns:
List of DelegateInfo objects, or an empty list if there are no delegates.
"""
result = self.query_runtime_api(
runtime_api="DelegateInfoRuntimeApi",
method="get_delegates",
params=[],
block=block,
)
if result:
return DelegateInfo.list_from_dicts(result)
else:
return []
def get_existential_deposit(self, block: Optional[int] = None) -> Optional[Balance]:
"""
Retrieves the existential deposit amount for the Bittensor blockchain.
The existential deposit is the minimum amount of TAO required for an account to exist on the blockchain.
Accounts with balances below this threshold can be reaped to conserve network resources.
Arguments:
block (Optional[int]): The blockchain block number for the query.
Returns:
The existential deposit amount.
The existential deposit is a fundamental economic parameter in the Bittensor network, ensuring efficient use of
storage and preventing the proliferation of dust accounts.
"""
result = self.substrate.get_constant(
module_name="Balances",
constant_name="ExistentialDeposit",
block_hash=self.determine_block_hash(block),
)
if result is None:
raise Exception("Unable to retrieve existential deposit amount.")
return Balance.from_rao(getattr(result, "value", 0))
def get_hotkey_owner(
self, hotkey_ss58: str, block: Optional[int] = None
) -> Optional[str]:
"""
Retrieves the owner of the given hotkey at a specific block hash.
This function queries the blockchain for the owner of the provided hotkey. If the hotkey does not exist at the
specified block hash, it returns None.
Arguments:
hotkey_ss58 (str): The SS58 address of the hotkey.
block (Optional[int]): The blockchain block number for the query.
Returns:
Optional[str]: The SS58 address of the owner if the hotkey exists, or None if it doesn't.
"""
hk_owner_query = self.substrate.query(
module="SubtensorModule",
storage_function="Owner",
params=[hotkey_ss58],
block_hash=self.determine_block_hash(block),
)
exists = False
if hk_owner_query:
exists = self.does_hotkey_exist(hotkey_ss58, block=block)