Add subtensor.is_subnet_active method

bittensor/core/async_subtensor.py

@@ -2582,6 +2582,35 @@ async def is_hotkey_registered_on_subnet(
             is not None
         )
 
+    async def is_subnet_active(
+        self,
+        netuid: int,
+        block: Optional[int] = None,
+        block_hash: Optional[str] = None,
+        reuse_block: bool = False,
+    ) -> bool:
+        """Verify if subnet with provided netuid is active.
+
+        Args:
+            netuid (int): The unique identifier of the subnet.
+            block (Optional[int]): The blockchain block number for the query.
+            block_hash (Optional[str]): The blockchain block_hash representation of block id.
+            reuse_block (bool): Whether to reuse the last-used block hash.
+
+        Returns:
+            True if subnet is active, False otherwise.
+
+        This means whether the `start_call` was initiated or not.
+        """
+        query = await self.query_subtensor(
+            name="FirstEmissionBlockNumber",
+            block=block,
+            block_hash=block_hash,
+            reuse_block=reuse_block,
+            params=[netuid],
+        )
+        return True if query and query.value > 0 else False
+
     async def last_drand_round(self) -> Optional[int]:
         """
         Retrieves the last drand round emitted in bittensor. This corresponds when committed weights will be revealed.

bittensor/core/subtensor.py

@@ -2023,6 +2023,25 @@ def is_hotkey_registered_on_subnet(
             is not None
         )
 
+    def is_subnet_active(self, netuid: int, block: Optional[int] = None) -> bool:
+        """Verify if subnet with provided netuid is active.
+
+        Args:
+            netuid (int): The unique identifier of the subnet.
+            block (Optional[int]): The blockchain block number for the query.
+
+        Returns:
+            True if subnet is active, False otherwise.
+
+        This means whether the `start_call` was initiated or not.
+        """
+        query = self.query_subtensor(
+            name="FirstEmissionBlockNumber",
+            block=block,
+            params=[netuid],
+        )
+        return True if query and query.value > 0 else False
+
     def last_drand_round(self) -> Optional[int]:
         """
         Retrieves the last drand round emitted in bittensor. This corresponds when committed weights will be revealed.

bittensor/core/subtensor_api/subnets.py

@@ -32,6 +32,7 @@ def __init__(self, subtensor: Union["_Subtensor", "_AsyncSubtensor"]):
         self.get_uid_for_hotkey_on_subnet = subtensor.get_uid_for_hotkey_on_subnet
         self.immunity_period = subtensor.immunity_period
         self.is_hotkey_registered_on_subnet = subtensor.is_hotkey_registered_on_subnet
+        self.is_subnet_active = subtensor.is_subnet_active
         self.max_weight_limit = subtensor.max_weight_limit
         self.min_allowed_weights = subtensor.min_allowed_weights
         self.recycle = subtensor.recycle

bittensor/core/subtensor_api/utils.py

@@ -108,6 +108,7 @@ def add_legacy_methods(subtensor: "SubtensorApi"):
     subtensor.is_hotkey_registered_on_subnet = (
         subtensor._subtensor.is_hotkey_registered_on_subnet
     )
+    subtensor.is_subnet_active = subtensor._subtensor.is_subnet_active
     subtensor.last_drand_round = subtensor._subtensor.last_drand_round
     subtensor.log_verbose = subtensor._subtensor.log_verbose
     subtensor.max_weight_limit = subtensor._subtensor.max_weight_limit

tests/e2e_tests/utils/e2e_test_utils.py

@@ -229,7 +229,7 @@ def validator(self, wallet, netuid):
 
 
 def wait_to_start_call(
-    subtensor: "bittensor.Subtensor",
+    subtensor: "bittensor.SubtensorApi",
     subnet_owner_wallet: "bittensor.Wallet",
     netuid: int,
     in_blocks: int = 10,
@@ -242,6 +242,11 @@ def wait_to_start_call(
         f"Current block: [blue]{subtensor.block}[/blue]."
     )
 
+    # make sure subnet isn't active
+    assert subtensor.subnets.is_subnet_active(netuid) is False, (
+        "Subnet is already active."
+    )
+
     # make sure we passed start_call limit
     subtensor.wait_for_block(subtensor.block + in_blocks + 1)
     status, message = subtensor.start_call(
@@ -251,6 +256,11 @@ def wait_to_start_call(
         wait_for_finalization=True,
     )
     assert status, message
+    # make sure subnet is active
+    assert subtensor.subnets.is_subnet_active(netuid), (
+        "Subnet did not activated after start call."
+    )
+
     return True
 
 

tests/unit_tests/test_subtensor.py

@@ -3633,3 +3633,32 @@ def test_get_subnet_validator_permits_is_none(subtensor, mocker):
     )
 
     assert result is None
+
+
+@pytest.mark.parametrize(
+    "query_return, expected",
+    [
+        [111, True],
+        [0, False],
+    ],
+)
+def test_is_subnet_active(subtensor, mocker, query_return, expected):
+    # preps
+    netuid = mocker.Mock()
+    block = mocker.Mock()
+    mocked_query_subtensor = mocker.MagicMock(
+        return_value=mocker.Mock(value=query_return)
+    )
+    subtensor.query_subtensor = mocked_query_subtensor
+
+    # call
+    result = subtensor.is_subnet_active(netuid=netuid, block=block)
+
+    # Asserts
+    mocked_query_subtensor.assert_called_once_with(
+        name="FirstEmissionBlockNumber",
+        block=block,
+        params=[netuid],
+    )
+
+    assert result == expected