Merge pull request #101 from Datura-ai/hotfix-main-bittensor

Hotfix main bittensor

Author: surcyf123

cortext/protocol.py

@@ -196,7 +196,7 @@ class StreamPrompting(bt.StreamingSynapse):
     )
 
     seed: int = pydantic.Field(
-        default="1234",
+        default=1234,
         title="Seed",
         description="Seed for text generation. This attribute is immutable and cannot be updated.",
     )

cursor/app.py

@@ -0,0 +1,30 @@
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import StreamingResponse
+import httpx
+import asyncio
+
+app = FastAPI()
+organic_server_url = "https://0.0.0.0:8000"
+
+
+@app.post("/apis.datura.ai/s18_sigma")
+async def stream_prompt():
+    try:
+        # Create an asynchronous HTTP client session
+        async with httpx.AsyncClient() as client:
+            # Make a streaming GET request to the external server
+            response = await client.stream("GET", organic_server_url)
+            response.raise_for_status()  # Raise an error for bad responses (4xx, 5xx)
+
+            # Define an async generator to read chunks of data from the external server
+            async def stream_generator():
+                async for chunk in response.aiter_bytes():
+                    yield chunk
+                    await asyncio.sleep(0)  # Allow other tasks to run
+
+            # Return a StreamingResponse that forwards the data from the external server
+            return StreamingResponse(stream_generator(), media_type="application/json")
+    except httpx.HTTPStatusError as e:
+        raise HTTPException(status_code=e.response.status_code, detail=str(e))
+    except httpx.RequestError as e:
+        raise HTTPException(status_code=500, detail="Failed to fetch data from external server")

server/app/curd.py

@@ -1,10 +1,13 @@
+import json
+import time
 import traceback
 
 import psycopg2
-import os
+from psycopg2.extras import RealDictCursor
 from typing import List
+
 from . import models, schemas
-from .database import cur, TABEL_NAME, conn, DATABASE_URL
+from .database import TABEL_NAME, DATABASE_URL
 from fastapi import HTTPException
 
 
@@ -20,10 +23,27 @@ def create_items(items: List[schemas.ItemCreate]):
     conn = psycopg2.connect(DATABASE_URL)
     # Create a cursor object to interact with the database
     cur = conn.cursor()
-    query = f"INSERT INTO {TABEL_NAME} (p_key, question, answer, provider, model, timestamp) VALUES (%s, %s, %s, %s, %s, %s)"
+    query = (f"INSERT INTO {TABEL_NAME} (p_key, question, answer, provider, model, timestamp, miner_hot_key, miner_uid"
+             f", score, similarity, vali_uid, timeout, time_taken, epoch_num, cycle_num, block_num"
+             f", name) VALUES (%s, %s, %s, %s, %s, %s"
+             f", %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)")
     datas = []
     for item in items:
-        datas.append((item.p_key, item.question, item.answer, item.provider, item.model, item.timestamp))
+        question = json.loads(item.question)
+        miner_uid = question.get("miner_info", {}).get("miner_id") or 99999
+        miner_hot_key = question.get("miner_info", {}).get("miner_hotkey") or ""
+        score = question.get("score") or 0
+        similarity = question.get("similarity") or 0
+        vali_uid = question.get("validator_info").get("vali_uid")
+        timeout = question.get("timeout")
+        time_taken = question.get("time_taken")
+        epoch_num = question.get("epoch_num")
+        cycle_num = question.get("cycle_num")
+        block_num = question.get("block_num")
+        name = question.get("name") or ""
+        datas.append((item.p_key, item.question, item.answer, item.provider, item.model, item.timestamp, miner_hot_key,
+                      miner_uid, score, similarity, vali_uid, timeout,
+                      time_taken, epoch_num, cycle_num, block_num, name))
     try:
         if conn.closed:
             print("connection is closed already")
@@ -35,27 +55,45 @@ def create_items(items: List[schemas.ItemCreate]):
         raise HTTPException(status_code=500, detail=f"Internal Server Error {err}")
 
 
-def get_items(skip: int = 0, limit: int = 10):
-    req_body = {
-        "filters": {
-            "min_score": 0,
-            "min_similarity": 120,
-            "model": "",
-            "provider": "",
-            "min_timestamp": 12345,
-            "max_timestamp": 12345
-        },
-        "search": 123 or "2FXABC",
-        "sort_by": "miner",
-        "sort_order": "desc"
-    }
+def get_items(req_body: models.RequestBody):
     conn = psycopg2.connect(DATABASE_URL)
-    # Create a cursor object to interact with the database
-    cur = conn.cursor()
-    query = f"SELECT * FROM {TABEL_NAME} offset {skip} limit {limit} ;"
-    cur.execute(query)
+    cur = conn.cursor(cursor_factory=RealDictCursor)
+    skip = req_body.skip
+    limit = req_body.limit
+
+    filter_by_miner_score = f"score>={req_body.filters.min_score}" if req_body.filters.min_score else ""
+    filter_by_miner_similarity = f"score>={req_body.filters.min_similarity}" if req_body.filters.min_similarity else ""
+    filter_by_provider = f"provider='{req_body.filters.provider}'" if req_body.filters.provider else ""
+    filter_by_model = f"model='{req_body.filters.model}'" if req_body.filters.model else ""
+    filter_by_min_timestamp = f"timestamp>={req_body.filters.min_timestamp}" if req_body.filters.min_timestamp else ""
+    filter_by_max_timestamp = f"timestamp<={req_body.filters.max_timestamp}" if req_body.filters.max_timestamp else ""
+    filter_by_epoch_num = f"epoch_num={req_body.filters.epoch_num}" if req_body.filters.epoch_num else ""
+    filter_by_block_num = f"block_num={req_body.filters.block_num}" if req_body.filters.block_num else ""
+    filter_by_cycle_num = f"cycle_num={req_body.filters.cycle_num}" if req_body.filters.cycle_num else ""
+    filter_by_name = f"name={req_body.filters.name}" if req_body.filters.name else ""
+    search_by_uid_or_hotkey = (f"miner_uid=%s" if str(req_body.search).isdigit()
+                               else f"miner_hot_key like %s") if req_body.search else ""
+    conditions = [filter_by_miner_score, filter_by_miner_similarity, filter_by_provider, filter_by_model,
+                  filter_by_min_timestamp,
+                  filter_by_max_timestamp, filter_by_epoch_num, filter_by_block_num, filter_by_cycle_num,
+                  filter_by_name, search_by_uid_or_hotkey]
+    conditions = [item for item in conditions if item]
+    conditions_query = " and ".join(conditions)
+    order_by = f"order by {req_body.sort_by} {req_body.sort_order}"
+    query = f"SELECT * FROM {TABEL_NAME} where {conditions_query} {order_by} limit {limit} offset {skip};"
+    print(query)
+    query_cnt = f"SELECT count(*) FROM {TABEL_NAME} where {conditions_query}"
+    start_time = time.time()
+    cur.execute(query, (f"%{req_body.search}%" if not str(req_body.search).isdigit() else str(req_body.search),))
+    print(f"execution query has been completed. {time.time() - start_time}")
     items = cur.fetchall()  # Fetch all results
-    return [item for item in items]
+    print(f"loaded to python object array. {time.time() - start_time}")
+    # cur.execute(query_cnt, (f"%{req_body.search}%" if not str(req_body.search).isdigit() else str(req_body.search),))
+    # cnt = cur.fetchone().get('count')
+    # print(f"all query execution has been succeed. {time.time() - start_time}")
+    cur.close()
+    conn.close()
+    return {"records": items, "limit": limit, "skip": skip}
 
 
 def get_item(p_key: int):

server/app/database.py

@@ -26,7 +26,18 @@ async def create_table(app):
             answer TEXT,
             provider VARCHAR(100),
             model VARCHAR(100),
-            timestamp FLOAT
+            timestamp FLOAT,
+            miner_hot_key VARCHAR(100),
+            miner_uid INTEGER,
+            score FLOAT,
+            similarity FLOAT,
+            vali_uid INTEGER,
+            timeout INTEGER,
+            time_taken INTEGER,
+            epoch_num INTEGER,
+            cycle_num INTEGER,
+            block_num INTEGER,
+            name VARCHAR(100)
         );
         """
 
@@ -35,6 +46,9 @@ async def create_table(app):
         conn.commit()  # Save changes
         create_index_query = f"""
         CREATE INDEX IF NOT EXISTS question_answer_index ON {TABEL_NAME} (provider, model);
+        CREATE INDEX IF NOT EXISTS miner_id_index ON {TABEL_NAME} (miner_uid);
+        CREATE INDEX IF NOT EXISTS miner_hot_key_index ON {TABEL_NAME} (miner_hot_key);
+        CREATE INDEX IF NOT EXISTS idx_score_sim_timestamp ON {TABEL_NAME} (score, similarity, timestamp);
         """
         cur.execute(create_index_query)
         conn.commit()

server/app/main.py

@@ -1,5 +1,6 @@
 from contextlib import asynccontextmanager
 from fastapi import FastAPI, Depends, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
 from . import curd, models, schemas
 from .database import create_table, conn, cur
 from typing import List
@@ -13,6 +14,10 @@ async def lifespan(app: FastAPI):
 
 
 app = FastAPI(lifespan=lifespan)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Allows all origins
+)
 
 
 @app.on_event("shutdown")
@@ -28,9 +33,9 @@ def create_item(items: List[schemas.ItemCreate]):
 
 
 # Read all items
-@app.get("/items")
-def read_items(skip: int = 0, limit: int = 10):
-    items = curd.get_items(skip=skip, limit=limit)
+@app.post("/items/search")
+def read_items(req_body: models.RequestBody):
+    items = curd.get_items(req_body)
     return items
 
 

server/app/models.py

@@ -3,16 +3,23 @@
 
 
 class Filters(BaseModel):
-    min_score: int = Field(0, ge=0)
-    min_similarity: int = Field(120, ge=0)
-    model: Optional[str] = Field(None)
-    provider: Optional[str] = Field(None)
-    min_timestamp: Optional[int] = Field(None, ge=0)
-    max_timestamp: Optional[int] = Field(None, ge=0)
+    min_score: float = Field(0, ge=0)
+    min_similarity: float = Field(0.0, ge=0)
+    model: Optional[str] = Field("")
+    provider: Optional[str] = Field("")
+    epoch_num: Optional[int] = Field("")
+    cycle_num: Optional[int] = Field("")
+    block_num: Optional[int] = Field("")
+    name: Optional[str] = Field("")
+    min_timestamp: Optional[float] = Field(0, ge=0)
+    max_timestamp: Optional[float] = Field(999999999999, ge=0)
+
 
 
 class RequestBody(BaseModel):
-    filters: Filters
-    search: Union[int, str] = Field(..., description="An integer ID or a string search key")
-    sort_by: str = Field("miner", description="Field to sort by")
-    sort_order: str = Field("desc", regex="^(asc|desc)$", description="Sorting order, 'asc' or 'desc'")
+    filters: Optional[Filters] = Field(..., description="filter for searching")
+    search: Optional[Union[int, str]] = Field(..., description="An integer ID or a string search key")
+    sort_by: Optional[str] = Field("miner_uid", description="Field to sort by")
+    sort_order: Optional[str] = Field("desc", pattern="^(asc|desc)$", description="Sorting order, 'asc' or 'desc'")
+    skip: Optional[int] = Field(0, description="skip for pagination")
+    limit: Optional[int] = Field(100, description="skip for pagination")

validators/weight_setter.py

@@ -1,12 +1,10 @@
 import asyncio
-import concurrent
 import random
 import threading
-import traceback
+import json
 
 import torch
 import time
-import requests
 
 from black.trans import defaultdict
 from substrateinterface import SubstrateInterface
@@ -15,7 +13,6 @@
 import bittensor as bt
 from bittensor import StreamingSynapse
 import cortext
-import json
 from starlette.types import Send
 
 from cortext.protocol import IsAlive, StreamPrompting, ImageResponse, Embeddings
@@ -26,6 +23,7 @@
 from validators.task_manager import TaskMgr
 from cortext.dendrite import CortexDendrite
 from cortext.axon import CortexAxon
+from fastapi import HTTPException
 
 scoring_organic_timeout = 60
 NUM_INTERVALS_PER_CYCLE = 10
@@ -483,6 +481,8 @@ async def embeddings(self, synapse: Embeddings) -> Embeddings:
 
     async def prompt(self, synapse: StreamPrompting) -> StreamingSynapse.BTStreamingResponse:
         bt.logging.info(f"Received {synapse}")
+        if len(json.dumps(synapse.messages)) > 1024:
+            raise HTTPException(status_code=413, detail="Request entity too large")
 
         async def _prompt(query_synapse: StreamPrompting, send: Send):
             query_synapse.deserialize_flag = False