Skip to content

Commit

Permalink
Porting.
Browse files Browse the repository at this point in the history
  • Loading branch information
thewhaleking committed Dec 17, 2024
1 parent 519bd0e commit 728754b
Show file tree
Hide file tree
Showing 3 changed files with 141 additions and 36 deletions.
81 changes: 58 additions & 23 deletions bittensor/core/async_subtensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
decode_account_id,
SubnetInfo,
PrometheusInfo,
ProposalVoteData,
)
from bittensor.core.extrinsics.asyncio.registration import register_extrinsic
from bittensor.core.extrinsics.asyncio.root import (
Expand Down Expand Up @@ -76,26 +77,6 @@ class ParamWithTypes(TypedDict):
type: str # ScaleType string of the parameter.


class ProposalVoteData:
index: int
threshold: int
ayes: list[str]
nays: list[str]
end: int

def __init__(self, proposal_dict: dict) -> None:
self.index = proposal_dict["index"]
self.threshold = proposal_dict["threshold"]
self.ayes = self.decode_ss58_tuples(proposal_dict["ayes"])
self.nays = self.decode_ss58_tuples(proposal_dict["nays"])
self.end = proposal_dict["end"]

@staticmethod
def decode_ss58_tuples(line: tuple):
"""Decodes a tuple of ss58 addresses formatted as bytes tuples."""
return [decode_account_id(line[x][0]) for x in range(len(line))]


def _decode_hex_identity_dict(info_dictionary: dict[str, Any]) -> dict[str, Any]:
"""Decodes a dictionary of hexadecimal identities."""
for k, v in info_dictionary.items():
Expand Down Expand Up @@ -1405,6 +1386,7 @@ async def sign_and_send_extrinsic(
wallet: "Wallet",
wait_for_inclusion: bool = True,
wait_for_finalization: bool = False,
sign_with: str = "coldkey",
) -> tuple[bool, str]:
"""
Helper method to sign and submit an extrinsic call to chain.
Expand All @@ -1414,13 +1396,19 @@ async def sign_and_send_extrinsic(
wallet (bittensor_wallet.Wallet): the wallet whose coldkey will be used to sign the extrinsic
wait_for_inclusion (bool): whether to wait until the extrinsic call is included on the chain
wait_for_finalization (bool): whether to wait until the extrinsic call is finalized on the chain
sign_with: the wallet attr to sign the extrinsic call [coldkey (default), hotkey, or coldkeypub]
Returns:
(success, error message)
"""
if sign_with not in ("coldkey", "hotkey", "coldkeypub"):
raise AttributeError(
f"'sign_with' must be either 'coldkey', 'hotkey' or 'coldkeypub', not '{sign_with}'"
)

extrinsic = await self.substrate.create_signed_extrinsic(
call=call, keypair=wallet.coldkey
) # sign with coldkey
call=call, keypair=getattr(wallet, sign_with)
)
try:
response = await self.substrate.submit_extrinsic(
extrinsic,
Expand All @@ -1437,24 +1425,38 @@ async def sign_and_send_extrinsic(
except SubstrateRequestException as e:
return False, format_error_message(e)

async def get_children(self, hotkey: str, netuid: int) -> tuple[bool, list, str]:
async def get_children(
self,
hotkey: str,
netuid: int,
block: Optional[int] = None,
block_hash: Optional[str] = None,
reuse_block: bool = False,
) -> tuple[bool, list, 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.
Args:
hotkey: The hotkey value.
netuid: The netuid value.
block: The block number to query. Do not specify if using block_hash or reuse_block.
block_hash: The hash of the blockchain block number for the query. Do not specify if using bloc or
reuse_block.
reuse_block: Whether to reuse the last-used blockchain hash. Do not set if using block_hash or reuse_block.
Returns:
A tuple containing a boolean indicating success or failure, a list of formatted children, and an error
message (if applicable)
"""
block_hash = await self._determine_block_hash(block, block_hash, reuse_block)
try:
children = await self.substrate.query(
module="SubtensorModule",
storage_function="ChildKeys",
params=[hotkey, netuid],
block_hash=block_hash,
reuse_block_hash=reuse_block,
)
if children:
formatted_children = []
Expand Down Expand Up @@ -2702,3 +2704,36 @@ async def unstake(
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
)

async def state_call(
self,
method: str,
data: str,
block: Optional[int] = None,
block_hash: Optional[str] = None,
reuse_block: bool = False,
) -> 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.
block_hash: The hash of the block to retrieve the parameter from. Do not specify if using block or
reuse_block
reuse_block: Whether to use the last-used block. Do not set if using block_hash or block.
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 = await self._determine_block_hash(block, block_hash, reuse_block)
return self.substrate.rpc_request(
method="state_call",
params=[method, data, block_hash] if block_hash else [method, data],
reuse_block_hash=reuse_block,
)
27 changes: 14 additions & 13 deletions bittensor/core/chain_data/proposal_vote_data.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,22 @@
from typing import TypedDict
from bittensor.core.chain_data.utils import decode_account_id


# Senate / Proposal data
class ProposalVoteData(TypedDict):
"""
This TypedDict represents the data structure for a proposal vote in the Senate.
Attributes:
index (int): The index of the proposal.
threshold (int): The threshold required for the proposal to pass.
ayes (List[str]): List of senators who voted 'aye'.
nays (List[str]): List of senators who voted 'nay'.
end (int): The ending timestamp of the voting period.
"""

class ProposalVoteData:
index: int
threshold: int
ayes: list[str]
nays: list[str]
end: int

def __init__(self, proposal_dict: dict) -> None:
self.index = proposal_dict["index"]
self.threshold = proposal_dict["threshold"]
self.ayes = self.decode_ss58_tuples(proposal_dict["ayes"])
self.nays = self.decode_ss58_tuples(proposal_dict["nays"])
self.end = proposal_dict["end"]

@staticmethod
def decode_ss58_tuples(line: tuple):
"""Decodes a tuple of ss58 addresses formatted as bytes tuples."""
return [decode_account_id(line[x][0]) for x in range(len(line))]
69 changes: 69 additions & 0 deletions bittensor/core/subtensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@
SubnetHyperparameters,
SubnetInfo,
StakeInfo,
ProposalVoteData,
decode_account_id,
)
from bittensor.core.config import Config
from bittensor.core.errors import SubstrateRequestException
Expand Down Expand Up @@ -2361,3 +2363,70 @@ def get_delegated(
return []

return DelegateInfo.delegated_list_from_vec_u8(bytes(result))

def get_vote_data(
self,
proposal_hash: str,
block: Optional[int] = None,
) -> Optional[ProposalVoteData]:
"""
Retrieves the voting data for a specific proposal on the Bittensor blockchain. This data includes information
about how senate members have voted on the proposal.
Args:
proposal_hash: The hash of the proposal for which voting data is requested.
block: The block number to query. Do not specify if using block_hash or reuse_block.
Returns:
An object containing the proposal's voting data, or `None` if not found.
This function is important for tracking and understanding the decision-making processes within the Bittensor
network, particularly how proposals are received and acted upon by the governing body.
"""
vote_data = self.substrate.query(
module="Triumvirate",
storage_function="Voting",
params=[proposal_hash],
block_hash=None if block is None else self.get_block_hash(block),
)
if vote_data is None:
return None
else:
return ProposalVoteData(vote_data)

def get_children(
self, hotkey: str, netuid: int, block: Optional[int] = None
) -> tuple[bool, list, 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.
Args:
hotkey: The hotkey value.
netuid: The netuid value.
block: the blockchain block number for the query
Returns:
A tuple containing a boolean indicating success or failure, a list of formatted children, and an error
message (if applicable)
"""
block_hash = None if block is None else self.get_block_hash(block)
try:
children = self.substrate.query(
module="SubtensorModule",
storage_function="ChildKeys",
params=[hotkey, netuid],
block_hash=block_hash,
)
if children:
formatted_children = []
for proportion, child in children:
# Convert U64 to int
formatted_child = decode_account_id(child[0])
int_proportion = int(proportion)
formatted_children.append((int_proportion, formatted_child))
return True, formatted_children, ""
else:
return True, [], ""
except SubstrateRequestException as e:
return False, [], format_error_message(e)

0 comments on commit 728754b

Please sign in to comment.