Fixes broken Subtensor methods

bittensor/core/async_subtensor.py

@@ -46,6 +46,7 @@
     format_error_message,
     decode_hex_identity_dict,
     validate_chain_endpoint,
+    hex_to_bytes,
 )
 from bittensor.utils.async_substrate_interface import (
     AsyncSubstrateInterface,
@@ -351,12 +352,7 @@ async def get_delegates(
             reuse_block=reuse_block,
         )
         if hex_bytes_result is not None:
-            try:
-                bytes_result = bytes.fromhex(hex_bytes_result[2:])
-            except ValueError:
-                bytes_result = bytes.fromhex(hex_bytes_result)
-
-            return DelegateInfo.list_from_vec_u8(bytes_result)
+            return DelegateInfo.list_from_vec_u8(hex_to_bytes(hex_bytes_result))
         else:
             return []
 
@@ -392,12 +388,7 @@ async def get_stake_info_for_coldkey(
         if hex_bytes_result is None:
             return []
 
-        try:
-            bytes_result = bytes.fromhex(hex_bytes_result[2:])
-        except ValueError:
-            bytes_result = bytes.fromhex(hex_bytes_result)
-
-        return StakeInfo.list_from_vec_u8(bytes_result)
+        return StakeInfo.list_from_vec_u8(hex_to_bytes(hex_bytes_result))
 
     async def get_stake_for_coldkey_and_hotkey(
         self, hotkey_ss58: str, coldkey_ss58: str, block_hash: Optional[str] = None
@@ -853,12 +844,7 @@ async def neurons_lite(
         if hex_bytes_result is None:
             return []
 
-        try:
-            bytes_result = bytes.fromhex(hex_bytes_result[2:])
-        except ValueError:
-            bytes_result = bytes.fromhex(hex_bytes_result)
-
-        return NeuronInfoLite.list_from_vec_u8(bytes_result)
+        return NeuronInfoLite.list_from_vec_u8(hex_to_bytes(hex_bytes_result))
 
     async def neuron_for_uid(
         self, uid: Optional[int], netuid: int, block_hash: Optional[str] = None
@@ -1170,12 +1156,7 @@ async def get_subnet_hyperparameters(
         if hex_bytes_result is None:
             return []
 
-        if hex_bytes_result.startswith("0x"):
-            bytes_result = bytes.fromhex(hex_bytes_result[2:])
-        else:
-            bytes_result = bytes.fromhex(hex_bytes_result)
-
-        return SubnetHyperparameters.from_vec_u8(bytes_result)
+        return SubnetHyperparameters.from_vec_u8(hex_to_bytes(hex_bytes_result))
 
     async def get_vote_data(
         self,

bittensor/core/subtensor.py

@@ -71,7 +71,13 @@
     transfer_extrinsic,
 )
 from bittensor.core.metagraph import Metagraph
-from bittensor.utils import networking, torch, ss58_to_vec_u8, u16_normalized_float
+from bittensor.utils import (
+    networking,
+    torch,
+    ss58_to_vec_u8,
+    u16_normalized_float,
+    hex_to_bytes,
+)
 from bittensor.utils.balance import Balance
 from bittensor.utils.btlogging import logging
 from bittensor.utils.registration import legacy_torch_api_compat
@@ -523,13 +529,11 @@ def query_runtime_api(
             return None
 
         return_type = call_definition["type"]
-
         as_scale_bytes = scalecodec.ScaleBytes(json_result["result"])
 
         rpc_runtime_config = RuntimeConfiguration()
         rpc_runtime_config.update_type_registry(load_type_registry_preset("legacy"))
         rpc_runtime_config.update_type_registry(custom_rpc_type_registry)
-
         obj = rpc_runtime_config.create_scale_object(return_type, as_scale_bytes)
         if obj.data.to_hex() == "0x0400":  # RPC returned None result
             return None
@@ -1227,12 +1231,7 @@ def get_subnet_hyperparameters(
         if hex_bytes_result is None:
             return []
 
-        if hex_bytes_result.startswith("0x"):
-            bytes_result = bytes.fromhex(hex_bytes_result[2:])
-        else:
-            bytes_result = bytes.fromhex(hex_bytes_result)
-
-        return SubnetHyperparameters.from_vec_u8(bytes_result)
+        return SubnetHyperparameters.from_vec_u8(hex_to_bytes(hex_bytes_result))
 
     # Community uses this method
     # Returns network ImmunityPeriod hyper parameter.
@@ -1308,10 +1307,13 @@ def get_commitment(self, netuid: int, uid: int, block: Optional[int] = None) ->
         hotkey = metagraph.hotkeys[uid]  # type: ignore
 
         metadata = get_metadata(self, netuid, hotkey, block)
-        commitment = metadata["info"]["fields"][0]  # type: ignore
-        hex_data = commitment[list(commitment.keys())[0]][2:]  # type: ignore
+        try:
+            commitment = metadata["info"]["fields"][0]  # type: ignore
+            hex_data = commitment[list(commitment.keys())[0]][2:]  # type: ignore
+            return bytes.fromhex(hex_data).decode()
 
-        return bytes.fromhex(hex_data).decode()
+        except TypeError:
+            return ""
 
     # Community uses this via `bittensor.utils.weight_utils.process_weights_for_netuid` function.
     def min_allowed_weights(
@@ -1367,7 +1369,7 @@ def get_prometheus_info(
             Optional[bittensor.core.chain_data.prometheus_info.PrometheusInfo]: A PrometheusInfo object containing the prometheus information, or ``None`` if the prometheus information is not found.
         """
         result = self.query_subtensor("Prometheus", block, [netuid, hotkey_ss58])
-        if result is not None and hasattr(result, "value"):
+        if result is not None and getattr(result, "value", None) is not None:
             return PrometheusInfo(
                 ip=networking.int_to_ip(result.value["ip"]),
                 ip_type=result.value["ip_type"],
@@ -1407,17 +1409,13 @@ def get_all_subnets_info(self, block: Optional[int] = None) -> list[SubnetInfo]:
 
         Gaining insights into the subnets' details assists in understanding the network's composition, the roles of different subnets, and their unique features.
         """
-        block_hash = None if block is None else self.substrate.get_block_hash(block)
-
-        json_body = self.substrate.rpc_request(
-            method="subnetInfo_getSubnetsInfo",  # custom rpc method
-            params=[block_hash] if block_hash else [],
+        hex_bytes_result = self.query_runtime_api(
+            "SubnetInfoRuntimeApi", "get_subnets_info", params=[], block=block
         )
-
-        if not (result := json_body.get("result", None)):
+        if not hex_bytes_result:
             return []
-
-        return SubnetInfo.list_from_vec_u8(result)
+        else:
+            return SubnetInfo.list_from_vec_u8(hex_to_bytes(hex_bytes_result))
 
     # Metagraph uses this method
     def bonds(
@@ -1561,12 +1559,7 @@ def neurons_lite(
         if hex_bytes_result is None:
             return []
 
-        if hex_bytes_result.startswith("0x"):
-            bytes_result = bytes.fromhex(hex_bytes_result[2:])
-        else:
-            bytes_result = bytes.fromhex(hex_bytes_result)
-
-        return NeuronInfoLite.list_from_vec_u8(bytes_result)  # type: ignore
+        return NeuronInfoLite.list_from_vec_u8(hex_to_bytes(hex_bytes_result))  # type: ignore
 
     # Used in the `neurons` method which is used in metagraph.py
     def weights(
@@ -1923,7 +1916,7 @@ def get_delegate_by_hotkey(
         if not (result := json_body.get("result", None)):
             return None
 
-        return DelegateInfo.from_vec_u8(result)
+        return DelegateInfo.from_vec_u8(bytes(result))
 
     # Subnet 27 uses this method name
     _do_serve_axon = do_serve_axon

bittensor/utils/__init__.py

@@ -403,3 +403,14 @@ def unlock_key(wallet: "Wallet", unlock_type="coldkey") -> "UnlockStatus":
     except KeyFileError:
         err_msg = f"{unlock_type.capitalize()} keyfile is corrupt, non-writable, or non-readable, or non-existent."
         return UnlockStatus(False, err_msg)
+
+
+def hex_to_bytes(hex_str: str) -> bytes:
+    """
+    Converts a hex-encoded string into bytes. Handles 0x-prefixed and non-prefixed hex-encoded strings.
+    """
+    if hex_str.startswith("0x"):
+        bytes_result = bytes.fromhex(hex_str[2:])
+    else:
+        bytes_result = bytes.fromhex(hex_str)
+    return bytes_result

tests/unit_tests/test_async_subtensor.py

@@ -351,7 +351,7 @@ async def test_get_delegates(subtensor, mocker, fake_hex_bytes_result, response)
 
 
 @pytest.mark.parametrize(
-    "fake_hex_bytes_result, response", [(None, []), ("zz001122", b"\xaa\xbb\xcc\xdd")]
+    "fake_hex_bytes_result, response", [(None, []), ("0x001122", b"\xaa\xbb\xcc\xdd")]
 )
 @pytest.mark.asyncio
 async def test_get_stake_info_for_coldkey(

tests/unit_tests/test_subtensor.py

@@ -2022,27 +2022,26 @@ def test_get_all_subnets_info_success(mocker, subtensor):
     """Test get_all_subnets_info returns correct data when subnet information is found."""
     # Prep
     block = 123
-    subnet_data = [1, 2, 3]  # Mocked response data
     mocker.patch.object(
         subtensor.substrate, "get_block_hash", return_value="mock_block_hash"
     )
-    mock_response = {"result": subnet_data}
-    mocker.patch.object(subtensor.substrate, "rpc_request", return_value=mock_response)
+    hex_bytes_result = "0x010203"
+    bytes_result = bytes.fromhex(hex_bytes_result[2:])
+    mocker.patch.object(subtensor, "query_runtime_api", return_value=hex_bytes_result)
     mocker.patch.object(
         subtensor_module.SubnetInfo,
         "list_from_vec_u8",
         return_value="list_from_vec_u80",
     )
 
     # Call
-    result = subtensor.get_all_subnets_info(block)
+    subtensor.get_all_subnets_info(block)
 
     # Asserts
-    subtensor.substrate.get_block_hash.assert_called_once_with(block)
-    subtensor.substrate.rpc_request.assert_called_once_with(
-        method="subnetInfo_getSubnetsInfo", params=["mock_block_hash"]
+    subtensor.query_runtime_api.assert_called_once_with(
+        "SubnetInfoRuntimeApi", "get_subnets_info", params=[], block=block
     )
-    subtensor_module.SubnetInfo.list_from_vec_u8.assert_called_once_with(subnet_data)
+    subtensor_module.SubnetInfo.list_from_vec_u8.assert_called_once_with(bytes_result)
 
 
 @pytest.mark.parametrize("result_", [[], None])
@@ -2053,18 +2052,17 @@ def test_get_all_subnets_info_no_data(mocker, subtensor, result_):
     mocker.patch.object(
         subtensor.substrate, "get_block_hash", return_value="mock_block_hash"
     )
-    mock_response = {"result": result_}
-    mocker.patch.object(subtensor.substrate, "rpc_request", return_value=mock_response)
     mocker.patch.object(subtensor_module.SubnetInfo, "list_from_vec_u8")
 
+    mocker.patch.object(subtensor, "query_runtime_api", return_value=result_)
+
     # Call
     result = subtensor.get_all_subnets_info(block)
 
     # Asserts
     assert result == []
-    subtensor.substrate.get_block_hash.assert_called_once_with(block)
-    subtensor.substrate.rpc_request.assert_called_once_with(
-        method="subnetInfo_getSubnetsInfo", params=["mock_block_hash"]
+    subtensor.query_runtime_api.assert_called_once_with(
+        "SubnetInfoRuntimeApi", "get_subnets_info", params=[], block=block
     )
     subtensor_module.SubnetInfo.list_from_vec_u8.assert_not_called()
 

tests/unit_tests/utils/test_utils.py

@@ -223,3 +223,15 @@ def test_unlock_key_errors(mocker, side_effect, response):
     result = utils.unlock_key(wallet=mock_wallet)
 
     assert result == response
+
+
+@pytest.mark.parametrize(
+    "hex_str, response",
+    [
+        ("5461796c6f72205377696674", b"Taylor Swift"),
+        ("0x5461796c6f72205377696674", b"Taylor Swift"),
+    ],
+)
+def test_hex_to_bytes(hex_str, response):
+    result = utils.hex_to_bytes(hex_str)
+    assert result == response