validator.py

import os
import time
import random
import argparse
import traceback
import hetu as ht

from protocol import Dummy


class Validator:
    def __init__(self):
        self.wallet = ht.Account.create()  # Use ETH wallet
        self.config = self.get_config()
        self.setup_logging()
        self.setup_hetutensor_objects()
        self.my_uid = self.metagraph.hotkeys.index(self.wallet.address)
        self.scores = [1.0] * len(self.metagraph.S)
        self.last_update = self.hetutensor.blocks_since_last_update(
            self.config.netuid, self.my_uid
        )
        self.tempo = self.hetutensor.tempo(self.config.netuid)
        self.moving_avg_scores = [1.0] * len(self.metagraph.S)
        self.alpha = 0.1

    def get_config(self):
        # Set up the configuration parser.
        parser = argparse.ArgumentParser()
        # TODO: Add your custom validator arguments to the parser.
        parser.add_argument(
            "--custom",
            default="my_custom_value",
            help="Adds a custom value to the parser.",
        )
        # Adds override arguments for network and netuid.
        parser.add_argument(
            "--netuid", type=int, default=1, help="The chain subnet uid."
        )
        # Adds hetutensor specific arguments.
        ht.hetutensor.add_args(parser)
        # Adds logging specific arguments.
        ht.logging.add_args(parser)
        # Parse the config.
        config = ht.config(parser)
        # Set up logging directory.
        eth_addr = self.wallet.address if hasattr(self, 'wallet') else 'eth_wallet'
        config.full_path = os.path.expanduser(
            f"{config.logging.logging_dir}/{eth_addr}/netuid{config.netuid}/validator"
        )
        # Ensure the logging directory exists.
        os.makedirs(config.full_path, exist_ok=True)
        return config

    def setup_logging(self):
        # Set up logging.
        ht.logging(config=self.config, logging_dir=self.config.full_path)
        ht.logging.info(
            f"Running validator for subnet: {self.config.netuid} on network: {self.config.hetutensor.network} with config:"
        )
        ht.logging.info(self.config)

    def setup_hetutensor_objects(self):
        # Build hetutensor validator objects.
        ht.logging.info("Setting up hetutensor objects.")

        # ETH wallet log
        ht.logging.info(f"Wallet: {self.wallet}")

        # Initialize hetutensor.
        self.hetutensor = ht.hetutensor(config=self.config)
        ht.logging.info(f"hetutensor: {self.hetutensor}")

        # Initialize dendrite.
        self.dendrite = ht.dendrite(wallet=self.wallet)
        ht.logging.info(f"Dendrite: {self.dendrite}")

        # Initialize metagraph.
        self.metagraph = self.hetutensor.metagraph(self.config.netuid)
        ht.logging.info(f"Metagraph: {self.metagraph}")

        # Connect the validator to the network.
        if hasattr(self.wallet, 'address') and self.wallet.address not in self.metagraph.hotkeys:
            ht.logging.error(
                f"Your validator: {self.wallet} is not registered to chain connection: {self.hetutensor} \nRun 'hetucli register' and try again."
            )
            exit()
        else:
            # Each validator gets a unique identity (UID) in the network.
            if hasattr(self.wallet, 'address'):
                self.my_subnet_uid = self.metagraph.hotkeys.index(self.wallet.address)
                ht.logging.info(f"Running validator on uid: {self.my_subnet_uid}")

        # Set up initial scoring weights for validation.
        ht.logging.info("Building validation weights.")
        self.scores = [1.0] * len(self.metagraph.S)
        ht.logging.info(f"Weights: {self.scores}")

    def run(self):
        # The Main Validation Loop.
        ht.logging.info("Starting validator loop.")
        while True:
            try:
                # time.sleep(int(self.hetutensor.tempo(self.config.netuid) * 0.25))
                # Create a synapse with the current step value.
                synapse = Dummy(dummy_input=random.randint(0, 100))

                # Broadcast a query to all miners on the network.
                responses = self.dendrite.query(
                    axons=self.metagraph.axons, synapse=synapse, timeout=12
                )
                ht.logging.info(f"sending input {synapse.dummy_input}")
                if responses:
                    responses = [
                        response.dummy_output
                        for response in responses
                        if response is not None
                    ]

                # Log the results.
                ht.logging.info(f"Received dummy responses: {responses}")

                # Adjust the length of moving_avg_scores to match the number of responses
                if len(self.moving_avg_scores) < len(responses):
                    self.moving_avg_scores.extend(
                        [1] * (len(responses) - len(self.moving_avg_scores))
                    )

                # Adjust the scores based on responses from miners and update moving average.
                for i, resp_i in enumerate(responses):
                    current_score = 1 if resp_i == synapse.dummy_input * 2 else 0
                    self.moving_avg_scores[i] = (
                        1 - self.alpha
                    ) * self.moving_avg_scores[i] + self.alpha * current_score

                ht.logging.info(f"Moving Average Scores: {self.moving_avg_scores}")
                self.last_update = self.hetutensor.blocks_since_last_update(
                    self.config.netuid, self.my_uid
                )

                # set weights once every tempo
                total = sum(self.moving_avg_scores)
                weights = [score / total for score in self.moving_avg_scores]
                ht.logging.info(f"[blue]Setting weights: {weights}[/blue]")
                # Update the incentive mechanism on the hetutensor blockchain.
                self.hetutensor.set_weights(
                    netuid=self.config.netuid,
                    wallet=self.wallet,
                    uids=self.metagraph.uids,
                    weights=weights,
                    wait_for_inclusion=True,
                    period=self.tempo  # Good for fast blocks - otherwise make sure to set proper period or remove this argument completely 
                )
                self.metagraph.sync()
                # sleep until next tempo
                time.sleep((((self.hetutensor.block // self.tempo) + 1) * self.tempo) + 1)

            except RuntimeError as e:
                ht.logging.error(e)
                traceback.print_exc()

            except KeyboardInterrupt:
                ht.logging.success("Keyboard interrupt detected. Exiting validator.")
                exit()


# Run the validator.
if __name__ == "__main__":
    validator = Validator()
    validator.run()