Porting.

Author: thewhaleking

bittensor/core/async_subtensor.py

@@ -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 (
@@ -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():
@@ -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.
@@ -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,
@@ -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 = []
@@ -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,
+        )

bittensor/core/chain_data/proposal_vote_data.py

@@ -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))]

bittensor/core/subtensor.py

@@ -32,6 +32,8 @@
     SubnetHyperparameters,
     SubnetInfo,
     StakeInfo,
+    ProposalVoteData,
+    decode_account_id,
 )
 from bittensor.core.config import Config
 from bittensor.core.errors import SubstrateRequestException
@@ -2376,3 +2378,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)