diff --git a/bittensor/core/async_subtensor.py b/bittensor/core/async_subtensor.py index 1dee258313..422fb0a6e4 100644 --- a/bittensor/core/async_subtensor.py +++ b/bittensor/core/async_subtensor.py @@ -580,6 +580,23 @@ async def all_subnets( subnets = DynamicInfo.list_from_dicts(query.decode()) return subnets + async def blocks_since_last_step( + self, netuid: int, block: Optional[int] = None + ) -> Optional[int]: + """Returns number of blocks since the last epoch of the subnet. + + Arguments: + netuid (int): The unique identifier of the subnetwork. + block: the block number for this query. + + Returns: + block number of the last step in the subnet. + """ + query = await self.query_subtensor( + name="BlocksSinceLastStep", block=block, params=[netuid] + ) + return query.value if query is not None and hasattr(query, "value") else query + async 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. @@ -3149,6 +3166,46 @@ async def get_timestamp( ).value return datetime.fromtimestamp(unix / 1000, tz=timezone.utc) + async def get_subnet_owner_hotkey( + self, netuid: int, block: Optional[int] = None + ) -> Optional[str]: + """ + Retrieves the hotkey of the subnet owner for a given network UID. + + This function queries the subtensor network to fetch the hotkey of the owner of a subnet specified by its + netuid. If no data is found or the query fails, the function returns None. + + Arguments: + netuid: The network UID of the subnet to fetch the owner's hotkey for. + block: The specific block number to query the data from. + + Returns: + The hotkey of the subnet owner if available; None otherwise. + """ + return await self.query_subtensor( + name="SubnetOwnerHotkey", params=[netuid], block=block + ) + + async def get_subnet_validator_permits( + self, netuid: int, block: Optional[int] = None + ) -> Optional[list[bool]]: + """ + Retrieves the list of validator permits for a given subnet as boolean values. + + Arguments: + netuid: The unique identifier of the subnetwork. + block: The blockchain block number for the query. + + Returns: + A list of boolean values representing validator permits, or None if not available. + """ + query = await self.query_subtensor( + name="ValidatorPermit", + params=[netuid], + block=block, + ) + return query.value if query is not None and hasattr(query, "value") else query + # Extrinsics helper ================================================================================================ async def sign_and_send_extrinsic( @@ -3172,6 +3229,9 @@ async def sign_and_send_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's keypair to use for the signing. Options are "coldkey", "hotkey", "coldkeypub" + use_nonce: unique identifier for the transaction related with hot/coldkey. + period: the period of the transaction as ERA part for transaction. Means how many blocks the transaction will be valid for. + nonce_key: the type on nonce to use. Options are "hotkey" or "coldkey". raise_error: raises relevant exception rather than returning `False` if unsuccessful. Returns: diff --git a/bittensor/core/subtensor.py b/bittensor/core/subtensor.py index 29fc857e2e..1268c481b6 100644 --- a/bittensor/core/subtensor.py +++ b/bittensor/core/subtensor.py @@ -388,6 +388,23 @@ def all_subnets(self, block: Optional[int] = None) -> Optional[list["DynamicInfo ) return DynamicInfo.list_from_dicts(query.decode()) + def blocks_since_last_step( + self, netuid: int, block: Optional[int] = None + ) -> Optional[int]: + """Returns number of blocks since the last epoch of the subnet. + + Arguments: + netuid (int): The unique identifier of the subnetwork. + block: the block number for this query. + + Returns: + block number of the last step in the subnet. + """ + query = self.query_subtensor( + name="BlocksSinceLastStep", block=block, params=[netuid] + ) + return query.value if query is not None and hasattr(query, "value") else query + 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. @@ -2575,6 +2592,46 @@ def get_timestamp(self, block: Optional[int] = None) -> datetime: unix = cast(ScaleObj, self.query_module("Timestamp", "Now", block=block)).value return datetime.fromtimestamp(unix / 1000, tz=timezone.utc) + def get_subnet_owner_hotkey( + self, netuid: int, block: Optional[int] = None + ) -> Optional[str]: + """ + Retrieves the hotkey of the subnet owner for a given network UID. + + This function queries the subtensor network to fetch the hotkey of the owner of a subnet specified by its + netuid. If no data is found or the query fails, the function returns None. + + Arguments: + netuid: The network UID of the subnet to fetch the owner's hotkey for. + block: The specific block number to query the data from. + + Returns: + The hotkey of the subnet owner if available; None otherwise. + """ + return self.query_subtensor( + name="SubnetOwnerHotkey", params=[netuid], block=block + ) + + def get_subnet_validator_permits( + self, netuid: int, block: Optional[int] = None + ) -> Optional[list[bool]]: + """ + Retrieves the list of validator permits for a given subnet as boolean values. + + Arguments: + netuid: The unique identifier of the subnetwork. + block: The blockchain block number for the query. + + Returns: + A list of boolean values representing validator permits, or None if not available. + """ + query = self.query_subtensor( + name="ValidatorPermit", + params=[netuid], + block=block, + ) + return query.value if query is not None and hasattr(query, "value") else query + # Extrinsics helper ================================================================================================ def sign_and_send_extrinsic( @@ -2598,6 +2655,9 @@ def sign_and_send_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's keypair to use for the signing. Options are "coldkey", "hotkey", "coldkeypub" + use_nonce: unique identifier for the transaction related with hot/coldkey. + period: the period of the transaction as ERA part for transaction. Means how many blocks the transaction will be valid for. + nonce_key: the type on nonce to use. Options are "hotkey" or "coldkey". raise_error: raises relevant exception rather than returning `False` if unsuccessful. Returns: diff --git a/tests/unit_tests/test_async_subtensor.py b/tests/unit_tests/test_async_subtensor.py index 0c4e4587e8..bb6bb5b49f 100644 --- a/tests/unit_tests/test_async_subtensor.py +++ b/tests/unit_tests/test_async_subtensor.py @@ -3168,3 +3168,138 @@ async def test_get_metagraph_info_subnet_not_exist(subtensor, mocker): assert result is None mocked_logger.assert_called_once_with(f"Subnet {netuid} does not exist.") + + +@pytest.mark.asyncio +async def test_blocks_since_last_step_with_value(subtensor, mocker): + """Test blocks_since_last_step returns correct value.""" + # preps + netuid = 1 + block = 123 + mocked_query_subtensor = mocker.AsyncMock() + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = await subtensor.blocks_since_last_step(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_awaited_once_with( + name="BlocksSinceLastStep", + block=block, + params=[netuid], + ) + + assert result == mocked_query_subtensor.return_value.value + + +@pytest.mark.asyncio +async def test_blocks_since_last_step_is_none(subtensor, mocker): + """Test blocks_since_last_step returns None correctly.""" + # preps + netuid = 1 + block = 123 + mocked_query_subtensor = mocker.AsyncMock(return_value=None) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = await subtensor.blocks_since_last_step(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_awaited_once_with( + name="BlocksSinceLastStep", + block=block, + params=[netuid], + ) + + assert result is None + + +@pytest.mark.asyncio +async def test_get_subnet_owner_hotkey_has_return(subtensor, mocker): + """Test get_subnet_owner_hotkey returns correct value.""" + # preps + netuid = 14 + block = 123 + expected_owner_hotkey = "owner_hotkey" + mocked_query_subtensor = mocker.AsyncMock(return_value=expected_owner_hotkey) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = await subtensor.get_subnet_owner_hotkey(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_awaited_once_with( + name="SubnetOwnerHotkey", + block=block, + params=[netuid], + ) + + assert result == expected_owner_hotkey + + +@pytest.mark.asyncio +async def test_get_subnet_owner_hotkey_is_none(subtensor, mocker): + """Test get_subnet_owner_hotkey returns None correctly.""" + # preps + netuid = 14 + block = 123 + mocked_query_subtensor = mocker.AsyncMock(return_value=None) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = await subtensor.get_subnet_owner_hotkey(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_awaited_once_with( + name="SubnetOwnerHotkey", + block=block, + params=[netuid], + ) + + assert result is None + + +@pytest.mark.asyncio +async def test_get_subnet_validator_permits_has_values(subtensor, mocker): + """Test get_subnet_validator_permits returns correct value.""" + # preps + netuid = 14 + block = 123 + expected_validator_permits = [False, True, False] + mocked_query_subtensor = mocker.AsyncMock(return_value=expected_validator_permits) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = await subtensor.get_subnet_validator_permits(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_awaited_once_with( + name="ValidatorPermit", + block=block, + params=[netuid], + ) + + assert result == expected_validator_permits + + +@pytest.mark.asyncio +async def test_get_subnet_validator_permits_is_none(subtensor, mocker): + """Test get_subnet_validator_permits returns correct value.""" + # preps + netuid = 14 + block = 123 + + mocked_query_subtensor = mocker.AsyncMock(return_value=None) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = await subtensor.get_subnet_validator_permits(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_awaited_once_with( + name="ValidatorPermit", + block=block, + params=[netuid], + ) + + assert result is None diff --git a/tests/unit_tests/test_subtensor.py b/tests/unit_tests/test_subtensor.py index 015884a831..ca6aad000d 100644 --- a/tests/unit_tests/test_subtensor.py +++ b/tests/unit_tests/test_subtensor.py @@ -3494,3 +3494,132 @@ def test_get_metagraph_info_subnet_not_exist(subtensor, mocker): assert result is None mocked_logger.assert_called_once_with(f"Subnet {netuid} does not exist.") + + +def test_blocks_since_last_step_with_value(subtensor, mocker): + """Test blocks_since_last_step returns correct value.""" + # preps + netuid = 1 + block = 123 + mocked_query_subtensor = mocker.MagicMock() + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = subtensor.blocks_since_last_step(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_called_once_with( + name="BlocksSinceLastStep", + block=block, + params=[netuid], + ) + + assert result == mocked_query_subtensor.return_value.value + + +def test_blocks_since_last_step_is_none(subtensor, mocker): + """Test blocks_since_last_step returns None correctly.""" + # preps + netuid = 1 + block = 123 + mocked_query_subtensor = mocker.MagicMock(return_value=None) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = subtensor.blocks_since_last_step(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_called_once_with( + name="BlocksSinceLastStep", + block=block, + params=[netuid], + ) + + assert result is None + + +def test_get_subnet_owner_hotkey_has_return(subtensor, mocker): + """Test get_subnet_owner_hotkey returns correct value.""" + # preps + netuid = 14 + block = 123 + expected_owner_hotkey = "owner_hotkey" + mocked_query_subtensor = mocker.MagicMock(return_value=expected_owner_hotkey) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = subtensor.get_subnet_owner_hotkey(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_called_once_with( + name="SubnetOwnerHotkey", + block=block, + params=[netuid], + ) + + assert result == expected_owner_hotkey + + +def test_get_subnet_owner_hotkey_is_none(subtensor, mocker): + """Test get_subnet_owner_hotkey returns None correctly.""" + # preps + netuid = 14 + block = 123 + mocked_query_subtensor = mocker.MagicMock(return_value=None) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = subtensor.get_subnet_owner_hotkey(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_called_once_with( + name="SubnetOwnerHotkey", + block=block, + params=[netuid], + ) + + assert result is None + + +def test_get_subnet_validator_permits_has_values(subtensor, mocker): + """Test get_subnet_validator_permits returns correct value.""" + # preps + netuid = 14 + block = 123 + expected_validator_permits = [False, True, False] + mocked_query_subtensor = mocker.MagicMock(return_value=expected_validator_permits) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = subtensor.get_subnet_validator_permits(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_called_once_with( + name="ValidatorPermit", + block=block, + params=[netuid], + ) + + assert result == expected_validator_permits + + +def test_get_subnet_validator_permits_is_none(subtensor, mocker): + """Test get_subnet_validator_permits returns correct value.""" + # preps + netuid = 14 + block = 123 + + mocked_query_subtensor = mocker.MagicMock(return_value=None) + subtensor.query_subtensor = mocked_query_subtensor + + # call + result = subtensor.get_subnet_validator_permits(netuid=netuid, block=block) + + # asserts + mocked_query_subtensor.assert_called_once_with( + name="ValidatorPermit", + block=block, + params=[netuid], + ) + + assert result is None