Set weights and check hotkey registered using CLI process

bittensor/check_if_hotkey_is_registered.py

@@ -0,0 +1,26 @@
+import sys
+import asyncio
+from bittensor.core.async_subtensor import AsyncSubtensor
+
+async def main():
+    if len(sys.argv) != 4:
+        print("Usage: python check_if_hotkey_is_registered.py <hotkey> <netuid> <subtensor_network>")
+        sys.exit(1)
+
+    hotkey = sys.argv[1]
+    netuid = int(sys.argv[2])
+    subtensor_network = sys.argv[3]
+
+    subtensor = AsyncSubtensor(network=subtensor_network)
+
+    is_registered = await subtensor.is_hotkey_registered(hotkey_ss58=hotkey, netuid=netuid)
+
+    if is_registered:
+        print(f"Hotkey {hotkey} is registered on subnet {netuid}")
+        sys.exit(0)
+    else:
+        print(f"Hotkey {hotkey} is NOT registered on subnet {netuid}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    asyncio.run(main())

bittensor/set_weights.py

@@ -0,0 +1,44 @@
+import sys
+import asyncio
+from bittensor.core.async_subtensor import AsyncSubtensor
+from bittensor_wallet.wallet import Wallet
+
+async def main():
+    if len(sys.argv) != 7:
+        print("Usage: python set_weights.py <hotkey> <netuid> <subtensor_network> <subtensor_address> <wallet_name> <hotkey_name>")
+        sys.exit(1)
+
+    target_hotkey = sys.argv[1]
+    netuid = int(sys.argv[2])
+    subtensor_network = sys.argv[3]
+    subtensor_address = sys.argv[4]
+    wallet_name = sys.argv[5]
+    hotkey_name = sys.argv[6]
+
+    wallet = Wallet(name=wallet_name, hotkey=hotkey_name)
+    subtensor = AsyncSubtensor(network=subtensor_network, fallback_endpoints=[subtensor_address])
+
+    uid = await subtensor.get_uid_for_hotkey_on_subnet(hotkey_ss58=target_hotkey, netuid=netuid)
+    if uid is None:
+        print(f"Hotkey {target_hotkey} not found on subnet {netuid}")
+        sys.exit(1)
+
+    print(f"Setting weight for hotkey {target_hotkey} (UID {uid}) to 1...")
+
+    success, message = await subtensor.set_weights(
+        wallet=wallet,
+        netuid=netuid,
+        uids=[uid],
+        weights=[1],
+        wait_for_inclusion=True,
+        wait_for_finalization=True
+    )
+
+    if success:
+        print(f"Successfully set weight for hotkey {target_hotkey} to 1")
+    else:
+        print(f"Failed to set weight: {message}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    asyncio.run(main())

utils/bittensor.py

@@ -1,20 +1,61 @@
 # NOTE ADAM: Subtensor bug (self.disable_third_party_loggers())
-from bittensor.core.async_subtensor import AsyncSubtensor
+import asyncio
 from bittensor_wallet.keypair import Keypair
 
 import api.config as config
 import utils.logger as logger
 
 
 
-subtensor = AsyncSubtensor(network=config.SUBTENSOR_NETWORK)
+async def check_if_hotkey_is_registered(hotkey: str) -> bool:
+    process = await asyncio.create_subprocess_exec(
+        "uv", "run", "bittensor/check_if_hotkey_is_registered.py",
+        hotkey,
+        str(config.NETUID),
+        config.SUBTENSOR_NETWORK,
+        stdout=asyncio.subprocess.DEVNULL,
+        stderr=asyncio.subprocess.DEVNULL
+    )
+    await process.wait()
+    return process.returncode == 0
 
+async def set_weights_from_mapping(weights_mapping: dict, netuid: int, subtensor_network: str, subtensor_address: str, wallet_name: str, hotkey_name: str, timeout_seconds: int) -> None:
+    if len(weights_mapping.keys()) != 1:
+        logger.error("Expected one hotkey in weights mapping")
+        return
 
+    hotkey = list(weights_mapping.keys())[0]
 
-async def check_if_hotkey_is_registered(hotkey: str) -> bool:
-    return await subtensor.is_hotkey_registered(hotkey_ss58=hotkey, netuid=config.NETUID)
+    process = await asyncio.create_subprocess_exec(
+        "uv", "run", "bittensor/set_weights.py",
+        hotkey,
+        str(netuid),
+        subtensor_network,
+        subtensor_address,
+        wallet_name,
+        hotkey_name,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE
+    )
 
+    try:
+        stdout, stderr = await asyncio.wait_for(
+            process.communicate(),
+            timeout=timeout_seconds
+        )
 
+        if stdout:
+            logger.info(stdout.decode().strip())
+        if stderr:
+            logger.error(stderr.decode().strip())
+    except asyncio.TimeoutError:
+        logger.error(f"Timeout setting weights after {timeout_seconds} seconds")
+        try:
+            process.kill()
+            await process.wait()
+        except:
+            pass
+        raise
 
 def validate_signed_timestamp(timestamp: int, signed_timestamp: str, hotkey: str) -> bool:
     try:

validator/main.py

@@ -1,6 +1,4 @@
 # NOTE ADAM: Subtensor bug (self.disable_third_party_loggers())
-from validator.set_weights import set_weights_from_mapping
-
 import os
 import sys
 import time
@@ -10,6 +8,7 @@
 import pathlib
 import traceback
 import utils.logger as logger
+import utils.bittensor
 import validator.config as config
 
 from typing import Any, Dict
@@ -78,9 +77,17 @@ async def set_weights_loop():
         weights_mapping = await get_ridges_platform("/scoring/weights", quiet=1)
 
         try:
-            await asyncio.wait_for(set_weights_from_mapping(weights_mapping), timeout=config.SET_WEIGHTS_TIMEOUT_SECONDS)
-        except asyncio.TimeoutError as e:
-            logger.error(f"asyncio.TimeoutError in set_weights_from_mapping(): {e}")
+            await utils.bittensor.set_weights_from_mapping(
+                weights_mapping=weights_mapping,
+                netuid=config.NETUID,
+                subtensor_network=config.SUBTENSOR_NETWORK,
+                subtensor_address=config.SUBTENSOR_ADDRESS,
+                wallet_name=config.VALIDATOR_WALLET_NAME,
+                hotkey_name=config.VALIDATOR_HOTKEY_NAME,
+                timeout_seconds=config.SET_WEIGHTS_TIMEOUT_SECONDS
+            )
+        except Exception as e:
+            logger.error(f"Error setting weights: {e}")
 
         await asyncio.sleep(config.SET_WEIGHTS_INTERVAL_SECONDS)
 

validator/set_weights.py

@@ -1,4 +1,5 @@
 # NOTE ADAM: Subtensor bug (self.disable_third_party_loggers())
+# NOTE: this file is Unused since subtensor.get_uid_for_hotkey_on_subnet keeps timing out after a few minutes
 from bittensor.core.async_subtensor import AsyncSubtensor
 
 import utils.logger as logger