Merge pull request #2887 from opentensor/feat/roman/selective_metagraph_back

SelectiveMetagraph back

Author: basfroman

bittensor/core/async_subtensor.py

@@ -22,6 +22,7 @@
     NeuronInfoLite,
     NeuronInfo,
     ProposalVoteData,
+    SelectiveMetagraphIndex,
     StakeInfo,
     SubnetHyperparameters,
     SubnetIdentity,
@@ -1639,37 +1640,70 @@ async def get_minimum_required_stake(self):
     async def get_metagraph_info(
         self,
         netuid: int,
+        field_indices: Optional[Union[list[SelectiveMetagraphIndex], list[int]]] = None,
         block: Optional[int] = None,
         block_hash: Optional[str] = None,
         reuse_block: bool = False,
     ) -> Optional[MetagraphInfo]:
         """
-        Retrieves the MetagraphInfo dataclass from the node for a single subnet (netuid)
+        Retrieves full or partial metagraph information for the specified subnet (netuid).
 
         Arguments:
-            netuid: The NetUID of the subnet.
+            netuid: The NetUID of the subnet to query.
+            field_indices: An optional list of SelectiveMetagraphIndex or int values specifying which fields to retrieve.
+                If not provided, all available fields will be returned.
             block: the block number at which to retrieve the hyperparameter. Do not specify if using block_hash or
                 reuse_block
             block_hash: The hash of blockchain block number for the query. Do not specify if using
                 block or reuse_block
             reuse_block: Whether to reuse the last-used block hash. Do not set if using block_hash or block.
 
         Returns:
-            MetagraphInfo dataclass
+            Optional[MetagraphInfo]: A MetagraphInfo object containing the requested subnet data, or None if the subnet
+                with the given netuid does not exist.
+
+        Example:
+            meta_info = await subtensor.get_metagraph_info(netuid=2)
+
+            partial_meta_info = await subtensor.get_metagraph_info(
+                netuid=2,
+                field_indices=[SelectiveMetagraphIndex.Name, SelectiveMetagraphIndex.OwnerHotkeys]
+            )
         """
         block_hash = await self.determine_block_hash(block, block_hash, reuse_block)
         if not block_hash and reuse_block:
             block_hash = self.substrate.last_block_hash
 
-        query = await self.substrate.runtime_call(
-            "SubnetInfoRuntimeApi",
-            "get_metagraph",
-            params=[netuid],
-            block_hash=block_hash,
-        )
+        if field_indices:
+            if isinstance(field_indices, list) and all(
+                isinstance(f, (SelectiveMetagraphIndex, int)) for f in field_indices
+            ):
+                indexes = [
+                    f.value if isinstance(f, SelectiveMetagraphIndex) else f
+                    for f in field_indices
+                ]
+            else:
+                raise ValueError(
+                    "`field_indices` must be a list of SelectiveMetagraphIndex enums or ints."
+                )
+
+            query = await self.substrate.runtime_call(
+                "SubnetInfoRuntimeApi",
+                "get_selective_metagraph",
+                params=[netuid, indexes if 0 in indexes else [0] + indexes],
+                block_hash=block_hash,
+            )
+        else:
+            query = await self.substrate.runtime_call(
+                "SubnetInfoRuntimeApi",
+                "get_metagraph",
+                params=[netuid],
+            )
+
         if query.value is None:
             logging.error(f"Subnet {netuid} does not exist.")
             return None
+
         return MetagraphInfo.from_dict(query.value)
 
     async def get_all_metagraphs_info(

bittensor/core/chain_data/metagraph_info.py

@@ -13,7 +13,7 @@
 from bittensor.utils.balance import Balance, fixed_to_float
 
 
-# to balance with unit (just shortcut)
+# to balance with unit (shortcut)
 def _tbwu(val: Optional[int], netuid: Optional[int] = 0) -> Optional[Balance]:
     """Returns a Balance object from a value and unit."""
     if val is None:
@@ -26,7 +26,9 @@ def _chr_str(codes: tuple[int]) -> str:
     return "".join(map(chr, codes))
 
 
-def process_nested(data: Union[tuple, dict], chr_transform):
+def process_nested(
+    data: Union[tuple, dict], chr_transform
+) -> Optional[Union[list, dict]]:
     """Processes nested data structures by applying a transformation function to their elements."""
     if isinstance(data, (list, tuple)):
         if len(data) > 0:
@@ -39,6 +41,7 @@ def process_nested(data: Union[tuple, dict], chr_transform):
         return {}
     elif isinstance(data, dict):
         return {k: chr_transform(v) for k, v in data.items()}
+    return None
 
 
 @dataclass
@@ -143,6 +146,9 @@ class MetagraphInfo(InfoBase):
         tuple[str, Balance]
     ]  # List of dividend payout in alpha via subnet.
 
+    # List of validators
+    validators: list[str]
+
     @classmethod
     def _from_dict(cls, decoded: dict) -> "MetagraphInfo":
         """Returns a MetagraphInfo object from decoded chain data."""
@@ -366,6 +372,9 @@ def _from_dict(cls, decoded: dict) -> "MetagraphInfo":
                 if decoded.get("alpha_dividends_per_hotkey") is not None
                 else None
             ),
+            validators=[v for v in decoded["validators"]]
+            if decoded.get("validators") is not None
+            else None,
         )
 
 
@@ -498,6 +507,7 @@ class SelectiveMetagraphIndex(Enum):
     TotalStake = 69
     TaoDividendsPerHotkey = 70
     AlphaDividendsPerHotkey = 71
+    Validators = 72
 
     @staticmethod
     def all_indices() -> list[int]:

bittensor/core/subtensor.py

@@ -21,6 +21,7 @@
     MetagraphInfo,
     NeuronInfo,
     NeuronInfoLite,
+    SelectiveMetagraphIndex,
     StakeInfo,
     SubnetInfo,
     SubnetIdentity,
@@ -1268,29 +1269,66 @@ def get_minimum_required_stake(self) -> Balance:
         return Balance.from_rao(getattr(result, "value", 0))
 
     def get_metagraph_info(
-        self, netuid: int, block: Optional[int] = None
+        self,
+        netuid: int,
+        field_indices: Optional[Union[list[SelectiveMetagraphIndex], list[int]]] = None,
+        block: Optional[int] = None,
     ) -> Optional[MetagraphInfo]:
         """
-        Retrieves the MetagraphInfo dataclass from the node for a single subnet (netuid)
+        Retrieves full or partial metagraph information for the specified subnet (netuid).
 
         Arguments:
-            netuid: The NetUID of the subnet.
-            block: the block number at which to retrieve the hyperparameter. Do not specify if using block_hash or
-                reuse_block
+            netuid: The NetUID of the subnet to query.
+            field_indices: An optional list of SelectiveMetagraphIndex or int values specifying which fields to retrieve.
+                If not provided, all available fields will be returned.
+            block: The block number at which to query the data. If not specified, the current block or one determined
+                via reuse_block or block_hash will be used.
 
         Returns:
-            MetagraphInfo dataclass
+            Optional[MetagraphInfo]: A MetagraphInfo object containing the requested subnet data, or None if the subnet
+                with the given netuid does not exist.
+
+        Example:
+            meta_info = subtensor.get_metagraph_info(netuid=2)
+
+            partial_meta_info = subtensor.get_metagraph_info(
+                netuid=2,
+                field_indices=[SelectiveMetagraphIndex.Name, SelectiveMetagraphIndex.OwnerHotkeys]
+            )
         """
         block_hash = self.determine_block_hash(block)
-        query = self.substrate.runtime_call(
-            "SubnetInfoRuntimeApi",
-            "get_metagraph",
-            params=[netuid],
-            block_hash=block_hash,
-        )
+
+        if field_indices:
+            if isinstance(field_indices, list) and all(
+                isinstance(f, (SelectiveMetagraphIndex, int)) for f in field_indices
+            ):
+                indexes = [
+                    f.value if isinstance(f, SelectiveMetagraphIndex) else f
+                    for f in field_indices
+                ]
+            else:
+                raise ValueError(
+                    "`field_indices` must be a list of SelectiveMetagraphIndex enums or ints."
+                )
+
+            query = self.substrate.runtime_call(
+                "SubnetInfoRuntimeApi",
+                "get_selective_metagraph",
+                params=[netuid, indexes if 0 in indexes else [0] + indexes],
+                block_hash=block_hash,
+            )
+        else:
+            query = self.substrate.runtime_call(
+                "SubnetInfoRuntimeApi",
+                "get_metagraph",
+                params=[netuid],
+                block_hash=block_hash,
+            )
+
         if query.value is None:
             logging.error(f"Subnet {netuid} does not exist.")
             return None
+
         return MetagraphInfo.from_dict(query.value)
 
     def get_all_metagraphs_info(

tests/e2e_tests/test_incentive.py

@@ -103,8 +103,25 @@ async def test_incentive(local_chain, subtensor, templates, alice_wallet, bob_wa
                 continue
             raise
 
-    # wait one tempo (fast block
-    subtensor.wait_for_block(subtensor.block + subtensor.tempo(alice_subnet_netuid))
+    # wait one tempo (fast block)
+    next_epoch_start_block = subtensor.subnets.get_next_epoch_start_block(
+        alice_subnet_netuid
+    )
+    subtensor.wait_for_block(next_epoch_start_block + tempo + 1)
+
+    validators = subtensor.metagraphs.get_metagraph_info(
+        alice_subnet_netuid, field_indices=[72]
+    ).validators
+
+    alice_uid = subtensor.subnets.get_uid_for_hotkey_on_subnet(
+        hotkey_ss58=alice_wallet.hotkey.ss58_address, netuid=alice_subnet_netuid
+    )
+    assert validators[alice_uid] == 1
+
+    bob_uid = subtensor.subnets.get_uid_for_hotkey_on_subnet(
+        hotkey_ss58=bob_wallet.hotkey.ss58_address, netuid=alice_subnet_netuid
+    )
+    assert validators[bob_uid] == 0
 
     while True:
         try: