Remove Unecessary Files

.gitignore

@@ -155,15 +155,8 @@ dmypy.json
 # Cython debug symbols
 cython_debug/
 
-# PyCharm
-#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
-#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
-#  and can be added to the global gitignore or merged into this file.  For a more nuclear
-#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
 
 testing/
 core
 app.config.js
-wandb
 .vscode

api.py

@@ -1,147 +0,0 @@
-
-import atexit
-import datetime
-
-from flask import Flask, request, jsonify
-from apscheduler.schedulers.background import BackgroundScheduler
-
-import utils
-
-app = Flask(__name__)
-
-
-# Global variables (saves time on loading data)
-state_vars = None
-reload_timestamp = datetime.datetime.now().strftime('%D %T')
-
-
-def load_data():
-    """
-    Reload the state variables
-    """
-    global state_vars, reload_timestamp
-    state_vars = utils.load_state_vars()
-
-    reload_timestamp = datetime.datetime.now().strftime('%D %T')
-
-    print(f'Reloaded data at {reload_timestamp}')
-
-
-def start_scheduler():
-    scheduler = BackgroundScheduler()
-    scheduler.add_job(func=load_data, trigger="interval", seconds=60*30)
-    scheduler.start()
-
-    # Shut down the scheduler when exiting the app
-    atexit.register(lambda: scheduler.shutdown())
-
-
-@app.route('/', methods=['GET'])
-def home():
-    return "Welcome to the Bittensor Protein Folding Leaderboard API!"
-
-
-@app.route('/updated', methods=['GET'])
-def updated():
-    return reload_timestamp
-
-
-@app.route('/data', methods=['GET'])
-@app.route('/data/<period>', methods=['GET'])
-def data(period=None):
-    """
-    Get the productivity metrics
-    """
-    assert period in ('24h', None), f"Invalid period: {period}. Must be '24h' or None."
-    df = state_vars["dataframe_24h"] if period == '24h' else state_vars["dataframe"]
-    return jsonify(
-        df.astype(str).to_dict(orient='records')
-    )
-
-@app.route('/productivity', methods=['GET'])
-@app.route('/productivity/<period>', methods=['GET'])
-def productivity_metrics(period=None):
-    """
-    Get the productivity metrics
-    """
-
-    assert period in ('24h', None), f"Invalid period: {period}. Must be '24h' or None."
-    df = state_vars["dataframe_24h"] if period == '24h' else state_vars["dataframe"]
-    return jsonify(
-        utils.get_productivity(df)
-    )
-
-
-@app.route('/throughput', methods=['GET'])
-@app.route('/throughput/<period>', methods=['GET'])
-def throughput_metrics(period=None):
-    """
-    Get the throughput metrics
-    """
-    assert period in ('24h', None), f"Invalid period: {period}. Must be '24h' or None."
-    df = state_vars["dataframe_24h"] if period == '24h' else state_vars["dataframe"]
-    return jsonify(utils.get_data_transferred(df))
-
-
-@app.route('/metagraph', methods=['GET'])
-def metagraph():
-    """
-    Get the metagraph data
-    Returns:
-    - metagraph_data: List of dicts (from pandas DataFrame)
-    """
-
-    df_m = state_vars["metagraph"]
-
-    return jsonify(
-        df_m.to_dict(orient='records')
-    )
-
-@app.route('/leaderboard', methods=['GET'])
-@app.route('/leaderboard/<entity>', methods=['GET'])
-@app.route('/leaderboard/<entity>/<ntop>', methods=['GET'])
-def leaderboard(entity='identity',ntop=10):
-    """
-    Get the leaderboard data
-    Returns:
-    - leaderboard_data: List of dicts (from pandas DataFrame)
-    """
-
-    assert entity in utils.ENTITY_CHOICES, f"Invalid entity choice: {entity}"
-
-    df_miners = utils.get_leaderboard(
-        state_vars["metagraph"],
-        ntop=int(ntop),
-        entity_choice=entity
-        )
-
-    return jsonify(
-        df_miners.to_dict(orient='records')
-    )
-
-@app.route('/validator', methods=['GET'])
-def validator():
-    """
-    Get the validator data
-    Returns:
-    - validator_data: List of dicts (from pandas DataFrame)
-    """
-    df_m = state_vars["metagraph"]
-    df_validators = df_m.loc[df_m.validator_trust > 0]
-
-    return jsonify(
-        df_validators.to_dict(orient='records')
-    )
-
-
-if __name__ == '__main__':
-
-    load_data()
-    start_scheduler()
-
-    app.run(host='0.0.0.0', port=5001, debug=True)
-    
-
-    # to test locally
-    # curl -X GET http://0.0.0.0:5001/data
-    

data/wandb/tzebw6rb.parquet

@@ -1,3 +0,0 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:97537f22180da6992b01c2ba7566fabc7c265d27c2c317eda2a191d52e08b843
-size 13624028

forward.py

@@ -1,249 +0,0 @@
-import time
-import sys
-import asyncio
-import numpy as np
-import bittensor as bt
-import traceback
-from typing import List, Dict, Awaitable
-from prompting.agent import HumanAgent
-from prompting.dendrite import DendriteResponseEvent
-from prompting.conversation import create_task
-from prompting.protocol import StreamPromptingSynapse
-from prompting.rewards import RewardResult
-from prompting.utils.uids import get_random_uids
-from prompting.utils.logging import log_event
-from prompting.utils.misc import async_log, serialize_exception_to_string
-from dataclasses import dataclass
-
-
-@async_log
-async def generate_reference(agent):
-    loop = asyncio.get_running_loop()
-    result = await loop.run_in_executor(
-        None, agent.task.generate_reference, agent.llm_pipeline
-    )
-    return result
-
-
-@async_log
-async def execute_dendrite_call(dendrite_call):
-    responses = await dendrite_call
-    return responses
-
-
-@dataclass
-class StreamResult:
-    synapse: StreamPromptingSynapse = None
-    exception: BaseException = None
-    uid: int = None
-
-
-async def process_response(uid: int, async_generator: Awaitable):
-    """Process a single response asynchronously."""
-    try:
-        chunk = None  # Initialize chunk with a default value
-        async for chunk in async_generator:  # most important loop, as this is where we acquire the final synapse.
-            bt.logging.debug(f"\nchunk for uid {uid}: {chunk}")
-
-        if chunk is not None:
-            synapse = chunk  # last object yielded is the synapse itself with completion filled
-
-            # Assuming chunk holds the last value yielded which should be a synapse
-            if isinstance(synapse, StreamPromptingSynapse):
-                return synapse
-
-        bt.logging.debug(
-            f"Synapse is not StreamPromptingSynapse. Miner uid {uid} completion set to '' "
-        )
-    except Exception as e:
-        # bt.logging.error(f"Error in generating reference or handling responses: {e}", exc_info=True)
-        traceback_details = traceback.format_exc()
-        bt.logging.error(
-            f"Error in generating reference or handling responses for uid {uid}: {e}\n{traceback_details}"
-        )
-
-        failed_synapse = StreamPromptingSynapse(
-            roles=["user"], messages=["failure"], completion=""
-        )
-
-        return failed_synapse
-
-
-@async_log
-async def handle_response(responses: Dict[int, Awaitable]) -> List[StreamResult]:
-    """The handle_response function is responsible for creating asyncio tasks around acquiring streamed miner chunks
-    and processing them asynchronously. It then pairs the results with their original UIDs and returns a list of StreamResults.
-
-    Args:
-        responses (Dict[int, Awaitable]): Responses contains awaitables that are used to acquire streamed miner chunks.
-
-    Raises:
-        ValueError
-
-    Returns:
-        List[StreamResult]: DataClass containing the synapse, exception, and uid
-    """
-    tasks_with_uid = [
-        (uid, responses[uid]) for uid, _ in responses.items()
-    ]  # Pair UIDs with their tasks
-
-    # Start tasks, preserving order and their associated UIDs
-    tasks = [process_response(uid, resp) for uid, resp in tasks_with_uid]
-
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-
-    mapped_results = []
-    # Pair each result with its original uid
-    for (uid, _), result in zip(tasks_with_uid, results):
-        # If the result is a StreamPromptingSynapse, the response was successful and the stream result is added without exceptions
-        if isinstance(result, StreamPromptingSynapse):
-            mapped_results.append(StreamResult(synapse=result, uid=uid))
-
-        # If the result is an exception, the response was unsuccessful and the stream result is added with the exception and an empty synapse
-        elif isinstance(result, BaseException):
-            failed_synapse = StreamPromptingSynapse(
-                roles=["user"], messages=["failure"], completion=""
-            )
-            mapped_results.append(
-                StreamResult(synapse=failed_synapse, exception=result, uid=uid)
-            )
-
-        # If the result is neither an error or a StreamSynapse, log the error and raise a ValueError
-        else:
-            bt.logging.error(f"Unexpected result type for UID {uid}: {result}")
-            raise ValueError(f"Unexpected result type for UID {uid}: {result}")
-
-    return mapped_results
-
-
-@async_log
-async def generate_reference(agent: HumanAgent):
-    loop = asyncio.get_running_loop()
-    result = await loop.run_in_executor(
-        None, agent.task.generate_reference, agent.llm_pipeline
-    )
-    return result
-
-
-def log_stream_results(stream_results: List[StreamResult]):
-    failed_responses = [
-        response for response in stream_results if response.exception is not None
-    ]
-    empty_responses = [
-        response
-        for response in stream_results
-        if response.exception is None and response.synapse.completion == ""
-    ]
-    non_empty_responses = [
-        response
-        for response in stream_results
-        if response.exception is None and response.synapse.completion != ""
-    ]
-
-    bt.logging.info(f"Total of non_empty responses: ({len(non_empty_responses)})")
-    bt.logging.info(f"Total of empty responses: ({len(empty_responses)})")
-    bt.logging.info(
-        f"Total of failed responses: ({len(failed_responses)}):\n {failed_responses}"
-    )
-
-    for failed_response in failed_responses:
-        formatted_exception = serialize_exception_to_string(failed_response.exception)
-        bt.logging.error(
-            f"Failed response for uid {failed_response.uid}: {formatted_exception}"
-        )
-
-
-async def run_step(
-    self, agent: HumanAgent, k: int, timeout: float, exclude: list = None
-):
-    """Executes a single step of the agent, which consists of:
-    - Getting a list of uids to query
-    - Querying the network
-    - Rewarding the network
-    - Updating the scores
-    - Logging the event
-
-    Args:
-        agent (HumanAgent): The agent to run the step for.
-        k (int): The number of uids to query.
-        timeout (float): The timeout for the queries.
-        exclude (list, optional): The list of uids to exclude from the query. Defaults to [].
-    """
-
-    bt.logging.debug("run_step", agent.task.name)
-
-    # Record event start time.
-    start_time = time.time()
-    # Get the list of uids to query for this step.
-    uids = get_random_uids(self, k=k, exclude=exclude or []).to(self.device)
-    uids_cpu = uids.cpu().tolist()
-
-    axons = [self.metagraph.axons[uid] for uid in uids]
-
-    # Directly call dendrite and process responses in parallel
-    streams_responses = await self.dendrite(
-        axons=axons,
-        synapse=StreamPromptingSynapse(roles=["user"], messages=[agent.challenge]),
-        timeout=timeout,
-        deserialize=False,
-        streaming=True,
-    )
-
-    # Prepare the task for handling stream responses
-    handle_stream_responses_task = asyncio.create_task(
-        handle_response(responses=dict(zip(uids_cpu, streams_responses)))
-    )
-
-    if not agent.task.static_reference:
-        reference_generation_task = generate_reference(agent)
-        _, stream_results = await asyncio.gather(
-            reference_generation_task, handle_stream_responses_task
-        )
-    else:
-        stream_results = await handle_stream_responses_task
-
-    log_stream_results(stream_results)
-
-    all_synapses_results = [stream_result.synapse for stream_result in stream_results]
-
-    # Encapsulate the responses in a response event (dataclass)
-    response_event = DendriteResponseEvent(
-        responses=all_synapses_results, uids=uids, timeout=timeout
-    )
-
-    bt.logging.info(f"Created DendriteResponseEvent:\n {response_event}")
-    # Reward the responses and get the reward result (dataclass)
-    # This contains a list of RewardEvents but can be exported as a dict (column-wise) for logging etc
-    reward_result = RewardResult(
-        self.reward_pipeline,
-        agent=agent,
-        response_event=response_event,
-        device=self.device,
-    )
-    bt.logging.info(f"Created RewardResult:\n {reward_result}")
-
-    # The original idea was that the agent is 'satisfied' when it gets a good enough response (e.g. reward critera is met, such as ROUGE>threshold)
-    agent.update_progress(
-        top_reward=reward_result.rewards.max(),
-        top_response=response_event.completions[reward_result.rewards.argmax()],
-    )
-
-    self.update_scores(reward_result.rewards, uids)
-
-    stream_results_uids = [stream_result.uid for stream_result in stream_results]
-    stream_results_exceptions = [
-        serialize_exception_to_string(stream_result.exception)
-        for stream_result in stream_results
-    ]
-    # Log the step event.
-    event = {
-        "block": self.block,
-        "step_time": time.time() - start_time,
-        "stream_results_uids": stream_results_uids,
-        "stream_results_exceptions": stream_results_exceptions,
-        **agent.__state_dict__(full=self.config.neuron.log_full),
-        **reward_result.__state_dict__(full=self.config.neuron.log_full),
-        **response_event.__state_dict__(),
-    }
-
-    return event

requirements.txt

@@ -6,5 +6,4 @@ streamlit
 plotly
 wandb
 numpy
-pandas 
-bittensor
+pandas

server.py

@@ -1,97 +0,0 @@
-import asyncio
-
-import bittensor as bt
-from aiohttp import web
-from aiohttp_apispec import (
-    docs,
-    request_schema,
-    response_schema,
-    setup_aiohttp_apispec,
-    validation_middleware,
-)
-
-from common import utils
-from common.middlewares import api_key_middleware, json_parsing_middleware
-from common.schemas import QueryChatSchema, StreamChunkSchema, StreamErrorSchema
-from validators import QueryValidatorParams, S1ValidatorAPI, ValidatorAPI
-
-
-@docs(tags=["Prompting API"], summary="Chat", description="Chat endpoint.")
-@request_schema(QueryChatSchema)
-@response_schema(StreamChunkSchema, 200)
-@response_schema(StreamErrorSchema, 400)
-async def chat(request: web.Request) -> web.StreamResponse:
-    """Chat endpoint for the validator"""
-    params = QueryValidatorParams.from_request(request)
-
-    # Access the validator from the application context
-    validator: ValidatorAPI = request.app["validator"]
-
-    response = await validator.query_validator(params)
-    return response
-
-
-@docs(
-    tags=["Prompting API"],
-    summary="Echo test",
-    description="Echo endpoint for testing purposes.",
-)
-@request_schema(QueryChatSchema)
-@response_schema(StreamChunkSchema, 200)
-@response_schema(StreamErrorSchema, 400)
-async def echo_stream(request: web.Request) -> web.StreamResponse:
-    return await utils.echo_stream(request)
-
-
-class ValidatorApplication(web.Application):
-    def __init__(self, validator_instance=None, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        self["validator"] = (
-            validator_instance if validator_instance else S1ValidatorAPI()
-        )
-
-        # Add middlewares to application
-        self.add_routes(
-            [
-                web.post("/chat/", chat),
-                web.post("/echo/", echo_stream),
-            ]
-        )
-        self.setup_openapi()
-        self.setup_middlewares()
-        # TODO: Enable rewarding and other features
-
-    def setup_middlewares(self):
-        self.middlewares.append(validation_middleware)
-        self.middlewares.append(json_parsing_middleware)
-        self.middlewares.append(api_key_middleware)
-
-    def setup_openapi(self):
-        setup_aiohttp_apispec(
-            app=self,
-            title="Prompting API",
-            url="/docs/swagger.json",
-            swagger_path="/docs",
-        )
-
-
-def main(run_aio_app=True, test=False) -> None:
-    loop = asyncio.get_event_loop()
-    port = 10000
-    if run_aio_app:
-        # Instantiate the application with the actual validator
-        bt.logging.info("Starting validator application.")
-        validator_app = ValidatorApplication()
-        bt.logging.success("Validator app initialized successfully", validator_app)
-
-        try:
-            web.run_app(validator_app, port=port, loop=loop)
-        except KeyboardInterrupt:
-            print("Keyboard interrupt detected. Exiting validator.")
-        finally:
-            pass
-
-
-if __name__ == "__main__":
-    main()

validators/__init__.py

@@ -1,4 +0,0 @@
-from .base import QueryValidatorParams, ValidatorAPI, MockValidator
-from .sn1_validator_wrapper import S1ValidatorAPI
-from .streamer import AsyncResponseDataStreamer
-from .stream_manager import StreamManager

validators/base.py

@@ -1,42 +0,0 @@
-from abc import ABC, abstractmethod
-from typing import List
-from dataclasses import dataclass
-from aiohttp.web import Response, Request, StreamResponse
-
-
-@dataclass
-class QueryValidatorParams:
-    k_miners: int
-    exclude: List[str]
-    roles: List[str]
-    messages: List[str]
-    timeout: int
-    prefer: str
-    request: Request
-    sampling_mode: str
-
-    @staticmethod
-    def from_request(request: Request):
-        data = request["data"]
-
-        return QueryValidatorParams(
-            k_miners=data.get("k", 10),
-            exclude=data.get("exclude", []),
-            roles=data["roles"],
-            messages=data["messages"],
-            timeout=data.get("timeout", 10),
-            prefer=data.get("prefer", "longest"),
-            request=request,
-            sampling_mode=data.get("sampling_mode", "random"),
-        )
-
-
-class ValidatorAPI(ABC):
-    @abstractmethod
-    async def query_validator(self, params: QueryValidatorParams) -> StreamResponse:
-        pass
-
-
-class MockValidator(ValidatorAPI):
-    async def query_validator(self, params: QueryValidatorParams) -> StreamResponse:
-        ...

validators/database.py

@@ -1,85 +0,0 @@
-import os
-import json
-import aiofiles
-import bittensor as bt
-from .streamer import ProcessedStreamResponse
-
-
-class LogDatabase:
-    """
-    A class to manage a log database stored as a JSONL (JSON Lines) file.
-
-    Attributes:
-        log_database_path (str): The path to the log database file.
-
-    Methods:
-        ensure_db_exists(file_path):
-            Ensures that the log database file exists. If it doesn't, an empty file is created.
-        
-        add_streams_to_db(stream_responses: ProcessedStreamResponse):
-            Asynchronously adds stream responses to the log database.
-        
-        append_dicts_to_file(file_path, dictionaries):
-            Asynchronously appends a list of dictionaries to the specified file.
-    """
-    
-    def __init__(self, log_database_path: str):
-        """
-        Initializes the LogDatabase with the given log database file path.
-
-        Args:
-            log_database_path (str): The path to the log database file.
-        """
-        self.log_database_path = log_database_path
-        self.ensure_db_exists(log_database_path)
-
-    def ensure_db_exists(self, file_path):
-        """
-        Ensures that the log database file exists. If it doesn't, creates an empty JSONL file.
-
-        Args:
-            file_path (str): The path to the log database file.
-        """
-        if not os.path.exists(file_path):
-            # Create an empty JSONL file
-            with open(file_path, "w") as file:
-                pass
-            # TODO: change log to debug
-            bt.logging.info(f"File '{file_path}' created.")
-        else:
-            bt.logging.info(f"File '{file_path}' already exists.")
-
-    async def add_streams_to_db(self, stream_responses: ProcessedStreamResponse):
-        """
-        Asynchronously adds stream responses to the log database.
-
-        Args:
-            stream_responses (ProcessedStreamResponse): A list of processed stream responses to add to the log database.
-
-        Raises:
-            Exception: If an error occurs while adding streams to the database.
-        """
-        bt.logging.info(f"Writing streams to the database...")
-        try:
-            stream_responses_dict = [
-                dict(stream_response) for stream_response in stream_responses
-            ]
-            await self.append_dicts_to_file(
-                self.log_database_path, stream_responses_dict
-            )
-            bt.logging.success("Streams added to the database.")
-        except Exception as e:
-            bt.logging.error(f"Error while adding streams to the database: {e}")
-            raise e
-
-    async def append_dicts_to_file(self, file_path, dictionaries):
-        """
-        Asynchronously appends a list of dictionaries to the specified file.
-
-        Args:
-            file_path (str): The path to the file where dictionaries will be appended.
-            dictionaries (list): A list of dictionaries to append to the file.
-        """
-        async with aiofiles.open(file_path, mode="a") as file:
-            for dictionary in dictionaries:
-                await file.write(json.dumps(dictionary) + "\n")

validators/sn1_validator_wrapper.py

@@ -1,66 +0,0 @@
-import asyncio
-import random
-import bittensor as bt
-from prompting.validator import Validator
-from prompting.utils.uids import get_random_uids
-from prompting.protocol import StreamPromptingSynapse
-from .base import QueryValidatorParams, ValidatorAPI
-from aiohttp.web_response import Response, StreamResponse
-from .streamer import AsyncResponseDataStreamer
-from .validator_utils import get_top_incentive_uids
-from .stream_manager import StreamManager
-
-
-class S1ValidatorAPI(ValidatorAPI):
-    def __init__(self):
-        self.validator = Validator()
-
-    def sample_uids(self, params: QueryValidatorParams):
-        if params.sampling_mode == "random":
-            uids = get_random_uids(
-                self.validator, k=params.k_miners, exclude=params.exclude or []
-            ).tolist()
-            return uids
-        if params.sampling_mode == "top_incentive":
-            metagraph = self.validator.metagraph
-            vpermit_tao_limit = self.validator.config.neuron.vpermit_tao_limit
-
-            top_uids = get_top_incentive_uids(
-                metagraph, k=params.k_miners, vpermit_tao_limit=vpermit_tao_limit
-            )
-
-            return top_uids
-
-    async def get_stream_response(self, params: QueryValidatorParams) -> StreamResponse:
-        # Guess the task name of current request
-        # task_name = utils.guess_task_name(params.messages[-1])
-
-        # Get the list of uids to query for this step.
-        uids = self.sample_uids(params)
-        axons = [self.validator.metagraph.axons[uid] for uid in uids]
-
-        # Make calls to the network with the prompt.
-        bt.logging.info(
-            f"Sampling dendrite by {params.sampling_mode} with roles {params.roles} and messages {params.messages}"
-        )
-
-        streams_responses = await self.validator.dendrite(
-            axons=axons,
-            synapse=StreamPromptingSynapse(
-                roles=params.roles, messages=params.messages
-            ),
-            timeout=params.timeout,
-            deserialize=False,
-            streaming=True,
-        )
-
-        # Creates a streamer from the selected stream
-        stream_manager = StreamManager()
-        selected_stream = await stream_manager.process_streams(
-            params.request, streams_responses, uids
-        )
-
-        return selected_stream
-
-    async def query_validator(self, params: QueryValidatorParams) -> Response:
-        return await self.get_stream_response(params)

validators/stream_manager.py

@@ -1,73 +0,0 @@
-import asyncio
-import bittensor as bt
-from .streamer import AsyncResponseDataStreamer
-from .database import LogDatabase
-from typing import List, AsyncIterator
-from aiohttp.web import Request
-
-
-class StreamManager:
-    """
-    A class to manage the processing of multiple asynchronous data streams and log their responses.
-
-    Attributes:
-        log_database (LogDatabase): The log database to store stream responses.
-
-    Methods:
-        process_streams(request, streams_responses, stream_uids):
-            Processes multiple asynchronous streams, logs their responses, and returns the selected stream response.
-    """
-    
-    def __init__(self, log_database_path: str = "requests_db.jsonl"):
-        """
-        Initializes the StreamManager with the given log database file path.
-
-        Args:
-            log_database_path (str): The path to the log database file, defaults to "requests_db.jsonl".
-        """
-        self.log_database = LogDatabase(log_database_path)
-
-    async def process_streams(
-        self,
-        request: Request,
-        streams_responses: List[AsyncIterator],
-        stream_uids: List[int],
-    ):
-        """
-        Processes multiple asynchronous streams, logs their responses, and returns the selected stream response (stream from first non-empty chunk).
-
-        Args:
-            request (Request): The web request object.
-            streams_responses (List[AsyncIterator]): A list of asynchronous iterators representing the streams.
-            stream_uids (List[int]): A list of unique IDs for the streams.
-
-        Returns:
-            ProcessedStreamResponse: The response from the selected stream.
-        """
-        lock = asyncio.Lock()
-
-        streamers = [
-            AsyncResponseDataStreamer(
-                async_iterator=stream, selected_uid=stream_uid, lock=lock
-            )
-            for stream, stream_uid in zip(streams_responses, stream_uids)
-        ]
-        completed_streams = await asyncio.gather(
-            *[streamer.stream(request) for streamer in streamers]
-        )
-
-        lock.release()
-        bt.logging.info(f"Streams from uids: {stream_uids} processing completed.")
-
-        await self.log_database.add_streams_to_db(completed_streams)
-        # Gets the first stream that acquired the lock, meaning the first stream that was able to return a non-empty chunk
-        selected_stream = next(
-            (
-                completed_stream
-                for streamer, completed_stream in zip(streamers, completed_streams)
-                if streamer.lock_acquired
-            ),
-            None,
-        )
-
-        return selected_stream

validators/streamer.py

@@ -1,271 +0,0 @@
-import json
-import time
-import asyncio
-import traceback
-import bittensor as bt
-from pydantic import BaseModel
-from datetime import datetime
-from typing import AsyncIterator, Optional, List, Union
-from aiohttp import web, web_response
-from prompting.protocol import StreamPromptingSynapse
-
-
-class StreamChunk(BaseModel):
-    """
-    A model representing a chunk of streaming data.
-
-    Attributes:
-        delta (str): The change in the stream.
-        finish_reason (Optional[str]): The reason for finishing the stream.
-        accumulated_chunks (List[str]): List of accumulated chunks.
-        accumulated_chunks_timings (List[float]): Timings for the accumulated chunks.
-        timestamp (str): The timestamp of the chunk.
-        sequence_number (int): The sequence number of the chunk.
-        selected_uid (int): The selected user ID.
-    """
-    delta: str
-    finish_reason: Optional[str]
-    accumulated_chunks: List[str]
-    accumulated_chunks_timings: List[float]
-    timestamp: str
-    sequence_number: int
-    selected_uid: int
-
-    def encode(self, encoding: str) -> bytes:
-        """
-        Encodes the StreamChunk instance to a JSON-formatted bytes object.
-
-        Args:
-            encoding (str): The encoding to use.
-
-        Returns:
-            bytes: The encoded JSON data.
-        """
-        data = json.dumps(self.dict(), indent=4)
-        return data.encode(encoding)
-
-
-class StreamError(BaseModel):
-    """
-    A model representing an error in the streaming data.
-
-    Attributes:
-        error (str): The error message.
-        timestamp (str): The timestamp of the error.
-        sequence_number (int): The sequence number at the time of error.
-        finish_reason (str): The reason for finishing the stream, defaults to "error".
-    """
-    error: str
-    timestamp: str
-    sequence_number: int
-    finish_reason: str = "error"
-
-    def encode(self, encoding: str) -> bytes:
-        data = json.dumps(self.dict(), indent=4)
-        return data.encode(encoding)
-
-
-ProcessedStreamResponse = Union[StreamChunk, StreamError]
-
-
-class AsyncResponseDataStreamer:
-    """
-    A class to manage asynchronous streaming of response data.
-
-    Attributes:
-        async_iterator (AsyncIterator): An asynchronous iterator for streaming data.
-        selected_uid (int): The selected user ID.
-        lock (asyncio.Lock): An asyncio lock to ensure exclusive access.
-        delay (float): Delay between processing chunks, defaults to 0.1 seconds.
-        accumulated_chunks (List[str]): List of accumulated chunks.
-        accumulated_chunks_timings (List[float]): Timings for the accumulated chunks.
-        finish_reason (str): The reason for finishing the stream.
-        sequence_number (int): The sequence number of the stream.
-        lock_acquired (bool): Flag indicating if the lock was acquired.
-    """
-    def __init__(
-        self,
-        async_iterator: AsyncIterator,
-        selected_uid: int,
-        lock: asyncio.Lock,
-        delay: float = 0.1,
-    ):
-        self.async_iterator = async_iterator
-        self.delay = delay
-        self.selected_uid = selected_uid
-        self.accumulated_chunks: List[str] = []
-        self.accumulated_chunks_timings: List[float] = []
-        self.finish_reason: str = None
-        self.sequence_number: int = 0
-        self.lock = lock
-        self.lock_acquired = False
-
-    def ensure_response_is_created(
-        self, initiated_response: web.StreamResponse
-    ) -> web.StreamResponse:
-        """
-        Ensures that a StreamResponse is created if it does not already exist.
-
-        Args:
-            initiated_response (web.StreamResponse): The initiated response.
-
-        Returns:
-            web.StreamResponse: The ensured response.
-        """
-        # Creates response if it was not created
-        if initiated_response == None:
-            initiated_response = web_response.StreamResponse(status=200, reason="OK")
-            initiated_response.headers["Content-Type"] = "application/json"
-            return initiated_response
-
-        return initiated_response
-
-    async def write_to_stream(
-        self,
-        request: web.Request,
-        initiated_response: web.StreamResponse,
-        stream_chunk: StreamChunk,
-        lock: asyncio.Lock,
-    ) -> web.StreamResponse:
-        """
-        Writes a stream chunk to the response if the lock is acquired.
-
-        Args:
-            request (web.Request): The web request object.
-            initiated_response (web.StreamResponse): The initiated response.
-            stream_chunk (StreamChunk): The chunk of stream data to write.
-            lock (asyncio.Lock): The lock to ensure exclusive access.
-
-        Returns:
-            web.StreamResponse: The response with the written chunk.
-        """
-        # Try to acquire the lock and sets the lock_acquired flag. Only the stream that acquires the lock should write to the response
-        if lock.locked() == False:
-            self.lock_acquired = await lock.acquire()
-
-        if initiated_response == None and self.lock_acquired:
-            initiated_response = self.ensure_response_is_created(initiated_response)
-            # Prepare and send the headers
-            await initiated_response.prepare(request)
-
-        if self.lock_acquired:
-            await initiated_response.write(stream_chunk.encode("utf-8"))
-        else:
-            bt.logging.debug(
-                f"Stream of uid {stream_chunk.selected_uid} was not the first to return, skipping..."
-            )
-
-        return initiated_response
-
-    async def stream(self, request: web.Request) -> ProcessedStreamResponse:
-        """
-        Streams data from the async iterator and writes it to the response.
-
-        Args:
-            request (web.Request): The web request object.
-
-        Returns:
-            ProcessedStreamResponse: The final processed stream response.
-
-        Raises:
-            ValueError: If the stream does not return a valid synapse.
-        """
-        try:
-            start_time = time.time()
-            client_response: web.Response = None
-            final_response: ProcessedStreamResponse
-
-            async for chunk in self.async_iterator:
-                if isinstance(chunk, str):
-                    # If chunk is empty, skip
-                    if not chunk:
-                        continue
-                    
-                    self.accumulated_chunks.append(chunk)
-                    self.accumulated_chunks_timings.append(time.time() - start_time)
-                    # Gets new response state
-                    self.sequence_number += 1
-                    new_response_state = self._create_chunk_response(
-                        chunk
-                    )
-                    # Writes the new response state to the response                    
-                    client_response = await self.write_to_stream(
-                        request, client_response, new_response_state, self.lock
-                    )
-
-            if chunk is not None and isinstance(chunk, StreamPromptingSynapse):
-                if len(self.accumulated_chunks) == 0:
-                    self.accumulated_chunks.append(chunk.completion)
-                    self.accumulated_chunks_timings.append(time.time() - start_time)
-
-                self.finish_reason = "completed"
-                self.sequence_number += 1
-                # Assuming the last chunk holds the last value yielded which should be a synapse with the completion filled
-                synapse = chunk
-                final_response = self._create_chunk_response(synapse.completion)
-
-                if synapse.completion:
-                    client_response = await self.write_to_stream(
-                        request, client_response, final_response, self.lock
-                    )
-            else:
-                raise ValueError("Stream did not return a valid synapse.")
-
-        except Exception as e:
-            bt.logging.error(
-                f"Encountered an error while processing stream for uid {self.selected_uid} get_stream_response:\n{traceback.format_exc()}"
-            )
-            error_response = self._create_error_response(str(e))
-            final_response = error_response
-
-            # Only the stream that acquires the lock should write the error response
-            if self.lock_acquired:
-                self.ensure_response_is_created(client_response)
-                client_response.set_status(500, reason="Internal error")
-                client_response.write(error_response.encode("utf-8"))
-        finally:
-            # Only the stream that acquires the lock should close the response
-            if self.lock_acquired:
-                self.ensure_response_is_created(client_response)
-                # Ensure to close the response properly
-                await client_response.write_eof()
-
-            return final_response
-
-    def _create_chunk_response(self, chunk: str) -> StreamChunk:
-        """
-        Creates a StreamChunk object with the current state.
-
-        :param chunk: List of strings representing the current chunk.
-        :return: StreamChunk object.
-        """
-        return StreamChunk(
-            delta=chunk,
-            finish_reason=self.finish_reason,
-            accumulated_chunks=self.accumulated_chunks,
-            accumulated_chunks_timings=self.accumulated_chunks_timings,
-            timestamp=self._current_timestamp(),
-            sequence_number=self.sequence_number,
-            selected_uid=self.selected_uid,
-        )
-
-    def _create_error_response(self, error_message: str) -> StreamError:
-        """
-        Creates a StreamError object.
-
-        :param error_message: Error message to include in the StreamError.
-        :return: StreamError object.
-        """
-        return StreamError(
-            error=error_message,
-            timestamp=self._current_timestamp(),
-            sequence_number=self.sequence_number,
-        )
-
-    def _current_timestamp(self) -> str:
-        """
-        Returns the current timestamp in ISO format.
-
-        :return: Current timestamp as a string.
-        """
-        return datetime.utcnow().isoformat()

validators/validator_utils.py

@@ -1,35 +0,0 @@
-from typing import List
-from prompting.utils.uids import check_uid_availability
-
-
-def get_top_incentive_uids(metagraph, k: int, vpermit_tao_limit: int) -> List[int]:
-    miners_uids = list(
-        map(
-            int,
-            filter(
-                lambda uid: check_uid_availability(metagraph, uid, vpermit_tao_limit),
-                metagraph.uids,
-            ),
-        )
-    )
-
-    # Builds a dictionary of uids and their corresponding incentives
-    all_miners_incentives = {
-        "miners_uids": miners_uids,
-        "incentives": list(map(lambda uid: metagraph.I[uid], miners_uids)),
-    }
-
-    # Zip the uids and their corresponding incentives into a list of tuples
-    uid_incentive_pairs = list(
-        zip(all_miners_incentives["miners_uids"], all_miners_incentives["incentives"])
-    )
-
-    # Sort the list of tuples by the incentive value in descending order
-    uid_incentive_pairs_sorted = sorted(
-        uid_incentive_pairs, key=lambda x: x[1], reverse=True
-    )
-
-    # Extract the top 10 uids
-    top_k_uids = [uid for uid, incentive in uid_incentive_pairs_sorted[:k]]
-
-    return top_k_uids